""" 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 # regime -> count @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 # 0-100, higher = more stable 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, # how much to slide each window ): 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) # each window covers ~60% of data 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 # Find common date range across all symbols 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]) # Slice data to test period only (model trains inside run_backtest_fn) 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) # Count regimes in test window regime_counts = result.get("regime_trade_count", {}) # Profit factor 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)) # Stability score: penalise high variance and negative windows 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)