feat(backtest): crypto live pool walk_forward pilot (3c)

pyproject.toml

@@ -9,7 +9,7 @@ description = "Live-pool rotation pipelines for crypto strategy runtime compatib
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@37c81901160c5b31127a27dba1c63944933fb6bf",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@0c69df08144872ccd1d8bf523738e80748d8d664",
   "pandas==3.0.3",
   "numpy==2.4.6",
   "requests==2.34.2",

requirements-lock.txt

@@ -1,4 +1,4 @@
-quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@37c81901160c5b31127a27dba1c63944933fb6bf
+quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@0c69df08144872ccd1d8bf523738e80748d8d664
 pandas==3.0.3
 numpy==2.4.6
 requests==2.34.2

requirements.txt

@@ -1,4 +1,4 @@
-quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@37c81901160c5b31127a27dba1c63944933fb6bf
+quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@0c69df08144872ccd1d8bf523738e80748d8d664
 pandas>=3.0.3
 numpy>=2.4.6,<2.5
 requests>=2.34.2

scripts/run_crypto_live_pool_walk_forward_pilot.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Pilot: run crypto_live_pool_rotation through BacktestOrchestrator.walk_forward()."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import date
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from src.strategy_lifecycle.orchestrator_runner import (  # noqa: E402
+    CryptoLivePoolBacktestRunner,
+    PROFILE_NAME,
+)
+
+DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = (
+    (date(2023, 6, 1), date(2024, 5, 31)),
+    (date(2024, 6, 1), date(2025, 5, 31)),
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Crypto live pool walk-forward pilot")
+    parser.add_argument("--output", type=Path, default=Path("crypto_live_pool_walk_forward_pilot.json"))
+    parser.add_argument("--synthetic-days", type=int, default=400)
+    args = parser.parse_args()
+
+    from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+    from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+    runner = CryptoLivePoolBacktestRunner(synthetic_days=args.synthetic_days)
+    params: dict[str, object] = {}
+    store = PerformanceStore(local_root=args.output.parent / ".wf_store")
+    orchestrator = BacktestOrchestrator(store=store)
+    orchestrator.register_runner("crypto", runner)
+
+    baseline = runner.run(PROFILE_NAME, params)
+    results = orchestrator.walk_forward(
+        PROFILE_NAME,
+        domain="crypto",
+        params=params,
+        windows=DEFAULT_WINDOWS,
+        param_set_id="crypto_live_pool_wf_pilot",
+    )
+    payload = {
+        "profile": PROFILE_NAME,
+        "baseline": {
+            "sharpe_ratio": baseline.sharpe_ratio,
+            "max_drawdown": baseline.max_drawdown,
+            "cagr": baseline.cagr,
+        },
+        "windows": [
+            {
+                "start": item.start_date.isoformat() if item.start_date else None,
+                "end": item.end_date.isoformat() if item.end_date else None,
+                "sharpe_ratio": item.sharpe_ratio,
+                "max_drawdown": item.max_drawdown,
+                "cagr": item.cagr,
+            }
+            for item in results
+        ],
+    }
+    args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/strategy_lifecycle/orchestrator_runner.py

@@ -0,0 +1,144 @@
+"""BacktestRunner adapter for crypto live pool rotation orchestrator integration."""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from typing import Any, Mapping
+
+import numpy as np
+import pandas as pd
+
+from src.backtest import run_single_backtest
+
+try:
+    from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult as QpkBacktestResult
+except ImportError:  # pragma: no cover
+    QpkBacktestResult = None  # type: ignore[misc, assignment]
+
+
+PROFILE_NAME = "crypto_live_pool_rotation"
+DEFAULT_MIN_HISTORY_DAYS = 120
+SUPPORTED_PROFILES = frozenset({PROFILE_NAME})
+
+DEFAULT_BACKTEST_CONFIG: dict[str, Any] = {
+    "strategy": {
+        "rebalance_frequency": "weekly",
+        "top_n": 2,
+        "weighting": "equal",
+        "signal_lag_days": 1,
+        "fee_bps": 10,
+        "slippage_bps": 5,
+    }
+}
+
+
+def _synthetic_panel(*, days: int = 1500, symbols: tuple[str, ...] = ("BTCUSDT", "ETHUSDT", "SOLUSDT")) -> pd.DataFrame:
+    dates = pd.date_range("2020-01-01", periods=days, freq="D")
+    index = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"])
+    panel = pd.DataFrame(index=index)
+    panel["in_universe"] = True
+    rng = np.random.default_rng(42)
+    rows: list[float] = []
+    for symbol in symbols:
+        price = 100.0 + hash(symbol) % 50
+        for _ in dates:
+            price *= 1.0 + float(rng.normal(0.001, 0.02))
+            rows.append(price)
+    panel["open"] = rows
+    scores: list[float] = []
+    for day_idx, _day in enumerate(dates):
+        for sym_idx, symbol in enumerate(symbols):
+            scores.append(float((day_idx + sym_idx * 17 + hash(symbol) % 11) % 100) / 100.0)
+    panel["final_score"] = scores
+    return panel.sort_index()
+
+
+def _slice_panel(panel: pd.DataFrame, *, start_date: date | None, end_date: date | None) -> pd.DataFrame:
+    frame = panel
+    level_dates = frame.index.get_level_values("date")
+    if start_date is not None:
+        frame = frame.loc[level_dates >= pd.Timestamp(start_date)]
+        level_dates = frame.index.get_level_values("date")
+    if end_date is not None:
+        frame = frame.loc[level_dates <= pd.Timestamp(end_date)]
+    return frame.sort_index()
+
+
+def _metrics_to_qpk_result(
+    *,
+    strategy_profile: str,
+    params: Mapping[str, Any],
+    metrics: Mapping[str, Any],
+    start_date: date | None,
+    end_date: date | None,
+    run_duration_seconds: float,
+) -> Any:
+    if QpkBacktestResult is None:
+        raise ImportError("quant_platform_kit is required to build BacktestResult")
+    cagr = float(metrics.get("CAGR") or 0.0)
+    max_drawdown = float(metrics.get("Max Drawdown") or 0.0)
+    calmar = abs(cagr / max_drawdown) if max_drawdown else None
+    return QpkBacktestResult(
+        strategy_profile=strategy_profile,
+        domain="crypto",
+        param_set_id="",
+        params=dict(params),
+        sharpe_ratio=float(metrics.get("Sharpe") or 0.0),
+        calmar_ratio=calmar,
+        max_drawdown=max_drawdown,
+        cagr=cagr,
+        volatility=float(metrics.get("Annualized Volatility") or 0.0),
+        win_rate=float(metrics.get("Win Rate") or 0.0),
+        start_date=start_date,
+        end_date=end_date,
+        observation_count=int(metrics.get("Trading Days") or metrics.get("days") or 0),
+        source_script="CryptoLivePoolPipelines.strategy_lifecycle.orchestrator_runner",
+        computed_at=datetime.now(timezone.utc).isoformat(),
+        run_duration_seconds=run_duration_seconds,
+    )
+
+
+class CryptoLivePoolBacktestRunner:
+    """Protocol-compatible BacktestRunner for crypto live pool rotation."""
+
+    def __init__(self, *, panel: pd.DataFrame | None = None, synthetic_days: int = 1600) -> None:
+        self._panel = panel
+        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)}"
+            )
+
+        panel = self._panel
+        if panel is None:
+            panel = _synthetic_panel(days=max(self._synthetic_days, DEFAULT_MIN_HISTORY_DAYS + 60))
+        sliced = _slice_panel(panel, start_date=start_date, end_date=end_date)
+        if sliced.empty:
+            raise ValueError("No panel rows for requested window")
+
+        started = datetime.now(timezone.utc)
+        result = run_single_backtest(sliced, "final_score", DEFAULT_BACKTEST_CONFIG)
+        elapsed = (datetime.now(timezone.utc) - started).total_seconds()
+        eval_dates = sliced.index.get_level_values("date")
+        metrics = dict(result.metrics)
+        metrics["days"] = int(len(result.returns.dropna()))
+        return _metrics_to_qpk_result(
+            strategy_profile=strategy_profile,
+            params=params,
+            metrics=result.metrics,
+            start_date=start_date or eval_dates.min().date(),
+            end_date=end_date or eval_dates.max().date(),
+            run_duration_seconds=elapsed,
+        )
+
+
+__all__ = ["PROFILE_NAME", "SUPPORTED_PROFILES", "CryptoLivePoolBacktestRunner"]

