feat(backtest): CN index ETF walk-forward pilot via BacktestOrchestrator

scripts/run_cn_index_etf_walk_forward_pilot.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Pilot: run cn_index_etf_tactical_rotation through BacktestOrchestrator.walk_forward()."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from cn_equity_strategies.backtest.orchestrator_runner import CnProxyBacktestRunner
+from cn_equity_strategies.strategies.cn_index_etf_tactical_rotation import (
+    DEFAULT_MIN_HISTORY_DAYS,
+    PROFILE_NAME,
+)
+
+DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = (
+    (date(2023, 6, 1), date(2024, 5, 31)),
+    (date(2024, 6, 1), date(2025, 5, 31)),
+    (date(2025, 6, 1), date(2026, 3, 31)),
+)
+
+
+def _baseline_full_window(runner: CnProxyBacktestRunner, params: dict[str, Any]) -> dict[str, Any]:
+    result = runner.run(PROFILE_NAME, params, start_date=None, end_date=None)
+    return {
+        "sharpe_ratio": result.sharpe_ratio,
+        "max_drawdown": result.max_drawdown,
+        "cagr": result.cagr,
+        "total_return": result.total_return,
+        "observation_count": result.observation_count,
+    }
+
+
+def run_pilot(*, compare_tolerance: float = 0.001) -> dict[str, Any]:
+    from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+    from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+    params = {"min_history_days": DEFAULT_MIN_HISTORY_DAYS}
+    runner = CnProxyBacktestRunner(synthetic_days=900)
+    baseline = _baseline_full_window(runner, params)
+
+    store = PerformanceStore(local_root=Path("/tmp/cn_equity_wf_pilot_store"))
+    orchestrator = BacktestOrchestrator(store=store)
+    orchestrator.register_runner("cn_equity", runner)
+    wf_results = orchestrator.walk_forward(
+        PROFILE_NAME,
+        domain="cn_equity",
+        params=params,
+        windows=DEFAULT_WINDOWS,
+        param_set_id="cn_index_etf_wf_pilot",
+    )
+
+    folds = [
+        {
+            "start_date": item.start_date.isoformat() if item.start_date else None,
+            "end_date": item.end_date.isoformat() if item.end_date else None,
+            "sharpe_ratio": item.sharpe_ratio,
+            "max_drawdown": item.max_drawdown,
+            "cagr": item.cagr,
+            "total_return": item.total_return,
+            "observation_count": item.observation_count,
+            "run_id": item.run_id,
+        }
+        for item in wf_results
+    ]
+    # Compare average fold CAGR/Sharpe against full-window baseline order of magnitude.
+    # For synthetic data we only assert walk_forward returns a non-empty list and
+    # that a full-window orchestrator.run stays within tolerance of direct runner.
+    direct = baseline
+    via_orch = orchestrator.run(
+        PROFILE_NAME,
+        domain="cn_equity",
+        params=params,
+        param_set_id="cn_index_etf_full_compare",
+    )
+    sharpe_delta = abs(float(via_orch.sharpe_ratio or 0.0) - float(direct["sharpe_ratio"] or 0.0))
+    mdd_delta = abs(float(via_orch.max_drawdown or 0.0) - float(direct["max_drawdown"] or 0.0))
+    within_tolerance = sharpe_delta <= compare_tolerance and mdd_delta <= compare_tolerance
+
+    return {
+        "strategy_profile": PROFILE_NAME,
+        "domain": "cn_equity",
+        "baseline": baseline,
+        "orchestrator_full_window": {
+            "sharpe_ratio": via_orch.sharpe_ratio,
+            "max_drawdown": via_orch.max_drawdown,
+            "cagr": via_orch.cagr,
+            "total_return": via_orch.total_return,
+            "observation_count": via_orch.observation_count,
+            "run_id": via_orch.run_id,
+        },
+        "walk_forward_folds": folds,
+        "compare": {
+            "sharpe_delta": sharpe_delta,
+            "max_drawdown_delta": mdd_delta,
+            "tolerance": compare_tolerance,
+            "within_tolerance": within_tolerance,
+        },
+        "source": "BacktestOrchestrator.walk_forward",
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="CN index ETF walk-forward pilot via BacktestOrchestrator.")
+    parser.add_argument("--json-output", type=Path)
+    parser.add_argument("--tolerance", type=float, default=0.001)
+    args = parser.parse_args()
+    payload = run_pilot(compare_tolerance=args.tolerance)
+    text = json.dumps(payload, indent=2, sort_keys=True, default=str)
+    if args.json_output:
+        args.json_output.parent.mkdir(parents=True, exist_ok=True)
+        args.json_output.write_text(text + "\n")
+    print(text)
+    if not payload["compare"]["within_tolerance"]:
+        raise SystemExit("orchestrator vs direct runner comparison exceeded tolerance")
+
+
+if __name__ == "__main__":
+    main()

src/cn_equity_strategies/backtest/orchestrator_runner.py

