| """
|
| ml/walk_forward.py — Enhanced walk-forward validation framework.
|
|
|
| Multiple rolling windows, cross-regime testing, stability scoring.
|
| """
|
| from __future__ import annotations
|
|
|
| import logging
|
| from dataclasses import dataclass, field
|
|
|
| import numpy as np
|
| import pandas as pd
|
|
|
| logger = logging.getLogger("trading_system.ml.walk_forward")
|
|
|
|
|
| @dataclass
|
| class WalkForwardResult:
|
| """Results from a single walk-forward window."""
|
| window_id: int
|
| train_start: str
|
| train_end: str
|
| test_start: str
|
| test_end: str
|
| total_return_pct: float
|
| sharpe_ratio: float
|
| sortino_ratio: float
|
| max_drawdown_pct: float
|
| win_rate_pct: float
|
| profit_factor: float
|
| total_trades: int
|
| regimes_encountered: dict
|
|
|
|
|
| @dataclass
|
| class WalkForwardSummary:
|
| """Aggregated walk-forward results."""
|
| windows: list[WalkForwardResult]
|
| avg_return_pct: float = 0.0
|
| avg_sharpe: float = 0.0
|
| avg_mdd: float = 0.0
|
| min_sharpe: float = 0.0
|
| max_mdd: float = 0.0
|
| sharpe_std: float = 0.0
|
| stability_score: float = 0.0
|
|
|
|
|
| class WalkForwardValidator:
|
| """Walk-forward validation with overlapping windows.
|
|
|
| Splits data into multiple train/test windows:
|
| [--train--|--test--]
|
| [--train--|--test--]
|
| [--train--|--test--]
|
|
|
| Parameters:
|
| train_pct: Fraction of each window for training (default 70%).
|
| n_windows: Number of overlapping windows.
|
| min_test_bars: Minimum bars in test period.
|
| """
|
|
|
| def __init__(
|
| self,
|
| n_windows: int = 5,
|
| train_pct: float = 0.70,
|
| min_test_bars: int = 100,
|
| step_pct: float = 0.15,
|
| ):
|
| self.n_windows = n_windows
|
| self.train_pct = train_pct
|
| self.min_test_bars = min_test_bars
|
| self.step_pct = step_pct
|
|
|
| def generate_windows(
|
| self,
|
| total_bars: int,
|
| ) -> list[tuple[int, int, int, int]]:
|
| """Generate (train_start, train_end, test_start, test_end) index tuples."""
|
| windows = []
|
| window_size = int(total_bars * 0.6)
|
| step = int(total_bars * self.step_pct)
|
| train_size = int(window_size * self.train_pct)
|
|
|
| for i in range(self.n_windows):
|
| start = i * step
|
| end = start + window_size
|
| if end > total_bars:
|
| end = total_bars
|
| if end - start < train_size + self.min_test_bars:
|
| break
|
|
|
| train_end = start + train_size
|
| test_start = train_end
|
| test_end = end
|
|
|
| windows.append((start, train_end, test_start, test_end))
|
|
|
| return windows
|
|
|
| def run_validation(
|
| self,
|
| run_backtest_fn,
|
| data_dict: dict,
|
| daily_data_dict: dict,
|
| symbols: list[str],
|
| **backtest_kwargs,
|
| ) -> WalkForwardSummary:
|
| """Run walk-forward validation using provided backtest function.
|
|
|
| Args:
|
| run_backtest_fn: Callable that takes (symbols, data, daily_data, **kwargs) -> result dict
|
| data_dict: {symbol: intraday DataFrame}
|
| daily_data_dict: {symbol: daily DataFrame}
|
| symbols: List of symbols to test.
|
| **backtest_kwargs: Additional kwargs for the backtest function.
|
|
|
| Returns: WalkForwardSummary with per-window and aggregated metrics.
|
| """
|
| from backtester.report import compute_metrics
|
| from backtester.regime_detector import classify_regimes
|
|
|
|
|
| all_dates = set()
|
| for sym in symbols:
|
| if sym in data_dict:
|
| all_dates.update(data_dict[sym].index)
|
| all_dates = sorted(all_dates)
|
| total_bars = len(all_dates)
|
|
|
| if total_bars < 300:
|
| logger.warning(f"Only {total_bars} bars — walk-forward needs at least 300")
|
| return WalkForwardSummary(windows=[])
|
|
|
| windows_indices = self.generate_windows(total_bars)
|
| results: list[WalkForwardResult] = []
|
|
|
| for wi, (tr_s, tr_e, te_s, te_e) in enumerate(windows_indices):
|
| train_dates = set(all_dates[tr_s:tr_e])
|
| test_dates = set(all_dates[te_s:te_e])
|
|
|
|
|
| test_data = {}
|
| test_daily = {}
|
| for sym in symbols:
|
| if sym in data_dict:
|
| df = data_dict[sym]
|
| mask = df.index.isin(test_dates)
|
| if mask.sum() >= self.min_test_bars:
|
| test_data[sym] = df[mask]
|
| if sym in daily_data_dict:
|
| test_daily[sym] = daily_data_dict[sym]
|
|
|
| if not test_data:
|
| continue
|
|
|
| try:
|
| result = run_backtest_fn(
|
| symbols=list(test_data.keys()),
|
| data=test_data,
|
| daily_data=test_daily,
|
| **backtest_kwargs,
|
| )
|
| metrics = compute_metrics(result)
|
|
|
|
|
| regime_counts = result.get("regime_trade_count", {})
|
|
|
|
|
| trades = result.get("trades", [])
|
| gross_profit = sum(t["pnl"] for t in trades if t["pnl"] > 0)
|
| gross_loss = abs(sum(t["pnl"] for t in trades if t["pnl"] <= 0))
|
| pf = gross_profit / gross_loss if gross_loss > 0 else 0.0
|
|
|
| wfr = WalkForwardResult(
|
| window_id=wi,
|
| train_start=str(all_dates[tr_s])[:10],
|
| train_end=str(all_dates[tr_e - 1])[:10],
|
| test_start=str(all_dates[te_s])[:10],
|
| test_end=str(all_dates[te_e - 1])[:10],
|
| total_return_pct=metrics["total_return_pct"],
|
| sharpe_ratio=metrics["sharpe_ratio"],
|
| sortino_ratio=metrics["sortino_ratio"],
|
| max_drawdown_pct=metrics["max_drawdown_pct"],
|
| win_rate_pct=metrics["win_rate_pct"],
|
| profit_factor=pf,
|
| total_trades=metrics["total_trades"],
|
| regimes_encountered=regime_counts,
|
| )
|
| results.append(wfr)
|
|
|
| except Exception as e:
|
| logger.error(f"Walk-forward window {wi} failed: {e}")
|
| continue
|
|
|
| return self._summarise(results)
|
|
|
| def _summarise(self, results: list[WalkForwardResult]) -> WalkForwardSummary:
|
| """Aggregate walk-forward results."""
|
| if not results:
|
| return WalkForwardSummary(windows=[])
|
|
|
| returns = [r.total_return_pct for r in results]
|
| sharpes = [r.sharpe_ratio for r in results]
|
| mdds = [r.max_drawdown_pct for r in results]
|
|
|
| avg_ret = float(np.mean(returns))
|
| avg_sharpe = float(np.mean(sharpes))
|
| avg_mdd = float(np.mean(mdds))
|
| min_sharpe = float(np.min(sharpes))
|
| max_mdd = float(np.max(mdds))
|
| sharpe_std = float(np.std(sharpes))
|
|
|
|
|
| positive_windows = sum(1 for r in returns if r > 0)
|
| consistency = positive_windows / len(returns) * 100
|
| sharpe_stability = max(0, 100 - sharpe_std * 50)
|
| dd_penalty = max(0, 100 - max_mdd * 5)
|
| stability = (consistency * 0.4 + sharpe_stability * 0.3 + dd_penalty * 0.3)
|
|
|
| return WalkForwardSummary(
|
| windows=results,
|
| avg_return_pct=round(avg_ret, 2),
|
| avg_sharpe=round(avg_sharpe, 2),
|
| avg_mdd=round(avg_mdd, 2),
|
| min_sharpe=round(min_sharpe, 2),
|
| max_mdd=round(max_mdd, 2),
|
| sharpe_std=round(sharpe_std, 2),
|
| stability_score=round(stability, 1),
|
| )
|
|
|
|
|
| def print_walk_forward_report(summary: WalkForwardSummary):
|
| """Pretty-print walk-forward validation results."""
|
| print()
|
| print("=" * 90)
|
| print(" WALK-FORWARD VALIDATION RESULTS")
|
| print("=" * 90)
|
|
|
| if not summary.windows:
|
| print(" No valid windows completed.")
|
| return
|
|
|
| print(f" Windows: {len(summary.windows)}")
|
| print(f" Avg Return: {summary.avg_return_pct:+.2f}%")
|
| print(f" Avg Sharpe: {summary.avg_sharpe:.2f} (std: {summary.sharpe_std:.2f})")
|
| print(f" Min Sharpe: {summary.min_sharpe:.2f}")
|
| print(f" Avg Max DD: {summary.avg_mdd:.2f}%")
|
| print(f" Worst DD: {summary.max_mdd:.2f}%")
|
| print(f" Stability Score: {summary.stability_score:.1f}/100")
|
|
|
| print()
|
| print(" PER-WINDOW BREAKDOWN:")
|
| print(f" {'#':>3} {'Test Period':>25} {'Return%':>9} {'Sharpe':>8} {'MDD%':>7} {'WR%':>6} {'PF':>6} {'Trades':>7}")
|
| print(" " + "-" * 70)
|
| for w in summary.windows:
|
| period = f"{w.test_start} → {w.test_end}"
|
| print(f" {w.window_id:>3} {period:>25} {w.total_return_pct:>+8.2f} "
|
| f"{w.sharpe_ratio:>8.2f} {w.max_drawdown_pct:>6.2f} "
|
| f"{w.win_rate_pct:>5.1f} {w.profit_factor:>5.2f} {w.total_trades:>7}")
|
|
|
| print("=" * 90)
|
|
|
