"""Shared modeling infrastructure for Ch11+ notebooks. Provides: - load_modeling_dataset(): Load features + temporal + labels, join, detect schema - load_configs(): Load model configs for a label from label YAML + presets - prepare_cv_folds(): Preprocess data into train/val folds (impute, scale) - ModelingDataset: Container for joined data with detected schema Cross-sectional IC computation lives in the library — call ``ml4t.diagnostic.metrics.cross_sectional_ic`` against a polars frame of (date, symbol, y_true, y_pred) directly. Usage: from utils.modeling import load_modeling_dataset, load_configs, prepare_cv_folds mds = load_modeling_dataset("etfs", "fwd_ret_21d") configs = load_configs("etfs", "fwd_ret_21d", family="linear") folds = prepare_cv_folds(mds.dataset.to_pandas(), mds.splits, ...) """ from __future__ import annotations import os import random import warnings from dataclasses import dataclass, field from pathlib import Path from typing import Any import numpy as np import pandas as pd import polars as pl import yaml from utils.artifact_specs import ( load_feature_spec, load_label_spec, resolve_label_buffer, resolve_market_semantics, resolve_storage_path, ) from utils.cv_splits import generate_cv_splits, make_wf_config RANDOM_SEED = 42 def seed_everything(seed: int = RANDOM_SEED) -> None: """Set all random seeds for full reproducibility (CPU + GPU). Must be called before any stochastic operation. For per-fold reproducibility, call again at the start of each fold with ``seed_everything(RANDOM_SEED + fold_id)``. """ random.seed(seed) np.random.seed(seed) os.environ["PYTHONHASHSEED"] = str(seed) try: import torch torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False except ImportError: pass # torch not needed for sklearn/numpy-only models (PCA, IPCA) from utils.paths import get_case_study_dir # Columns that are identifiers, not features ID_COLS = { "date", "timestamp", "asset", "symbol", "stock_id", "product", "position", "instrument_id", } # Meta columns that may leak or are composites (not independent features) META_LEAK = { "underlying_price", "instr_mid", "instr_bid", "instr_ask", "ls_signal", "risk_adj_score", } @dataclass class ModelingDataset: """Container for a fully-joined modeling dataset with detected schema.""" dataset: pl.DataFrame feature_names: list[str] label_col: str date_col: str entity_cols: list[str] join_cols: list[str] splits: list[dict[str, Any]] label_buffer: str cv_config: Any = None # WalkForwardConfig (optional, avoids hard import dep) task_type: str = "regression" # "regression" or "classification" num_classes: int = 0 # 0 for regression, 2+ for classification class_values: list = field(default_factory=list) # sorted unique values for classification temporal_by_fold: pd.DataFrame | None = None # Per-fold temporal features (has 'fold' column) temporal_keys: list[str] = field(default_factory=list) # Join keys for temporal features temporal_feature_names: list[str] = field(default_factory=list) # Temporal feature column names # Continuous-return label that classification predictions are scored against. # None for regression labels. When set, the column lives in ``dataset`` and # downstream IC computation must use it instead of the binary ``label_col``. eval_label_col: str | None = None # --------------------------------------------------------------------------- # CV / Protocol Configuration # --------------------------------------------------------------------------- @dataclass class WalkForwardConfig: """Config for walk-forward cross-validation. Compatible with ml4t.diagnostic.splitters.WalkForwardCV. """ n_splits: int train_size: str test_size: str embargo_td: str label_horizon: str timestamp_col: str = "timestamp" calendar_id: str | None = None test_start: str | None = None test_end: str | None = None def model_dump(self) -> dict: """Return fields as a dict (Pydantic-compatible API).""" from dataclasses import asdict return asdict(self) def to_json(self, path: str | Path) -> None: """Serialize config to a JSON file (Pydantic-compatible API).""" import json path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w") as f: json.dump(self.model_dump(), f, indent=2) @classmethod def from_json(cls, path: str | Path) -> WalkForwardConfig: """Deserialize config from a JSON file.""" import json with open(path) as f: data = json.load(f) return cls(**data) def load_protocol(case_study_id: str) -> dict: """Load evaluation protocol from config/setup.yaml. Returns dict with n_splits, train_size, test_size, holdout, leakage_guards, calendar. """ path = get_case_study_dir(case_study_id) / "config" / "setup.yaml" if not path.exists(): msg = f"Setup not found: {path}" raise FileNotFoundError(msg) with open(path) as f: setup = yaml.safe_load(f) ev = setup.get("evaluation", {}) labels = setup.get("labels", {}) market_semantics = resolve_market_semantics(case_study_id, setup) protocol = { "n_splits": ev.get("n_splits", 5), "train_size": ev.get("train_size", "5Y"), "test_size": ev.get("val_size", "1Y"), "step_size": ev.get("step_size", "1Y"), "calendar": market_semantics.get("calendar") or ev.get("calendar", "NYSE"), "holdout": { "start": ev.get("holdout_start"), "end": ev.get("holdout_end"), }, "leakage_guards": {}, } # Parse label horizon from buffer string (e.g. "21D", "8h", "15T", "60m", "1M") buffer = labels.get("buffer", "21D") if buffer.endswith("D"): protocol["leakage_guards"]["label_horizon_days"] = int(buffer[:-1]) elif buffer.endswith(("h", "H")): protocol["leakage_guards"]["label_horizon_hours"] = int(buffer[:-1]) elif buffer.endswith(("T",)) or (buffer.endswith("m") and not buffer.endswith("M")): protocol["leakage_guards"]["label_horizon_minutes"] = int(buffer[:-1]) elif buffer.endswith("M"): # Monthly: approximate as 30 calendar days. # pd.Timedelta rejects 'M' as ambiguous; ml4t-diagnostic needs a library fix # to support calendar-month durations natively (see ml4t-diagnostic-dev/bugs/). protocol["leakage_guards"]["label_horizon_days"] = int(buffer[:-1]) * 30 return protocol def _label_horizon_to_iso(leakage_guards: dict) -> str: """Convert label horizon to ISO 8601 duration string.""" if "label_horizon_days" in leakage_guards: return f"P{leakage_guards['label_horizon_days']}D" if "label_horizon_hours" in leakage_guards: return f"PT{leakage_guards['label_horizon_hours']}H" if "label_horizon_minutes" in leakage_guards: return f"PT{leakage_guards['label_horizon_minutes']}M" return "P0D" def get_cv_config(case_study_id: str) -> WalkForwardConfig: """Load walk-forward CV config from case study setup.yaml. This is the single entry point for CV configuration. Reads from ``case_studies/{id}/config/setup.yaml``. """ protocol = load_protocol(case_study_id) embargo = _label_horizon_to_iso(protocol.get("leakage_guards", {})) holdout = protocol.get("holdout", {}) return WalkForwardConfig( n_splits=protocol["n_splits"], train_size=protocol["train_size"], test_size=protocol["test_size"], embargo_td=embargo, label_horizon=embargo, calendar_id=protocol.get("calendar", "NYSE"), test_start=str(holdout["start"]) if holdout.get("start") else None, test_end=str(holdout["end"]) if holdout.get("end") else None, ) # Classification label prefixes _CLASSIFICATION_PREFIXES = ("fwd_dir_", "fwd_class_", "fwd_tb_", "fwd_carry_") def detect_label_type(label_col: str, label_series: pl.Series) -> tuple[str, int, list]: """Detect regression vs classification from label name and values. Returns (task_type, num_classes, class_values). - task_type: "regression" or "classification" - num_classes: 0 for regression, 2+ for classification - class_values: sorted unique values for classification, [] for regression """ is_classification = any(label_col.startswith(p) for p in _CLASSIFICATION_PREFIXES) # Also classify if integer dtype with few unique values if not is_classification and label_series.dtype in (pl.Int8, pl.Int16, pl.Int32, pl.Int64): if label_series.drop_nulls().n_unique() <= 10: is_classification = True if not is_classification: return "regression", 0, [] unique_vals = sorted(label_series.drop_nulls().unique().to_list()) return "classification", len(unique_vals), unique_vals def get_classification_eval_label(case_study_id: str, label: str) -> str: """Return the continuous return that a classification label is derived from. Classification predictions (probabilities or class-expected-values) must be scored against the underlying continuous return, not the binary/categorical label itself: ``Spearman(score, fwd_continuous_return)`` is the proper IC, while ``Spearman(score, binary_label)`` collapses to ``2·(AUC − 0.5)``. The mapping is declared per-case-study under ``labels.classification_eval_label`` in ``config/setup.yaml``. There is no runtime inference — every classification label must be registered explicitly. Parameters ---------- case_study_id : str Case study identifier (e.g., ``"us_firm_characteristics"``). label : str Classification label name (e.g., ``"fwd_class_1m"``). Returns ------- str Continuous-return label name (e.g., ``"fwd_ret_1m"``). Raises ------ KeyError If ``labels.classification_eval_label[label]`` is missing from ``setup.yaml``. """ from utils import CASE_STUDIES_DIR setup_path = CASE_STUDIES_DIR / case_study_id / "config" / "setup.yaml" setup = yaml.safe_load(setup_path.read_text()) mapping = (setup.get("labels") or {}).get("classification_eval_label") or {} if label not in mapping: raise KeyError( f"labels.classification_eval_label[{label!r}] not declared in " f"case_studies/{case_study_id}/config/setup.yaml. Add an entry mapping " f"the binary/categorical label to its source continuous-return label " f"(e.g., {label}: fwd_ret_1m). IC for classification predictions is " f"computed against the continuous return — there is no silent fallback." ) return str(mapping[label]) def load_modeling_dataset( case_study_id: str, primary_label: str, max_symbols: int = 0, symbols: list[str] | None = None, ) -> ModelingDataset: """Load and join features + temporal + labels for a case study. This is the canonical data loading function for ALL Ch11+ modeling notebooks. It handles schema detection (date vs timestamp, asset vs stock_id vs product), temporal join-key casting, and universe reduction. Parameters ---------- case_study_id : str Case study identifier (e.g., "etfs", "crypto_perps_funding"). primary_label : str Label file stem (e.g., "fwd_ret_21d"). max_symbols : int, default 0 Universe reduction for fast development. 0 = all symbols. symbols : list of str, optional Explicit symbol whitelist. When given, the universe is restricted to exactly these symbols (intersected with what is available) and ``max_symbols`` is ignored. Used by tests to pin a small universe that is guaranteed to exist in the reduced test-data (e.g. the Darts base return series), rather than the top-by-history selection ``max_symbols`` makes — which can pick symbols absent from a sampled data set. Returns ------- ModelingDataset Container with joined dataset, detected schema, and CV splits. """ case_dir = get_case_study_dir(case_study_id) financial_spec = load_feature_spec(case_study_id, "financial") temporal_spec = load_feature_spec(case_study_id, "model_based") label_spec = load_label_spec(case_study_id, primary_label) # Check prerequisites exist before loading features_path = resolve_storage_path( case_study_id, financial_spec, "features/financial.parquet" ) label_path = resolve_storage_path(case_study_id, label_spec, f"labels/{primary_label}.parquet") missing = [] if not features_path.exists(): missing.append(("features/financial.parquet", "03_financial_features")) if not label_path.exists(): missing.append((f"labels/{primary_label}.parquet", "02_labels")) if not (case_dir / "config" / "setup.yaml").exists(): missing.append(("config/setup.yaml", None)) if missing: print(f"\n Missing prerequisites for '{case_study_id}' modeling:\n") for path, producer in missing: if producer is None: print(f" {path} (canonical hand-curated file — ensure committed)") else: print(f" {path} (run {producer}.py first)") first_producer = next((p for _, p in missing if p is not None), None) if first_producer is not None: print("\n Example:") print(f" uv run python case_studies/{case_study_id}/{first_producer}.py\n") raise FileNotFoundError( f"Missing prerequisites for '{case_study_id}': " + ", ".join(p for p, _ in missing) ) # Load artifacts features = pl.read_parquet(features_path) temporal_path = resolve_storage_path( case_study_id, temporal_spec, "features/model_based.parquet" ) temporal = pl.read_parquet(temporal_path) if temporal_path.exists() else None labels = pl.read_parquet(label_path) # Auto-detect label column (the non-ID column in the label file) label_col = [c for c in labels.columns if c not in ID_COLS][0] # Detect date column from features feature_keys = sorted(set(features.columns) & ID_COLS) date_col = "timestamp" if "timestamp" in feature_keys else "date" alt_date = "timestamp" if date_col == "date" else "date" # Normalize date column names across DataFrames if alt_date in labels.columns and date_col not in labels.columns: labels = labels.rename({alt_date: date_col}) if temporal is not None and alt_date in temporal.columns and date_col not in temporal.columns: temporal = temporal.rename({alt_date: date_col}) # Detect join columns label_keys = sorted(set(labels.columns) & ID_COLS) join_cols = sorted(set(feature_keys) & set(label_keys)) entity_cols = [c for c in join_cols if c != date_col] # Filter out constant entity columns (e.g. instrument_id='straddle_30d_atm') # that break cross-sectional IC computation by collapsing all entities into one group. # NOTE: join_cols retains ALL shared ID columns for data integrity during joins; # entity_cols is filtered separately for IC computation only. entity_cols = [c for c in entity_cols if features[c].n_unique() > 1] # Sort by cardinality descending so the primary entity (most unique values) # comes first. Important when downstream code uses entity_cols[0] for IC # (e.g., CME futures: 'product' has 30 values vs 'position' has 3). entity_cols = sorted(entity_cols, key=lambda c: features[c].n_unique(), reverse=True) # Join features + temporal (left join to keep all feature rows) temporal_by_fold_pd = None _temporal_keys = [] _temporal_feature_names = [] if temporal is not None: _temporal_keys = sorted(set(temporal.columns) & set(feature_keys)) casts = { k: features.schema[k] for k in _temporal_keys if temporal.schema[k] != features.schema[k] } if casts: temporal = temporal.cast(casts) if "fold" in temporal.columns: # Per-fold temporal features — join fold 0 as placeholder for schema, # store full per-fold data for fold-aware preparation functions. _temporal_feature_names = [ c for c in temporal.columns if c not in set(_temporal_keys) | {"fold"} ] fold_ids = sorted(temporal["fold"].unique().to_list()) placeholder_fold = fold_ids[0] placeholder = temporal.filter(pl.col("fold") == placeholder_fold).drop("fold") placeholder_dedup = placeholder.unique(subset=_temporal_keys, keep="last") dataset = features.join(placeholder_dedup, on=_temporal_keys, how="left", suffix="_t") del placeholder, placeholder_dedup # Convert to pandas for fold-preparation functions temporal_by_fold_pd = temporal.to_pandas() else: # Legacy: single feature set, join directly temporal_dedup = temporal.unique(subset=_temporal_keys, keep="last") dataset = features.join(temporal_dedup, on=_temporal_keys, how="left", suffix="_t") del temporal_dedup else: dataset = features # Inner-join with labels (drops rows without labels) dataset = dataset.join(labels, on=join_cols, how="inner") # Drop any meta columns that leaked in drop_cols = [c for c in dataset.columns if c in META_LEAK] if drop_cols: dataset = dataset.drop(drop_cols) # Optional universe reduction if symbols and entity_cols: primary_entity = entity_cols[0] dataset = dataset.filter(pl.col(primary_entity).is_in(list(symbols))) elif max_symbols > 0 and entity_cols: primary_entity = entity_cols[0] top = dataset.group_by(primary_entity).len().sort("len", descending=True).head(max_symbols) dataset = dataset.filter(pl.col(primary_entity).is_in(top[primary_entity])) # Feature columns = everything except IDs and label feature_names = [c for c in dataset.columns if c not in ID_COLS and c != label_col] # CV splits — read buffer from setup.yaml (explicit, handles non-standard labels) setup = yaml.safe_load((case_dir / "config" / "setup.yaml").read_text()) label_buffer = resolve_label_buffer(case_study_id, primary_label, setup) if not label_buffer: raise ValueError( f"No explicit label buffer found for '{primary_label}' in " f"case_studies/{case_study_id}/config/setup.yaml. " f"Add buffer to labels.buffer (primary) or labels.variant_buffers (variants)." ) splits = generate_cv_splits( dataset, case_study_id=case_study_id, label_buffer=label_buffer, date_col=date_col, ) # WalkForwardConfig for library integration # Normalize month-based buffers to days (pd.Timedelta rejects 'M' as ambiguous) wf_horizon = label_buffer if wf_horizon and wf_horizon.endswith("M") and wf_horizon[:-1].isdigit(): wf_horizon = f"{int(wf_horizon[:-1]) * 30}D" try: cv_config = make_wf_config(case_study_id, label_horizon=wf_horizon, date_col=date_col) except Exception as exc: warnings.warn(f"WalkForwardConfig creation failed for {case_study_id}: {exc}", stacklevel=2) cv_config = None # Detect label type (regression vs classification) task_type, num_classes, class_values = detect_label_type(label_col, dataset[label_col]) # Classification labels: load the continuous-return label they were derived # from so IC can be computed against returns rather than the binary target. eval_label_col: str | None = None if task_type == "classification": eval_label_col = get_classification_eval_label(case_study_id, label_col) eval_label_path = resolve_storage_path( case_study_id, load_label_spec(case_study_id, eval_label_col), f"labels/{eval_label_col}.parquet", ) if not eval_label_path.exists(): raise FileNotFoundError( f"Eval label parquet missing for classification label {label_col!r}: " f"expected {eval_label_path}. Generate it via 02_labels.py or update " f"labels.classification_eval_label[{label_col}] in setup.yaml." ) eval_labels = pl.read_parquet(eval_label_path) # Normalize date column name if needed if alt_date in eval_labels.columns and date_col not in eval_labels.columns: eval_labels = eval_labels.rename({alt_date: date_col}) # Inner-join eval column on the same join_cols (drops rows missing eval) eval_join_cols = sorted(set(eval_labels.columns) & set(dataset.columns) & ID_COLS) eval_labels = eval_labels.select([*eval_join_cols, eval_label_col]) dataset = dataset.join(eval_labels, on=eval_join_cols, how="left") # Refresh feature_names so the eval label column is not used as a feature feature_names = [ c for c in dataset.columns if c not in ID_COLS and c not in {label_col, eval_label_col} ] return ModelingDataset( dataset=dataset, feature_names=feature_names, label_col=label_col, date_col=date_col, entity_cols=entity_cols, join_cols=join_cols, splits=splits, label_buffer=label_buffer, cv_config=cv_config, task_type=task_type, num_classes=num_classes, class_values=class_values, temporal_by_fold=temporal_by_fold_pd, temporal_keys=_temporal_keys, temporal_feature_names=_temporal_feature_names, eval_label_col=eval_label_col, ) def append_holdout_fold_if_needed( mds: ModelingDataset, prediction_split: str, case_study_id: str, ) -> None: """Append a holdout fold to ``mds.splits`` when ``prediction_split=='holdout'``. Mirrors the etfs reference pattern at ``case_studies/etfs/04_model_based_features.py`` lines 109-116: train on everything from the first CV fold's ``train_start`` through ``holdout_start``, validate on ``[holdout_start, holdout_end]``. The combined fold becomes fold N+1, so downstream code iterating ``mds.splits`` produces one holdout prediction set per (training run, config) pair without any other change to the training loop. Idempotent — if the trailing fold already covers the holdout window (val_end matches setup.yaml's holdout_end), no fold is appended. Use case: nasdaq100 v4 winner-only holdout regeneration in Session 4 of the v4 sweep program. The val sweep keeps ``prediction_split= 'validation'`` (no holdout fold); the single retrain at the end uses ``prediction_split='holdout'``. """ if prediction_split != "holdout": return if not mds.splits: msg = ( "ModelingDataset.splits is empty; cannot derive train_start for " "the holdout fold. Verify that load_modeling_dataset produced " "at least one CV fold for this case study." ) raise RuntimeError(msg) path = get_case_study_dir(case_study_id) / "config" / "setup.yaml" with open(path) as f: setup = yaml.safe_load(f) ev = setup.get("evaluation", {}) holdout_start = str(ev.get("holdout_start", "")).strip() holdout_end = str(ev.get("holdout_end", "")).strip() if not holdout_start or not holdout_end: msg = ( f"setup.yaml::evaluation.holdout_start/holdout_end missing for " f"case_studies/{case_study_id}; cannot append holdout fold." ) raise KeyError(msg) # Use pd.Timestamp boundaries to match generate_cv_splits (which stores # train/val_start/end as Timestamps). Mixing raw strings with Timestamps # made the idempotency check below never match (str(Timestamp) != YAML # string) and risked a tz-naive/aware comparison on the pandas filter path. ho_start_ts = pd.Timestamp(holdout_start) ho_end_ts = pd.Timestamp(holdout_end) trailing = mds.splits[-1] if ( trailing.get("val_end") is not None and pd.Timestamp(trailing.get("val_end")) == ho_end_ts and pd.Timestamp(trailing.get("val_start")) == ho_start_ts ): return # already covered holdout_fold = { "fold": len(mds.splits), "train_start": pd.Timestamp(mds.splits[0]["train_start"]), "train_end": ho_start_ts, "val_start": ho_start_ts, "val_end": ho_end_ts, } mds.splits.append(holdout_fold) # --------------------------------------------------------------------------- # Config loading # --------------------------------------------------------------------------- class ConfigError(Exception): """Raised when model config loading or instantiation fails.""" # Model type directory → family mapping _MODEL_TYPE_TO_FAMILY: dict[str, str] = { "ols": "linear", "ridge": "linear", "lasso": "linear", "elastic_net": "linear", "logistic": "linear", "lgb": "gbm", "lstm": "deep_learning", "tcn": "deep_learning", "tsmixer": "deep_learning", "nlinear": "deep_learning", "nbeats": "deep_learning", "patchtst": "deep_learning", "tabm": "tabular_dl", "pca": "latent_factors", "ipca": "latent_factors", "cae": "latent_factors", "sae": "latent_factors", "sdf": "latent_factors", "dml": "causal_dml", } # Model type directory → library mapping _MODEL_TYPE_TO_LIBRARY: dict[str, str] = { "ols": "sklearn", "ridge": "sklearn", "lasso": "sklearn", "elastic_net": "sklearn", "logistic": "sklearn", "lgb": "lightgbm", "lstm": "pytorch", "tcn": "pytorch", "tsmixer": "pytorch", "nlinear": "pytorch", "nbeats": "darts", "patchtst": "pytorch", "tabm": "tabm", "pca": "sklearn", "ipca": "sklearn", "cae": "pytorch", "sae": "pytorch", "sdf": "pytorch", "dml": "causal_ml", } # Config names that override the directory-default library _CONFIG_LIBRARY_OVERRIDES: dict[str, str] = { "tabpfn": "tabpfn", "tsmixer": "darts", } def _enrich_config(preset: dict, preset_path: Path) -> dict: """Inject config_name, family, library from the file path. These were previously stored in the YAML but are derivable from the file system location: ``config/{model_type}/{config_name}.yaml``. """ config_name = preset_path.stem model_type = preset_path.parent.name preset["config_name"] = config_name preset.setdefault("family", _MODEL_TYPE_TO_FAMILY.get(model_type, model_type)) preset.setdefault( "library", _CONFIG_LIBRARY_OVERRIDES.get(config_name, _MODEL_TYPE_TO_LIBRARY.get(model_type, "")), ) return preset def _find_preset(config_root: Path, name: str) -> Path | None: """Search all config/{model_type}/ subdirs for a preset YAML.""" matches = list(config_root.glob(f"*/{name}.yaml")) return matches[0] if matches else None def load_configs( case_study_id: str, label: str, family: str, ) -> list[dict[str, Any]]: """Load model configurations for a label and family. Reads the training menu file (e.g., ``config/training/fwd_ret_21d.yaml``) to get the list of config names for the given family, then loads each referenced preset from ``case_studies/config/{model_type}/{config_name}.yaml``. Parameters ---------- case_study_id : str Case study identifier (e.g., "etfs"). label : str Label name (e.g., "fwd_ret_21d"). Must match a YAML file in ``config/training/``. family : str Model family (e.g., "linear", "gbm", "deep_learning"). Returns ------- list[dict] Each dict has keys: config_name, family, library, model_class, params. Raises ------ ConfigError If the training menu file or a referenced preset is missing. """ case_dir = get_case_study_dir(case_study_id) label_config_path = case_dir / "config" / "training" / f"{label}.yaml" if not label_config_path.exists(): raise ConfigError( f"No training config file found: {label_config_path}\n" f"Create it with the config names you want to run.\n" f"See case_studies/config/ for available presets." ) label_config = yaml.safe_load(label_config_path.read_text()) config_names = label_config.get(family, []) if not config_names: raise ConfigError( f"No '{family}' configs listed in {label_config_path}\n" f"Add config names under the '{family}:' key." ) config_root = case_dir.parent / "config" configs = [] for name in config_names: preset_path = _find_preset(config_root, name) if preset_path is None: raise ConfigError( f"Preset not found: {name}.yaml in {config_root}/*/\n" f"Referenced in {label_config_path} under '{family}'." ) preset = yaml.safe_load(preset_path.read_text()) _enrich_config(preset, preset_path) configs.append(preset) return configs def resolve_linear_params( cfg: dict[str, Any], X_train: np.ndarray, y_train: np.ndarray, ) -> dict[str, Any]: """Resolve a linear preset's constructor params for one training fold. Most presets carry an absolute ``alpha`` and are returned unchanged. L1 presets (Lasso, ElasticNet) may instead specify ``alpha_frac`` — a fraction of the data-derived degeneracy boundary ``alpha_max`` (the smallest alpha that drives every coefficient to zero). Because ``alpha_max`` is computed per fold from the standardized design, ``alpha_frac`` auto-calibrates the regularization grid to each case study and fold, so a fraction < 1 always yields at least one non-zero coefficient. This avoids hardcoding absolute alphas that land above ``alpha_max`` for low-signal case studies (which would silently drop the entire L1 family as degenerate). For Lasso the degeneracy boundary is ``max|Xᵀ(y − ȳ)| / n``; for ElasticNet it is that quantity divided by ``l1_ratio`` (only the L1 term gates entry). Parameters ---------- cfg : dict Preset config with a ``params`` dict (e.g. from :func:`load_configs`). X_train, y_train : np.ndarray Standardized training design and target for the current fold. Returns ------- dict Constructor params with ``alpha`` resolved and ``alpha_frac`` removed. """ params = dict(cfg["params"]) frac = params.pop("alpha_frac", None) if frac is None: return params n = X_train.shape[0] residual = y_train - y_train.mean() alpha_max = float(np.max(np.abs(X_train.T @ residual))) / n l1_ratio = params.get("l1_ratio", 1.0) params["alpha"] = frac * alpha_max / l1_ratio return params # --------------------------------------------------------------------------- # CV fold preparation # --------------------------------------------------------------------------- def _replace_temporal_columns( dataset_pd: pd.DataFrame, mask: np.ndarray, temporal_by_fold: pd.DataFrame, temporal_keys: list[str], temporal_feature_names: list[str], fold_id: int, ) -> pd.DataFrame: """Replace temporal feature columns in a dataset slice with fold-specific values. Returns a copy of the masked rows with temporal columns overwritten. """ rows = dataset_pd.loc[mask].copy() fold_temp = temporal_by_fold[temporal_by_fold["fold"] == fold_id].drop(columns=["fold"]) fold_temp = fold_temp.drop_duplicates(subset=temporal_keys, keep="last") # Drop old temporal columns and merge fold-specific ones rows = rows.drop(columns=temporal_feature_names, errors="ignore") rows = rows.merge(fold_temp, on=temporal_keys, how="left") return rows def prepare_cv_folds( dataset_pd: pd.DataFrame, splits: list[dict[str, Any]], feature_names: list[str], label_col: str, date_col: str, entity_col: str | None, temporal_by_fold: pd.DataFrame | None = None, temporal_keys: list[str] | None = None, temporal_feature_names: list[str] | None = None, train_sample_frac: float = 1.0, eval_label_col: str | None = None, ) -> list[dict[str, Any]]: """Preprocess data into train/val folds with imputation and scaling. For each CV split: filter by date range, drop NaN labels, impute missing features (median), and standardize. Returns a list of fold dicts ready for model fitting. Parameters ---------- dataset_pd : pd.DataFrame Full dataset (pandas) with features, label, date, entity columns. splits : list[dict] Walk-forward splits with fold, train_start, train_end, val_start, val_end. feature_names : list[str] Feature column names. label_col : str Target column name. date_col : str Date/timestamp column name. entity_col : str or None Entity column for cross-sectional IC. None for single-entity data. temporal_by_fold : pd.DataFrame, optional Per-fold temporal features with a 'fold' column. When provided, each fold's temporal feature columns are replaced with fold-specific values (fit on that fold's training data only — no look-ahead). temporal_keys : list[str], optional Join keys for temporal features (e.g., ['timestamp', 'symbol']). temporal_feature_names : list[str], optional Temporal feature column names to replace per fold. Returns ------- list[dict] Each dict has keys: fold, X_train, X_val, y_train, y_val, meta (val metadata DataFrame), dates, entities, n_train, n_val. """ from sklearn.impute import SimpleImputer from sklearn.pipeline import make_pipeline from sklearn.preprocessing import StandardScaler has_fold_temporal = temporal_by_fold is not None and temporal_keys and temporal_feature_names dates_series = dataset_pd[date_col] folds = [] for split in splits: fold_id = split["fold"] val_start = split.get("val_start", split.get("test_start")) val_end = split.get("val_end", split.get("test_end")) train_mask = (dates_series >= split["train_start"]) & (dates_series <= split["train_end"]) val_mask = (dates_series >= val_start) & (dates_series <= val_end) if has_fold_temporal: train_rows = _replace_temporal_columns( dataset_pd, train_mask, temporal_by_fold, temporal_keys, temporal_feature_names, fold_id, ) val_rows = _replace_temporal_columns( dataset_pd, val_mask, temporal_by_fold, temporal_keys, temporal_feature_names, fold_id, ) X_train = train_rows[feature_names].values y_train = train_rows[label_col].values X_val = val_rows[feature_names].values y_val = val_rows[label_col].values y_eval = val_rows[eval_label_col].values if eval_label_col else None del train_rows, val_rows else: X_train = dataset_pd.loc[train_mask, feature_names].values y_train = dataset_pd.loc[train_mask, label_col].values X_val = dataset_pd.loc[val_mask, feature_names].values y_val = dataset_pd.loc[val_mask, label_col].values y_eval = dataset_pd.loc[val_mask, eval_label_col].values if eval_label_col else None if len(X_train) == 0 or len(X_val) == 0: print(f" Fold {fold_id}: SKIP (empty train={len(X_train)}, val={len(X_val)})") continue # Drop rows where label is NaN train_valid = ~np.isnan(y_train) val_valid = ~np.isnan(y_val) X_train, y_train = X_train[train_valid], y_train[train_valid] X_val, y_val = X_val[val_valid], y_val[val_valid] if y_eval is not None: y_eval = y_eval[val_valid] # Optional train subsample (never touch val — OOS IC uses full val slice). # Walk-forward CV structure is preserved; only within-fold row density # is reduced. Seed tied to fold_id for reproducibility. if 0.0 < train_sample_frac < 1.0 and len(X_train) > 0: n_keep = max(1, int(len(X_train) * train_sample_frac)) rng = np.random.default_rng(RANDOM_SEED + fold_id) keep_idx = rng.choice(len(X_train), size=n_keep, replace=False) keep_idx.sort() X_train = X_train[keep_idx] y_train = y_train[keep_idx] # Preprocessing: median imputation + standard scaling preprocessor = make_pipeline(SimpleImputer(strategy="median"), StandardScaler()) X_train_s = preprocessor.fit_transform(X_train) X_val_s = preprocessor.transform(X_val) X_train_s = np.nan_to_num(X_train_s, nan=0.0, posinf=0.0, neginf=0.0) X_val_s = np.nan_to_num(X_val_s, nan=0.0, posinf=0.0, neginf=0.0) val_meta = dataset_pd.loc[val_mask].iloc[val_valid.nonzero()[0]] folds.append( { "fold": fold_id, "X_train": X_train_s, "X_val": X_val_s, "y_train": y_train, "y_val": y_val, "y_eval": y_eval, "meta": val_meta, "dates": val_meta[date_col].values, "entities": val_meta[entity_col].values if entity_col else None, "n_train": len(X_train), "n_val": len(X_val), } ) return folds def prepare_single_fold( dataset: pl.DataFrame | pd.DataFrame, split: dict[str, Any], feature_names: list[str], label_col: str, date_col: str, entity_col: str | None, temporal_by_fold: pd.DataFrame | None = None, temporal_keys: list[str] | None = None, temporal_feature_names: list[str] | None = None, *, train_sample_frac: float = 1.0, eval_label_col: str | None = None, ) -> dict[str, Any] | None: """Preprocess a single CV fold — impute, scale, return arrays. Same logic as ``prepare_cv_folds`` but for ONE split at a time. Use this for large datasets where materializing all folds at once would exceed available memory. Accepts either Polars or pandas DataFrames. When given Polars, only the fold's rows are converted to numpy — avoiding a full-dataset pandas copy (~5GB for us_equities_panel). Parameters ---------- train_sample_frac : float, optional Fraction of training rows to keep (1.0 = all). Same semantics as ``prepare_cv_folds`` / ``prepare_gbm_folds``: validation is never sampled, seed is tied to fold_id for reproducibility. Returns None if the fold is empty (no train or val rows). """ from sklearn.impute import SimpleImputer from sklearn.pipeline import make_pipeline from sklearn.preprocessing import StandardScaler fold_id = split["fold"] val_start = split.get("val_start", split.get("test_start")) val_end = split.get("val_end", split.get("test_end")) _has_fold_temporal = temporal_by_fold is not None and temporal_keys and temporal_feature_names if isinstance(dataset, pl.DataFrame): # Polars path — slice directly to numpy, no pandas intermediate # Cast string split boundaries to match the column's temporal dtype col_dtype = dataset.schema[date_col] _cast = pl.lit # default: pass through if col_dtype == pl.Date: import datetime def _cast(s): return pl.lit(datetime.date.fromisoformat(str(s)[:10])) elif col_dtype in (pl.Datetime, pl.Datetime("us"), pl.Datetime("ns"), pl.Datetime("ms")): import datetime def _cast(s, _dt=col_dtype): # Cast to the column's exact dtype to preserve resolution + timezone # (e.g. crypto's timestamps are Datetime('ms', 'UTC')) try: d = datetime.datetime.fromisoformat(str(s)) except ValueError: d = datetime.datetime.fromisoformat(f"{str(s)[:10]}T00:00:00") return pl.lit(d).cast(_dt) train_df = dataset.filter( (pl.col(date_col) >= _cast(split["train_start"])) & (pl.col(date_col) <= _cast(split["train_end"])) ) val_df = dataset.filter( (pl.col(date_col) >= _cast(val_start)) & (pl.col(date_col) <= _cast(val_end)) ) if len(train_df) == 0 or len(val_df) == 0: print(f" Fold {fold_id}: SKIP (empty train={len(train_df)}, val={len(val_df)})") return None # Replace temporal columns with fold-specific values if available if _has_fold_temporal: fold_temp_pd = temporal_by_fold[temporal_by_fold["fold"] == fold_id].drop( columns=["fold"] ) fold_temp_pl = pl.from_pandas(fold_temp_pd) fold_temp_pl = fold_temp_pl.unique(subset=temporal_keys, keep="last") # Cast temporal keys to match dataset dtypes for k in temporal_keys: if k in fold_temp_pl.columns and fold_temp_pl.schema[k] != train_df.schema[k]: fold_temp_pl = fold_temp_pl.cast({k: train_df.schema[k]}) for df_name in ("train_df", "val_df"): df = train_df if df_name == "train_df" else val_df df = df.drop(temporal_feature_names) df = df.join(fold_temp_pl, on=temporal_keys, how="left") if df_name == "train_df": train_df = df else: val_df = df # Drop rows where label is null train_df = train_df.filter(pl.col(label_col).is_not_null() & pl.col(label_col).is_not_nan()) val_df = val_df.filter(pl.col(label_col).is_not_null() & pl.col(label_col).is_not_nan()) X_train = train_df.select(feature_names).to_numpy() y_train = train_df[label_col].to_numpy() X_val = val_df.select(feature_names).to_numpy() y_val = val_df[label_col].to_numpy() y_eval = val_df[eval_label_col].to_numpy() if eval_label_col else None # Keep ID columns for val metadata (needed for IC computation + prediction assembly) id_cols = [c for c in dataset.columns if c in ID_COLS] val_meta_pl = val_df.select(id_cols) dates = val_df[date_col].to_numpy() entities = val_df[entity_col].to_numpy() if entity_col else None del train_df, val_df else: # Pandas path (used by prepare_cv_folds callers) dates_series = dataset[date_col] train_mask = (dates_series >= split["train_start"]) & (dates_series <= split["train_end"]) val_mask = (dates_series >= val_start) & (dates_series <= val_end) if _has_fold_temporal: train_rows = _replace_temporal_columns( dataset, train_mask, temporal_by_fold, temporal_keys, temporal_feature_names, fold_id, ) val_rows = _replace_temporal_columns( dataset, val_mask, temporal_by_fold, temporal_keys, temporal_feature_names, fold_id, ) X_train = train_rows[feature_names].values y_train = train_rows[label_col].values X_val = val_rows[feature_names].values y_val = val_rows[label_col].values y_eval = val_rows[eval_label_col].values if eval_label_col else None else: X_train = dataset.loc[train_mask, feature_names].values y_train = dataset.loc[train_mask, label_col].values X_val = dataset.loc[val_mask, feature_names].values y_val = dataset.loc[val_mask, label_col].values y_eval = dataset.loc[val_mask, eval_label_col].values if eval_label_col else None if len(X_train) == 0 or len(X_val) == 0: print(f" Fold {fold_id}: SKIP (empty train={len(X_train)}, val={len(X_val)})") return None train_valid = ~np.isnan(y_train) val_valid = ~np.isnan(y_val) X_train, y_train = X_train[train_valid], y_train[train_valid] X_val, y_val = X_val[val_valid], y_val[val_valid] if y_eval is not None: y_eval = y_eval[val_valid] if _has_fold_temporal: val_meta_pd = val_rows.iloc[val_valid.nonzero()[0]] else: val_meta_pd = dataset.loc[val_mask].iloc[val_valid.nonzero()[0]] dates = val_meta_pd[date_col].values entities = val_meta_pd[entity_col].values if entity_col else None val_meta_pl = None # Will use val_meta_pd below # Optional train subsample (never touch val — OOS IC uses full val slice). # Seed tied to fold_id for reproducibility. if 0.0 < train_sample_frac < 1.0 and len(X_train) > 0: n_keep = max(1, int(len(X_train) * train_sample_frac)) rng = np.random.default_rng(RANDOM_SEED + fold_id) keep_idx = rng.choice(len(X_train), size=n_keep, replace=False) keep_idx.sort() X_train = X_train[keep_idx] y_train = y_train[keep_idx] # Preprocessing: median imputation + standard scaling preprocessor = make_pipeline(SimpleImputer(strategy="median"), StandardScaler()) X_train_s = preprocessor.fit_transform(X_train) X_val_s = preprocessor.transform(X_val) X_train_s = np.nan_to_num(X_train_s, nan=0.0, posinf=0.0, neginf=0.0) X_val_s = np.nan_to_num(X_val_s, nan=0.0, posinf=0.0, neginf=0.0) n_train, n_val = len(X_train), len(X_val) del X_train, X_val # Free unscaled arrays return { "fold": fold_id, "X_train": X_train_s, "X_val": X_val_s, "y_train": y_train, "y_val": y_val, "y_eval": y_eval, "meta_pl": val_meta_pl, # Polars DataFrame (if Polars input) "meta": val_meta_pd if val_meta_pl is None else None, # pandas (if pandas input) "dates": dates, "entities": entities, "n_train": n_train, "n_val": n_val, } # --------------------------------------------------------------------------- # IC computation # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Classification metrics # --------------------------------------------------------------------------- def compute_classification_metrics( y_true: np.ndarray, y_score: np.ndarray, class_values: list, ) -> dict[str, float]: """Compute classification metrics from predictions. Handles both binary and multiclass ordinal labels. The ``y_score`` array is the expected value of class probabilities (``proba @ class_values``), which is how classification predictions are stored in this project. For binary {0, 1}: y_score IS p(class=1). For binary with other values: y_score is linearly transformed from probabilities. For multiclass ordinal: y_score is a weighted ranking score. Parameters ---------- y_true : np.ndarray True class labels (integer values matching ``class_values``). y_score : np.ndarray Predicted scores (expected value of class probabilities). class_values : list Sorted unique class values (e.g. [0, 1] or [-1, 0, 1]). Returns ------- dict[str, float] Metric name → value. Keys depend on binary vs multiclass. """ from sklearn.metrics import ( accuracy_score, balanced_accuracy_score, brier_score_loss, log_loss, roc_auc_score, ) valid = np.isfinite(y_score) & np.isfinite(y_true) if valid.sum() < 10: return {} yt = y_true[valid] ys = y_score[valid] cv = sorted(class_values) n_classes = len(cv) metrics: dict[str, float] = {} if n_classes == 2: # Binary classification # Convert expected value back to p(positive_class) c0, c1 = cv[0], cv[1] # expected_value = p * c1 + (1 - p) * c0 => p = (ev - c0) / (c1 - c0) span = c1 - c0 if span > 0: p1 = (ys - c0) / span else: return {} p1 = np.clip(p1, 1e-15, 1 - 1e-15) y_binary = (yt == c1).astype(int) y_pred_class = (p1 >= 0.5).astype(int) metrics["auc_roc"] = float(roc_auc_score(y_binary, p1)) metrics["log_loss"] = float(log_loss(y_binary, p1)) metrics["brier_score"] = float(brier_score_loss(y_binary, p1)) metrics["accuracy"] = float(accuracy_score(y_binary, y_pred_class)) metrics["balanced_accuracy"] = float(balanced_accuracy_score(y_binary, y_pred_class)) # AUC-PR (average precision) — import separately since it can fail try: from sklearn.metrics import average_precision_score metrics["auc_pr"] = float(average_precision_score(y_binary, p1)) except Exception as exc: warnings.warn(f"AUC-PR computation failed: {exc}", stacklevel=2) else: # Multiclass ordinal: assign to nearest class value y_pred_class_vals = np.array([cv[np.argmin(np.abs(np.array(cv) - s))] for s in ys]) y_pred_class = y_pred_class_vals metrics["accuracy"] = float(accuracy_score(yt, y_pred_class)) metrics["balanced_accuracy"] = float(balanced_accuracy_score(yt, y_pred_class)) # AUC via ranking score against each binary split # For ordinal {-1,0,1}: treat expected value as ranking score # One-vs-rest AUC using the score as ordinal ranking for c in cv: y_bin = (yt == c).astype(int) if y_bin.sum() > 0 and y_bin.sum() < len(y_bin): # Higher score → more likely to be higher class score = ys if c == max(cv) else -ys if c == min(cv) else np.abs(ys) auc = float(roc_auc_score(y_bin, score)) metrics[f"auc_class_{c}"] = auc return metrics