@@ -0,0 +1,188 @@
+"""BacktestRunner adapter that wraps CN proxy backtests for QuantPlatformKit."""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from typing import Any, Mapping
+
+import pandas as pd
+
+from cn_equity_strategies.backtest.proxy_simulator import ProxyBacktestConfig, run_proxy_backtest
+from cn_equity_strategies.strategies.cn_index_etf_tactical_rotation import (
+    DEFAULT_MIN_HISTORY_DAYS,
+    PROFILE_NAME as CN_INDEX_ETF_TACTICAL_ROTATION_PROFILE,
+    build_target_weights,
+    extract_managed_symbols,
+)
+
+try:
+    from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+except ImportError:  # pragma: no cover - exercised only without QPK installed
+    BacktestResult = None  # type: ignore[misc, assignment]
+
+
+SUPPORTED_PROFILES = frozenset({CN_INDEX_ETF_TACTICAL_ROTATION_PROFILE})
+
+
+def _synthetic_market_history(*, days: int = 900, start: str = "2022-01-03") -> pd.DataFrame:
+    dates = pd.bdate_range(start, periods=days)
+    symbols = tuple(extract_managed_symbols())
+    rates = {symbol: 1.0002 + (idx * 0.00005) for idx, symbol in enumerate(symbols)}
+    rows: list[dict[str, object]] = []
+    for symbol in symbols:
+        price = 10.0 + (hash(symbol) % 7)
+        rate = rates.get(symbol, 1.0002)
+        for idx, day in enumerate(dates):
+            price *= rate
+            close = price * (1.0 + 0.03 * ((idx % 7) - 3) / 7)
+            rows.append({"date": day, "symbol": symbol, "close": close})
+    return pd.DataFrame(rows)
+
+
+def _slice_history(
+    market_history: pd.DataFrame,
+    *,
+    start_date: date | None,
+    end_date: date | None,
+    lookback_days: int = 0,
+) -> pd.DataFrame:
+    frame = market_history.copy()
+    frame["date"] = pd.to_datetime(frame["date"], utc=False).dt.tz_localize(None).dt.normalize()
+    if start_date is not None:
+        # Keep lookback before the evaluation window so warm-up indicators can form.
+        effective_start = pd.Timestamp(start_date) - pd.tseries.offsets.BDay(max(int(lookback_days), 0))
+        frame = frame[frame["date"] >= effective_start]
+    if end_date is not None:
+        frame = frame[frame["date"] <= pd.Timestamp(end_date)]
+    return frame.sort_values(["date", "symbol"]).reset_index(drop=True)
+
+
+def _signal_fn(history: Any, **kwargs: Any):
+    return build_target_weights(history, **kwargs)
+
+
+def _metrics_to_backtest_result(
+    *,
+    strategy_profile: str,
+    params: Mapping[str, Any],
+    metrics: Mapping[str, Any],
+    start_date: date | None,
+    end_date: date | None,
+) -> Any:
+    if BacktestResult is None:
+        raise ImportError("quant_platform_kit is required to build BacktestResult")
+    days = int(metrics.get("days") or 0)
+    annual_return = float(metrics.get("annual_return") or 0.0)
+    max_drawdown = float(metrics.get("max_drawdown") or 0.0)
+    annual_vol = float(metrics.get("annual_volatility") or 0.0)
+    sharpe = float(metrics.get("sharpe_ratio") or 0.0)
+    total_return = float(metrics.get("total_return") or 0.0)
+    calmar = abs(annual_return / max_drawdown) if max_drawdown else None
+    return BacktestResult(
+        strategy_profile=strategy_profile,
+        domain="cn_equity",
+        param_set_id="",
+        params=dict(params),
+        sharpe_ratio=sharpe,
+        calmar_ratio=calmar,
+        max_drawdown=max_drawdown,
+        cagr=annual_return,
+        volatility=annual_vol,
+        total_return=total_return,
+        start_date=start_date,
+        end_date=end_date,
+        observation_count=days,
+        source_script="cn_equity_strategies.backtest.orchestrator_runner",
+        computed_at=datetime.now(timezone.utc).isoformat(),
+    )
+
+
+class CnProxyBacktestRunner:
+    """Protocol-compatible BacktestRunner for CN ETF proxy strategies."""
+
+    def __init__(
+        self,
+        *,
+        market_history: pd.DataFrame | None = None,
+        initial_cash: float = 1_000_000.0,
+        synthetic_days: int = 500,
+    ) -> None:
+        self._market_history = market_history
+        self._initial_cash = float(initial_cash)
+        self._synthetic_days = int(synthetic_days)
+
+    def run(
+        self,
+        strategy_profile: str,
+        params: Mapping[str, Any],
+        start_date: date | None = None,
+        end_date: date | None = None,
+    ) -> Any:
+        if strategy_profile not in SUPPORTED_PROFILES:
+            raise ValueError(
+                f"Unsupported strategy_profile={strategy_profile!r}; "
+                f"supported={sorted(SUPPORTED_PROFILES)}"
+            )
+
+        min_history_days = int(params.get("min_history_days", DEFAULT_MIN_HISTORY_DAYS))
+        history = self._market_history
+        if history is None:
+            # Ensure synthetic history covers lookback + each fold window.
+            history = _synthetic_market_history(days=max(self._synthetic_days, min_history_days + 400))
+        # +5 business-day buffer: proxy simulator signals on the day *before*
+        # the first eligible rebalance, so exact min_history_days is one short.
+        sliced = _slice_history(
+            history,
+            start_date=start_date,
+            end_date=end_date,
+            lookback_days=min_history_days + 5,
+        )
+        if sliced.empty:
+            raise ValueError("No market history rows for requested window")
+
+        started = datetime.now(timezone.utc)
+        result = run_proxy_backtest(
+            sliced,
+            _signal_fn,
+            config=ProxyBacktestConfig(
+                initial_cash=self._initial_cash,
+                min_history_days=min_history_days,
+            ),
+            strategy_kwargs={"min_history_days": min_history_days},
+        )
+        elapsed = (datetime.now(timezone.utc) - started).total_seconds()
+        eval_frame = sliced
+        if start_date is not None:
+            eval_frame = sliced[sliced["date"] >= pd.Timestamp(start_date)]
+        payload = _metrics_to_backtest_result(
+            strategy_profile=strategy_profile,
+            params=params,
+            metrics=result.metrics,
+            start_date=start_date or (eval_frame["date"].min().date() if not eval_frame.empty else None),
+            end_date=end_date or (eval_frame["date"].max().date() if not eval_frame.empty else None),
+        )
+        # BacktestResult is frozen; rebuild with duration via replace-like path
+        return BacktestResult(
+            strategy_profile=payload.strategy_profile,
+            domain=payload.domain,
+            param_set_id=payload.param_set_id,
+            params=dict(payload.params),
+            sharpe_ratio=payload.sharpe_ratio,
+            calmar_ratio=payload.calmar_ratio,
+            max_drawdown=payload.max_drawdown,
+            cagr=payload.cagr,
+            volatility=payload.volatility,
+            total_return=payload.total_return,
+            start_date=payload.start_date,
+            end_date=payload.end_date,
+            observation_count=payload.observation_count,
+            source_script=payload.source_script,
+            computed_at=payload.computed_at,
+            run_duration_seconds=elapsed,
+        )
+
+
+__all__ = [
+    "SUPPORTED_PROFILES",
+    "CnProxyBacktestRunner",
+]

