feat(backtest): task 3c crypto orchestrator walk-forward CLI

scripts/research_crypto_proxy_orchestrator_backtest.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Generic crypto orchestrator research entrypoint (task 3c)."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from crypto_strategies.backtest.orchestrator_runner import (  # noqa: E402
+    DEFAULT_MIN_HISTORY_DAYS,
+    PROFILE_NAME,
+    SUPPORTED_PROFILES,
+    CryptoLivePoolBacktestRunner,
+)
+from scripts.run_walk_forward_backtest import run_walk_forward  # noqa: E402
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Crypto orchestrator research backtest.")
+    parser.add_argument("--profile", default=PROFILE_NAME)
+    parser.add_argument("--list-profiles", action="store_true")
+    parser.add_argument("--mode", choices=("single", "walk_forward"), default="walk_forward")
+    parser.add_argument("--synthetic-days", type=int, default=1600)
+    parser.add_argument("--json-output", type=Path)
+    args = parser.parse_args()
+
+    if args.list_profiles:
+        print(json.dumps({"profiles": sorted(SUPPORTED_PROFILES)}, indent=2))
+        return 0
+
+    if args.mode == "walk_forward":
+        payload = run_walk_forward(profile=args.profile, synthetic_days=args.synthetic_days)
+    else:
+        runner = CryptoLivePoolBacktestRunner(synthetic_days=args.synthetic_days)
+        params = {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7}
+        result = runner.run(args.profile, params)
+        payload = {
+            "profile": args.profile,
+            "metrics": {
+                "sharpe_ratio": result.sharpe_ratio,
+                "max_drawdown": result.max_drawdown,
+                "cagr": result.cagr,
+            },
+            "source": "CryptoLivePoolBacktestRunner",
+        }
+
+    text = json.dumps(payload, indent=2, sort_keys=True, default=str)
+    if args.json_output:
+        args.json_output.write_text(text + "\n")
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_walk_forward_backtest.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Run walk-forward backtests via QuantPlatformKit BacktestOrchestrator."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from crypto_strategies.backtest.orchestrator_runner import (
+    DEFAULT_MIN_HISTORY_DAYS,
+    PROFILE_NAME,
+    SUPPORTED_PROFILES,
+    CryptoLivePoolBacktestRunner,
+)
+
+DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = (
+    (date(2023, 6, 1), date(2024, 5, 31)),
+    (date(2024, 6, 1), date(2025, 5, 31)),
+)
+
+PROFILE_DEFAULTS: dict[str, dict[str, Any]] = {
+    PROFILE_NAME: {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7},
+}
+
+
+def _result_payload(item: Any) -> dict[str, Any]:
+    return {
+        "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": getattr(item, "run_id", None),
+    }
+
+
+def run_walk_forward(
+    *,
+    profile: str,
+    windows: tuple[tuple[date, date], ...] = DEFAULT_WINDOWS,
+    synthetic_days: int = 1600,
+    store_root: Path | None = None,
+) -> dict[str, Any]:
+    from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+    from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+    if profile not in SUPPORTED_PROFILES:
+        raise ValueError(f"unsupported profile={profile!r}; supported={sorted(SUPPORTED_PROFILES)}")
+
+    params = dict(PROFILE_DEFAULTS.get(profile, {"min_history_days": DEFAULT_MIN_HISTORY_DAYS}))
+    runner = CryptoLivePoolBacktestRunner(synthetic_days=synthetic_days)
+    store = PerformanceStore(local_root=store_root or Path("/tmp/crypto_wf_store"))
+    orchestrator = BacktestOrchestrator(store=store)
+    orchestrator.register_runner("crypto", runner)
+
+    baseline = runner.run(profile, params)
+    wf_results = orchestrator.walk_forward(
+        profile,
+        domain="crypto",
+        params=params,
+        windows=windows,
+        param_set_id=f"{profile}_wf",
+    )
+    return {
+        "strategy_profile": profile,
+        "domain": "crypto",
+        "baseline": _result_payload(baseline),
+        "walk_forward_folds": [_result_payload(item) for item in wf_results],
+        "source": "BacktestOrchestrator.walk_forward",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Crypto walk-forward backtest via BacktestOrchestrator.")
+    parser.add_argument("--profile", default=PROFILE_NAME)
+    parser.add_argument("--list-profiles", action="store_true")
+    parser.add_argument("--json-output", type=Path)
+    parser.add_argument("--synthetic-days", type=int, default=1600)
+    parser.add_argument("--store-root", type=Path)
+    args = parser.parse_args()
+
+    if args.list_profiles:
+        print(json.dumps({"profiles": sorted(SUPPORTED_PROFILES)}, indent=2))
+        return 0
+
+    payload = run_walk_forward(
+        profile=args.profile,
+        synthetic_days=args.synthetic_days,
+        store_root=args.store_root,
+    )
+    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)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/crypto_strategies/backtest/live_pool_simulator.py