tests/test_monthly_publish_workflow_config.py

@@ -9,7 +9,7 @@
 README_ZH_PATH = PROJECT_ROOT / "README.zh-CN.md"
 QPK_DEPENDENCY = (
     "quant-platform-kit @ "
-    "git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@37c81901160c5b31127a27dba1c63944933fb6bf"
+    "git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@0c69df08144872ccd1d8bf523738e80748d8d664"
 )
 
 

tests/test_orchestrator_runner.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from datetime import date
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from src.strategy_lifecycle.orchestrator_runner import (  # noqa: E402
+    PROFILE_NAME,
+    SUPPORTED_PROFILES,
+    CryptoLivePoolBacktestRunner,
+)
+
+
+class CryptoOrchestratorRunnerTests(unittest.TestCase):
+    def test_supported_profile(self) -> None:
+        self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES)
+
+    def test_run_returns_backtest_result(self) -> None:
+        runner = CryptoLivePoolBacktestRunner(synthetic_days=1600)
+        result = runner.run(
+            PROFILE_NAME,
+            {},
+            start_date=date(2023, 6, 1),
+            end_date=date(2024, 3, 1),
+        )
+        self.assertEqual(result.strategy_profile, PROFILE_NAME)
+        self.assertEqual(result.domain, "crypto")
+        self.assertIsNotNone(result.sharpe_ratio)
+
+    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("crypto", CryptoLivePoolBacktestRunner(synthetic_days=1600))
+            windows = (
+                (date(2023, 6, 1), date(2023, 12, 31)),
+                (date(2024, 1, 1), date(2024, 6, 30)),
+            )
+            results = orchestrator.walk_forward(
+                PROFILE_NAME,
+                domain="crypto",
+                params={},
+                windows=windows,
+            )
+            self.assertEqual(len(results), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()