"""Core backtest execution — engine-first, used by BOTH demo and sweep notebooks. This module provides a single ``run_backtest()`` function that: 1. Converts predictions to target weights via strategy_spec["signal"] 2. Dispatches to engine or vectorized path 3. Optionally registers the result in registry.db 4. Returns a unified result object The key invariant is that **sweep notebooks call the same function as demo notebooks**. There is no separate vectorized reimplementation for sweeps. Usage:: from case_studies.utils.backtest_runner import run_backtest result = run_backtest( case_study="etfs", prediction_hash="abc123", strategy_spec=spec, prices=prices, predictions=predictions, ) """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Any import numpy as np import polars as pl from case_studies.utils.backtest_loaders import BacktestConfig, get_backtest_config from case_studies.utils.backtest_presets import ( apply_calendar_session_enforcement, ensure_backtest_spec, runtime_backtest_config, strategy_view, ) from case_studies.utils.signals import build_target_weights_from_config # --------------------------------------------------------------------------- # Periods per year for Sharpe annualization # --------------------------------------------------------------------------- # Periods per year for Sharpe annualization. Used by the vectorized path # where each return observation corresponds to one rebalance period. _PERIODS_PER_YEAR: dict[str, float] = { "monthly_month_end": 12, "weekly": 52, "weekly_friday": 52, "weekly_friday_close": 52, "daily": 252, "daily_close": 252, "daily_ny_close": 252, "8_hour_funding_aligned": 365 * 3, # 3 observations per calendar day "15_min": 252 * 26, # ~26 fifteen-min bars per NYSE session "15_minute": 252 * 26, # alias "30_min": 252 * 13, # ~13 thirty-min bars per NYSE session "30_minute": 252 * 13, "1_hour": 252 * 6.5, # 6.5 hours per NYSE session "1_hourly": 252 * 6.5, "4_hour": 252 * 1.625, # ~1.625 four-hour bars per NYSE session "4_hourly": 252 * 1.625, } # Calendar name → exchange_calendars MIC code _CALENDAR_TO_XCAL: dict[str, str] = { "NYSE": "XNYS", "CME": "us_futures", "FX": "24/5", "crypto": "24/7", } # Cache for calendar session counts _calendar_ppy_cache: dict[str, int] = {} def calendar_periods_per_year(calendar: str) -> int: """Get trading days per year for a calendar using exchange_calendars. Computes the average number of sessions over a 10-year window (2015-2024) and caches the result. """ if calendar in _calendar_ppy_cache: return _calendar_ppy_cache[calendar] xcal_name = _CALENDAR_TO_XCAL.get(calendar, calendar) try: import exchange_calendars as xcals cal = xcals.get_calendar(xcal_name) total = sum( len(cal.sessions_in_range(f"{y}-01-01", f"{y}-12-31")) for y in range(2015, 2025) ) ppy = round(total / 10) except Exception: # Fallback if exchange_calendars unavailable or calendar unknown ppy = 252 _calendar_ppy_cache[calendar] = ppy return ppy # --------------------------------------------------------------------------- # Portfolio metrics via ml4t-diagnostic # --------------------------------------------------------------------------- def compute_portfolio_metrics( returns: np.ndarray, *, periods_per_year: int = 252, case_study: str | None = None, label: str | None = None, uncertainty: bool = True, uncertainty_n_boot: int = 1000, uncertainty_seed: int = 0, trim_leading_zeros: bool = False, ) -> dict[str, float]: """Compute portfolio metrics using ml4t-diagnostic. Replaces hand-rolled Sharpe/drawdown/etc. with the library's validated implementation. When ``uncertainty=True`` (default) the returned dict is extended with block-bootstrap CIs, Lo/LdP-2025 Sharpe SE, Newey-West HAC SE for annualized return, and PSR p-value — driven by :func:`case_studies.utils.uncertainty.compute_backtest_uncertainty`. Parameters ---------- returns : np.ndarray Array of period returns (daily or per-rebalance). periods_per_year : int Annualization factor (252 for daily, 52 for weekly, etc.). case_study, label : optional Used by the block-length resolver to pick rebalance_step from setup.yaml. uncertainty : bool, default True If False, skip the bootstrap (fast path for sweep inner loops). uncertainty_n_boot, uncertainty_seed : int Bootstrap configuration. trim_leading_zeros : bool, default False Legacy first-non-zero strip. Kept for callers that pass pre-canonical return series (e.g., raw engine output without canonical-window slice). Production callers (``_run_engine`` and the retrofit pipeline) pass ``False`` because they slice to the canonical (cs, label, split) window first, which preserves real "no-trade" days at the start of the window as legitimate zero-return periods rather than stripping them. Returns ------- dict[str, float] Metric name → value. Keys match the existing backtest_metrics schema, plus uncertainty columns when ``uncertainty=True``. """ from ml4t.diagnostic.evaluation import PortfolioAnalysis if trim_leading_zeros and len(returns) > 0: nonzero = np.flatnonzero(np.asarray(returns) != 0.0) if len(nonzero) > 0: returns = returns[nonzero[0] :] if len(returns) < 2: return { "sharpe": 0.0, "sortino": 0.0, "total_return": 0.0, "max_drawdown": 0.0, "cagr": 0.0, "calmar": 0.0, "volatility": 0.0, "win_rate": 0.0, "omega": 0.0, "var_95": 0.0, "cvar_95": 0.0, "stability": 0.0, "skewness": 0.0, "kurtosis": 0.0, "tail_ratio": 0.0, "n_periods": int(len(returns)), } analysis = PortfolioAnalysis(returns=returns, periods_per_year=periods_per_year) pm = analysis.compute_summary_stats() def _safe(v: float) -> float: """Sanitize metric value: handle complex, inf, nan.""" if isinstance(v, complex): v = v.real if not np.isfinite(v): return 0.0 return float(v) out = { "sharpe": _safe(pm.sharpe_ratio), "sortino": _safe(pm.sortino_ratio), "total_return": _safe(pm.total_return), "max_drawdown": _safe(pm.max_drawdown), "cagr": _safe(pm.annual_return), "calmar": _safe(pm.calmar_ratio), "volatility": _safe(pm.annual_volatility), "win_rate": _safe(pm.win_rate), "omega": _safe(pm.omega_ratio), "var_95": _safe(pm.var_95), "cvar_95": _safe(pm.cvar_95), "stability": _safe(pm.stability), "skewness": _safe(pm.skewness), "kurtosis": _safe(pm.kurtosis), "tail_ratio": _safe(pm.tail_ratio), "n_periods": int(len(returns)), } if uncertainty and len(returns) >= 4: try: from case_studies.utils.uncertainty import compute_backtest_uncertainty unc = compute_backtest_uncertainty( returns, periods_per_year=periods_per_year, case_study=case_study, label=label, n_boot=uncertainty_n_boot, seed=uncertainty_seed, ) out.update(unc) except Exception as exc: # pragma: no cover - never block point estimates import warnings warnings.warn( f"compute_backtest_uncertainty failed: {exc}; point metrics returned without CIs", stacklevel=2, ) return out # --------------------------------------------------------------------------- # Result container # --------------------------------------------------------------------------- @dataclass class BacktestRunResult: """Unified result from both engine and vectorized paths.""" daily_returns: pl.DataFrame # [timestamp, daily_return] metrics: dict[str, float] strategy_spec: dict prediction_hash: str backtest_hash: str | None = None # Engine-only fields engine_result: Any = None # BacktestResult from ml4t-backtest weights: pl.DataFrame | None = None execution_mode: str = "engine" # --------------------------------------------------------------------------- # Weight precomputation (for risk sweep reuse) # --------------------------------------------------------------------------- def precompute_weights( predictions: pl.DataFrame, strategy_spec: dict, prices: pl.DataFrame, *, label: str = "", case_study: str = "", ) -> pl.DataFrame: """Compute allocation weights from a strategy spec, without running the engine. Use this to avoid redundant MVO/HRP computation in Ch19 risk sweeps where the same allocation weights are tested with different risk overlays. Returns ------- pl.DataFrame Weights [timestamp, symbol, weight] ready for ``run_backtest(precomputed_weights=...)``. """ predictions = normalize_prediction_columns(predictions) strategy = strategy_view(strategy_spec) signal_config = strategy["signal"] rebal_spec = strategy.get("rebalance", {}) weights = build_target_weights_from_config(predictions, signal_config) alloc_spec = strategy.get("allocation") if alloc_spec: cadence = strategy.get("rebalance", {}).get("cadence", "") weights = _apply_allocation( weights, predictions, prices, alloc_spec, cadence=cadence, label=label, case_study=case_study, ) return weights # --------------------------------------------------------------------------- # Strategy spec construction # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Prediction normalization # --------------------------------------------------------------------------- def normalize_prediction_columns(df: pl.DataFrame) -> pl.DataFrame: """Normalize prediction columns to canonical [timestamp, symbol, y_score, ...].""" renames = {} # Time column: date → timestamp if "timestamp" not in df.columns and "date" in df.columns: renames["date"] = "timestamp" # Entity column: asset/product/stock_id/entity → symbol if "symbol" not in df.columns: for col in ("asset", "product", "stock_id", "entity"): if col in df.columns: renames[col] = "symbol" break # Score column if "y_score" not in df.columns: if "prediction" in df.columns: renames["prediction"] = "y_score" if "y_true" not in df.columns and "actual" in df.columns: renames["actual"] = "y_true" if "fold_id" not in df.columns and "fold" in df.columns: renames["fold"] = "fold_id" if renames: df = df.rename(renames) # Cast types if "timestamp" in df.columns: ts_dtype = df.schema["timestamp"] if ts_dtype == pl.Date: df = df.with_columns(pl.col("timestamp").cast(pl.Datetime("us"))) elif ts_dtype in (pl.String, pl.Utf8): df = df.with_columns(pl.col("timestamp").str.to_datetime().cast(pl.Datetime("us"))) elif hasattr(ts_dtype, "time_zone") and ts_dtype.time_zone: df = df.with_columns(pl.col("timestamp").dt.replace_time_zone(None)) if "symbol" in df.columns and df.schema["symbol"] != pl.String: df = df.with_columns(pl.col("symbol").cast(pl.String)) return df # Tolerant-by-design cap on (timestamp, symbol) join misses between a # classification label and its continuous-return counterpart. >10% null # rate indicates a regeneration mismatch between the two label parquets # (not source-data sparsity), and is escalated to a hard error rather # than silently dropping rows from the backtest. Callers operating in # a legitimately high-null regime can override via the ``max_null_rate`` # parameter on ``substitute_continuous_return_for_classification``. _MAX_NULL_RATE = 0.10 # Polars integer dtypes for symbol id columns (e.g., us_firm ``stock_id``). # Used by ``_align_symbol_dtype`` to detect numeric-vs-string mismatches. _INT_SYMBOL_DTYPES = ( pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64, pl.Int8, pl.Int16, pl.Int32, pl.Int64, ) def _align_symbol_dtype( target: pl.DataFrame, other: pl.DataFrame, *, case_study: str, target_side: str, other_side: str, ) -> pl.DataFrame: """Cast ``other['symbol']`` to ``target['symbol'].dtype``, failing loudly. Polars will silently raise ``InvalidOperationError`` when a string column with real tickers (e.g. ``"AAPL"``) is cast to integer — the error message names neither the case study nor the column origin, making diagnostics painful. This helper detects the pl.Utf8 ↔ integer mismatch and surfaces a context-rich error before the cast, keeping the same behavior for compatible cases (same dtype, or same-kind cast). """ target_dtype = target["symbol"].dtype other_dtype = other["symbol"].dtype if other_dtype == target_dtype: return other target_is_int = target_dtype in _INT_SYMBOL_DTYPES other_is_str = other_dtype in (pl.Utf8, pl.String) other_is_int = other_dtype in _INT_SYMBOL_DTYPES target_is_str = target_dtype in (pl.Utf8, pl.String) if target_is_int and other_is_str: # Probe: every value must parse as the target integer dtype. try: return other.with_columns(pl.col("symbol").cast(target_dtype)) except Exception as exc: # noqa: BLE001 — surface Polars's opaque error raise TypeError( f"_align_symbol_dtype: incompatible symbol representations for " f"case_study={case_study!r}: {target_side}.symbol is " f"{target_dtype} (numeric ids) but {other_side}.symbol is " f"{other_dtype} (likely tickers, not parseable as integer). " f"Underlying Polars error: {exc}" ) from exc if other_is_int and target_is_str: return other.with_columns(pl.col("symbol").cast(target_dtype)) # Same-kind cast (e.g., Int32 → Int64, Utf8 → String alias). return other.with_columns(pl.col("symbol").cast(target_dtype)) def substitute_continuous_return_for_classification( predictions: pl.DataFrame, case_study: str, label: str, *, max_null_rate: float = _MAX_NULL_RATE, ) -> pl.DataFrame: """Replace binary y_true with the underlying continuous return for classification labels. The vectorized backtest computes ``gross_ret = weight * y_true``. For regression labels y_true is the forward return; for classification labels (fwd_class_*, fwd_dir_*) it is the binary class indicator, so the product collapses into a position-weighted accuracy proxy rather than economic P&L. We substitute y_true with the continuous return declared in setup.yaml::labels.classification_eval_label. Returns predictions unchanged when ``label`` is not registered as a classification target (i.e., regression labels pass through). """ if not label: return predictions from pathlib import Path as _Path import yaml as _yaml from utils import CASE_STUDIES_DIR setup_path = _Path(CASE_STUDIES_DIR) / case_study / "config" / "setup.yaml" if not setup_path.exists(): return predictions setup = _yaml.safe_load(setup_path.read_text()) mapping = (setup.get("labels") or {}).get("classification_eval_label") or {} if label not in mapping: return predictions eval_label = str(mapping[label]) eval_path = _Path(CASE_STUDIES_DIR) / case_study / "labels" / f"{eval_label}.parquet" if not eval_path.exists(): raise FileNotFoundError( f"Continuous-return label {eval_label!r} expected at {eval_path} " f"for classification label {label!r} but not found. Required so the " f"vectorized backtest can compute economic P&L instead of weight × binary." ) eval_df = pl.read_parquet(eval_path).select(["timestamp", "symbol", eval_label]) # Dedupe-assert eval_df on the join key before the left join. A duplicate # (timestamp, symbol) row in the continuous-return parquet would fan out # ``predictions`` silently, inflating downstream weight × y_true into a # wrong-but-plausible P&L (the very failure mode this function is meant # to prevent on the classification path). eval_h0 = eval_df.height eval_h_uniq = eval_df.unique(subset=["timestamp", "symbol"]).height if eval_h_uniq != eval_h0: raise ValueError( f"substitute_continuous_return_for_classification: continuous-return " f"label parquet at {eval_path} has {eval_h0 - eval_h_uniq} duplicate " f"(timestamp, symbol) rows ({eval_h_uniq} unique). Re-run the upstream " f"label step for case_study={case_study!r} to produce a unique-keyed " f"parquet." ) eval_df = eval_df.unique(subset=["timestamp", "symbol"], keep="first") # Harmonize join-key dtypes to match the (already-normalized) predictions frame. if eval_df["timestamp"].dtype != predictions["timestamp"].dtype: if eval_df["timestamp"].dtype == pl.Date: eval_df = eval_df.with_columns(pl.col("timestamp").cast(pl.Datetime("us"))) eval_df = eval_df.cast({"timestamp": predictions["timestamp"].dtype}) eval_df = _align_symbol_dtype( predictions, eval_df, case_study=case_study, target_side="predictions", other_side=f"labels/{eval_label}.parquet", ) pred_h0 = predictions.height joined = ( predictions.drop("y_true") .join(eval_df, on=["timestamp", "symbol"], how="left") .rename({eval_label: "y_true"}) ) # Height-assert: ``left`` should never produce more rows than the left frame # carried in. Belt-and-suspenders for the dedupe assertion above. if joined.height != pred_h0: raise RuntimeError( f"substitute_continuous_return_for_classification: left join " f"changed row count {pred_h0} -> {joined.height} for case_study=" f"{case_study!r} label={label!r}. eval_df keys are not unique " f"on (timestamp, symbol) after dedupe — internal invariant broken." ) n_null = int(joined["y_true"].null_count()) if n_null > 0: n_total = joined.height null_rate = n_null / n_total # Tolerant-by-design caps at ``max_null_rate`` (default # ``_MAX_NULL_RATE`` = 10%); above that, raise. >10% null rate # indicates a regeneration mismatch between the classification and # continuous-return parquets, not source-data sparsity. if null_rate > max_null_rate: raise ValueError( f"substitute_continuous_return_for_classification: " f"{n_null}/{n_total} ({null_rate:.2%}) predictions for " f"classification label {label!r} have no matching {eval_label!r} " f"value after join on (timestamp, symbol); exceeds " f"max_null_rate={max_null_rate:.2%}. Null rate above " f"{max_null_rate:.0%} indicates a regeneration mismatch " f"between the classification and continuous-return label " f"parquets; re-run the upstream label step for {case_study!r}." ) print( f" WARN substitute_continuous_return_for_classification: " f"{n_null}/{n_total} ({null_rate:.4%}) predictions for " f"classification label {label!r} have no matching {eval_label!r} " f"value after join on (timestamp, symbol); dropping those rows." ) joined = joined.filter(pl.col("y_true").is_not_null()) return joined def _apply_cost_feasible_filter( predictions: pl.DataFrame, case_study: str, prediction_hash: str | None, ) -> pl.DataFrame: """Restrict predictions to the frozen, per-split cost-feasible universe. The split is resolved from the prediction set's registry entry; the symbol list is read from ``setup.yaml::universe.cost_feasible.{split}``. Raises if the split cannot be resolved or the list is absent — a silent full-universe fallback would change the registered result. """ from pathlib import Path as _Path import yaml as _yaml from case_studies.utils.cv_window import lookup_split from utils import CASE_STUDIES_DIR if not prediction_hash: raise ValueError( "universe_filter='cost_feasible' requires a prediction_hash to " f"resolve the split for case_study={case_study!r}; got none." ) split = lookup_split(case_study, prediction_hash) if split not in ("validation", "holdout"): raise ValueError( f"universe_filter='cost_feasible' could not resolve split for " f"prediction_hash={prediction_hash!r} (case_study={case_study!r}); " f"lookup_split returned {split!r}. The prediction set must be " f"registered with a 'validation' or 'holdout' split first." ) setup = _yaml.safe_load( (_Path(CASE_STUDIES_DIR) / case_study / "config" / "setup.yaml").read_text() ) symbols = (((setup.get("universe") or {}).get("cost_feasible")) or {}).get(split) if not symbols: raise KeyError( f"setup.yaml::universe.cost_feasible.{split} missing/empty for " f"case_study={case_study!r}; required when " f"signal.universe_filter='cost_feasible'." ) filtered = predictions.filter(pl.col("symbol").is_in(list(symbols))) if filtered.is_empty() and not predictions.is_empty(): raise ValueError( f"universe_filter='cost_feasible' produced an empty frame for " f"case_study={case_study!r} split={split!r}: the prediction set's " f"symbols do not intersect the frozen cost-feasible list (e.g. a " f"point-in-time ticker mismatch like FB/META). Refusing to run a " f"zero-row backtest — same 'no silent fallback' intent as above." ) return filtered def apply_universe_filter( predictions: pl.DataFrame, prices: pl.DataFrame, case_study: str, signal_config: dict | None, prediction_hash: str | None = None, ) -> pl.DataFrame: """Apply spec-declared universe restriction to predictions before backtest. When ``signal_config["universe_filter"] == "liquid"`` (sp500_options rung-3 in the O'Donovan-Yu / Muravyev-Pearson HTM cost cascade), the backtest must restrict each rebalance date to the tightest-quoted subset of the universe. The quantile lives in ``setup.yaml::backtest.sweep.htm_cost_cascade.liquid_quantile``; the spread column is ``instr_rel_spread`` on the prices frame. When ``signal_config["universe_filter"] == "cost_feasible"`` (nasdaq100_microstructure), the backtest restricts predictions to a FROZEN, per-split symbol list committed under ``setup.yaml::universe.cost_feasible.{validation,holdout}``. The list is the cost-feasible universe — the cheapest-to-trade names by round-trip cost, profiled strictly before each window (no look-ahead — see ``build_cost_feasible_universe.py``); the split is resolved from the prediction set's registry entry via ``lookup_split``. Like ``liquid``, only the filter *name* enters the backtest hash, not the resolved symbols. Returns predictions unchanged when no filter applies. Built into ``run_backtest`` so any caller — sweep notebooks, ``generate_holdout``, ad-hoc scripts — gets the same filter as the bespoke sp500_options pipeline, driven purely by the strategy spec. """ if not signal_config: return predictions uf = str(signal_config.get("universe_filter", "")).strip().lower() if uf in ("", "full", "none"): return predictions if uf == "cost_feasible": return _apply_cost_feasible_filter(predictions, case_study, prediction_hash) if uf != "liquid": raise ValueError( f"universe_filter={uf!r} not supported. Allowed: 'liquid', 'cost_feasible', or 'full'." ) if "instr_rel_spread" not in prices.columns: raise ValueError( f"universe_filter='liquid' requires 'instr_rel_spread' on the prices " f"frame for case_study={case_study!r}; got columns={list(prices.columns)}." ) from pathlib import Path as _Path import yaml as _yaml from utils import CASE_STUDIES_DIR setup = _yaml.safe_load( (_Path(CASE_STUDIES_DIR) / case_study / "config" / "setup.yaml").read_text() ) cascade = (((setup.get("backtest") or {}).get("sweep") or {}).get("htm_cost_cascade")) or {} if "liquid_quantile" not in cascade: raise KeyError( f"setup.yaml::backtest.sweep.htm_cost_cascade.liquid_quantile missing for " f"case_study={case_study!r}; required when signal.universe_filter='liquid'." ) liquid_quantile = float(cascade["liquid_quantile"]) # Daily quantile of relative half-spread; ties broken with rank('min'). # Collapse timestamp to the date grain before grouping so any caller # supplying sub-daily or unnormalized intraday bars still produces a # within-date rank rather than a within-bar rank (mirrors the bespoke # sp500_options sweep). Dedupe ``(date, symbol)`` to one row per # (date, symbol) — taking the min half-spread when multiple bars share # a date — so the rank denominator is symbol-count, not bar-count. half = ( prices.select( pl.col("timestamp").cast(pl.Date).alias("_date"), pl.col("symbol"), (pl.col("instr_rel_spread") / 2).alias("_hs"), ) .group_by(["_date", "symbol"]) .agg(pl.col("_hs").min()) ) liquid_keys = ( half.with_columns( (pl.col("_hs").rank("min").over("_date") / pl.col("_hs").count().over("_date")).alias( "_q" ) ) .filter(pl.col("_q") <= liquid_quantile) .select([pl.col("_date").alias("timestamp"), pl.col("symbol")]) ) if liquid_keys["timestamp"].dtype != predictions["timestamp"].dtype: # Predictions stamps are typically Datetime("us") at midnight; cast # back from Date so the semi-join key types match exactly. if predictions["timestamp"].dtype == pl.Datetime("us"): liquid_keys = liquid_keys.with_columns(pl.col("timestamp").cast(pl.Datetime("us"))) else: liquid_keys = liquid_keys.cast({"timestamp": predictions["timestamp"].dtype}) liquid_keys = _align_symbol_dtype( predictions, liquid_keys, case_study=case_study, target_side="predictions", other_side="prices", ) return predictions.join(liquid_keys, on=["timestamp", "symbol"], how="semi") # --------------------------------------------------------------------------- # Core backtest function # --------------------------------------------------------------------------- def run_backtest( case_study: str, prediction_hash: str, strategy_spec: dict, *, prices: pl.DataFrame, predictions: pl.DataFrame, label: str = "", register: bool = True, initial_cash: float = 1_000_000.0, calendar: str = "NYSE", precomputed_weights: pl.DataFrame | None = None, force_rebacktest: bool = False, contract_specs: dict | None = None, ) -> BacktestRunResult: """Core backtest: predictions -> weights -> engine/vectorized -> result. This is the SINGLE entry point for ALL backtests — demo, sweep, and downstream chapters. Sweep notebooks call this in a loop with different strategy_specs; they never contain backtest math themselves. Parameters ---------- case_study : str Case study identifier (e.g., "etfs"). prediction_hash : str Hash of the prediction set being backtested. strategy_spec : dict Identity-defining configuration with signal, execution, costs sections. prices : pl.DataFrame Price data [timestamp, symbol, open, high, low, close, volume]. predictions : pl.DataFrame Predictions [timestamp, symbol, y_score, y_true, ...]. label : str Label name (used for thinning in vectorized mode). register : bool Whether to register the result in registry.db. initial_cash : float Starting portfolio value. calendar : str Trading calendar for daily return aggregation. precomputed_weights : pl.DataFrame, optional Pre-computed allocation weights [timestamp, symbol, weight]. When provided, skips signal computation and allocation — goes straight to engine/vectorized with these weights. Use this in Ch19 risk sweeps where allocation is identical across risk variants (avoids re-running expensive MVO/HRP per variant). contract_specs : dict, optional Per-asset contract specifications (futures multipliers, tick sizes). Pass for futures case studies to get correct P&L scaling. Returns ------- BacktestRunResult Unified result with daily_returns, metrics, and optional engine_result. """ import time from datetime import UTC _bt_started_at = datetime.now(UTC).isoformat() _bt_t0 = time.perf_counter() # 0. Normalize prediction columns to canonical schema, then for # classification labels replace the binary y_true with the underlying # continuous return so weight × y_true produces economic P&L rather # than a position-weighted accuracy proxy (see # ``substitute_continuous_return_for_classification`` docstring). predictions = normalize_prediction_columns(predictions) predictions = substitute_continuous_return_for_classification(predictions, case_study, label) strategy_spec = ensure_backtest_spec( case_study, get_backtest_config(case_study), strategy_spec, prices=prices, prediction_hash=prediction_hash, initial_cash=initial_cash, ) # Re-source initial_cash from the canonical spec. ensure_backtest_spec's # idempotent-canonical branch preserves an existing backtest_config.cash.initial # (typically $100K from setup.yaml) without overwriting it from the function # arg. The broker starts at that spec value; the RiskManager must initialize # its high-water-mark from the same number or it sees a fictitious 90% # drawdown on bar 1 when the function-arg default ($1M) diverges from the # spec ($100K) — halting the strategy before any trade is placed. initial_cash = float(strategy_spec["backtest_config"]["cash"]["initial"]) strategy = strategy_view(strategy_spec) # Apply spec-declared universe restriction (e.g., sp500_options rung-3 # 'liquid' subset). Driven purely by strategy.signal.universe_filter so # the bespoke sweep notebooks and generic generate_holdout share the # same code path. predictions = apply_universe_filter( predictions, prices, case_study, strategy.get("signal") or {}, prediction_hash=prediction_hash, ) # Skip-if-complete: if the backtest_hash already has complete artifacts, # return the cached result instead of re-running (unless force_rebacktest). if register and not force_rebacktest: from case_studies.utils.registry import backtest_dir as _bt_dir_fn from case_studies.utils.registry import backtest_run_status from case_studies.utils.registry.store import _case_dir, _open_registry _bt_status = backtest_run_status(case_study, prediction_hash, strategy_spec) if _bt_status.complete: _cached_dir = _bt_dir_fn(case_study, _bt_status.backtest_hash) _cached_returns = _cached_dir / "daily_returns.parquet" if _cached_returns.exists(): print(f" SKIP backtest ({_bt_status.summary()}) — reusing cached result") cached_df = pl.read_parquet(_cached_returns) # Load cached metrics from registry _db = _open_registry(_case_dir(case_study)) try: _metric_cols = [ r[1] for r in _db.execute("PRAGMA table_info(backtest_metrics)").fetchall() ] _metric_cols = [ c for c in _metric_cols if c not in ("backtest_hash", "computed_at") ] if _metric_cols: _q = f"SELECT {', '.join(_metric_cols)} FROM backtest_metrics WHERE backtest_hash = ?" _row = _db.execute(_q, (_bt_status.backtest_hash,)).fetchone() cached_metrics = dict(zip(_metric_cols, _row, strict=True)) if _row else {} else: cached_metrics = {} finally: _db.close() return BacktestRunResult( daily_returns=cached_df, metrics=cached_metrics, strategy_spec=strategy_spec, prediction_hash=prediction_hash, backtest_hash=_bt_status.backtest_hash, engine_result=None, weights=precomputed_weights, execution_mode=strategy.get("rebalance", {}).get("mode", "unknown"), ) signal_config = strategy["signal"] rebal_spec = strategy.get("rebalance", {}) if precomputed_weights is not None: # Skip signal + allocation — use provided weights directly weights = precomputed_weights elif signal_config.get("method") == "slot_persistent_signal_exit": # Slot selection IS the allocation — Ch17 allocator stage is skipped # because the slot mechanism's `weight_per_slot` plays the role of # the cheap allocator. See case_studies/utils/slot_strategy.py for # the mechanism and rules/standards docs for the design call. from case_studies.utils.slot_strategy import build_persistent_slot_weights_hybrid slot_kwargs: dict = {} for required in ("long_q", "lookback_days", "bars_per_day", "max_slots", "hold_bars"): # Reject both absent and explicit-None values so a hand-wired # signal_config surfaces here rather than as a far-off TypeError # inside build_persistent_slot_weights_hybrid. if signal_config.get(required) is None: msg = ( f"signal method 'slot_persistent_signal_exit' requires a " f"non-null {required!r} in signal_config; got {sorted(signal_config)}" ) raise KeyError(msg) slot_kwargs[required] = signal_config[required] for opt in ( "weight_per_slot", "exit_signal_q", "take_profit", "stop_loss", "pred_freshness_max_min", "direction", ): if signal_config.get(opt) is not None: slot_kwargs[opt] = signal_config[opt] weights, _slot_stats = build_persistent_slot_weights_hybrid( predictions, prices, **slot_kwargs, ) else: # 1. Convert predictions to target weights weights = build_target_weights_from_config(predictions, signal_config) # 1b. Apply allocation method if specified (Ch17+) alloc_spec = strategy.get("allocation") if alloc_spec: weights = _apply_allocation( weights, predictions, prices, alloc_spec, cadence=rebal_spec.get("cadence", ""), label=label, case_study=case_study, prediction_hash=prediction_hash, ) # 2. Dispatch to engine or vectorized bt_cfg = strategy_spec["backtest_config"] commission_block = bt_cfg["commission"] slippage_block = bt_cfg["slippage"] if commission_block.get("model") == "per_share": cost_spec = { "model": "per_share_plus_spread", "per_share": float(commission_block["per_share"]), "default_half_spread_usd": float(slippage_block.get("spread", 0.0)), "asset_spreads": dict(slippage_block.get("spread_by_asset", {}) or {}), "spread_convention": slippage_block.get("spread_convention", "half_spread"), } else: cost_spec = { "model": "percentage", "commission_bps": float(commission_block["rate"]) * 10_000.0, "slippage_bps": float(slippage_block["rate"]) * 10_000.0, } if rebal_spec["mode"] == "vectorized": # sp500_options HTM short-straddle uses a dedicated multi-cohort daily-MTM # backtest path: overlapping 5-cohort book, per-cohort daily premium + hedge # P&L, entry-spread + hedge-rebalance transaction costs. The simple # weights × y_true vectorized path cannot express this strategy because # y_true is a single 30-day return, not a daily P&L series. if case_study == "sp500_options" and label == "ret_to_expiry": result = _run_htm_daily_mtm( case_study=case_study, predictions=predictions, signal_config=signal_config, initial_cash=initial_cash, risk_spec=strategy.get("risk", {}), allocation_spec=strategy.get("allocation", {}), label=label, prediction_hash=prediction_hash, ) else: result = _run_vectorized( weights=weights, predictions=predictions, prices=prices, cost_spec=cost_spec, cadence=rebal_spec.get("cadence", ""), label=label, case_study=case_study, initial_cash=initial_cash, risk_spec=strategy.get("risk", {}), prediction_hash=prediction_hash, ) else: allow_short = signal_config.get("long_short", False) or ( str(signal_config.get("direction", "long_only")).strip().lower() == "short_only" ) result = _run_engine( weights=weights, prices=prices, predictions=predictions, strategy_spec=strategy_spec, rebalance_spec=rebal_spec, risk_spec=strategy.get("risk", {}), allow_short=allow_short, initial_cash=initial_cash, calendar=calendar, contract_specs=contract_specs, case_study=case_study, label=label, ) # Build metrics dict metrics = result["metrics"] # Build daily returns DataFrame daily_returns = result["daily_returns"] # Extract trade log, fills, equity, portfolio state from engine result # (all None for vectorized path) trades_df = result.get("trades_df") fills_df = result.get("fills_df") equity_df = result.get("equity_df") portfolio_state_df = result.get("portfolio_state_df") # 3. Register backtest_hash = None if register: from case_studies.utils.registry import ( compute_backtest_fold_metrics, register_backtest_fold_metrics, register_backtest_run, ) _bt_elapsed_s = time.perf_counter() - _bt_t0 backtest_hash = register_backtest_run( case_study, prediction_hash, strategy_spec, returns=daily_returns, trades=trades_df, fills=fills_df, equity=equity_df, portfolio_state=portfolio_state_df, weights=weights, metrics=metrics, started_at=_bt_started_at, elapsed_s=_bt_elapsed_s, ) # Compute and register per-fold backtest metrics cadence = rebal_spec.get("cadence", "daily") ppy = int(_PERIODS_PER_YEAR.get(cadence, 252)) fold_metrics = compute_backtest_fold_metrics( daily_returns, case_study, label=label, periods_per_year=ppy, ) if fold_metrics: register_backtest_fold_metrics(case_study, backtest_hash, fold_metrics) return BacktestRunResult( daily_returns=daily_returns, metrics=metrics, strategy_spec=strategy_spec, prediction_hash=prediction_hash, backtest_hash=backtest_hash, engine_result=result.get("engine_result"), weights=weights, execution_mode=rebal_spec["mode"], ) # --------------------------------------------------------------------------- # Engine path # --------------------------------------------------------------------------- def _run_engine( weights: pl.DataFrame, prices: pl.DataFrame, predictions: pl.DataFrame, strategy_spec: dict, rebalance_spec: dict, risk_spec: dict, allow_short: bool, initial_cash: float, calendar: str, contract_specs: dict | None = None, *, case_study: str | None = None, label: str | None = None, ) -> dict: """Run backtest via ml4t-backtest Engine.""" from ml4t.backtest import DataFeed, Engine, RebalanceConfig, Strategy, TargetWeightExecutor from case_studies.utils.backtest_loaders import ( extract_daily_returns_frame, infer_session_alignment, ) config = runtime_backtest_config(strategy_spec) profile_rebalance_mode = config.rebalance_mode # Session enforcement — drop bars outside trading sessions (e.g., CME # Saturdays). Idempotent with the same mutation applied in # ``ensure_backtest_spec``; kept here as belt-and-suspenders so the # engine always sees the right value even if a spec is passed in raw. apply_calendar_session_enforcement(config, calendar) # Pre-compute weight dict from DataFrame weight_dict: dict[datetime, dict[str, float]] = {} for row in weights.iter_rows(named=True): ts = row["timestamp"] if ts not in weight_dict: weight_dict[ts] = {} if row["weight"] != 0: weight_dict[ts][row["symbol"]] = row["weight"] # Resolve calendar-aware rebalance schedule, then thin by the label's # non-overlapping step from setup.yaml::labels.rebalance_step. Mirrors # the same two-step thinning that thin_to_rebalance_dates() applies on # the vectorized path (see backtest_loaders.thin_to_rebalance_dates). # Without this, multi-step labels (e.g. fwd_ret_60m on a 15m cadence # with step=4) over-rebalance by step×. # # The schedule is derived from the canonical *prediction* timeline, not # from weight_dict.keys(). Allocation-class methods (score_weighted, # HRP, MVO, inverse_vol, risk_parity) pre-thin to non-overlapping # rebalance dates inside _apply_allocation via thin_to_rebalance_dates; # if we resolved the schedule from those already-sparse weight keys and # applied gather_every(step) again, we'd thin by step² and trade # ~step× too rarely. The on_data callback already gates on # ``timestamp in weight_dict``, so dates without weights are skipped. from case_studies.utils.backtest_loaders import ( get_rebalance_step, resolve_rebalance_timestamps, ) cadence = rebalance_spec.get("cadence", "monthly_month_end") all_pred_ts = pl.Series("ts", predictions["timestamp"].unique().sort().to_list()) schedule_dates = resolve_rebalance_timestamps(all_pred_ts, cadence, calendar) if case_study and label: step = get_rebalance_step(case_study, label) if step > 1: schedule_dates = schedule_dates.gather_every(step) rebalance_schedule = set(schedule_dates.to_list()) # Build risk components from spec (Ch19) position_rules = _build_position_rules(risk_spec) risk_manager = _build_risk_manager(risk_spec, initial_cash) # Rebalance thresholds are sourced from setup.yaml::backtest.rebalance and # always present in the canonical strategy.rebalance block (populated by # ensure_backtest_spec()). min_weight_change = float(rebalance_spec["min_weight_change"]) min_trade_value = float(rebalance_spec["min_trade_value"]) # Build strategy class _PrecomputedStrategy(Strategy): def __init__(self): self._rules_set = False self.executor = TargetWeightExecutor( config=RebalanceConfig( min_trade_value=min_trade_value, min_weight_change=min_weight_change, allow_fractional=None, # Defer to broker.share_type (profile) allow_short=allow_short, rebalance_mode=profile_rebalance_mode, ) ) def on_data(self, timestamp, data, context, broker): # Set position rules on broker (once, first bar) if not self._rules_set: if position_rules: broker.set_position_rules(position_rules) self._rules_set = True # Check portfolio-level limits (each bar) if risk_manager: positions = {a: p.market_value for a, p in broker.positions.items()} risk_results = risk_manager.update( equity=broker.get_account_value(), positions=positions, timestamp=timestamp, broker=broker, ) # Two guards on purpose: the liquidate check catches a bar # where the manager flattened but left is_halted=False, while # is_halted catches a prior-bar halt; neither subsumes the other. if any(result.action == "liquidate" for result in risk_results): return if risk_manager.is_halted: return # Calendar-aware schedule: only rebalance on resolved dates if timestamp not in rebalance_schedule: return if timestamp in weight_dict: targets = {a: w for a, w in weight_dict[timestamp].items() if a in data} if targets: self.executor.execute(targets, data, broker) # Resolve the canonical (cs, label, split) window — same window for every # strategy on the same (cs, label, split). Callers pre-window `prices` via # load_backtest_prices_for(cs, label, split=...) so the parquet read is # row-group-pruned; the engine asserts the price range stays within the # canonical window. Falls back to predictions.min/max only when no # canonical window can be derived (label without CV folds, sentinel # prediction_hash, etc.). from case_studies.utils.cv_window import canonical_window, lookup_split prices_ts_dtype = prices.schema["timestamp"] window = None if case_study and label: prediction_hash = ( strategy_spec.get("backtest_config", {}).get("metadata", {}).get("prediction_hash") ) split = lookup_split(case_study, prediction_hash) if prediction_hash else None if split is not None: window = canonical_window(case_study, label, split=split) # If split is unknown (no prediction_hash in metadata or unrecognized # split label) we deliberately fall through to the predictions.min/max # branch below rather than silently mis-windowing a holdout backtest # against the validation window. if window is not None: win_start, win_end = window # Compare on the date component so calendar-edge drift (parquet starts # 2024-01-02 when win_start=2024-01-01 because Jan 1 is a holiday) is # tolerated. The upper-bound assertion fires when prices EXTEND past # the canonical window — i.e. the caller forgot to pre-window the # right edge. The lower bound is intentionally NOT asserted: callers # may load earlier-than-canonical prefix history when a rolling-vol # allocator (inverse_vol / risk_parity / hrp / mvo_ledoit_wolf) needs # warmup so the first rebalance has data-driven (not median-imputed) # weights. The daily_returns frame is sliced to [win_start, win_end] # below regardless of how wide the load was. prices_dates = prices["timestamp"].dt.date() prices_min_date = prices_dates.min() prices_max_date = prices_dates.max() if prices_min_date is None or prices_max_date is None: raise RuntimeError( f"Empty prices frame for cs={case_study} label={label} " f"split={split} — canonical window [{win_start}, {win_end}]." ) if prices_max_date > win_end: raise AssertionError( f"Prices not pre-windowed for cs={case_study} label={label} " f"split={split}: canonical window [{win_start}, {win_end}], " f"prices range [{prices_min_date}, {prices_max_date}] — " f"upper bound exceeded. Pass end_date to load_backtest_prices() " f"or call load_backtest_prices_for(cs, label, split=split)." ) elif predictions.height > 0: # Fallback when canonical window unavailable: still slice to the # predictions' span so demo notebooks with sentinel prediction_hash # don't process pre-history. pred_ts = predictions["timestamp"] if pred_ts.dtype != prices_ts_dtype: pred_ts = pred_ts.cast(prices_ts_dtype) prices = prices.filter( (pl.col("timestamp") >= pred_ts.min()) & (pl.col("timestamp") <= pred_ts.max()) ) # signals_df is intentionally omitted: _PrecomputedStrategy reads # weight_dict directly, so routing predictions through the bar iterator # would waste hot-path memory. feed = DataFeed(prices_df=prices, feed_spec=config.feed_spec) strategy = _PrecomputedStrategy() engine = Engine.from_config(feed, strategy, config, contract_specs=contract_specs) engine_result = engine.run() # Extract daily returns session_aligned = infer_session_alignment(calendar) daily_df = extract_daily_returns_frame( engine_result, calendar=calendar, session_aligned=session_aligned, ) # Slice the persisted daily-returns frame to the canonical (cs, label, # split) window — same window as the price-trim above, so every # (cs, label, split) produces a daily_returns parquet covering the same # dates regardless of which strategy was run. Date-component compare so # intraday bars on win_end aren't dropped by midnight promotion. if window is not None: daily_df = daily_df.filter( (pl.col("timestamp").dt.date() >= window[0]) & (pl.col("timestamp").dt.date() <= window[1]) ) returns_arr = daily_df["daily_return"].to_numpy() ppy = calendar_periods_per_year(calendar) metrics = compute_portfolio_metrics(returns_arr, periods_per_year=ppy, trim_leading_zeros=False) # Engine-specific metrics (execution details not derivable from returns) m = engine_result.metrics metrics["num_trades"] = m.get("num_trades", 0) metrics["total_commission"] = m.get("total_commission", 0.0) metrics["total_slippage"] = m.get("total_slippage", 0.0) # avg_turnover: target-weight semantics (sum_i |Δw_i| averaged over the daily # timeline, 0 on non-rebalance days). Same formula as the vectorized path so # the registry column has consistent meaning across both engines. Skipping the # engine's own `m["avg_turnover"]` (notional/equity) — that value is unbounded # for leveraged products (cme_futures multipliers inflate it 10⁴–10⁵×) and # mixes incompatibly with vectorized-path rows on the same column. if weights.height > 0: weights_sorted = weights.sort("symbol", "timestamp").with_columns( abs_change=( pl.col("weight") - pl.col("weight").shift(1).over("symbol").fill_null(0.0) ).abs(), ) turnover_by_ts = weights_sorted.group_by("timestamp").agg( turnover=pl.col("abs_change").sum() ) # Align to daily timeline so non-rebalance days contribute 0 to the mean # (matches port_ret.join(turnover) in the vectorized path). turnover_aligned = daily_df.join( turnover_by_ts.with_columns(pl.col("timestamp").cast(daily_df.schema["timestamp"])), on="timestamp", how="left", ).with_columns(pl.col("turnover").fill_null(0.0)) mean_turnover = turnover_aligned["turnover"].mean() metrics["avg_turnover"] = float(mean_turnover) if mean_turnover is not None else 0.0 else: metrics["avg_turnover"] = 0.0 # Extract trade log trades_df = None if engine_result.trades: try: trades_df = engine_result.to_trades_dataframe() except Exception as e: import logging logging.getLogger(__name__).warning("Trade extraction failed: %s", e) # Extract fill-level execution records (quote-aware since backtest b11) fills_df = None if engine_result.fills: try: fills_df = engine_result.to_fills_dataframe() except Exception as e: import logging logging.getLogger(__name__).warning("Fills extraction failed: %s", e) # Extract equity curve and portfolio state (bar-level resolution) equity_df = None portfolio_state_df = None try: equity_df = engine_result.to_equity_dataframe() portfolio_state_df = engine_result.to_portfolio_state_dataframe() except Exception as e: import logging logging.getLogger(__name__).warning("Equity/portfolio state extraction failed: %s", e) return { "daily_returns": daily_df, "metrics": metrics, "engine_result": engine_result, "trades_df": trades_df, "fills_df": fills_df, "equity_df": equity_df, "portfolio_state_df": portfolio_state_df, } # --------------------------------------------------------------------------- # Hold-to-expiry daily-MTM path (sp500_options / ret_to_expiry) # --------------------------------------------------------------------------- def _run_htm_daily_mtm( case_study: str, predictions: pl.DataFrame, signal_config: dict, initial_cash: float, risk_spec: dict | None = None, allocation_spec: dict | None = None, label: str | None = None, prediction_hash: str | None = None, ) -> dict: """Dispatch wrapper for the hold-to-expiry daily-MTM short-straddle backtest. Delegates to ``case_studies.sp500_options._htm_backtest.run_htm_daily_mtm``, which implements the multi-cohort daily-MTM accounting: - Friday entry of top-K short straddles, ~30-day DTE. - Daily delta hedge via the underlying stock (threshold rehedging). - Cash-settle at expiry (no market exit, no exit bid-ask). - Full transaction costs: entry option spread (bid-ask on both legs) on cohort entry day; hedge-trade spread + equity commission on every hedge rebalance day. - Book size = 5 concurrent cohorts × 1/5 capital each (fully invested). The entry-and-weighting scheme is read from ``signal_config`` (same shape as the vectorized signal dispatcher): ``method`` + ``top_k`` / ``percentile``. Returns the same shape as ``_run_vectorized``: ``{daily_returns, metrics}`` where ``daily_returns`` has columns ``[timestamp, daily_return]`` so the registry write path treats it identically to any other backtest. """ from pathlib import Path import yaml from case_studies.sp500_options._htm_backtest import run_htm_daily_mtm from utils import CASE_STUDIES_DIR from utils.paths import REPO_ROOT cs_dir = CASE_STUDIES_DIR / case_study labels_dir = cs_dir / "labels" # Anchor on REPO_ROOT — same convention as every other case-study data # path. Resolving relative to cwd masked real "data missing" errors as # cwd-mismatch fallbacks pointing at a different (also-missing) path. raw_options_dir = REPO_ROOT / "data" / "equities" / "market" / "sp500" / "options_straddles_raw" method = str(signal_config.get("method", "equal_weight_top_k")) top_k = int(signal_config.get("top_k", 20)) percentile = float(signal_config.get("percentile", 90.0)) exit_at_max_days = signal_config.get("exit_at_max_days") if exit_at_max_days is not None: exit_at_max_days = int(exit_at_max_days) # For round-trip mode (exit_at_max_days set), weekly entry with a 10-day # hold yields ~2 concurrent cohorts, not 5. Caller can override via # signal_config.n_roll; default is the HTM-expiry value (5). from case_studies.sp500_options._htm_backtest import N_ROLL_DEFAULT n_roll = int(signal_config.get("n_roll", N_ROLL_DEFAULT)) # Read cost/risk parameters from setup.yaml so the wrapper does not # silently drop them. Required keys raise KeyError; missing optional keys # fall through to run_htm_daily_mtm's defaults. setup = yaml.safe_load((cs_dir / "config" / "setup.yaml").read_text()) cost_components = setup["costs"]["components"] delta_threshold = float(setup["hedging_protocol"]["delta_threshold"]) hedge_spread_bps = float(cost_components["hedge_spread"]["estimate_bps_of_notional"]) equity_commission_per_share = float(cost_components["commission"]["equity_per_share"]) option_commission_per_contract = float(cost_components["commission"]["option_per_contract"]) result = run_htm_daily_mtm( case_study=case_study, predictions=predictions, labels_dir=labels_dir, raw_options_dir=raw_options_dir, method=method, top_k=top_k, percentile=percentile, exit_at_max_days=exit_at_max_days, n_roll=n_roll, delta_threshold=delta_threshold, hedge_spread_bps=hedge_spread_bps, equity_commission_per_share=equity_commission_per_share, option_commission_per_contract=option_commission_per_contract, allocation_spec=allocation_spec, ) port = result["daily_returns"] metrics = result["metrics"] # Slice port to canonical (cs, label, split) window so daily_returns and # the aux cost-accounting metrics (cumulative_entry_cost, n_rebalance_dates, # etc.) all reflect the same canonical window. Mirrors _run_engine and # _run_vectorized — same drifting-parquet bug otherwise. The cohort fields # (entry_cost_day, n_open, etc.) are filtered with the same `date` slice # because the multi-cohort daily-MTM book emits one row per holding date. sliced = False if prediction_hash and case_study and label: from case_studies.utils.cv_window import canonical_window, lookup_split split = lookup_split(case_study, prediction_hash) if split is not None: window = canonical_window(case_study, label, split=split) if window is not None: win_start, win_end = window port_filtered = port.filter( (pl.col("date").cast(pl.Date) >= win_start) & (pl.col("date").cast(pl.Date) <= win_end) ) if port_filtered.is_empty(): raise RuntimeError( f"Canonical window [{win_start}, {win_end}] for " f"cs={case_study} label={label} split={split} produced " f"empty port (HTM daily-MTM; port span " f"{port['date'].min()} → {port['date'].max()})." ) if port_filtered.height != port.height: sliced = True port = port_filtered # Shape the return like _run_vectorized so the registry writer is agnostic. daily_returns = port.select( pl.col("date").cast(pl.Datetime("us")).alias("timestamp"), pl.col("portfolio_ret").alias("daily_return"), ) # When the canonical-window slice actually trimmed rows, recompute the # returns-based metric set so the registry Sharpe/CAGR/volatility/etc. # reflect the sliced daily_returns rather than the inner function's # pre-slice values. Aux cohort metrics (cumulative_entry_cost, etc.) are # recomputed from sliced port below regardless of slice. if sliced: from case_studies.sp500_options._htm_backtest import _compute_metrics metrics.update(_compute_metrics(port)) # Optional portfolio-level risk overlay (Ch19). Same mechanism as vectorized # path: operates on the daily return series post-hoc. if risk_spec: from case_studies.sp500_options._htm_backtest import _compute_metrics port_for_risk = daily_returns.rename({"daily_return": "net_ret"}) port_for_risk = _apply_vectorized_risk(port_for_risk, risk_spec) daily_returns = port_for_risk.select( pl.col("timestamp"), pl.col("net_ret").alias("daily_return") ) # Recompute the full metric set from the post-overlay return series so # cagr/max_drawdown/volatility/etc. reflect the same series as Sharpe. post = daily_returns.rename({"daily_return": "portfolio_ret"}) metrics.update(_compute_metrics(post)) # Final unified metric pass: replace HTM-internal Sharpe/Sortino/etc. with # the canonical ml4t.diagnostic.PortfolioAnalysis values so HTM metrics are # comparable to engine/vectorized paths AND include the uncertainty bands # (sharpe_se_lo, sharpe_ci95_lo/hi, sortino_ci95_*, ann_return_hac_se + # ci95, max_dd_ci95_*, calmar_ci95_*, psr_pvalue, bootstrap_block_length/n). # HTM uses daily MTM on NYSE sessions → periods_per_year = 252. Operates on # the FINAL daily_returns (post-slice, post-risk-overlay) so the persisted # parquet and the registered metrics are derived from the same series. returns_arr = daily_returns["daily_return"].to_numpy() metrics.update( compute_portfolio_metrics( returns_arr, periods_per_year=252, case_study=case_study, label=label, uncertainty=True, ) ) # Number of distinct rebalance events (= entry days with any new cohort). # `n_open.sum()` is the count of cohort-days, kept under a distinct key. metrics["n_rebalance_dates"] = int((port["entry_cost_day"] > 0).sum()) metrics["cohort_days_open"] = int(port["n_open"].sum()) metrics["avg_cohorts_open"] = float(port["n_open"].mean()) metrics["cumulative_entry_cost"] = float(port["entry_cost_day"].sum()) metrics["cumulative_hedge_cost"] = float(port["hedge_cost_day"].sum()) if "exit_cost_day" in port.columns: metrics["cumulative_exit_cost"] = float(port["exit_cost_day"].sum()) return { "daily_returns": daily_returns, "metrics": metrics, } # --------------------------------------------------------------------------- # Vectorized path (for 3 special case studies) # --------------------------------------------------------------------------- def _run_vectorized( weights: pl.DataFrame, predictions: pl.DataFrame, prices: pl.DataFrame, cost_spec: dict, cadence: str, label: str, case_study: str, initial_cash: float, risk_spec: dict | None = None, prediction_hash: str | None = None, ) -> dict: """Run vectorized backtest (weight × forward return - costs). Used for us_firm_characteristics, sp500_options, nasdaq100_microstructure. Cost dispatch supports two models: * percentage — fractional drag = turnover × (commission_bps + slippage_bps) / 1e4 * per_share_plus_spread — fractional drag = sum_i(|Δw_i| × (per_share + half_spread_i) / price_i), which models per-share commission and per-asset half-spread slippage. The |Δshares_i| × cost_per_share_i / NAV identity reduces to the form above because |Δshares_i| = |Δw_i| × NAV / price_i and NAV cancels. Portfolio-level risk overlays are applied post-hoc via ``_apply_vectorized_risk``. Only ``max_drawdown`` is supported — it models an intraday exit at the threshold with explicit slippage. ``daily_loss`` is refused on this path: an honest per-bar halt requires intraday position tracking that the close-to-close return series cannot express. Position-level rules (stop-loss, trailing stop) likewise cannot be applied in vectorized mode. """ from case_studies.utils.backtest_loaders import get_rebalance_step, thin_to_rebalance_dates # Thin predictions to non-overlapping periods. Step is declared per-label # in the case study's setup.yaml under labels.rebalance_step. step = get_rebalance_step(case_study, label) thinned = thin_to_rebalance_dates(predictions, cadence=cadence, step=step) # Re-compute weights on thinned predictions # (The weights were computed on full predictions; we need to recompute # or filter to thinned timestamps) rebalance_dates = thinned["timestamp"].unique() # Semi-join to filter — avoids Polars is_in precision mismatch rebal_df = pl.DataFrame({"timestamp": rebalance_dates}) if rebal_df["timestamp"].dtype != weights["timestamp"].dtype: rebal_df = rebal_df.cast({"timestamp": weights["timestamp"].dtype}) weights_thinned = weights.join(rebal_df, on="timestamp", how="semi") # Harmonize timestamp dtypes before join thinned_sel = thinned.select(["timestamp", "symbol", "y_true"]) if weights_thinned["timestamp"].dtype != thinned_sel["timestamp"].dtype: thinned_sel = thinned_sel.cast({"timestamp": weights_thinned["timestamp"].dtype}) # Join weights with forward returns bt = weights_thinned.join( thinned_sel, on=["timestamp", "symbol"], how="inner", ) # Portfolio returns per period port_ret = ( bt.group_by("timestamp") .agg( gross_ret=(pl.col("weight") * pl.col("y_true")).sum(), n_positions=pl.len(), ) .sort("timestamp") ) # Compute per-symbol weight changes and aggregate turnover for diagnostics weights_sorted = weights_thinned.sort("timestamp", "symbol").with_columns( abs_change=( pl.col("weight") - pl.col("weight").shift(1).over("symbol").fill_null(0.0) ).abs(), ) turnover = weights_sorted.group_by("timestamp").agg(turnover=pl.col("abs_change").sum()) port_ret = port_ret.join(turnover, on="timestamp", how="left").with_columns( pl.col("turnover").fill_null(0.0) ) # Apply costs — dispatch on cost_spec.model cost_model = cost_spec.get("model", "percentage") if cost_model == "per_share_plus_spread": per_share = float(cost_spec["per_share"]) default_hs = float(cost_spec.get("default_half_spread_usd", 0.0)) asset_spreads = cost_spec.get("asset_spreads", {}) or {} if "close" in prices.columns: price_col = "close" elif "mid" in prices.columns: price_col = "mid" else: raise ValueError( "per_share_plus_spread cost model requires a 'close' or 'mid' " f"column on the prices frame; got columns={list(prices.columns)}" ) prices_sel = prices.select(["timestamp", "symbol", pl.col(price_col).alias("_px")]) if prices_sel["timestamp"].dtype != weights_sorted["timestamp"].dtype: prices_sel = prices_sel.cast({"timestamp": weights_sorted["timestamp"].dtype}) wc_priced = weights_sorted.join(prices_sel, on=["timestamp", "symbol"], how="left") # Per-asset half-spread map; default fallback for symbols not in map if asset_spreads: wc_priced = wc_priced.with_columns( _hs=pl.col("symbol").replace_strict( asset_spreads, default=default_hs, return_dtype=pl.Float64 ) ) else: wc_priced = wc_priced.with_columns(_hs=pl.lit(default_hs, dtype=pl.Float64)) # Fractional cost drag per period: sum_i(|Δw_i| × (per_share + hs_i) / price_i) # Skip rows where price is null (symbol not in prices frame); they contribute 0. cost_drag = ( wc_priced.with_columns( _drag=pl.when(pl.col("_px").is_not_null() & (pl.col("_px") > 0)) .then(pl.col("abs_change") * (per_share + pl.col("_hs")) / pl.col("_px")) .otherwise(0.0) ) .group_by("timestamp") .agg(cost_drag=pl.col("_drag").sum()) ) port_ret = port_ret.join(cost_drag, on="timestamp", how="left").with_columns( pl.col("cost_drag").fill_null(0.0), net_ret=pl.col("gross_ret") - pl.col("cost_drag"), ) else: cost_rate = ( float(cost_spec.get("commission_bps", 0.0)) + float(cost_spec.get("slippage_bps", 0.0)) ) / 10_000 port_ret = port_ret.with_columns( net_ret=pl.col("gross_ret") - pl.col("turnover") * cost_rate, ) # Slice port_ret to canonical (cs, label, split) window so every strategy # on the same (cs, label, split) produces a daily_returns parquet covering # the same dates regardless of which predictions span which dates. Mirrors # _run_engine slice at lines 740-786 + 805-816. Without this, vectorized # daily_returns drift by the prediction window's left/right edges and # cross-config comparisons aren't apples-to-apples. Slice happens BEFORE # the risk overlay so the drawdown breaker only fires on canonical-window # losses (not on stale pre-canonical drawdowns). if prediction_hash and case_study and label: from case_studies.utils.cv_window import canonical_window, lookup_split split = lookup_split(case_study, prediction_hash) if split is not None: window = canonical_window(case_study, label, split=split) if window is not None: win_start, win_end = window port_ret_filtered = port_ret.filter( (pl.col("timestamp").cast(pl.Date) >= win_start) & (pl.col("timestamp").cast(pl.Date) <= win_end) ) if port_ret_filtered.is_empty(): raise RuntimeError( f"Canonical window [{win_start}, {win_end}] for " f"cs={case_study} label={label} split={split} produced " f"empty port_ret (vectorized path; port_ret span " f"{port_ret['timestamp'].min()} → {port_ret['timestamp'].max()})." ) port_ret = port_ret_filtered # Apply portfolio-level risk overlays (post-hoc on return series) if risk_spec: port_ret = _apply_vectorized_risk(port_ret, risk_spec) # Daily returns DataFrame daily_returns = port_ret.select( pl.col("timestamp"), pl.col("net_ret").alias("daily_return"), ) # Portfolio metrics via ml4t-diagnostic returns_arr = daily_returns["daily_return"].to_numpy() n = len(returns_arr) # Annualization: use cadence when known, else estimate from data span periods_per_year = int(_PERIODS_PER_YEAR.get(cadence, 0)) if not periods_per_year and n > 1: all_ts = daily_returns["timestamp"].unique().sort() span_secs = float((all_ts[-1] - all_ts[0]).total_seconds()) span_years = span_secs / (365.25 * 86400) periods_per_year = int(n / span_years) if span_years > 0.01 else 252 metrics = compute_portfolio_metrics(returns_arr, periods_per_year=periods_per_year or 252) # Vectorized-specific metrics (not derivable from returns alone) avg_turnover = float(port_ret["turnover"].mean()) if n > 0 else 0.0 metrics["avg_turnover"] = avg_turnover metrics["n_periods"] = n return { "daily_returns": daily_returns, "metrics": metrics, } # --------------------------------------------------------------------------- # Vectorized risk overlays (Ch19) — portfolio-level limits only # --------------------------------------------------------------------------- def _apply_vectorized_risk(port_ret: pl.DataFrame, risk_spec: dict) -> pl.DataFrame: """Apply portfolio-level risk limits to a close-to-close return series. Used by the vectorized + HTM dispatch paths (us_firm_characteristics, sp500_options/ret_to_expiry). Engine-path case studies use ``ml4t.backtest.risk.RiskManager`` via ``_build_risk_manager`` and never enter this function. Supported limits: ``max_drawdown``: model an intraday exit at the drawdown threshold. Find the first close where cumulative drawdown crosses ``-threshold``; replace that bar's return with the equity move from the prior close down to ``peak * (1 - threshold) * (1 - breach_slippage)``; zero every subsequent bar. Default ``breach_slippage`` = 50 bps; configurable via ``risk_spec['breach_slippage']``. Refused (raises ``ValueError``): ``daily_loss``: a vectorized close-to-close series cannot implement a per-bar daily-loss halt without lookahead — zeroing the breach bar's loss while keeping every winning bar inflates Sharpe to infinity in the limit. Use ``max_drawdown`` or move the CS to the engine path (which has proper ``DailyLossLimit`` halt-on-update semantics through ``ml4t.backtest.risk``). """ limits = risk_spec.get("portfolio_limits", []) if not limits: return port_ret dd_threshold = None for lc in limits: ltype = lc["type"] if ltype == "max_drawdown": dd_threshold = lc["threshold"] elif ltype == "daily_loss": raise ValueError( "daily_loss portfolio limit is not supported on the " "vectorized/HTM path: the only honest implementation needs " "intraday position tracking (engine path's " "ml4t.backtest.risk.DailyLossLimit). Drop it from the sweep " "config or move the case study to the engine path." ) returns = port_ret["net_ret"].to_numpy().copy() if dd_threshold is not None: breach_slippage = float(risk_spec.get("breach_slippage", 0.005)) cum = np.cumprod(1 + returns) peak = np.maximum.accumulate(cum) drawdowns = cum / peak - 1.0 breach_idx = np.where(drawdowns < -abs(dd_threshold))[0] if len(breach_idx) > 0: i = int(breach_idx[0]) prior_eq = float(cum[i - 1]) if i > 0 else 1.0 # Exit at peak * (1 - threshold), then take breach_slippage on the # exit. Equity at exit = peak[i] * (1 - threshold) * (1 - slip). exit_eq = float(peak[i]) * (1.0 - abs(dd_threshold)) * (1.0 - breach_slippage) returns[i] = exit_eq / prior_eq - 1.0 returns[i + 1 :] = 0.0 return port_ret.with_columns(pl.Series("net_ret", returns)) # --------------------------------------------------------------------------- # Allocation dispatch (Ch17) # --------------------------------------------------------------------------- def _apply_allocation( weights: pl.DataFrame, predictions: pl.DataFrame, prices: pl.DataFrame, alloc_spec: dict, *, cadence: str = "", label: str = "", case_study: str = "", prediction_hash: str | None = None, ) -> pl.DataFrame: """Post-process signal weights with an allocation method. Dispatches to utils.allocation functions based on alloc_spec["method"]. The signal weights determine asset SELECTION (which assets are in the portfolio); the allocation method determines SIZING (how much weight each gets). """ method = alloc_spec.get("method", "equal_weight") top_k = int(alloc_spec.get("top_k", weights["symbol"].n_unique())) long_short = bool(alloc_spec.get("long_short", False)) if method == "equal_weight": return weights from case_studies.utils.allocation import ( _cap_weights, compute_conformal_weights, compute_hrp_weights, compute_inverse_vol_weights, compute_mvo_weights, compute_risk_parity_weights, ) # Harmonize timestamp + symbol dtypes before joins. us_firm predictions # carry symbol as UInt32 (stock_id) while prices use String; without this, # downstream is_in/join on symbol silently produces empty results. The # symbol cast is routed through ``_align_symbol_dtype`` so a real # ticker-vs-id mismatch surfaces with case-study context instead of an # opaque Polars ``InvalidOperationError``. ts_dtype = weights["timestamp"].dtype if predictions["timestamp"].dtype != ts_dtype: predictions = predictions.cast({"timestamp": ts_dtype}) predictions = _align_symbol_dtype( weights, predictions, case_study=case_study, target_side="weights", other_side="predictions", ) if prices["timestamp"].dtype != ts_dtype: prices = prices.cast({"timestamp": ts_dtype}) prices = _align_symbol_dtype( weights, prices, case_study=case_study, target_side="weights", other_side="prices", ) # Filter predictions to only the assets selected by the signal step selected_keys = weights.select(["timestamp", "symbol"]).unique() filtered_preds = predictions.join(selected_keys, on=["timestamp", "symbol"], how="inner") # Allocation only matters on actual rebalance dates. Without cadence-aware # thinning, covariance-based allocators solve the same optimization on every # prediction timestamp even when the engine only rebalances weekly/monthly. from case_studies.utils.backtest_loaders import get_rebalance_step, thin_to_rebalance_dates if not case_study or not label: raise ValueError( "_apply_allocation requires both case_study and label to look up " "labels.rebalance_step from setup.yaml. Pass them from the caller." ) step = get_rebalance_step(case_study, label) rebal_preds = thin_to_rebalance_dates(filtered_preds, cadence=cadence, step=step) # Max weight cap — applied after all covariance-based allocators max_weight = float(alloc_spec.get("max_weight", 0.0)) if method == "score_weighted": from case_studies.utils.signals import build_target_weights result = build_target_weights( rebal_preds, method="score_weighted_top_k", top_k=top_k, long_short=long_short, ) if max_weight > 0: result = _cap_weights(result, max_weight) return result if method == "conformal_weighted": if not prediction_hash: raise ValueError( "conformal_weighted allocation requires prediction_hash; " "caller must pass it through _apply_allocation." ) from case_studies.utils.conformal import load_conformal_widths alpha = float(alloc_spec.get("alpha", 0.20)) widths = load_conformal_widths(case_study, prediction_hash, alpha=alpha) floor_q = float(alloc_spec.get("floor_quantile", 0.01)) result = compute_conformal_weights( rebal_preds, widths, top_k, long_short=long_short, floor_quantile=floor_q, ) if max_weight > 0: result = _cap_weights(result, max_weight) return result vol_window = int(alloc_spec.get("vol_window", alloc_spec.get("lookback", 63))) if method == "inverse_vol": result = compute_inverse_vol_weights( rebal_preds, prices, top_k, vol_window=vol_window, long_short=long_short ) elif method == "risk_parity": result = compute_risk_parity_weights( rebal_preds, prices, top_k, vol_window=vol_window, long_short=long_short ) elif method in ("mvo", "mvo_ledoit_wolf"): lookback = int(alloc_spec.get("lookback", 126)) mvo_max_weight = max_weight if max_weight > 0 else 1.0 result = compute_mvo_weights( rebal_preds, prices, top_k, lookback=lookback, max_weight=mvo_max_weight, long_short=long_short, ) elif method == "hrp": result = compute_hrp_weights( rebal_preds, prices, top_k, vol_window=vol_window, long_short=long_short ) else: import logging logging.getLogger(__name__).warning( "Unknown allocation method '%s', returning signal weights", method ) return weights if max_weight > 0: result = _cap_weights(result, max_weight) return result # --------------------------------------------------------------------------- # Risk rules (Ch19) — engine-level integration # --------------------------------------------------------------------------- def _build_position_rules(risk_spec: dict): """Create ml4t-backtest PositionRule objects from risk spec. Supports: stop_loss, trailing_stop, time_exit. Returns a RuleChain (multiple rules) or single rule, or None. """ rules_config = risk_spec.get("position_rules", []) if not rules_config: return None from ml4t.backtest.risk import RuleChain, StopLoss, TimeExit, TrailingStop rules = [] for rc in rules_config: rtype = rc["type"] if rtype == "stop_loss": rules.append(StopLoss(pct=rc["threshold"])) elif rtype == "trailing_stop": rules.append(TrailingStop(pct=rc["threshold"])) elif rtype == "time_exit": rules.append(TimeExit(max_bars=rc["bars"])) if not rules: return None return RuleChain(rules) if len(rules) > 1 else rules[0] def _build_risk_manager(risk_spec: dict, initial_cash: float): """Create RiskManager with portfolio-level limits from risk spec. Supports: max_drawdown, daily_loss. Returns initialized RiskManager, or None. """ limits_config = risk_spec.get("portfolio_limits", []) if not limits_config: return None from ml4t.backtest.risk import DailyLossLimit, MaxDrawdownLimit, RiskManager limits = [] for lc in limits_config: ltype = lc["type"] if ltype == "max_drawdown": limits.append(MaxDrawdownLimit(max_drawdown=lc["threshold"])) elif ltype == "daily_loss": limits.append(DailyLossLimit(max_daily_loss_pct=lc["threshold"])) if not limits: return None rm = RiskManager(limits=limits) rm.initialize(initial_cash) return rm # --------------------------------------------------------------------------- # Convenience: run random-signal plumbing test # --------------------------------------------------------------------------- def run_plumbing_test( case_study: str, prices: pl.DataFrame, strategy_spec: dict, *, n_assets: int | None = None, top_k: int = 20, seed: int = 42, initial_cash: float = 1_000_000.0, calendar: str = "NYSE", contract_specs: dict | None = None, ) -> float: """Run a random-signal backtest. Returns Sharpe ratio (should be ~0). This validates the backtest pipeline produces no spurious alpha from random inputs. """ strategy_spec = ensure_backtest_spec( case_study, get_backtest_config(case_study), strategy_spec, prices=prices, prediction_hash="plumbing_test", initial_cash=initial_cash, ) strategy = strategy_view(strategy_spec) rebal_spec = strategy.get("rebalance", {}) if rebal_spec["mode"] == "vectorized": # Generate random weights timestamps = prices["timestamp"].unique().sort() symbols = prices["symbol"].unique().sort().to_list() rng = np.random.default_rng(seed) rows = [] k = min(top_k, len(symbols)) for ts in timestamps: selected = rng.choice(symbols, size=k, replace=False) w = 1.0 / k for s in selected: rows.append({"timestamp": ts, "symbol": s, "weight": w}) random_weights = pl.DataFrame(rows) # Need y_true for vectorized path — use prices to get returns # This is a simplified plumbing test for vectorized return 0.0 # Vectorized plumbing test is in the notebook # Engine plumbing test from ml4t.backtest import DataFeed, Engine, RebalanceConfig, Strategy, TargetWeightExecutor config = runtime_backtest_config(strategy_spec) signal_config = strategy["signal"] long_short = bool(signal_config.get("long_short", False)) signal_direction = str(signal_config.get("direction", "long_only")).strip().lower() allow_short = long_short or signal_direction == "short_only" # Calendar-aware rebalance schedule for random signal from case_studies.utils.backtest_loaders import resolve_rebalance_timestamps cadence = rebal_spec.get("cadence", "monthly_month_end") price_ts = prices["timestamp"].unique().sort() plumbing_schedule = set(resolve_rebalance_timestamps(price_ts, cadence, calendar).to_list()) asset_list = sorted(prices["symbol"].unique().to_list()) k = min(top_k, len(asset_list)) rng = np.random.default_rng(seed) class _RandomStrategy(Strategy): def __init__(self): self.executor = TargetWeightExecutor( config=RebalanceConfig( min_trade_value=100.0, min_weight_change=0.005, allow_fractional=None, # Defer to broker.share_type (profile) allow_short=allow_short, ) ) def on_data(self, timestamp, data, context, broker): if timestamp not in plumbing_schedule: return available = [a for a in asset_list if a in data] if not available: return if signal_direction == "short_only": selected = rng.choice(available, size=min(k, len(available)), replace=False) weight = -1.0 / len(selected) targets = {a: weight for a in selected} elif long_short: side_k = min(k, len(available) // 2) if side_k == 0: return selected = rng.choice(available, size=side_k * 2, replace=False).tolist() longs = selected[:side_k] shorts = selected[side_k:] long_weight = 1.0 / len(longs) short_weight = -1.0 / len(shorts) targets = {a: long_weight for a in longs} targets.update({a: short_weight for a in shorts}) else: selected = rng.choice(available, size=min(k, len(available)), replace=False) weight = 1.0 / len(selected) targets = {a: weight for a in selected} self.executor.execute(targets, data, broker) feed = DataFeed(prices_df=prices, feed_spec=config.feed_spec) strategy = _RandomStrategy() engine = Engine.from_config(feed, strategy, config, contract_specs=contract_specs) result = engine.run() return result.metrics.get("sharpe", 0.0)