wehub-resource-sync
576 lines
22 KiB
Python
576 lines
22 KiB
Python
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 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
|
|
|
|
|
|
_EXECUTION_MODE_BY_DELAY = {
|
|
"NEXT_BAR_OPEN": "next_bar",
|
|
"MONDAY_OPEN": "next_bar",
|
|
"1_BAR": "next_bar",
|
|
"AT_FUNDING_TIMESTAMP": "same_bar",
|
|
}
|
|
|
|
|
|
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,
|
|
) -> 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``).
|
|
"""
|
|
if is_backtest_spec(strategy_spec):
|
|
spec = deepcopy(strategy_spec)
|
|
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 "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)
|
|
),
|
|
}
|
|
strategy = {
|
|
"signal": deepcopy(strategy_spec.get("signal", {})),
|
|
"rebalance": rebalance,
|
|
}
|
|
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",
|
|
)
|
|
|
|
|
|
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,
|
|
chapter: str | None = None,
|
|
execution_mode: str | None = None,
|
|
min_weight_change: float | None = None,
|
|
min_trade_value: float | None = None,
|
|
) -> dict[str, Any]:
|
|
strategy_spec: dict[str, Any] = {
|
|
"signal": deepcopy(signal),
|
|
"execution": {
|
|
"mode": (
|
|
execution_mode
|
|
if execution_mode is not None
|
|
else "vectorized"
|
|
if case_study in {"us_firm_characteristics", "sp500_options"}
|
|
else "engine"
|
|
),
|
|
"engine_preset": "realistic",
|
|
"cadence": case_config.cadence,
|
|
"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": _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),
|
|
}
|