feat(strategy_lifecycle): add walk-forward and sensitivity to BacktestOrchestrator (#192)

Extend BacktestOrchestrator with walk_forward() for multi-window runs and
sensitivity() for basic param-grid sweeps, plus SensitivityReport contract
and unit tests covering run, walk_forward, and sensitivity paths.

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 3 people

src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py

@@ -5,11 +5,12 @@
 
 from __future__ import annotations
 
+import itertools
 import uuid
 from datetime import date, datetime, timezone
-from typing import Any, Mapping, Protocol, runtime_checkable
+from typing import Any, Mapping, Protocol, Sequence, runtime_checkable
 
-from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult, SensitivityReport
 from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
 
 
@@ -140,3 +141,120 @@ def run(
     def run_latest(self, strategy_profile: str, *, domain: str) -> BacktestResult | None:
         """Load the latest persisted backtest result for a strategy."""
         return self._store.load_latest_backtest(domain, strategy_profile)
+
+    def walk_forward(
+        self,
+        strategy_profile: str,
+        *,
+        domain: str,
+        params: Mapping[str, Any],
+        windows: Sequence[tuple[date | None, date | None]],
+        param_set_id: str = "",
+        param_version: int = 1,
+    ) -> list[BacktestResult]:
+        """Run backtests across multiple time windows.
+
+        Args:
+            strategy_profile: Canonical strategy profile.
+            domain: Market domain.
+            params: Strategy parameters shared across windows.
+            windows: Sequence of (start_date, end_date) pairs, one per fold.
+            param_set_id: Base identifier for parameter set metadata.
+            param_version: Version number for this parameter set.
+
+        Returns:
+            One BacktestResult per window, in order.
+
+        Raises:
+            ValueError: If windows is empty or no runner is registered.
+        """
+        if not windows:
+            raise ValueError("windows must contain at least one (start_date, end_date) pair")
+
+        base_id = param_set_id or _run_id()
+        results: list[BacktestResult] = []
+        for idx, (start_date, end_date) in enumerate(windows):
+            results.append(
+                self.run(
+                    strategy_profile,
+                    domain=domain,
+                    params=params,
+                    param_set_id=f"{base_id}_wf{idx}",
+                    param_version=param_version,
+                    start_date=start_date,
+                    end_date=end_date,
+                )
+            )
+        return results
+
+    def sensitivity(
+        self,
+        strategy_profile: str,
+        *,
+        domain: str,
+        base_params: Mapping[str, Any],
+        param_ranges: Mapping[str, Sequence[Any]],
+        start_date: date | None = None,
+        end_date: date | None = None,
+        max_combinations: int = 500,
+    ) -> SensitivityReport:
+        """Run a basic parameter-grid sensitivity sweep.
+
+        Expands ``param_ranges`` via Cartesian product, merging each combination
+        onto ``base_params``, and runs one backtest per combination.
+
+        Args:
+            strategy_profile: Canonical strategy profile.
+            domain: Market domain.
+            base_params: Fixed parameters applied to every combination.
+            param_ranges: Per-parameter value lists to sweep.
+            start_date: Backtest start date.
+            end_date: Backtest end date.
+            max_combinations: Upper bound on grid size (subsamples if exceeded).
+
+        Returns:
+            SensitivityReport with one BacktestResult per combination tried.
+
+        Raises:
+            ValueError: If param_ranges is empty or no runner is registered.
+        """
+        if not param_ranges:
+            raise ValueError("param_ranges must contain at least one parameter dimension")
+
+        keys = sorted(param_ranges.keys())
+        value_lists = [list(param_ranges[k]) for k in keys]
+        total = 1
+        for values in value_lists:
+            total *= len(values)
+
+        combos: list[dict[str, Any]] = []
+        for idx, combo in enumerate(itertools.product(*value_lists)):
+            if total > max_combinations and idx % max(1, total // max_combinations) != 0:
+                continue
+            merged = dict(base_params)
+            merged.update(dict(zip(keys, combo)))
+            combos.append(merged)
+            if len(combos) >= max_combinations:
+                break
+
+        results: list[BacktestResult] = []
+        for idx, combo_params in enumerate(combos):
+            results.append(
+                self.run(
+                    strategy_profile,
+                    domain=domain,
+                    params=combo_params,
+                    param_set_id=f"{strategy_profile}_sens_{idx}",
+                    param_version=1,
+                    start_date=start_date,
+                    end_date=end_date,
+                )
+            )
+
+        return SensitivityReport(
+            strategy_profile=strategy_profile,
+            domain=domain,
+            base_params=dict(base_params),
+            results=tuple(results),
+            combination_count=len(results),
+        )

src/quant_platform_kit/strategy_lifecycle/contracts.py

@@ -282,6 +282,26 @@ def to_dict(self) -> dict[str, object]:
         }
 
 
+@dataclass(frozen=True)
+class SensitivityReport:
+    """Results from a parameter-grid sensitivity sweep."""
+
+    strategy_profile: str
+    domain: str
+    base_params: Mapping[str, Any]
+    results: tuple[BacktestResult, ...] = ()
+    combination_count: int = 0
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "strategy_profile": self.strategy_profile,
+            "domain": self.domain,
+            "base_params": dict(self.base_params),
+            "combination_count": self.combination_count,
+            "results": [r.to_dict() for r in self.results],
+        }
+
+
 @dataclass(frozen=True)
 class ParamSearchSpace:
     """Definition of the search space for one strategy's parameters."""

tests/test_backtest_orchestrator.py

@@ -0,0 +1,162 @@
+"""Tests for strategy_lifecycle.backtest_orchestrator."""
+
+from __future__ import annotations
+
+from datetime import date
+import tempfile
+import unittest
+from pathlib import Path
+from typing import Any, Mapping
+
+from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+
+class _RecordingRunner:
+    """Mock BacktestRunner that records calls and returns deterministic metrics."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def run(
+        self,
+        strategy_profile: str,
+        params: Mapping[str, Any],
+        start_date: date | None = None,
+        end_date: date | None = None,
+    ) -> BacktestResult:
+        self.calls.append(
+            {
+                "strategy_profile": strategy_profile,
+                "params": dict(params),
+                "start_date": start_date,
+                "end_date": end_date,
+            }
+        )
+        lookback = int(params.get("lookback", 20))
+        sharpe = 1.0 + lookback * 0.01
+        return BacktestResult(
+            strategy_profile=strategy_profile,
+            domain="us_equity",
+            param_set_id="mock",
+            params=dict(params),
+            sharpe_ratio=sharpe,
+            cagr=0.12,
+            max_drawdown=-0.08,
+            start_date=start_date,
+            end_date=end_date,
+            observation_count=252,
+        )
+
+
+class BacktestOrchestratorTests(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = PerformanceStore(local_root=Path(self.tmp.name))
+        self.orchestrator = BacktestOrchestrator(store=self.store)
+        self.runner = _RecordingRunner()
+        self.orchestrator.register_runner("us_equity", self.runner)
+
+    def tearDown(self) -> None:
+        self.tmp.cleanup()
+
+    def test_run_enriches_and_persists(self) -> None:
+        result = self.orchestrator.run(
+            "test_strat",
+            domain="us_equity",
+            params={"lookback": 30},
+            start_date=date(2020, 1, 1),
+            end_date=date(2024, 12, 31),
+        )
+        self.assertEqual(result.strategy_profile, "test_strat")
+        self.assertEqual(result.domain, "us_equity")
+        self.assertEqual(result.params, {"lookback": 30})
+        self.assertAlmostEqual(result.sharpe_ratio, 1.3)
+        self.assertTrue(result.run_id)
+        self.assertTrue(result.computed_at)
+        self.assertEqual(result.source_script, "backtest_orchestrator")
+
+    def test_run_raises_without_runner(self) -> None:
+        with self.assertRaises(ValueError):
+            self.orchestrator.run("test_strat", domain="cn_equity", params={})
+
+    def test_walk_forward_runs_each_window(self) -> None:
+        windows = [
+            (date(2020, 1, 1), date(2021, 12, 31)),
+            (date(2022, 1, 1), date(2023, 12, 31)),
+            (date(2024, 1, 1), date(2024, 12, 31)),
+        ]
+        results = self.orchestrator.walk_forward(
+            "test_strat",
+            domain="us_equity",
+            params={"lookback": 20},
+            windows=windows,
+            param_set_id="wf_test",
+        )
+        self.assertEqual(len(results), 3)
+        self.assertEqual(len(self.runner.calls), 3)
+        for idx, (result, window) in enumerate(zip(results, windows)):
+            self.assertEqual(result.start_date, window[0])
+            self.assertEqual(result.end_date, window[1])
+            self.assertEqual(result.param_set_id, f"wf_test_wf{idx}")
+        self.assertEqual(
+            [self.runner.calls[i]["start_date"] for i in range(3)],
+            [w[0] for w in windows],
+        )
+
+    def test_walk_forward_empty_windows_raises(self) -> None:
+        with self.assertRaises(ValueError):
+            self.orchestrator.walk_forward(
+                "test_strat",
+                domain="us_equity",
+                params={},
+                windows=[],
+            )
+
+    def test_sensitivity_runs_param_grid(self) -> None:
+        report = self.orchestrator.sensitivity(
+            "test_strat",
+            domain="us_equity",
+            base_params={"top_n": 2},
+            param_ranges={"lookback": [10, 20, 30]},
+            start_date=date(2020, 1, 1),
+            end_date=date(2024, 12, 31),
+        )
+        self.assertEqual(report.combination_count, 3)
+        self.assertEqual(len(report.results), 3)
+        self.assertEqual(report.strategy_profile, "test_strat")
+        self.assertEqual(report.base_params, {"top_n": 2})
+        sharpes = [r.sharpe_ratio for r in report.results]
+        self.assertEqual(sharpes, [1.1, 1.2, 1.3])
+        for call, result in zip(self.runner.calls, report.results):
+            self.assertEqual(call["params"]["top_n"], 2)
+            self.assertIn("lookback", call["params"])
+            self.assertEqual(result.params["top_n"], 2)
+
+    def test_sensitivity_two_dim_grid(self) -> None:
+        report = self.orchestrator.sensitivity(
+            "test_strat",
+            domain="us_equity",
+            base_params={},
+            param_ranges={"lookback": [10, 20], "top_n": [2, 3]},
+        )
+        self.assertEqual(report.combination_count, 4)
+        lookbacks = sorted({r.params["lookback"] for r in report.results})
+        top_ns = sorted({r.params["top_n"] for r in report.results})
+        self.assertEqual(lookbacks, [10, 20])
+        self.assertEqual(top_ns, [2, 3])
+
+    def test_sensitivity_empty_ranges_raises(self) -> None:
+        with self.assertRaises(ValueError):
+            self.orchestrator.sensitivity(
+                "test_strat",
+                domain="us_equity",
+                base_params={},
+                param_ranges={},
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()