tests/test_orchestrator_runner.py

@@ -0,0 +1,64 @@
+"""Tests for CnProxyBacktestRunner + BacktestOrchestrator integration."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import date
+from pathlib import Path
+
+from cn_equity_strategies.backtest.orchestrator_runner import CnProxyBacktestRunner, SUPPORTED_PROFILES
+from cn_equity_strategies.strategies.cn_index_etf_tactical_rotation import (
+    DEFAULT_MIN_HISTORY_DAYS,
+    PROFILE_NAME,
+)
+
+
+class CnProxyBacktestRunnerTests(unittest.TestCase):
+    def test_supported_profile_includes_index_etf(self) -> None:
+        self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES)
+
+    def test_run_returns_backtest_result(self) -> None:
+        runner = CnProxyBacktestRunner(synthetic_days=400)
+        result = runner.run(
+            PROFILE_NAME,
+            {"min_history_days": DEFAULT_MIN_HISTORY_DAYS},
+            start_date=date(2023, 6, 1),
+            end_date=date(2024, 6, 1),
+        )
+        self.assertEqual(result.strategy_profile, PROFILE_NAME)
+        self.assertEqual(result.domain, "cn_equity")
+        self.assertIsNotNone(result.sharpe_ratio)
+        self.assertGreater(result.observation_count, 0)
+
+    def test_unsupported_profile_raises(self) -> None:
+        runner = CnProxyBacktestRunner(synthetic_days=100)
+        with self.assertRaises(ValueError):
+            runner.run("unknown_profile", {})
+
+
+class WalkForwardPilotTests(unittest.TestCase):
+    def test_walk_forward_produces_one_result_per_window(self) -> None:
+        from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+        from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            orchestrator = BacktestOrchestrator(store=store)
+            orchestrator.register_runner("cn_equity", CnProxyBacktestRunner(synthetic_days=700))
+            windows = (
+                (date(2023, 6, 1), date(2023, 12, 31)),
+                (date(2024, 1, 1), date(2024, 6, 30)),
+            )
+            results = orchestrator.walk_forward(
+                PROFILE_NAME,
+                domain="cn_equity",
+                params={"min_history_days": DEFAULT_MIN_HISTORY_DAYS},
+                windows=windows,
+            )
+            self.assertEqual(len(results), 2)
+            self.assertTrue(all(item.strategy_profile == PROFILE_NAME for item in results))
+
+
+if __name__ == "__main__":
+    unittest.main()