校准风险控制并配置回测扫描
代码 《交易机器学习》
总结
该配置模块介绍一种框架，用于为各案例研究声明回测扫描设置。它从配置文件读取入场方案、配置方法、交易成本网格和风险控制；如果缺少必需配置，则明确报错。模块还集中管理执行默认值和配置器回看窗口，使投资组合配置方法使用一致的估计期。可选的分阶段筛选可以先测试一组基线策略，再对较小的排名子集进行更广泛的比较。
一种风险控制方法估计不同资产和持有期限下的不利价格变动，再利用选定百分位数上的中位数变动设定追踪止损阈值。该流程会跳过价格历史过短的资产，并删除重复阈值。模块还区分横截面多空筛选与账户做空权限，避免错误解读期权策略。这些工具用于构建比较性回测；本文提供的是实现逻辑，并非任何特定配置器、止损或策略能够改善表现的证据。
核心观点
- 每个案例研究的回测扫描维度和执行默认值都在配置中声明。
- 可根据不同资产的不利变动中位数估算追踪止损水平。
- 使用统一回看期有助于在一致基础上比较基于矩的配置器。
- 筛选模式与卖空权限是不同概念，不应混为一谈。
- 分阶段扫描可先对基线子集排序，再对选定预测结果进行额外测试。
标签
全文
# sweep_config.py
```py
"""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
from case_studies.utils.warning_policy import warn_the_reader
# ---------------------------------------------------------------------------
# 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):
# Resolve through get_case_study_dir so setup.yaml is read from the same place
# the model stages read it: the source tree in production, or ML4T_OUTPUT_DIR
# when set (tests, and the create_experiment workflow where the reader edits
# backtest.sweep/costs/execution in their isolated experiment copy).
from utils.paths import get_case_study_dir
# create=False: reading config must never materialize the case-study dir. A
# side-effect mkdir under ML4T_OUTPUT_DIR would make create_experiment think
# the experiment already exists.
return get_case_study_dir(case_study, create=False) / "config" / "setup.yaml"
def _load_setup(case_study: str) -> dict:
return yaml.safe_load(_setup_path(case_study).read_text())
def get_declared_long_short(case_study: str) -> bool:
"""Whether a case study's declared selection mode is cross-sectional long-short.
Delegates to ``get_backtest_config``, which reads it off
``mapping.position_state_space``. That is the selection mode, not
``account.allow_short_selling``: the account flag is an execution permission, and
the two come apart. ``sp500_options`` sets the permission because it sells
straddles, but its state space is ``short_straddle_hedged`` - a one-sided short
that builds no cross-sectional short sleeve, and its backtest passes
``long_short: False`` accordingly. Sizing its concentration grid off the account
flag would halve a ceiling that nothing halves at run time.
"""
from case_studies.utils.backtest_loaders import get_backtest_config
return bool(get_backtest_config(case_study).long_short)
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 (equal-weight baseline)
"allocation": 10, # top-10 model configs by baseline Sharpe
"cost_sensitivity": 1, # top-1 of {signal+allocation} per label
"risk_overlay": 1, # top-1 of {signal+allocation} per label
}
_DEFAULT_CHECKPOINTS_PER_CONFIG = 1
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 | 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.
For ``allocation`` the unit counted is a *model config* — one ``(family,
config_name)`` pair — not a prediction. See ``get_checkpoints_per_config``
for how many checkpoints each advancing config contributes.
"""
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 top_n_cap(top_n: int) -> int | None:
"""Return the cap ``top_n`` asks for, or ``None`` when it asks for every candidate.
``0`` is how ``top_n_predictions`` spells "all of them". ``_STAGE_DEFAULTS["signal"]``
is 0, every shipped ``setup.yaml`` declares ``signal: 0`` under the comment "all signal
predictions", and the four signal-stage notebooks read it that way, each guarding its
truncation with ``if TOP_N_PREDICTIONS > 0``.
The allocation stage had no such reading, so the same number meant the opposite one stage
later: ``0`` reached a SQL ``LIMIT 0`` and a ``.head(0)`` and selected nothing. Taking the
cap from here keeps one number meaning one thing at every stage. A negative width is a
caller error rather than a second spelling of "all", because SQLite's ``LIMIT -1`` means
no limit and a width arithmetic produced would otherwise pass through as one.
"""
width = int(top_n)
if width < 0:
raise ValueError(f"top_n must be 0 (every candidate) or positive, got {top_n!r}")
return width or None
def get_checkpoints_per_config(case_study: str) -> int:
"""Return how many checkpoints each advancing model config contributes.
Reads ``backtest.sweep.checkpoints_per_config``. The default of 1 means a
config enters the allocation sweep through its single best checkpoint, so
a long checkpoint sweep of one model cannot crowd out other model families.
"""
value = load_sweep(case_study).get("checkpoints_per_config")
if value is None:
return _DEFAULT_CHECKPOINTS_PER_CONFIG
return int(value)
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_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)
# How many registered prediction sets the width resolver reads before it decides.
# One is enough where they agree and cannot detect it where they do not.
_RANKED_WIDTH_SAMPLE = 8
def ranked_cross_section_width(
case_study: str, label: str, *, split: str = "validation"
) -> int | None:
"""Distinct symbols in the predictions an entry-scheme sweep will rank.
``None`` when nothing is registered yet for ``(label, split)``, when the
artifacts behind the registered sets are missing, or when the sampled sets do
not agree on a width. A case study whose model stages have not run has no
ranked cross-section, and inventing one would refuse a sweep that has not
been set up wrong.
The registry is sampled rather than asked about the population a particular
caller is sweeping, and those are different questions: `cme_futures/13_backtest`
selects official or preview candidates, and a width taken from an unrelated
set could remove a concentration that is feasible for the ones actually being
swept. So the resolver only speaks when the sample is unanimous - up to eight
registered sets, all reporting the same width - and otherwise returns ``None``
and leaves the caller's own number standing. A caller that knows its
population passes ``ranked_width`` and skips this entirely.
Unanimity is the common case, not a hope: measured 2026-09-07 over six
registered sets per label in etfs, us_firm_characteristics,
sp500_equity_option_analytics and us_equities_panel, every label had exactly
one width - 99, 3,708, 549 and 3,147 respectively.
Distinct symbols over the window, which is the same count the callers take
off the price panel, so the comparison in ``get_entry_schemes_for`` is between
two measurements of the same kind. The number that literally bounds a ranking
is the per-date cross-section, which is smaller - 88 against 99 for etfs,
2,032 against 3,708 for us_firm_characteristics - and moving the rule onto it
would change which concentrations are feasible today. That is a separate
decision about the rule; this function is about which set the rule reads.
"""
import sqlite3
from case_studies.utils.registry.store import _case_dir, _registry_db_path, _run_log_dir
case_dir = _case_dir(case_study)
db_path = _registry_db_path(case_dir)
if not db_path.is_file():
return None
try:
with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) as db:
rows = db.execute(
"""
SELECT p.prediction_hash
FROM prediction_sets p
JOIN training_runs t ON t.training_hash = p.training_hash
WHERE t.label = ? AND p.split = ?
LIMIT ?
""",
(label, split, _RANKED_WIDTH_SAMPLE),
).fetchall()
except sqlite3.Error:
return None
widths: set[int] = set()
for (prediction_hash,) in rows:
artifact = _run_log_dir(case_dir) / "predictions" / prediction_hash / "predictions.parquet"
if not artifact.is_file():
continue
try:
widths.add(
int(pl.scan_parquet(artifact).select(pl.col("symbol").n_unique()).collect().item())
)
except (pl.exceptions.PolarsError, OSError, ValueError):
return None
if not widths:
return None
if len(widths) > 1:
warn_the_reader(
f"{case_study}/{label}: registered prediction sets disagree on the ranked "
f"cross-section ({sorted(widths)}), so the entry-scheme feasibility rule falls "
"back to the caller's price-panel width. Pass ranked_width= with the width of "
"the population actually being swept.",
source="ranked_width_from_registry",
key=(case_study, label, tuple(sorted(widths))),
)
return None
return widths.pop()
def get_entry_schemes_for(
case_study: str,
label: str,
n_assets: int,
long_short: bool,
*,
ranked_width: int | None = None,
) -> 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.
``n_assets`` is the width of the price panel. What bounds a ranked selection
is the prediction cross-section, and the two are not the same number: every
caller measures ``n_assets`` off the panel it loaded, while the ranking runs
over whatever the fitting stages scored. ``ranked_width`` is that second
number, resolved from the registry when the caller does not supply it, and
the feasibility rule below applies to the narrower of the two.
They agree on `main` today, which is why nothing is currently wrong and why
the check could not fire on the condition it exists to catch
. Any change that narrows the fitting stages
without narrowing the panel - a new preview tier, a per-notebook
``max_symbols``, a family that scores a subset - reproduces the same
failure: twelve backtests with a Sharpe, zero trades, and
four notebooks failing several stages downstream of the cause.
"""
sweep = load_sweep(case_study)
schemes: list[dict] = []
if ranked_width is None:
ranked_width = ranked_cross_section_width(case_study, label)
tradeable = n_assets if ranked_width is None else min(n_assets, ranked_width)
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)}"
)
declared_top_k = [int(k) for k in top_k_by_label.get(label, [])]
top_k_schemes: list[dict] = []
for k in declared_top_k:
# k == tradeable holds the whole cross-section = equal-weight benchmark, not
# a prediction-based portfolio; exclude it to match get_top_k_values_for.
if k >= tradeable:
continue
# Long and short selections must be disjoint. Keep the declared grid and
# exclude only members that the available cross-section cannot realize.
if long_short and 2 * k > tradeable:
continue
top_k_schemes.append(
{
"name": f"ew_top{k}",
"method": "equal_weight_top_k",
"top_k": k,
"long_short": long_short,
}
)
# Raised here rather than left to the caller's empty-list guard, because only
# this scope holds both numbers. A caller that reports `n_assets` back is
# reporting the number the check did not use, which is how the original
# failure presented. The panel-too-narrow case (`tradeable == n_assets`) is
# left to that guard, which names the right number for it.
if declared_top_k and not top_k_schemes and tradeable < n_assets:
raise ValueError(
f"no concentration declared for {case_study}/{label} can be realized: "
f"backtest.sweep.top_k_grid declares {sorted(declared_top_k)}, and the "
f"predictions this sweep ranks carry {tradeable} symbols against a price "
f"panel of {n_assets}. Every k is at or above the ranked cross-section, so "
"each would hold all of it rather than select from it, and every backtest "
"in the sweep would record num_trades = 0 while still reporting a Sharpe. "
"Narrow the price panel and the fitting stages together, or declare a "
"smaller concentration."
)
schemes.extend(top_k_schemes)
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:
# Same ceiling as `get_top_k_values_for`, and it has to be
# applied here too: this branch sets `long_short` off its own
# `direction` axis, so a `k` in `(n_assets // 2, n_assets)`
# would register as `ewtopk_ls_k<k>` and run at the clamped
# half. `direction` is per-scheme, so use it rather than the
# case study's declared mode.
if top_k >= n_assets or (ls and 2 * 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,
*,
long_short: bool | None = None,
ranked_width: int | None = None,
) -> 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, and ``ValueError``
when the filter empties the grid.
A long-short strategy takes ``k`` names on each side, so its ceiling is half
the universe, not all of it. ``signals.py`` clamps ``eff_k`` to
``n_assets // 2`` when ``long_short`` is set, and ``top_k`` is identity-bearing
through ``plan_backtests(signal=...)`` - so a ``k`` in the half-open interval
``(n_assets // 2, n_assets)`` passes this filter, registers under the declared
``k``, and runs at the clamped one. Two such values register distinct identities
over a byte-identical weight series. ``get_entry_schemes_for`` has excluded that
interval since it took a ``long_short`` argument; this function is the other half
of the same grid and has to agree with it.
``long_short`` defaults to the case study's declared selection mode - the same
``mapping.position_state_space`` that ``get_backtest_config`` reads, which is what
every caller already passes to ``get_entry_schemes_for`` as ``bt_config.long_short``.
Not ``account.allow_short_selling``: that is an execution permission, and
``sp500_options`` holds it while selecting long-only. Pass it explicitly to override.
An empty grid is never a legitimate result: the caller multiplies it into
a sweep size, so zero concentrations means zero backtests, and the sweep
loop then completes without registering anything while still reporting
itself done. The downstream risk-overlay notebook is the only thing that
notices, and it blames the operator for not having run this stage. Raise
here instead, where the cause - a universe cap smaller than the smallest
declared k - is still visible.
``ranked_width`` is the prediction cross-section, resolved from the registry
when not supplied, and the filter runs against the narrower of it and
``n_assets`` for the reason given under ``get_entry_schemes_for``
. The two functions are halves of one grid and
have to filter it the same way or a sweep and its plumbing test disagree
about which concentrations exist.
"""
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"
)
if long_short is None:
long_short = get_declared_long_short(case_study)
if ranked_width is None:
ranked_width = ranked_cross_section_width(case_study, label)
tradeable = n_assets if ranked_width is None else min(n_assets, ranked_width)
ceiling = tradeable // 2 if long_short else tradeable
values = [
int(k) for k in grid if int(k) < tradeable and not (long_short and 2 * int(k) > tradeable)
]
if not values:
smallest = min(int(k) for k in grid)
# What the universe has to hold, which is not the same number in the two
# cases. Long-short takes `k` names on each side, so it needs `2k`, and
# saying "raise MAX_SYMBOLS above k" there sends the reader to a cap that
# cannot fix it - or worse, to a cap the fixture cannot reach at all.
needed = (
f"at least {2 * smallest} names for the smallest declared k of {smallest} "
f"({smallest} long and {smallest} short)"
if long_short
else f"more than {smallest} names for the smallest declared k"
)
side = f"long-short ceiling ({ceiling})" if long_short else "universe"
raise ValueError(
f"top_k_grid[{label!r}] = {list(grid)} is empty after filtering against "
f"{tradeable} tradeable names for case_studies/{case_study} (price panel "
f"{n_assets}, ranked cross-section {ranked_width}): every declared k holds "
f"the whole {side}. A cross-sectional sweep here needs {needed}. Raise the "
f"universe cap (MAX_SYMBOLS) if the panel has them, widen the panel if it "
f"does not, or declare a smaller k."
)
return values
_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):
if ppy is None:
raise RuntimeError("calendar lookback resolution requires periods_per_year")
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. Two names are supported there: ``"liquid"`` (the
sp500_options bottom-quantile half-spread subset) and ``"cost_feasible"``
(the nasdaq100_microstructure frozen per-split symbol list under
``setup.yaml::universe.cost_feasible``); anything else raises.
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 ``15_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 [])
def get_signal_passes_for(case_study: str) -> dict | None:
"""Return the declared two-pass plan for the signal stage, or ``None``.
Shape, from ``backtest.sweep.signal_passes`` in ``setup.yaml``::
signal_passes:
baseline_schemes: [ew_top5, ew_top10, ew_top20]
baseline_universe: cost_feasible
mechanism_top_n: 8
reference_schemes: [ew_top5, ew_top10, ew_top20]
reference_universe: full
Pass 1 runs ``baseline_schemes`` on ``baseline_universe`` over every
admissible prediction. Pass 2 runs every remaining entry scheme on
``baseline_universe``, plus ``reference_schemes`` on ``reference_universe``,
over the ``mechanism_top_n`` predictions with the highest pass-1 Sharpe.
A case study that declares no block gets ``None`` and its sweep is the
single pass over the full cross-product, which is what every case study
other than ``nasdaq100_microstructure`` runs.
Universe names are normalized the way ``get_universe_filters_for``
normalizes them: ``full`` and ``none`` become ``None``, because a ``None``
written into a spec is not the same thing as an absent key and every row
registered before the universe axis existed carries the absent one.
Raises on a block that is present but unusable, rather than returning a
plan that quietly sweeps nothing: an empty ``baseline_schemes`` would run
pass 1 over zero arms, rank nothing, and hand pass 2 an empty selection,
which reports a completed sweep and registers no rows.
"""
sweep = load_sweep(case_study)
block = sweep.get("signal_passes")
if block is None:
return None
def _universe(key: str) -> str | None:
raw = block.get(key)
if raw is None:
return None
return None if str(raw).lower() in ("full", "none") else str(raw)
baseline = [str(x) for x in (block.get("baseline_schemes") or [])]
reference = [str(x) for x in (block.get("reference_schemes") or [])]
top_n = int(block.get("mechanism_top_n") or 0)
if not baseline or top_n <= 0:
msg = (
f"backtest.sweep.signal_passes for {case_study} declares "
f"baseline_schemes={baseline} and mechanism_top_n={top_n}; both must be "
"non-empty or the sweep ranks nothing and pass 2 selects nothing. Remove "
"the block to sweep the full cross-product instead."
)
raise ValueError(msg)
return {
"baseline_schemes": baseline,
"baseline_universe": _universe("baseline_universe"),
"mechanism_top_n": top_n,
"reference_schemes": reference,
"reference_universe": _universe("reference_universe"),
}
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。