diff --git a/scripts/research_crypto_proxy_orchestrator_backtest.py b/scripts/research_crypto_proxy_orchestrator_backtest.py new file mode 100644 index 0000000..e229c58 --- /dev/null +++ b/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()) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py new file mode 100644 index 0000000..1ef8c7b --- /dev/null +++ b/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()) diff --git a/src/crypto_strategies/backtest/__init__.py b/src/crypto_strategies/backtest/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/crypto_strategies/backtest/live_pool_simulator.py b/src/crypto_strategies/backtest/live_pool_simulator.py new file mode 100644 index 0000000..f3f8301 --- /dev/null +++ b/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) diff --git a/src/crypto_strategies/backtest/orchestrator_runner.py b/src/crypto_strategies/backtest/orchestrator_runner.py new file mode 100644 index 0000000..cdb1f06 --- /dev/null +++ b/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"] diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py new file mode 100644 index 0000000..c21999d --- /dev/null +++ b/tests/test_orchestrator_runner.py @@ -0,0 +1,55 @@ +"""Tests for CryptoLivePoolBacktestRunner + BacktestOrchestrator integration.""" + +from __future__ import annotations + +import tempfile +import unittest +from datetime import date +from pathlib import Path + +from crypto_strategies.backtest.orchestrator_runner import ( + 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, + {"min_history_days": 120, "top_n": 2, "rebalance_every": 7}, + start_date=date(2023, 6, 1), + end_date=date(2024, 6, 1), + ) + self.assertEqual(result.strategy_profile, PROFILE_NAME) + self.assertEqual(result.domain, "crypto") + self.assertGreater(result.observation_count, 0) + + 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={"min_history_days": 120, "top_n": 2, "rebalance_every": 7}, + windows=windows, + ) + self.assertEqual(len(results), 2) + + +if __name__ == "__main__": + unittest.main()