wehub-resource-sync
776 lines
32 KiB
Python
776 lines
32 KiB
Python
"""Central sweep configuration for Ch16-20 parametric backtesting.
|
||
|
||
The Ch16-19 sweep grid (entry schemes, allocators, cost grid, risk controls)
|
||
is declared per-case-study under ``backtest.sweep`` in each case study's
|
||
``config/setup.yaml``. ``load_sweep(case_study)`` and the ``*_for`` / ``get_*``
|
||
helpers read that block and synthesize dispatcher-shaped config dicts. No
|
||
implicit fallback — the block is required, and ``KeyError`` is raised when
|
||
missing.
|
||
|
||
Usage:
|
||
from case_studies.utils.sweep_config import (
|
||
get_entry_schemes_for, get_top_k_values_for,
|
||
get_allocators, get_cost_grid_bps,
|
||
get_position_risk_controls, get_portfolio_risk_controls,
|
||
)
|
||
|
||
schemes = get_entry_schemes_for(
|
||
"us_firm_characteristics", label="fwd_ret_1m",
|
||
n_assets=2500, long_short=True,
|
||
)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import polars as pl
|
||
import yaml
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# MAE/MFE-calibrated risk controls (Ch19)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def calibrate_trailing_stops(
|
||
prices: pl.DataFrame,
|
||
horizons: list[int] | None = None,
|
||
percentiles: tuple[float, ...] = (10, 25),
|
||
asset_col: str = "symbol",
|
||
time_col: str = "timestamp",
|
||
price_col: str = "close",
|
||
) -> list[dict]:
|
||
"""Compute case-study-specific trailing stop thresholds from MAE percentiles.
|
||
|
||
Uses the ml4t-diagnostic excursion module to analyze how far prices
|
||
typically draw down over various holding horizons, then converts
|
||
MAE percentiles into trailing stop thresholds.
|
||
|
||
Args:
|
||
prices: Long-format DataFrame with [timestamp, symbol, close] (at minimum).
|
||
horizons: Bar horizons to analyze. Default uses [10, 20, 40] for daily data.
|
||
percentiles: MAE percentiles to convert to stops. Default (10, 25) gives
|
||
tight and moderate thresholds.
|
||
asset_col: Name of the asset identifier column.
|
||
time_col: Name of the timestamp column.
|
||
price_col: Name of the close price column.
|
||
|
||
Returns:
|
||
List of risk control dicts in the same shape as the declared
|
||
``backtest.sweep.risk_controls.position`` entries in setup.yaml,
|
||
with names like ``trailing_mae_p10_h20`` (10th percentile MAE at
|
||
20-bar horizon).
|
||
"""
|
||
import numpy as np
|
||
from ml4t.diagnostic.evaluation.excursion import analyze_excursions
|
||
|
||
if horizons is None:
|
||
horizons = [10, 20, 40]
|
||
|
||
# Aggregate across all assets for a universe-level excursion profile
|
||
# Key: (horizon, percentile) → list of per-asset MAE values
|
||
symbols = prices[asset_col].unique().to_list()
|
||
mae_by_hp: dict[tuple[int, float], list[float]] = {
|
||
(h, p): [] for h in horizons for p in percentiles
|
||
}
|
||
|
||
for sym in symbols:
|
||
sym_prices = (
|
||
prices.filter(pl.col(asset_col) == sym)
|
||
.sort(time_col)
|
||
.get_column(price_col)
|
||
.drop_nulls()
|
||
)
|
||
if len(sym_prices) < max(horizons) + 10:
|
||
continue
|
||
|
||
result = analyze_excursions(
|
||
sym_prices,
|
||
horizons=horizons,
|
||
percentiles=list(percentiles),
|
||
)
|
||
|
||
for h in horizons:
|
||
if h not in result.statistics:
|
||
continue
|
||
stats = result.statistics[h]
|
||
for p in percentiles:
|
||
val = stats.mae_percentiles.get(p)
|
||
if val is not None and not np.isnan(val):
|
||
mae_by_hp[(h, p)].append(abs(val))
|
||
|
||
# Build calibrated trailing stop configs from aggregated MAE percentiles
|
||
controls: list[dict] = []
|
||
seen_thresholds: set[float] = set()
|
||
|
||
for h in horizons:
|
||
for p in percentiles:
|
||
vals = mae_by_hp[(h, p)]
|
||
if not vals:
|
||
continue
|
||
# Median across all assets gives a robust universe-level threshold
|
||
threshold = float(np.median(vals))
|
||
threshold = round(threshold, 3)
|
||
if threshold <= 0 or threshold in seen_thresholds:
|
||
continue
|
||
seen_thresholds.add(threshold)
|
||
pct_str = f"{threshold * 100:.1f}".replace(".", "p")
|
||
controls.append(
|
||
{
|
||
"name": f"trailing_mae_p{int(p)}_h{h}_{pct_str}pct",
|
||
"type": "trailing_stop",
|
||
"threshold": threshold,
|
||
"calibration": {"horizon": h, "percentile": p, "source": "mae"},
|
||
}
|
||
)
|
||
|
||
# Sort by threshold for readable sweep ordering
|
||
controls.sort(key=lambda c: c["threshold"])
|
||
return controls
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# setup.yaml-driven sweep loader (Ch16-19)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def load_sweep(case_study: str) -> dict:
|
||
"""Return the ``backtest.sweep`` block from a case study's setup.yaml.
|
||
|
||
The block declares the Ch16-19 sweep grid (signal selection, allocators,
|
||
cost grid, risk controls) per-case-study. Raises ``KeyError`` if missing —
|
||
there is no fallback to module-level constants. See
|
||
``case_studies/us_firm_characteristics/config/setup.yaml`` for the schema.
|
||
"""
|
||
setup_path = _setup_path(case_study)
|
||
setup = yaml.safe_load(setup_path.read_text())
|
||
sweep = (setup.get("backtest") or {}).get("sweep")
|
||
if sweep is None:
|
||
raise KeyError(
|
||
f"backtest.sweep missing from case_studies/{case_study}/config/"
|
||
"setup.yaml — the Ch16-19 sweep grid must be declared explicitly. "
|
||
"See case_studies/us_firm_characteristics/config/setup.yaml for "
|
||
"the schema."
|
||
)
|
||
return sweep
|
||
|
||
|
||
def _setup_path(case_study: str):
|
||
from utils import CASE_STUDIES_DIR
|
||
|
||
return CASE_STUDIES_DIR / case_study / "config" / "setup.yaml"
|
||
|
||
|
||
def _load_setup(case_study: str) -> dict:
|
||
return yaml.safe_load(_setup_path(case_study).read_text())
|
||
|
||
|
||
def get_execution_defaults(case_study: str) -> dict:
|
||
"""Return the ``execution:`` block from setup.yaml.
|
||
|
||
Single source of truth for engine-level defaults — ``initial_cash``,
|
||
``share_type``, ``allocator_lookback``. Raises ``KeyError`` if missing;
|
||
the Ch16-19 notebooks must not declare local INITIAL_CASH constants.
|
||
"""
|
||
setup = _load_setup(case_study)
|
||
block = setup.get("execution")
|
||
if block is None:
|
||
raise KeyError(
|
||
f"execution: block missing from case_studies/{case_study}/config/"
|
||
"setup.yaml — must declare initial_cash, share_type, allocator_lookback."
|
||
)
|
||
return block
|
||
|
||
|
||
def get_allocator_lookback(case_study: str) -> int:
|
||
"""Return the CS-level lookback (bars-of-underlying) for moment-based allocators.
|
||
|
||
Read from ``setup.yaml::execution.allocator_lookback``. Applied uniformly
|
||
to inverse_vol, risk_parity, hrp, mvo_ledoit_wolf so allocators compete
|
||
on method, not on window choice. Bars count on the price DataFrame the
|
||
allocator consumes (typically daily even when rebalance is monthly).
|
||
"""
|
||
exec_block = get_execution_defaults(case_study)
|
||
lb = exec_block.get("allocator_lookback")
|
||
if lb is None:
|
||
raise KeyError(
|
||
f"execution.allocator_lookback missing from case_studies/{case_study}/"
|
||
"config/setup.yaml — required for moment-based allocators."
|
||
)
|
||
return int(lb)
|
||
|
||
|
||
_STAGE_DEFAULTS = {
|
||
"signal": 0, # 0 = all predictions
|
||
"allocation": 10, # legacy uniform-allocator key (notebooks pre-tier-split)
|
||
"allocation_cheap": 0, # post-tier-split: cheap allocators see all signal preds
|
||
"allocation_expensive": 10, # post-tier-split: expensive allocators see top-10 by signal Sharpe
|
||
"cost_sensitivity": 1, # top-1 of {signal+allocation} per label
|
||
"risk_overlay": 1, # top-1 of {signal+allocation} per label
|
||
}
|
||
|
||
_DEFAULT_EXPENSIVE_ALLOCATORS: tuple[str, ...] = ("risk_parity", "mvo_ledoit_wolf", "hrp")
|
||
|
||
|
||
def get_top_n_predictions(case_study: str, stage: str) -> int:
|
||
"""Return the top-N predictions to feed into ``stage`` from the upstream stage.
|
||
|
||
Reads ``backtest.sweep.top_n_predictions[stage]`` with safe fallbacks.
|
||
``stage`` is one of ``signal | allocation | allocation_cheap | allocation_expensive
|
||
| cost_sensitivity | risk_overlay``. Unknown stage names raise ``ValueError``
|
||
rather than ``KeyError`` so the stack trace clearly distinguishes a typo'd
|
||
lookup from a missing key in the YAML.
|
||
"""
|
||
if stage not in _STAGE_DEFAULTS:
|
||
raise ValueError(f"unknown stage {stage!r}; expected one of {sorted(_STAGE_DEFAULTS)}")
|
||
block = load_sweep(case_study).get("top_n_predictions") or {}
|
||
return int(block.get(stage, _STAGE_DEFAULTS[stage]))
|
||
|
||
|
||
def get_expensive_allocators_skip(case_study: str) -> bool:
|
||
"""Return whether MVO/HRP should be skipped at Ch17 (intraday escape hatch)."""
|
||
return bool(load_sweep(case_study).get("expensive_allocators_skip", False))
|
||
|
||
|
||
def get_expensive_allocators(case_study: str) -> tuple[str, ...]:
|
||
"""Return the allocator names routed through the ``allocation_expensive`` tier.
|
||
|
||
Reads ``backtest.sweep.expensive_allocators`` from setup.yaml. Falls back
|
||
to the canonical default ``(risk_parity, mvo_ledoit_wolf, hrp)`` if the
|
||
key is missing (legacy setup.yaml without the tier split). Allocators
|
||
not in this list are routed through ``allocation_cheap`` and see all
|
||
signal-stage predictions.
|
||
"""
|
||
block = load_sweep(case_study).get("expensive_allocators")
|
||
if block is None:
|
||
return _DEFAULT_EXPENSIVE_ALLOCATORS
|
||
return tuple(str(a) for a in block)
|
||
|
||
|
||
def get_cost_grid_half_spread_usd(case_study: str) -> list[float]:
|
||
"""Return the half-spread (USD per share) grid for per-share cost-regime sweep.
|
||
|
||
Used by Ch18 cost sensitivity for CSes whose declared cost model is
|
||
``per_share_plus_spread`` (etfs, nasdaq100_microstructure). Returns an
|
||
empty list when the key is absent — the bps grid is the only cost
|
||
dimension for that CS.
|
||
"""
|
||
block = load_sweep(case_study).get("cost_grid_half_spread_usd")
|
||
if block is None:
|
||
return []
|
||
return [float(v) for v in block]
|
||
|
||
|
||
def get_per_share_commission(case_study: str, default: float = 0.0035) -> float:
|
||
"""Return the per-share commission (USD/share) from ``costs.per_share``.
|
||
|
||
Single source of truth for the per-share companion cost regime in Ch18.
|
||
For CSes whose headline cost model is bps (e.g. sp500_equity_option_analytics)
|
||
the ``costs.per_share`` key may be absent; ``default`` (IBKR Pro Tiered top
|
||
tier, $0.0035/share) is returned in that case. Replaces ad-hoc
|
||
``open(setup.yaml)["costs"]["per_share"]`` reads in cost notebooks.
|
||
"""
|
||
costs = _load_setup(case_study).get("costs") or {}
|
||
return float(costs.get("per_share", default))
|
||
|
||
|
||
def get_cadence_sweep(case_study: str) -> list[str]:
|
||
"""Return the alternative-cadence list for the Ch18 cadence × cost heatmap.
|
||
|
||
Read from ``backtest.sweep.cadence_sweep`` in setup.yaml. Used by CSes
|
||
that explore rebalance-cadence sensitivity (nasdaq100_microstructure).
|
||
Returns an empty list when the key is absent — the CS does not run a
|
||
cadence sweep. Tokens are the engine's cadence vocabulary (e.g.
|
||
``15_minute``, ``1_hour``).
|
||
"""
|
||
block = load_sweep(case_study).get("cadence_sweep")
|
||
if block is None:
|
||
return []
|
||
return [str(c) for c in block]
|
||
|
||
|
||
# --- Calendar-aware allocator lookback resolution ---------------------------
|
||
|
||
# Calendar tokens we recognize in setup.yaml allocator entries. The numeric
|
||
# multiplier is read from the token prefix (e.g., ``3M`` → 3 × month).
|
||
_CALENDAR_UNITS_PER_YEAR = {
|
||
"Y": 1,
|
||
"M": 12,
|
||
"W": 52,
|
||
"D": None, # special-cased to periods_per_year (handles intraday cadences)
|
||
}
|
||
|
||
|
||
def _resolve_calendar_lookback(value, periods_per_year: float) -> int:
|
||
"""Translate a ``"3M"``/``"6M"``/``"1Y"`` token into bars-of-underlying.
|
||
|
||
``periods_per_year`` is interpreted as the **price-data bars per year**
|
||
(not the Sharpe-annualization factor — for most CSes the two coincide
|
||
because both equal 252 when underlying prices are daily). The allocator
|
||
consumes the raw ``prices`` DataFrame whose row cadence is the price
|
||
cadence, not the rebalance cadence; rolling-vol windows are measured in
|
||
those rows.
|
||
|
||
Cases where the two definitions diverge:
|
||
- crypto_perps_funding: ``evaluation.periods_per_year=365`` is the
|
||
daily-equivalent annualization factor, but the underlying data is
|
||
8-hourly (1095 bars/yr). Override per-CS via a future
|
||
``evaluation.bars_per_year`` field; until then prefer literal bars
|
||
in setup.yaml for non-daily-underlying CSes.
|
||
- nasdaq100_microstructure: same situation (intraday underlying).
|
||
|
||
Integer values pass through unchanged (legacy ``vol_window: 63``).
|
||
"""
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
if not isinstance(value, str):
|
||
raise TypeError(
|
||
f"lookback must be int or calendar string, got {type(value).__name__}: {value!r}"
|
||
)
|
||
token = value.strip().upper()
|
||
if not token:
|
||
raise ValueError(f"empty lookback string: {value!r}")
|
||
unit = token[-1]
|
||
if unit not in _CALENDAR_UNITS_PER_YEAR:
|
||
raise ValueError(
|
||
f"unsupported lookback unit {unit!r} in {value!r}; expected one of Y/M/W/D"
|
||
)
|
||
try:
|
||
n = int(token[:-1])
|
||
except ValueError as exc:
|
||
raise ValueError(f"cannot parse lookback prefix in {value!r}") from exc
|
||
units_per_year = _CALENDAR_UNITS_PER_YEAR[unit]
|
||
if units_per_year is None:
|
||
# D → use periods_per_year directly (assumes ppy counts daily bars;
|
||
# intraday CSes that nonetheless evaluate at daily MTM keep ppy=252)
|
||
bars_per_unit = periods_per_year / 252.0
|
||
return max(1, int(round(n * bars_per_unit)))
|
||
bars_per_unit = periods_per_year / units_per_year
|
||
return max(1, int(round(n * bars_per_unit)))
|
||
|
||
|
||
def _periods_per_year_for(case_study: str) -> float:
|
||
setup = _load_setup(case_study)
|
||
ppy = (setup.get("evaluation") or {}).get("periods_per_year")
|
||
if ppy is None:
|
||
raise KeyError(
|
||
f"evaluation.periods_per_year missing from case_studies/{case_study}/"
|
||
"config/setup.yaml — required for calendar-aware allocator lookbacks."
|
||
)
|
||
return float(ppy)
|
||
|
||
|
||
def get_entry_schemes_for(
|
||
case_study: str,
|
||
label: str,
|
||
n_assets: int,
|
||
long_short: bool,
|
||
) -> list[dict]:
|
||
"""Synthesize Ch16 entry schemes for ``(case_study, label)`` from setup.yaml.
|
||
|
||
Reads ``backtest.sweep.{top_k_grid, percentile_grid, quantile_grid}``
|
||
keyed by label, produces one scheme dict per (axis × value), filtered for
|
||
feasibility against ``n_assets``. Output dicts match the shape consumed
|
||
by ``Ch16 backtest notebooks`` (one scheme per (axis, value)).
|
||
|
||
Quantile schemes (``quintile_long_short`` / ``decile_long_short``) carry
|
||
``long_short=True`` regardless of the ``long_short`` argument — they are
|
||
inherently long-short by construction. ``long_short`` controls only the
|
||
sign of top-k / percentile schemes.
|
||
|
||
When the case study declares a ``backtest.sweep.signal_nasdaq100`` block
|
||
(the nasdaq100 v4 slot-mechanism sweep), schemes from that block are
|
||
appended — see ``get_signal_nasdaq100_schemes_for`` for the cross-product.
|
||
"""
|
||
sweep = load_sweep(case_study)
|
||
schemes: list[dict] = []
|
||
|
||
top_k_by_label = sweep.get("top_k_grid") or {}
|
||
pct_by_label = sweep.get("percentile_grid") or {}
|
||
qnt_by_label = sweep.get("quantile_grid") or {}
|
||
|
||
# Strict label gate. If any of the three grids is declared in the YAML
|
||
# but the label appears in none of them, raise — silently returning an
|
||
# empty scheme list lets a typo'd LABEL papermill parameter register
|
||
# zero backtests with no warning (Ch16/13 loops over schemes have no
|
||
# else-clause). The legacy "no backtest.sweep block at all" path is
|
||
# preserved: when all three grids are empty/absent, we fall through and
|
||
# return [] (callers that explicitly opt into the legacy
|
||
# ``get_entry_schemes(...)`` helper keep working).
|
||
any_grid_declared = (
|
||
"top_k_grid" in sweep or "percentile_grid" in sweep or "quantile_grid" in sweep
|
||
)
|
||
label_known = label in top_k_by_label or label in pct_by_label or label in qnt_by_label
|
||
if any_grid_declared and not label_known:
|
||
raise KeyError(
|
||
f"label {label!r} not declared in any of backtest.sweep.{{top_k_grid, "
|
||
f"percentile_grid, quantile_grid}} for case_studies/{case_study}/"
|
||
f"config/setup.yaml; known labels: top_k={sorted(top_k_by_label)}, "
|
||
f"pct={sorted(pct_by_label)}, qnt={sorted(qnt_by_label)}"
|
||
)
|
||
|
||
for k in top_k_by_label.get(label, []):
|
||
k = int(k)
|
||
# k == n_assets holds the whole universe = equal-weight benchmark, not a
|
||
# prediction-based portfolio; exclude it to match get_top_k_values_for.
|
||
if k >= n_assets:
|
||
continue
|
||
schemes.append(
|
||
{
|
||
"name": f"ew_top{k}",
|
||
"method": "equal_weight_top_k",
|
||
"top_k": k,
|
||
"long_short": long_short,
|
||
}
|
||
)
|
||
|
||
for p in pct_by_label.get(label, []):
|
||
p = float(p)
|
||
schemes.append(
|
||
{
|
||
"name": f"cs_pct{int(p)}",
|
||
"method": "cross_sectional_percentile",
|
||
"percentile": p,
|
||
"long_short": long_short,
|
||
}
|
||
)
|
||
|
||
for n_q in qnt_by_label.get(label, []):
|
||
n_q = int(n_q)
|
||
if n_q == 5:
|
||
name, method = "quintile_ls", "quintile_long_short"
|
||
elif n_q == 10:
|
||
name, method = "decile_ls", "decile_long_short"
|
||
else:
|
||
raise ValueError(
|
||
f"quantile_grid only supports n_quantiles ∈ {{5, 10}}; "
|
||
f"got {n_q} for label {label!r} in case_studies/{case_study}/"
|
||
f"config/setup.yaml. The backtest dispatcher has no method for "
|
||
f"q{n_q}_long_short."
|
||
)
|
||
if n_assets < 2 * n_q:
|
||
continue
|
||
schemes.append(
|
||
{
|
||
"name": name,
|
||
"method": method,
|
||
"n_quantiles": n_q,
|
||
"long_short": True,
|
||
}
|
||
)
|
||
|
||
# nasdaq100 v4 slot mechanism — appended when the block is present.
|
||
if "signal_nasdaq100" in sweep:
|
||
schemes.extend(get_signal_nasdaq100_schemes_for(case_study, label, n_assets))
|
||
|
||
return schemes
|
||
|
||
|
||
def get_signal_nasdaq100_schemes_for(
|
||
case_study: str,
|
||
label: str,
|
||
n_assets: int,
|
||
) -> list[dict]:
|
||
"""Expand the ``backtest.sweep.signal_nasdaq100`` block into entry schemes.
|
||
|
||
Block shape (all keys required unless noted)::
|
||
|
||
signal_nasdaq100:
|
||
selection_method: [slot_persistent_signal_exit, eq_w_topk]
|
||
long_q: [0.90, 0.95, 0.99] # slot only — entry quantile
|
||
direction: [long_only, long_short]
|
||
max_slots: [5, 10, 20] # slot only — concurrent holdings
|
||
hold_bars: [8, 16, 32] # slot only — max-hold backstop
|
||
exit_signal_q: [null, 0.30, ...] # slot only — stay threshold, null disables
|
||
pred_freshness_max_min: 14 # slot only — backward-asof tolerance
|
||
bars_per_day_grid: [14] # slot only — execution cadence
|
||
top_k_grid: [5, 10, 20] # eq_w_topk only — top-k holdings
|
||
lookback_days: 21 # slot only — rolling window depth
|
||
|
||
Slot mechanism is single-direction (the slot book cannot be both long
|
||
and short simultaneously); slot × long_short combinations are dropped.
|
||
eq_w_topk supports both directions via the canonical long_short axis.
|
||
|
||
Schemes carry a ``name`` derived from the cross-product coordinates so
|
||
the registry rows are distinguishable. ``selection_method_config`` keys
|
||
are flattened into the scheme dict so the existing 14_backtest.py loop
|
||
passes them through to ``run_backtest`` unchanged.
|
||
"""
|
||
sweep = load_sweep(case_study)
|
||
block = sweep.get("signal_nasdaq100")
|
||
if block is None:
|
||
msg = (
|
||
f"backtest.sweep.signal_nasdaq100 missing from "
|
||
f"case_studies/{case_study}/config/setup.yaml"
|
||
)
|
||
raise KeyError(msg)
|
||
|
||
methods = list(block.get("selection_method", []))
|
||
if not methods:
|
||
msg = (
|
||
f"signal_nasdaq100.selection_method must list at least one method "
|
||
f"for case_studies/{case_study}"
|
||
)
|
||
raise ValueError(msg)
|
||
long_qs = [float(v) for v in block.get("long_q", [])]
|
||
directions = list(block.get("direction", []))
|
||
max_slots_grid = [int(v) for v in block.get("max_slots", [])]
|
||
hold_bars_grid = [int(v) for v in block.get("hold_bars", [])]
|
||
# Tolerate a scalar/`null` exit_signal_q (e.g. ``exit_signal_q: null``)
|
||
# instead of raising an opaque ``list(None)`` TypeError far from the config.
|
||
raw_exit = block.get("exit_signal_q", [None])
|
||
exit_qs = list(raw_exit) if isinstance(raw_exit, list) else [raw_exit]
|
||
pred_freshness = block.get("pred_freshness_max_min")
|
||
bpd_grid = [int(v) for v in block.get("bars_per_day_grid", [])]
|
||
top_k_grid = [int(v) for v in block.get("top_k_grid", [])]
|
||
lookback_days = int(block.get("lookback_days", 21))
|
||
|
||
# Fail loudly on unknown directions and on missing/typo'd required keys: a
|
||
# YAML typo (e.g. ``max_slot:`` or ``bars_per_day:``) would otherwise leave
|
||
# the corresponding grid empty, silently collapse the cross-product to zero
|
||
# schemes, and register zero backtests — the failure mode the strict-label
|
||
# gate exists to prevent.
|
||
_allowed_dirs = {"long_only", "short_only", "long_short"}
|
||
bad_dirs = [d for d in directions if d not in _allowed_dirs]
|
||
if bad_dirs:
|
||
msg = (
|
||
f"signal_nasdaq100.direction has unknown value(s) {bad_dirs} for "
|
||
f"case_studies/{case_study}; allowed: {sorted(_allowed_dirs)}"
|
||
)
|
||
raise ValueError(msg)
|
||
_required: dict[str, list] = {"direction": directions}
|
||
if "slot_persistent_signal_exit" in methods:
|
||
_required.update(
|
||
long_q=long_qs, max_slots=max_slots_grid,
|
||
hold_bars=hold_bars_grid, bars_per_day_grid=bpd_grid,
|
||
)
|
||
if "eq_w_topk" in methods:
|
||
_required["top_k_grid"] = top_k_grid
|
||
missing = sorted(k for k, v in _required.items() if not v)
|
||
if missing:
|
||
msg = (
|
||
f"signal_nasdaq100 is missing/empty required key(s) {missing} for "
|
||
f"case_studies/{case_study} given selection_method={methods}; "
|
||
f"check for a YAML typo in the sweep block."
|
||
)
|
||
raise ValueError(msg)
|
||
|
||
schemes: list[dict] = []
|
||
for method in methods:
|
||
n_before = len(schemes)
|
||
if method == "slot_persistent_signal_exit":
|
||
for long_q in long_qs:
|
||
for direction in directions:
|
||
# slot books are single-direction by construction:
|
||
# long_only/short_only are supported, long_short is dropped.
|
||
if direction not in ("long_only", "short_only"):
|
||
continue
|
||
for max_slots in max_slots_grid:
|
||
if max_slots >= n_assets:
|
||
continue
|
||
for hold_bars in hold_bars_grid:
|
||
for exit_q in exit_qs:
|
||
exit_q_norm = None if exit_q is None else float(exit_q)
|
||
if exit_q_norm is not None and exit_q_norm >= long_q:
|
||
continue
|
||
for bpd in bpd_grid:
|
||
eq_tag = (
|
||
"noexit"
|
||
if exit_q_norm is None
|
||
else f"q{int(exit_q_norm * 100):02d}"
|
||
)
|
||
name = (
|
||
f"slot_{direction[0]}_lq{int(long_q * 100):02d}"
|
||
f"_s{max_slots}_h{hold_bars}_{eq_tag}_b{bpd}"
|
||
)
|
||
schemes.append(
|
||
{
|
||
"name": name,
|
||
"method": "slot_persistent_signal_exit",
|
||
"long_q": long_q,
|
||
"lookback_days": lookback_days,
|
||
"bars_per_day": bpd,
|
||
"max_slots": max_slots,
|
||
"hold_bars": hold_bars,
|
||
"exit_signal_q": exit_q_norm,
|
||
"pred_freshness_max_min": pred_freshness,
|
||
"direction": direction,
|
||
"long_short": False,
|
||
}
|
||
)
|
||
elif method == "eq_w_topk":
|
||
for direction in directions:
|
||
ls = direction == "long_short"
|
||
for top_k in top_k_grid:
|
||
if top_k >= n_assets:
|
||
continue
|
||
dtag = "ls" if ls else direction[0]
|
||
schemes.append(
|
||
{
|
||
"name": f"ewtopk_{dtag}_k{top_k}",
|
||
"method": "equal_weight_top_k",
|
||
"top_k": top_k,
|
||
"long_short": ls,
|
||
"direction": "long_only" if ls else direction,
|
||
}
|
||
)
|
||
else:
|
||
msg = (
|
||
f"signal_nasdaq100.selection_method has unknown method "
|
||
f"{method!r} for case_studies/{case_study}; supported: "
|
||
f"slot_persistent_signal_exit, eq_w_topk"
|
||
)
|
||
raise ValueError(msg)
|
||
# A requested method that expands to zero schemes is the silent-zero
|
||
# failure the validation exists to catch (e.g. a slot-only block with
|
||
# direction=[long_short], which slot drops, or every grid value filtered
|
||
# out by max_slots/top_k >= n_assets). Fail loudly instead.
|
||
if len(schemes) == n_before:
|
||
msg = (
|
||
f"signal_nasdaq100 method {method!r} produced zero schemes for "
|
||
f"case_studies/{case_study} (n_assets={n_assets}); every grid "
|
||
f"combination was filtered out — check direction/max_slots/top_k."
|
||
)
|
||
raise ValueError(msg)
|
||
return schemes
|
||
|
||
|
||
def get_top_k_values_for(
|
||
case_study: str,
|
||
label: str,
|
||
n_assets: int,
|
||
) -> list[int]:
|
||
"""Return the top-K grid for ``(case_study, label)`` used by Ch17.
|
||
|
||
Filters out k >= n_assets (holding everything is the equal-weight
|
||
benchmark, not a prediction-based portfolio). Raises ``KeyError`` if
|
||
``backtest.sweep.top_k_grid[label]`` is not declared.
|
||
"""
|
||
sweep = load_sweep(case_study)
|
||
grid = (sweep.get("top_k_grid") or {}).get(label)
|
||
if grid is None:
|
||
raise KeyError(
|
||
f"backtest.sweep.top_k_grid[{label!r}] not declared in "
|
||
f"case_studies/{case_study}/config/setup.yaml"
|
||
)
|
||
return [int(k) for k in grid if int(k) < n_assets]
|
||
|
||
|
||
_MOMENT_ALLOCATORS = {"inverse_vol", "risk_parity", "hrp", "mvo_ledoit_wolf", "mvo"}
|
||
_LOOKBACK_KEYS = ("vol_window", "lookback")
|
||
|
||
|
||
def get_allocators(case_study: str) -> list[dict]:
|
||
"""Return the Ch17 allocator configs (lookback-injected from setup.yaml).
|
||
|
||
Each dict matches the shape consumed by
|
||
``case_studies.utils.backtest_runner._apply_allocation``:
|
||
``{"method": str, ...kwargs}``. Common kwargs: ``vol_window``,
|
||
``lookback``, ``max_weight``.
|
||
|
||
Moment-based allocators (inverse_vol, risk_parity, hrp, mvo_ledoit_wolf)
|
||
receive ``vol_window``/``lookback`` from the CS-level
|
||
``execution.allocator_lookback`` — a single window keeps allocators
|
||
comparable. Calendar-string fallback (``"3M"``/``"6M"``) is still
|
||
supported on individual entries when an override is needed, but the
|
||
standard path is the CS-level lookback.
|
||
|
||
The ``name`` key in setup.yaml is human-readable metadata only; it is
|
||
stripped here so the dispatcher sees a stable spec shape and the
|
||
allocation-stage registry hash is reproducible.
|
||
"""
|
||
raw = load_sweep(case_study).get("allocators") or []
|
||
cs_lookback = None
|
||
if any(a.get("method") in _MOMENT_ALLOCATORS for a in raw):
|
||
cs_lookback = get_allocator_lookback(case_study)
|
||
ppy = _periods_per_year_for(case_study) if _needs_calendar_resolve(raw) else None
|
||
resolved = []
|
||
for entry in raw:
|
||
out = {k: v for k, v in entry.items() if k != "name"}
|
||
# CS-level lookback injection for moment-based allocators that don't
|
||
# carry an explicit per-entry override.
|
||
if out.get("method") in _MOMENT_ALLOCATORS and cs_lookback is not None:
|
||
if out["method"] in {"mvo", "mvo_ledoit_wolf"}:
|
||
out.setdefault("lookback", cs_lookback)
|
||
else:
|
||
out.setdefault("vol_window", cs_lookback)
|
||
# Calendar-string overrides (``"3M"`` etc.) resolved against ppy.
|
||
for key in _LOOKBACK_KEYS:
|
||
if key in out and isinstance(out[key], str):
|
||
out[key] = _resolve_calendar_lookback(out[key], ppy)
|
||
resolved.append(out)
|
||
return resolved
|
||
|
||
|
||
def _needs_calendar_resolve(allocators: list[dict]) -> bool:
|
||
return any(isinstance(a.get(k), str) for a in allocators for k in _LOOKBACK_KEYS)
|
||
|
||
|
||
def get_allocator_label(alloc: dict) -> str:
|
||
"""Return a human-readable label for an allocator dict (``alloc['method']``)."""
|
||
return str(alloc.get("method", "unknown"))
|
||
|
||
|
||
def get_cost_grid_bps(case_study: str) -> list[float]:
|
||
"""Return the Ch18 cost-sweep grid (bps; commission + slippage combined)."""
|
||
return [float(c) for c in (load_sweep(case_study).get("cost_grid_bps") or [])]
|
||
|
||
|
||
def get_htm_cost_cascade(case_study: str) -> dict:
|
||
"""Return the Ch18 HTM cost-cascade block (sp500_options only).
|
||
|
||
The cascade dispatches the O'Donovan & Yu (2025) hold-to-expiry cost
|
||
analysis: entry-only half-spread fractions, optionally restricted to a
|
||
liquid-universe subset (rung-3). Raises ``KeyError`` if the block is
|
||
missing — case studies that use the standard bps regime should call
|
||
:func:`get_cost_grid_bps` instead.
|
||
"""
|
||
block = load_sweep(case_study).get("htm_cost_cascade")
|
||
if block is None:
|
||
raise KeyError(
|
||
f"backtest.sweep.htm_cost_cascade missing from "
|
||
f"case_studies/{case_study}/config/setup.yaml — only the "
|
||
f"HTM-cascade case studies (sp500_options) declare this block."
|
||
)
|
||
return block
|
||
|
||
|
||
def get_universe_filters_for(case_study: str) -> list[str | None]:
|
||
"""Return the list of ``strategy.signal.universe_filter`` axis values to sweep.
|
||
|
||
``None`` represents the full universe (no filter applied, equivalent to
|
||
``apply_universe_filter`` returning predictions unchanged). Other values
|
||
are passed through to ``apply_universe_filter`` in ``backtest_runner.py``
|
||
where they drive a spec-declared universe restriction at the
|
||
rebalance-date grain (currently only ``"liquid"`` is supported, for the
|
||
sp500_options bottom-quantile half-spread subset).
|
||
|
||
Sourced from ``backtest.sweep.universe_filter`` in ``setup.yaml``: a
|
||
single scalar value pinning the canonical sweep to one universe. For
|
||
sp500_options this is ``"liquid"`` (the only economic universe for the
|
||
HTM straddle strategy after costs); absent everywhere else, which
|
||
yields ``[None]``. ``"full"`` and ``"none"`` are normalized to ``None``
|
||
so pre-universe-axis registry rows (which carry no
|
||
``signal.universe_filter`` in their spec) remain hash-stable.
|
||
|
||
Note: ``backtest.sweep.htm_cost_cascade.universes`` is a separate
|
||
Ch18-only block consumed directly by ``14_costs.py`` via
|
||
``get_htm_cost_cascade``; it is the cost-comparison axis (full vs
|
||
liquid) and does NOT participate in the canonical rank-1 sweep.
|
||
"""
|
||
sweep = load_sweep(case_study)
|
||
uf = sweep.get("universe_filter")
|
||
if uf is not None:
|
||
return [(None if str(uf).lower() in ("full", "none") else str(uf))]
|
||
return [None]
|
||
|
||
|
||
def get_position_risk_controls(case_study: str) -> list[dict]:
|
||
"""Return the Ch19 position-level risk controls (engine case studies only)."""
|
||
risk = load_sweep(case_study).get("risk_controls") or {}
|
||
return list(risk.get("position") or [])
|
||
|
||
|
||
def get_portfolio_risk_controls(case_study: str) -> list[dict]:
|
||
"""Return the Ch19 portfolio-level risk controls (all case studies)."""
|
||
risk = load_sweep(case_study).get("risk_controls") or {}
|
||
return list(risk.get("portfolio") or [])
|