@@ -0,0 +1,88 @@
+"""Minimal synthetic live-pool rotation backtest for orchestrator integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import numpy as np
+import pandas as pd
+
+
+@dataclass(frozen=True)
+class LivePoolBacktestResult:
+    metrics: dict[str, float]
+    returns: pd.Series
+
+
+def _performance_metrics(returns: pd.Series) -> dict[str, float]:
+    clean = returns.dropna()
+    if clean.empty:
+        return {
+            "CAGR": 0.0,
+            "Max Drawdown": 0.0,
+            "Sharpe": 0.0,
+            "Annualized Volatility": 0.0,
+            "Win Rate": 0.0,
+            "Trading Days": 0.0,
+        }
+    equity = (1.0 + clean).cumprod()
+    total_return = float(equity.iloc[-1] - 1.0)
+    years = max(len(clean) / 365.25, 1.0 / 365.25)
+    cagr = float((equity.iloc[-1]) ** (1.0 / years) - 1.0) if equity.iloc[-1] > 0 else 0.0
+    vol = float(clean.std(ddof=0) * np.sqrt(365.25))
+    sharpe = float((clean.mean() * 365.25) / vol) if vol > 0 else 0.0
+    drawdown = float((equity / equity.cummax() - 1.0).min())
+    win_rate = float((clean > 0).mean())
+    return {
+        "CAGR": cagr,
+        "Max Drawdown": drawdown,
+        "Sharpe": sharpe,
+        "Annualized Volatility": vol,
+        "Win Rate": win_rate,
+        "Trading Days": float(len(clean)),
+        "total_return": total_return,
+    }
+
+
+def run_live_pool_rotation_backtest(
+    panel: pd.DataFrame,
+    *,
+    score_column: str = "final_score",
+    top_n: int = 2,
+    rebalance_every: int = 7,
+) -> LivePoolBacktestResult:
+    """Long-only equal-weight top-N rotation on a synthetic/live panel."""
+    dates = sorted(panel.index.get_level_values("date").unique())
+    symbols = sorted(panel.loc[panel["in_universe"]].index.get_level_values("symbol").unique())
+    if not dates or not symbols:
+        empty = pd.Series(dtype=float)
+        return LivePoolBacktestResult(metrics=_performance_metrics(empty), returns=empty)
+
+    open_matrix = (
+        panel.reset_index()
+        .pivot(index="date", columns="symbol", values="open")
+        .reindex(index=dates, columns=symbols)
+        .astype(float)
+    )
+    open_returns = open_matrix.shift(-1).div(open_matrix).sub(1.0).fillna(0.0)
+
+    weights = pd.Series(0.0, index=symbols, dtype=float)
+    daily_returns: list[float] = []
+    for idx, day in enumerate(dates[:-1]):
+        if idx % rebalance_every == 0:
+            snapshot = panel.xs(day, level="date")
+            ranked = (
+                snapshot[snapshot["in_universe"]]
+                .sort_values(score_column, ascending=False)
+                .head(top_n)
+            )
+            weights[:] = 0.0
+            if not ranked.empty:
+                weight = 1.0 / len(ranked)
+                for symbol in ranked.index:
+                    weights.loc[symbol] = weight
+        daily_returns.append(float((weights * open_returns.loc[day]).sum()))
+
+    returns = pd.Series(daily_returns, index=pd.DatetimeIndex(dates[:-1]))
+    return LivePoolBacktestResult(metrics=_performance_metrics(returns), returns=returns)

src/crypto_strategies/backtest/orchestrator_runner.py

@@ -0,0 +1,135 @@
+"""BacktestRunner adapter for crypto live pool rotation."""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from typing import Any, Mapping
+
+import numpy as np
+import pandas as pd
+
+from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest
+
+try:
+    from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+except ImportError:  # pragma: no cover
+    BacktestResult = None  # type: ignore[misc, assignment]
+
+
+PROFILE_NAME = "crypto_live_pool_rotation"
+DEFAULT_MIN_HISTORY_DAYS = 120
+SUPPORTED_PROFILES = frozenset({PROFILE_NAME})
+
+
+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:
+    level_dates = panel.index.get_level_values("date")
+    frame = panel
+    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_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 BacktestResult 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 BacktestResult(
+        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 0),
+        source_script="crypto_strategies.backtest.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_live_pool_rotation_backtest(
+            sliced,
+            top_n=int(params.get("top_n", 2)),
+            rebalance_every=int(params.get("rebalance_every", 7)),
+        )
+        elapsed = (datetime.now(timezone.utc) - started).total_seconds()
+        eval_dates = sliced.index.get_level_values("date")
+        return _metrics_to_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"]