feat(backtester): add walk-forward analysis engine

1 files changed, 145 insertions, 0 deletions

diff --git a/services/backtester/src/backtester/walk_forward.py b/services/backtester/src/backtester/walk_forward.py
new file mode 100644
index 0000000..fe6d020
--- /dev/null
+++ b/services/backtester/src/backtester/walk_forward.py
@@ -0,0 +1,145 @@
+"""Walk-forward analysis for strategy parameter optimization."""
+from dataclasses import dataclass, field
+from decimal import Decimal
+from typing import Callable
+
+from shared.models import Candle
+from backtester.engine import BacktestEngine, BacktestResult, StrategyProtocol
+
+
+@dataclass
+class WalkForwardWindow:
+    """Result for a single in-sample/out-of-sample window."""
+    window_index: int
+    in_sample_result: BacktestResult
+    out_of_sample_result: BacktestResult
+    best_params: dict
+
+
+@dataclass
+class WalkForwardResult:
+    """Aggregated walk-forward analysis results."""
+    strategy_name: str
+    symbol: str
+    num_windows: int
+    windows: list[WalkForwardWindow] = field(default_factory=list)
+
+    @property
+    def out_of_sample_profit_pct(self) -> float:
+        """Combined out-of-sample profit percentage."""
+        if not self.windows:
+            return 0.0
+        total_profit = sum(float(w.out_of_sample_result.profit) for w in self.windows)
+        initial = float(self.windows[0].out_of_sample_result.initial_balance)
+        return (total_profit / initial * 100) if initial > 0 else 0.0
+
+    @property
+    def in_sample_profit_pct(self) -> float:
+        if not self.windows:
+            return 0.0
+        total_profit = sum(float(w.in_sample_result.profit) for w in self.windows)
+        initial = float(self.windows[0].in_sample_result.initial_balance)
+        return (total_profit / initial * 100) if initial > 0 else 0.0
+
+    @property
+    def efficiency_ratio(self) -> float:
+        """Out-of-sample / in-sample performance ratio.
+        Close to 1.0 = robust. Much less = overfitting."""
+        if self.in_sample_profit_pct == 0:
+            return 0.0
+        return self.out_of_sample_profit_pct / self.in_sample_profit_pct
+
+
+class WalkForwardEngine:
+    """Runs walk-forward analysis on a strategy.
+
+    Splits candle data into N rolling windows. For each window:
+    1. In-sample: Try each param set, pick the best by profit
+    2. Out-of-sample: Run best params on held-out data
+    """
+
+    def __init__(
+        self,
+        strategy_factory: Callable[[], StrategyProtocol],
+        param_grid: list[dict],
+        initial_balance: Decimal = Decimal("10000"),
+        num_windows: int = 5,
+        in_sample_pct: float = 0.7,
+    ) -> None:
+        self._strategy_factory = strategy_factory
+        self._param_grid = param_grid
+        self._initial_balance = initial_balance
+        self._num_windows = num_windows
+        self._in_sample_pct = in_sample_pct
+
+    def run(self, candles: list[Candle]) -> WalkForwardResult:
+        """Run walk-forward analysis over the candle data."""
+        if not candles or not self._param_grid:
+            strategy = self._strategy_factory()
+            return WalkForwardResult(
+                strategy_name=strategy.name,
+                symbol=candles[0].symbol if candles else "",
+                num_windows=0,
+            )
+
+        total = len(candles)
+        window_size = total // self._num_windows
+        if window_size < 10:
+            # Not enough data for meaningful walk-forward
+            strategy = self._strategy_factory()
+            return WalkForwardResult(
+                strategy_name=strategy.name,
+                symbol=candles[0].symbol if candles else "",
+                num_windows=0,
+            )
+
+        windows = []
+        strategy_name = self._strategy_factory().name
+        symbol = candles[0].symbol
+
+        for i in range(self._num_windows):
+            start = i * window_size
+            end = min(start + window_size, total)
+            window_candles = candles[start:end]
+
+            split = int(len(window_candles) * self._in_sample_pct)
+            in_sample = window_candles[:split]
+            out_of_sample = window_candles[split:]
+
+            if len(in_sample) < 5 or len(out_of_sample) < 5:
+                continue
+
+            # Optimize on in-sample
+            best_params = {}
+            best_profit = Decimal("-999999")
+            best_is_result = None
+
+            for params in self._param_grid:
+                strategy = self._strategy_factory()
+                strategy.configure(params)
+                engine = BacktestEngine(strategy, self._initial_balance)
+                result = engine.run(in_sample)
+                if result.profit > best_profit:
+                    best_profit = result.profit
+                    best_params = params
+                    best_is_result = result
+
+            # Validate on out-of-sample with best params
+            strategy = self._strategy_factory()
+            strategy.configure(best_params)
+            engine = BacktestEngine(strategy, self._initial_balance)
+            oos_result = engine.run(out_of_sample)
+
+            windows.append(WalkForwardWindow(
+                window_index=i,
+                in_sample_result=best_is_result,
+                out_of_sample_result=oos_result,
+                best_params=best_params,
+            ))
+
+        return WalkForwardResult(
+            strategy_name=strategy_name,
+            symbol=symbol,
+            num_windows=len(windows),
+            windows=windows,
+        )