100 lines
5.3 KiB
Python
100 lines
5.3 KiB
Python
"""Cumulative ablation, chronological holdouts and rolling walk-forward reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
import pandas as pd
|
|
|
|
from .backtest import BacktestResult, run_backtest
|
|
from .config import StrategyConfig, ThaiMarketRules, TradingCosts
|
|
|
|
|
|
ABLATIONS = [
|
|
("A 21DMA only", {}),
|
|
("B + RS", {"enable_rs": True}),
|
|
("C + Market Regime", {"enable_market_regime": True, "require_market_support": True}),
|
|
("D + MCO", {"enable_mco": True}),
|
|
("E + MCSI", {"enable_mcsi": True}),
|
|
("F + Higher Low", {"enable_higher_low": True}),
|
|
("G + Weekly Structure", {"enable_weekly": True}),
|
|
("H + ATR Extension Filter", {"enable_atr_filter": True}),
|
|
]
|
|
|
|
|
|
def run_ablations(features_by_symbol: dict[str, pd.DataFrame], market: pd.DataFrame, *,
|
|
base: StrategyConfig | None = None, rules: ThaiMarketRules | None = None,
|
|
costs: TradingCosts | None = None) -> tuple[pd.DataFrame, dict[str, BacktestResult]]:
|
|
base = base or StrategyConfig()
|
|
results: dict[str, BacktestResult] = {}
|
|
rows = []
|
|
toggles = {"enable_rs": False, "enable_market_regime": False, "enable_mco": False,
|
|
"enable_mcsi": False, "enable_higher_low": False, "enable_weekly": False,
|
|
"enable_atr_filter": False, "require_market_support": False}
|
|
for label, changes in ABLATIONS:
|
|
toggles.update(changes)
|
|
strategy = replace(base, **toggles)
|
|
result = run_backtest(features_by_symbol, market, strategy=strategy, rules=rules, costs=costs)
|
|
results[label] = result
|
|
rows.append({"experiment": label, **result.metrics, "trades": len(result.trades)})
|
|
return pd.DataFrame(rows), results
|
|
|
|
|
|
def chronological_splits(dates: pd.Index, *, train_fraction: float = 0.60,
|
|
validation_fraction: float = 0.20) -> dict[str, tuple[pd.Timestamp, pd.Timestamp]]:
|
|
if len(dates) < 10 or not 0 < train_fraction < 1 or not 0 < validation_fraction < 1 or train_fraction + validation_fraction >= 1:
|
|
raise ValueError("invalid date count or train/validation fractions")
|
|
ordered = pd.DatetimeIndex(sorted(pd.Timestamp(d) for d in dates))
|
|
train_end = max(1, int(len(ordered) * train_fraction))
|
|
validation_end = max(train_end + 1, int(len(ordered) * (train_fraction + validation_fraction)))
|
|
validation_end = min(validation_end, len(ordered) - 1)
|
|
return {
|
|
"train": (ordered[0], ordered[train_end - 1]),
|
|
"validation": (ordered[train_end], ordered[validation_end - 1]),
|
|
"out_of_sample": (ordered[validation_end], ordered[-1]),
|
|
}
|
|
|
|
|
|
def run_chronological_holdouts(features_by_symbol: dict[str, pd.DataFrame], market: pd.DataFrame, *,
|
|
strategy: StrategyConfig | None = None,
|
|
rules: ThaiMarketRules | None = None,
|
|
costs: TradingCosts | None = None) -> tuple[pd.DataFrame, dict[str, BacktestResult]]:
|
|
dates = pd.DatetimeIndex(sorted(set().union(*(frame.index for frame in features_by_symbol.values()))))
|
|
splits = chronological_splits(dates)
|
|
results = {}
|
|
rows = []
|
|
for label, (start, end) in splits.items():
|
|
result = run_backtest(features_by_symbol, market, strategy=strategy, rules=rules, costs=costs,
|
|
start=start.isoformat(), end=(end + pd.Timedelta(days=1)).isoformat())
|
|
results[label] = result
|
|
rows.append({"period": label, "start": start.date().isoformat(), "end": end.date().isoformat(),
|
|
**result.metrics, "trades": len(result.trades)})
|
|
return pd.DataFrame(rows), results
|
|
|
|
|
|
def rolling_walk_forward(features_by_symbol: dict[str, pd.DataFrame], market: pd.DataFrame, *,
|
|
train_sessions: int = 252, validation_sessions: int = 63,
|
|
test_sessions: int = 63, step_sessions: int = 63,
|
|
strategy: StrategyConfig | None = None,
|
|
rules: ThaiMarketRules | None = None,
|
|
costs: TradingCosts | None = None) -> pd.DataFrame:
|
|
dates = pd.DatetimeIndex(sorted(set().union(*(frame.index for frame in features_by_symbol.values()))))
|
|
if min(train_sessions, validation_sessions, test_sessions, step_sessions) < 1:
|
|
raise ValueError("walk-forward window lengths must be positive")
|
|
rows = []
|
|
start = 0
|
|
fold = 1
|
|
while start + train_sessions + validation_sessions + test_sessions <= len(dates):
|
|
train_end = start + train_sessions
|
|
val_end = train_end + validation_sessions
|
|
test_end = val_end + test_sessions
|
|
# Parameters remain fixed: training and validation are reported separately, not optimized here.
|
|
for label, begin, finish in [("train", start, train_end), ("validation", train_end, val_end), ("out_of_sample", val_end, test_end)]:
|
|
result = run_backtest(features_by_symbol, market, strategy=strategy, rules=rules, costs=costs,
|
|
start=dates[begin].isoformat(), end=(dates[finish - 1] + pd.Timedelta(days=1)).isoformat())
|
|
rows.append({"fold": fold, "period": label, "start": dates[begin].date().isoformat(),
|
|
"end": dates[finish - 1].date().isoformat(), **result.metrics, "trades": len(result.trades)})
|
|
fold += 1
|
|
start += step_sessions
|
|
return pd.DataFrame(rows)
|