Предустановки бэктеста для исполнения, издержек и идентификации запусков
Код Machine Learning for Trading
Сводка
В коде описано, как настройки бэктеста определяются по конфигурации примера и как при идентификации запуска учитываются изменения торгуемого набора и возраста прогнозов. Главный вывод: любой фильтр, изменяющий портфель или набор прогнозов, должен быть записан в спецификации стратегии; иначе разные запуски могут иметь одинаковый хеш и повторно использовать результаты друг друга. Идентификатор набора задаётся контрольной суммой отсортированных символов, а описание возраста прогнозов фиксирует ограничение и удалённые строки и символы.
Конструктор предустановок сопоставляет заданные задержки с исполнением на том же или следующем баре, определяет частоту данных и настраивает счёт, комиссии и проскальзывание. Он также поддерживает комиссию за акцию с проскальзыванием на основе спреда, включая размер позиции целым числом акций и предотвращение повторного списания спреда, когда при исполнении уже используются цены bid/ask. В коде приведён измеренный пример одинаковых результатов для сокращённого и полного наборов до добавления отслеживания идентификаторов, а также пример выравнивания прогнозов, где ограничение возраста удалило строки. Эти примеры объясняют необходимость защитных проверок, но относятся к указанным наборам данных; сам код — инфраструктура, а не доказательство прибыльности торговой стратегии.
Ключевые идеи
- Сокращённый торгуемый набор должен входить в идентификатор запуска, чтобы результат для одного набора символов нельзя было использовать для другого.
- В описании возраста прогнозов следует фиксировать фактически удалённые строки: при переоценке могут получиться разные наборы прогнозов с одним ограничением возраста.
- Маркеры времени исполнения сопоставляются с явными режимами исполнения на том же или следующем баре.
- Комиссия за акцию и проскальзывание на основе спреда позволяют задавать издержки точнее, чем единая процентная ставка.
- Если исполнение по ценам на стороне котировки уже учитывает спред, повторный учёт того же спреда как проскальзывания приведёт к двойному учёту.
Теги
Полный текст
# backtest_presets.py
```py
from __future__ import annotations
from copy import deepcopy
from pathlib import Path
from typing import Any
import polars as pl
import yaml
from case_studies.utils.backtest_loaders import (
VECTORIZED_CASE_STUDIES,
declared_rebalance_step,
declares_rebalance_step,
)
from case_studies.utils.backtest_loaders import BacktestConfig as CaseStudyBacktestConfig
from utils.paths import get_case_study_dir
try:
from ml4t.backtest import BacktestConfig as EngineBacktestConfig
except (ImportError, ModuleNotFoundError): # pragma: no cover - import depends on env
EngineBacktestConfig = None
# The token names the bar a signal is filled on, not the price within it. Every
# entry here that maps to "next_bar" fills on the bar after the decision bar; which
# price that is belongs to the dataset. `NEXT_SESSION_CLOSE` is `sp500_options`,
# whose AlgoSeek option chain carries one end-of-session quote per contract per day
# and no open, so the next bar's only price is its close.
_EXECUTION_MODE_BY_DELAY = {
"NEXT_BAR_OPEN": "next_bar",
"NEXT_SESSION_CLOSE": "next_bar",
"MONDAY_OPEN": "next_bar",
"1_BAR": "next_bar",
"AT_FUNDING_TIMESTAMP": "same_bar",
}
def traded_universe_declaration(prices: pl.DataFrame) -> dict[str, Any]:
"""Describe the symbol set a run can hold, for the caller to put in its spec.
``MAX_SYMBOLS`` reduces the price panel and nothing else. On the engine path that
bounds what the run can trade, because the engine cannot fill an unpriced name; on
the vectorized path it bounds nothing, because ``gross_ret = weight * y_true`` is
computed from the predictions and ``prices`` supplies only the rebalance calendar.
Either way the reduction never reached ``backtest_hash``, so a reduced run and a
full run over the same predictions hashed alike and the second was served the
first's result. Measured on us_firm_characteristics/11_backtest, 2026-08-24: 8
predictions x 4 schemes at 300 symbols and at 3,708 gave bit-identical Sharpe, CAGR
and drawdown across all 32 backtests.
The declaration carries a digest of the sorted symbol list rather than its length,
because ``{A, B}`` and ``{A, C}`` are two symbols each and two different portfolios.
It goes in ``strategy.signal``, which is hashed whole, so a caller that declares one
gets an identity of its own and can neither be served nor serve a full-universe row.
``run_backtest`` reads it back through ``apply_traded_universe``: it checks the panel
it was handed against this digest and narrows the predictions to it, so the run
trades what its identity says on both paths.
Call it only when the panel was deliberately reduced. A caller that declares nothing
produces byte-identical specs to before this key existed, which is what leaves every
registered backtest at the identity it was written under.
"""
from case_studies.utils.registry.specs import canonical_json, compute_hash
if "symbol" not in prices.columns:
raise ValueError(
"traded_universe_declaration needs a 'symbol' column on the price panel; "
f"got columns={list(prices.columns)}"
)
symbols = sorted(prices.get_column("symbol").drop_nulls().unique().to_list())
if not symbols:
raise ValueError("traded_universe_declaration was handed an empty price panel")
return {
"n_symbols": len(symbols),
"digest": compute_hash(canonical_json({"symbols": symbols})),
}
def prediction_age_declaration(
*, max_age_sessions: int, dropped: int, kept: int, symbols_dropped: int
) -> dict[str, Any]:
"""Describe a prediction-age bound that removed rows, for the caller to put in its spec.
The sibling of :func:`traded_universe_declaration`, and the same defect it exists for.
Aligning predictions onto a coarser bar with a backward as-of join reuses a symbol's last
score for as long as the panel runs after its series ends, and bounding that age changes
what the backtest is computed from while ``prediction_hash`` and the strategy spec stay
identical - so a run with the bound and a run without it hash alike, and the second is
served the first's result. Measured on `nasdaq100_microstructure/17_costs`: the bound
removes 19,667 of 303,641 aligned rows at 30-minute cadence.
Carries the counts and not only the bound, because two runs can declare the same bound
and drop different rows: the prediction set behind them moves on every refit, and a run
whose universe left earlier is a different portfolio at the same tolerance.
Declare it only when the bound actually removed something. A caller that drops nothing
passes ``None`` and produces byte-identical specs to before this key existed, which is
what leaves every registered backtest at the identity it was written under.
"""
if dropped <= 0:
raise ValueError(
"prediction_age_declaration is for a bound that removed rows; a bound that "
"removed none must declare nothing, so the run keeps the identity it had."
)
return {
"max_age_sessions": int(max_age_sessions),
"dropped": int(dropped),
"kept": int(kept),
"symbols_dropped": int(symbols_dropped),
}
def resolve_execution_mode(fill_timing: str):
"""Map fill_timing string to ExecutionMode enum.
Raises ValueError for unknown tokens instead of silently degrading.
"""
from ml4t.backtest import ExecutionMode
token = fill_timing.upper().replace(" ", "_")
mode_str = _EXECUTION_MODE_BY_DELAY.get(token)
if mode_str is None:
raise ValueError(
f"Unknown execution delay '{fill_timing}'. "
f"Known values: {sorted(_EXECUTION_MODE_BY_DELAY.keys())}"
)
return ExecutionMode.NEXT_BAR if mode_str == "next_bar" else ExecutionMode.SAME_BAR
def preset_path(case_study: str) -> Path:
"""Path to the source-controlled backtest preset.
Always reads from the source repo (never ML4T_OUTPUT_DIR), since
config/backtest/base.yaml is checked-in source, not runtime data.
"""
from utils.paths import get_case_study_source_dir
return get_case_study_source_dir(case_study) / "config" / "backtest" / "base.yaml"
def load_backtest_preset(case_study: str) -> dict[str, Any]:
path = preset_path(case_study)
with path.open() as f:
data = yaml.safe_load(f) or {}
if not isinstance(data, dict):
raise TypeError(f"Backtest preset at {path} must be a mapping")
return data
def _infer_data_frequency(cadence: str) -> str:
token = cadence.lower()
if "15" in token:
return "15m"
if "30" in token:
return "30m"
if "1_hour" in token or "hourly" in token or token == "1h":
return "1h"
if "8_hour" in token or "funding" in token:
return "irregular"
return "daily"
def _build_feed_spec(
case_study: str,
prices: pl.DataFrame,
case_config: CaseStudyBacktestConfig,
) -> dict[str, Any]:
columns = set(prices.columns)
feed = {
"timestamp_col": "timestamp",
"entity_col": "symbol",
"open_col": "open" if "open" in columns else None,
"high_col": "high" if "high" in columns else None,
"low_col": "low" if "low" in columns else None,
"close_col": "close" if "close" in columns else None,
"price_col": "price" if "price" in columns else ("close" if "close" in columns else None),
"volume_col": "volume" if "volume" in columns else None,
"bid_col": "bid" if "bid" in columns else None,
"ask_col": "ask" if "ask" in columns else None,
"mid_col": "mid" if "mid" in columns else None,
"calendar": case_config.calendar,
"data_frequency": _infer_data_frequency(case_config.cadence),
"timezone": "UTC",
}
if case_study in {"sp500_options", "sp500_equity_option_analytics"}:
feed["bar_type"] = "quote"
return {k: v for k, v in feed.items() if v is not None}
def build_resolved_backtest_config(
case_study: str,
case_config: CaseStudyBacktestConfig,
strategy_spec: dict[str, Any],
*,
prices: pl.DataFrame,
initial_cash: float,
) -> EngineBacktestConfig:
if EngineBacktestConfig is None: # pragma: no cover - import depends on env
raise ImportError("ml4t-backtest is required for backtest preset resolution")
preset = deepcopy(load_backtest_preset(case_study))
preset.setdefault("account", {})
preset.setdefault("execution", {})
preset.setdefault("commission", {})
preset.setdefault("slippage", {})
preset.setdefault("cash", {})
preset.setdefault("calendar", {})
preset.setdefault("position_sizing", {})
fill_timing = strategy_spec.get("execution", {}).get(
"fill_timing"
) or case_config.execution_delay.upper().replace(" ", "_")
execution_mode = _EXECUTION_MODE_BY_DELAY.get(fill_timing, "next_bar")
signal_spec = strategy_spec.get("signal", {})
signal_direction = str(signal_spec.get("direction", "long_only")).strip().lower()
allow_short = bool(signal_spec.get("long_short", case_config.long_short)) or (
signal_direction == "short_only"
)
preset["account"]["allow_short_selling"] = allow_short
preset["execution"]["execution_mode"] = execution_mode
preset["cash"]["initial"] = float(initial_cash)
costs = strategy_spec.get("costs", {})
cost_model = costs.get("model", "percentage")
if cost_model == "percentage":
commission_bps = float(costs.get("commission_bps", case_config.commission_bps))
slippage_bps = float(costs.get("slippage_bps", case_config.slippage_bps))
preset["commission"]["model"] = "percentage"
preset["commission"]["rate"] = commission_bps / 10_000.0
preset["slippage"]["model"] = "percentage"
preset["slippage"]["rate"] = slippage_bps / 10_000.0
elif cost_model == "per_share_plus_spread":
# IB-style realistic equity costs: per-share commission, integer-share
# sizing, half-spread slippage in dollars per share. Per-asset spreads
# can be supplied directly (asset_spreads dict in setup.yaml) or via a
# parquet artifact (asset_spreads_source) measured from quote data.
preset["commission"]["model"] = "per_share"
preset["commission"]["per_share"] = float(costs["per_share"])
preset["commission"]["minimum"] = float(costs.get("minimum", 0.35))
# When execution_price is quote_side, the fill price already includes
# the bid/ask half-spread relative to mid (FillEngine returns
# ask for BUY / bid for SELL via broker.QUOTE_SIDE). Wiring the same
# measured half-spread into the slippage layer on top of that would
# charge the spread twice. Use a zero slippage layer in that case;
# per-share commission still applies independently.
execution_price = preset.get("execution", {}).get("execution_price")
if execution_price == "quote_side":
preset["slippage"]["model"] = "percentage"
preset["slippage"]["rate"] = 0.0
else:
preset["slippage"]["model"] = "spread"
preset["slippage"]["spread_convention"] = costs.get("spread_convention", "half_spread")
spread_by_asset: dict[str, float] = {}
asset_spreads_source = costs.get("asset_spreads_source")
if asset_spreads_source:
spreads_path = get_case_study_dir(case_study, create=False) / asset_spreads_source
spread_col = costs.get("asset_spreads_column", "median_half_spread_usd")
spreads_df = pl.read_parquet(spreads_path)
spread_by_asset = dict(
zip(
spreads_df["symbol"].to_list(),
[float(x) for x in spreads_df[spread_col].to_list()],
)
)
elif "asset_spreads" in costs:
spread_by_asset = {str(k): float(v) for k, v in costs["asset_spreads"].items()}
if spread_by_asset:
preset["slippage"]["spread_by_asset"] = spread_by_asset
if "default_half_spread_usd" in costs:
preset["slippage"]["spread"] = float(costs["default_half_spread_usd"])
else:
raise ValueError(
f"Unknown costs.model {cost_model!r}. Supported: 'percentage', 'per_share_plus_spread'."
)
# Share-type comes from setup.yaml::execution.share_type via case_config —
# never hardcoded per-cost-model branch. Falls back to the preset JSON's
# value (if any) when case_config.share_type is the default placeholder.
share_type = getattr(case_config, "share_type", None)
if share_type:
preset["position_sizing"]["share_type"] = share_type
preset["calendar"]["calendar"] = case_config.calendar
preset["calendar"].setdefault("data_frequency", _infer_data_frequency(case_config.cadence))
derived_feed = _build_feed_spec(case_study, prices, case_config)
explicit_feed = preset.get("feed", {})
preset["feed"] = {
**derived_feed,
**{k: v for k, v in explicit_feed.items() if v is not None},
}
metadata = dict(preset.get("metadata", {}))
metadata.update(
{
"case_study": case_study,
"chapter": strategy_spec.get("chapter"),
"cadence": strategy_spec.get("execution", {}).get("cadence", case_config.cadence),
"fill_timing": fill_timing,
"preset_path": str(preset_path(case_study)),
"signal_direction": signal_direction,
}
)
preset["metadata"] = metadata
return EngineBacktestConfig.from_dict(preset, preset_name=preset_path(case_study).stem)
def _serialize_backtest_config(config: EngineBacktestConfig | dict[str, Any]) -> dict[str, Any]:
if hasattr(config, "to_dict"):
return dict(config.to_dict())
return dict(config)
def runtime_backtest_config(spec: dict[str, Any]) -> EngineBacktestConfig:
if EngineBacktestConfig is None: # pragma: no cover - import depends on env
raise ImportError("ml4t-backtest is required for backtest config resolution")
runtime = EngineBacktestConfig.from_dict(spec["backtest_config"])
spec["_runtime_backtest_config"] = runtime
return runtime
# Calendars that require session enforcement (drop bars outside trading
# sessions, e.g. CME Saturdays). Mirror of the rule in backtest_runner._run_engine;
# applied at spec construction so plan-time hashes match registered hashes.
SESSION_ENFORCED_CALENDARS = frozenset({"CME", "us_futures"})
def apply_calendar_session_enforcement(config: EngineBacktestConfig, calendar: str | None) -> None:
"""Set ``enforce_sessions=True`` on ``config`` when ``calendar`` requires it.
Without this, ``ensure_backtest_spec`` would produce a plan-time spec
with ``enforce_sessions=False`` while ``_run_engine`` later mutates the
same runtime to ``True``, breaking ``_runtime_backtest_config`` hash
stability.
"""
if calendar in SESSION_ENFORCED_CALENDARS:
config.enforce_sessions = True
def serializable_backtest_spec(spec: dict[str, Any]) -> dict[str, Any]:
clean = deepcopy(spec)
clean.pop("_runtime_backtest_config", None)
if "backtest_config" in clean:
clean["backtest_config"] = _serialize_backtest_config(clean["backtest_config"])
return clean
def is_backtest_spec(spec: dict[str, Any]) -> bool:
"""Return True if ``spec`` is in canonical form (has ``strategy`` + ``backtest_config``)."""
return spec.get("version") == 2 and "strategy" in spec and "backtest_config" in spec
def ensure_backtest_spec(
case_study: str,
case_config: CaseStudyBacktestConfig,
strategy_spec: dict[str, Any],
*,
prices: pl.DataFrame,
prediction_hash: str,
initial_cash: float,
traded_universe: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Normalize ``strategy_spec`` to the canonical backtest spec form.
Idempotent: if ``strategy_spec`` is already canonical, it is returned with
a refreshed ``_runtime_backtest_config``. Otherwise, a flat strategy_spec
(with ``signal`` / ``execution`` / ``costs`` / etc. blocks) is projected
into the canonical envelope: ``strategy.{signal, rebalance, allocation, risk}``
plus a resolved ``backtest_config`` block.
Rebalance thresholds (``min_weight_change``, ``min_trade_value``) are
always populated in ``strategy.rebalance`` — taken from ``execution.*``
when present, otherwise from ``case_config`` (which sources them from
``setup.yaml::backtest.rebalance.default``).
``traded_universe`` is the same declaration ``build_backtest_spec`` takes, for the
same reason and with the same rule: build it with ``traded_universe_declaration``
when the price panel was deliberately reduced, and it lands in ``strategy.signal``,
which is hashed whole, so the reduction reaches ``backtest_hash`` instead of hashing
like the full run over the same predictions. It is
written only when a caller declares one, on both the carried-forward and the projected
path, so every existing call site produces byte-identical specs and no registered
backtest re-keys. A declaration handed here overwrites one inherited from the spec
being normalized, because the panel this run was given is the universe it trades;
``apply_traded_universe`` then checks that claim against the panel in the runner.
"""
if is_backtest_spec(strategy_spec):
spec = deepcopy(strategy_spec)
if case_study == "sp500_options":
spec.setdefault("strategy", {}).setdefault("signal", {}).setdefault(
"schedule_contract", SP500_OPTIONS_SCHEDULE_CONTRACT
)
spec.setdefault(
"chapter", spec.get("backtest_config", {}).get("metadata", {}).get("chapter")
)
# Ensure rebalance thresholds are populated; specs that omit them
# would otherwise raise KeyError on `rebalance_spec["min_weight_change"]`.
rb = spec.setdefault("strategy", {}).setdefault("rebalance", {})
rb.setdefault("min_weight_change", float(getattr(case_config, "min_weight_change", 0.005)))
rb.setdefault("min_trade_value", float(getattr(case_config, "min_trade_value", 100.0)))
if traded_universe is not None:
spec["strategy"].setdefault("signal", {})["traded_universe"] = deepcopy(traded_universe)
if "backtest_config" in spec:
# Always overwrite metadata.prediction_hash with the caller's
# argument — the spec may have been cloned from another run
# (e.g. Ch20 holdout reuses the validation rank-1 spec with a
# fresh holdout pred_hash), and downstream split-resolution
# depends on the metadata reflecting the prediction set actually
# being backtested.
metadata = spec["backtest_config"].setdefault("metadata", {})
metadata["prediction_hash"] = prediction_hash
runtime = EngineBacktestConfig.from_dict(spec["backtest_config"])
# Only the session-enforced calendars actually mutate ``runtime``
# here; for all other calendars the from_dict/to_dict round-trip
# would be a silent re-serialize that could perturb hashes for
# any CS whose dict shape differs from the dataclass defaults
# (None→0 normalization, dropped unknown keys, etc.). Confine the
# ``backtest_config`` overwrite to the CSes where it is needed.
if case_config.calendar in SESSION_ENFORCED_CALENDARS:
apply_calendar_session_enforcement(runtime, case_config.calendar)
# Re-serialize so the canonical ``backtest_config`` matches
# the runtime; preserve every caller-supplied metadata key
# by merging the original metadata back over the dataclass's
# serialized view (the dataclass typically pins a schema
# and drops unknown keys).
original_metadata = dict(metadata)
rebuilt = runtime.to_dict()
rebuilt_metadata = dict(rebuilt.get("metadata") or {})
rebuilt_metadata.update(original_metadata)
rebuilt["metadata"] = rebuilt_metadata
spec["backtest_config"] = rebuilt
spec["_runtime_backtest_config"] = runtime
return spec
execution = deepcopy(strategy_spec.get("execution", {}))
rebalance = {
"mode": execution.get("mode", "engine"),
"cadence": execution.get("cadence", case_config.cadence),
"min_weight_change": float(
execution["min_weight_change"]
if "min_weight_change" in execution
else getattr(case_config, "min_weight_change", 0.005)
),
"min_trade_value": float(
execution["min_trade_value"]
if "min_trade_value" in execution
else getattr(case_config, "min_trade_value", 100.0)
),
}
if "step" in execution:
rebalance["step"] = int(execution["step"])
strategy = {
"signal": deepcopy(strategy_spec.get("signal", {})),
"rebalance": rebalance,
}
if case_study == "sp500_options":
strategy["signal"].setdefault("schedule_contract", SP500_OPTIONS_SCHEDULE_CONTRACT)
if traded_universe is not None:
strategy["signal"]["traded_universe"] = deepcopy(traded_universe)
if "allocation" in strategy_spec:
strategy["allocation"] = deepcopy(strategy_spec["allocation"])
if "risk" in strategy_spec:
strategy["risk"] = deepcopy(strategy_spec["risk"])
resolved_config = build_resolved_backtest_config(
case_study,
case_config,
strategy_spec,
prices=prices,
initial_cash=initial_cash,
)
resolved_config.metadata["prediction_hash"] = prediction_hash
apply_calendar_session_enforcement(resolved_config, case_config.calendar)
resolved_config_dict = resolved_config.to_dict()
return {
"version": 2,
"chapter": strategy_spec.get("chapter"),
"preset_id": f"{case_study}:base",
"strategy": strategy,
"backtest_config": resolved_config_dict,
"_runtime_backtest_config": resolved_config,
}
_COST_PASSTHROUGH_KEYS = (
"model",
"per_share",
"minimum",
"max_pct",
"asset_spreads_source",
"asset_spreads_column",
"asset_spreads",
"default_half_spread_usd",
"spread_convention",
)
SP500_OPTIONS_SCHEDULE_CONTRACT = "last_available_session_per_iso_week_v1"
def _costs_block_from_case_config(
case_config: CaseStudyBacktestConfig,
) -> dict[str, Any]:
"""Build the strategy_spec.costs block from the case config.
For the percentage model (default), emit the bps form. For the
per_share_plus_spread model, forward the full costs schema from setup.yaml
so build_resolved_backtest_config can dispatch.
"""
raw_costs = getattr(case_config, "raw_costs", None) or {}
cost_model = raw_costs.get("model", "percentage")
if cost_model == "per_share_plus_spread":
return {key: deepcopy(raw_costs[key]) for key in _COST_PASSTHROUGH_KEYS if key in raw_costs}
return {
"commission_bps": case_config.commission_bps,
"slippage_bps": case_config.slippage_bps,
}
def build_backtest_spec(
case_study: str,
case_config: CaseStudyBacktestConfig,
*,
prices: pl.DataFrame,
prediction_hash: str,
initial_cash: float,
signal: dict[str, Any],
allocation: dict[str, Any] | None = None,
risk: dict[str, Any] | None = None,
costs: dict[str, Any] | None = None,
chapter: str | None = None,
execution_mode: str | None = None,
min_weight_change: float | None = None,
min_trade_value: float | None = None,
label: str | None = None,
traded_universe: dict[str, Any] | None = None,
prediction_age: dict[str, Any] | None = None,
) -> dict[str, Any]:
# A case study that declares per-label cadences must be told which label it is building for.
# Defaulting to the case-study cadence here would put the spec on the wrong grid and register
# it as if it were right, which is the failure this parameter exists to prevent - so it is a
# refusal, not a fallback. Case studies that declare no override are unaffected.
if getattr(case_config, "cadence_by_label", None) and not label:
raise ValueError(
f"{case_study} declares decision.cadence_by_label "
f"({sorted(case_config.cadence_by_label)}) so build_backtest_spec needs a non-empty "
"label=; "
"pass the label this spec is being built for."
)
# The same refusal for the step, and for a sharper reason: the step is emitted below only
# when a label resolves one, while `run_backtest` stamps the declared step onto whatever
# spec it is handed. So an unlabelled call here does not build a spec on the wrong grid -
# it builds a spec that hashes to an identity no run ever registers, and a caller that
# pre-hashes it to decide what to skip finds every registered row missing
# . Measured on us_firm_characteristics: 2,276 of 2,276
# registered baseline rows invisible to `11_backtest`'s own skip check.
if declares_rebalance_step(case_study) and not label:
raise ValueError(
f"{case_study} declares labels.rebalance_step, which is part of the backtest "
"identity, so build_backtest_spec needs a non-empty label=; pass the label this "
"spec is being built for. Without it the spec hashes to an identity no run "
"registers."
)
resolved_signal = deepcopy(signal)
if case_study == "sp500_options":
resolved_signal.setdefault("schedule_contract", SP500_OPTIONS_SCHEDULE_CONTRACT)
# The universe a reduced run trades, from `traded_universe_declaration`. It sits in the
# signal block because that block is hashed whole, so the reduction reaches
# `backtest_hash` without a second place to keep in step with it. Emitted only when the
# caller declares one - the rule `cadence` and `step` already follow above - so a full
# run produces byte-identical specs to before this parameter existed and every
# registered backtest keeps the identity it was written under.
if traded_universe is not None:
resolved_signal["traded_universe"] = deepcopy(traded_universe)
# The prediction-age bound, from `prediction_age_declaration`, on the same terms and in
# the same block for the same reason: it changes what the run is computed from without
# touching `prediction_hash`, so without this a bounded run and an unbounded one over the
# same predictions hash alike and the second is served the first's result.
if prediction_age is not None:
resolved_signal["prediction_age"] = deepcopy(prediction_age)
strategy_spec: dict[str, Any] = {
"signal": resolved_signal,
"execution": {
# `VECTORIZED_CASE_STUDIES` rather than a second copy of the same set. The
# membership was written out here as well, and the two are read by different
# things: this one decides the dispatch in `backtest_runner`, and the constant
# decides `IS_VECTORIZED` in four risk-management notebooks. Adding a case study
# to one and not the other gives a notebook the wrong branch with nothing raising.
"mode": (
execution_mode
if execution_mode is not None
else "vectorized"
if case_study in VECTORIZED_CASE_STUDIES
else "engine"
),
"engine_preset": "realistic",
# Per-label when the case study declares one. `cadence_for` returns the case-study
# default for an unlabelled caller and for a label with no override, so a case study
# that declares nothing produces byte-identical specs to before this parameter existed.
"cadence": case_config.cadence_for(label),
# The step composes with the cadence to decide which slots are traded, so it
# belongs to the identity beside it. Emitted only
# when the case study declares one, so a case study that declares nothing
# produces byte-identical specs to before this key existed - the same rule
# `cadence_for` follows above.
**(
{"step": _step}
if (_step := declared_rebalance_step(case_study, label)) is not None
else {}
),
"fill_timing": case_config.execution_delay.upper().replace(" ", "_"),
"min_weight_change": (
min_weight_change
if min_weight_change is not None
else getattr(case_config, "min_weight_change", 0.005)
),
"min_trade_value": (
min_trade_value
if min_trade_value is not None
else getattr(case_config, "min_trade_value", 100.0)
),
},
"costs": (
deepcopy(costs) if costs is not None else _costs_block_from_case_config(case_config)
),
}
if chapter is not None:
strategy_spec["chapter"] = chapter
if allocation is not None:
strategy_spec["allocation"] = deepcopy(allocation)
if risk is not None:
strategy_spec["risk"] = deepcopy(risk)
return ensure_backtest_spec(
case_study,
case_config,
strategy_spec,
prices=prices,
prediction_hash=prediction_hash,
initial_cash=initial_cash,
)
def clone_backtest_spec(spec: dict[str, Any]) -> dict[str, Any]:
cloned = serializable_backtest_spec(spec)
if is_backtest_spec(cloned):
cloned["_runtime_backtest_config"] = EngineBacktestConfig.from_dict(
cloned["backtest_config"]
)
return cloned
def set_backtest_costs_bps(
spec: dict[str, Any],
*,
commission_bps: float,
slippage_bps: float,
) -> dict[str, Any]:
if not is_backtest_spec(spec):
updated = deepcopy(spec)
updated["costs"] = {
"commission_bps": commission_bps,
"slippage_bps": slippage_bps,
}
return updated
updated = deepcopy(spec)
bt_cfg = updated["backtest_config"]
bt_cfg["commission"] = {
"model": "percentage",
"rate": commission_bps / 10_000.0,
}
bt_cfg["slippage"] = {
"model": "percentage",
"rate": slippage_bps / 10_000.0,
}
updated["_runtime_backtest_config"] = EngineBacktestConfig.from_dict(bt_cfg)
return updated
def set_backtest_costs_per_share(
spec: dict[str, Any],
*,
per_share: float,
default_half_spread_usd: float,
asset_spreads: dict[str, float] | None = None,
spread_convention: str = "half_spread",
minimum: float = 0.0,
) -> dict[str, Any]:
"""Mutate spec to use per-share commission + spread slippage.
Switches the engine commission/slippage models from `percentage` to
`per_share` / `spread`. Safe to call on a spec that originally used the
percentage model — the previous rate fields are replaced. Used by the
cost-sensitivity sweep to walk the per-share+spread regime alongside
the bps regime for case studies whose dataset supports it (those with
prices and integer-share semantics).
`minimum` is the per-order commission floor in dollars and defaults to
`0.0` (no floor). Pass `minimum=0.35` to match the IBKR Pro per-order
floor used by `build_resolved_backtest_config`.
"""
if not is_backtest_spec(spec):
updated = deepcopy(spec)
updated["costs"] = {
"model": "per_share_plus_spread",
"per_share": float(per_share),
"default_half_spread_usd": float(default_half_spread_usd),
"asset_spreads": dict(asset_spreads or {}),
"spread_convention": spread_convention,
"minimum": float(minimum),
}
return updated
updated = deepcopy(spec)
bt_cfg = updated["backtest_config"]
bt_cfg["commission"] = {
"model": "per_share",
"rate": 0.0,
"per_share": float(per_share),
"minimum": float(minimum),
"per_trade": 0.0,
}
bt_cfg["slippage"] = {
"model": "spread",
"rate": 0.0,
"spread": float(default_half_spread_usd),
"spread_convention": spread_convention,
}
if asset_spreads:
bt_cfg["slippage"]["spread_by_asset"] = {str(k): float(v) for k, v in asset_spreads.items()}
bt_cfg.setdefault("position_sizing", {})["share_type"] = "integer"
updated["_runtime_backtest_config"] = EngineBacktestConfig.from_dict(bt_cfg)
return updated
def strategy_view(spec: dict[str, Any]) -> dict[str, Any]:
return spec["strategy"] if is_backtest_spec(spec) else spec
def cost_view(spec: dict[str, Any]) -> dict[str, Any]:
if not is_backtest_spec(spec):
return spec.get("costs", {})
cfg = spec.get("backtest_config", {})
commission = cfg.get("commission", {})
slippage = cfg.get("slippage", {})
return {
"commission_bps": round(float(commission.get("rate", 0.0)) * 10_000.0, 10),
"slippage_bps": round(float(slippage.get("rate", 0.0)) * 10_000.0, 10),
}
```Полный текст с указанием источника опубликован на условиях его лицензии. Лицензия: MIT
Это краткое изложение подготовлено исследовательским агентом Stratmill по оригиналу и не является его копией.