From ad072d8adc8f0fabe0efb30edc99ea966700cd97 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 10 Jul 2026 03:36:17 +0800 Subject: [PATCH] feat(backtest): add crypto_equity_combo orchestrator migration (wave 3) Wire CryptoEquityComboBacktestRunner, combo simulator, walk-forward CLI, and research script --orchestrator path following HK equity combo pattern. Co-Authored-By: Claude Co-authored-by: Cursor --- scripts/research_crypto_combo_backtest.py | 46 ++++- ...arch_crypto_proxy_orchestrator_backtest.py | 13 +- scripts/run_walk_forward_backtest.py | 17 +- .../backtest/combo_simulator.py | 175 ++++++++++++++++++ .../backtest/orchestrator_research.py | 99 ++++++++++ .../backtest/orchestrator_runner.py | 141 +++++++++++++- tests/test_combo_orchestrator.py | 53 ++++++ tests/test_orchestrator_runner.py | 60 +++++- 8 files changed, 592 insertions(+), 12 deletions(-) create mode 100644 src/crypto_strategies/backtest/combo_simulator.py create mode 100644 src/crypto_strategies/backtest/orchestrator_research.py create mode 100644 tests/test_combo_orchestrator.py diff --git a/scripts/research_crypto_combo_backtest.py b/scripts/research_crypto_combo_backtest.py index 5c6fdd2..e85be0d 100644 --- a/scripts/research_crypto_combo_backtest.py +++ b/scripts/research_crypto_combo_backtest.py @@ -207,12 +207,37 @@ def load_crypto_data() -> pd.DataFrame: def run_backtest( prices: pd.DataFrame, + *, + orchestrator: bool = False, ) -> dict[str, dict[str, Any]]: """Run the three-strategy backtest. Returns nested dict keyed by strategy name, each containing an equity curve DataFrame and per-period metrics. """ + if orchestrator: + from crypto_strategies.backtest.orchestrator_research import run_combo_profile_backtest + from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME + + rows = [] + for day in prices.index: + rows.append({"date": day, "symbol": "BTCUSDT", "close": float(prices.loc[day, "btc_close"])}) + rows.append({"date": day, "symbol": "ETHUSDT", "close": float(prices.loc[day, "eth_close"])}) + market_history = pd.DataFrame(rows) + payload = run_combo_profile_backtest( + PROFILE_NAME, + market_history=market_history, + params={"combo_mode": "dynamic"}, + ) + return { + "orchestrator": { + "equity": pd.Series(dtype=float), + "metrics": payload["metrics"], + "profile": payload["profile"], + "source": payload["source"], + } + } + btc_close = prices["btc_close"].dropna() eth_close = prices["eth_close"].dropna() @@ -475,6 +500,11 @@ def main() -> None: action="store_true", help="Output results as JSON to stdout", ) + parser.add_argument( + "--orchestrator", + action="store_true", + help="Thin path via CryptoEquityComboBacktestRunner (single dynamic combo window).", + ) args = parser.parse_args() print("Loading crypto price data via yfinance ...", file=sys.stderr) @@ -485,9 +515,23 @@ def main() -> None: ) print("Running backtest simulation ...", file=sys.stderr) - results = run_backtest(prices) + results = run_backtest(prices, orchestrator=args.orchestrator) print(" Done.", file=sys.stderr) + if args.orchestrator: + payload = results["orchestrator"] + text = json.dumps( + { + "profile": payload["profile"], + "metrics": payload["metrics"], + "source": payload["source"], + "orchestrator": True, + }, + indent=2, + ) + print(text) + return + if args.json_output: # Strip equity curves for JSON output (too large) json_results: dict[str, Any] = {} diff --git a/scripts/research_crypto_proxy_orchestrator_backtest.py b/scripts/research_crypto_proxy_orchestrator_backtest.py index e229c58..c79f58b 100644 --- a/scripts/research_crypto_proxy_orchestrator_backtest.py +++ b/scripts/research_crypto_proxy_orchestrator_backtest.py @@ -14,11 +14,13 @@ sys.path.insert(0, str(SRC)) from crypto_strategies.backtest.orchestrator_runner import ( # noqa: E402 + COMBO_DEFAULT_MIN_HISTORY_DAYS, DEFAULT_MIN_HISTORY_DAYS, PROFILE_NAME, SUPPORTED_PROFILES, - CryptoLivePoolBacktestRunner, + build_backtest_runner, ) +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE from scripts.run_walk_forward_backtest import run_walk_forward # noqa: E402 @@ -38,8 +40,11 @@ def main() -> int: 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} + runner = build_backtest_runner(args.profile, synthetic_days=args.synthetic_days) + if args.profile == CRYPTO_EQUITY_COMBO_PROFILE: + params = {"min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, "combo_mode": "dynamic"} + else: + params = {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7} result = runner.run(args.profile, params) payload = { "profile": args.profile, @@ -48,7 +53,7 @@ def main() -> int: "max_drawdown": result.max_drawdown, "cagr": result.cagr, }, - "source": "CryptoLivePoolBacktestRunner", + "source": type(runner).__name__, } text = json.dumps(payload, indent=2, sort_keys=True, default=str) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 1ef8c7b..3cde4b0 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -10,11 +10,13 @@ from typing import Any from crypto_strategies.backtest.orchestrator_runner import ( + COMBO_DEFAULT_MIN_HISTORY_DAYS, DEFAULT_MIN_HISTORY_DAYS, PROFILE_NAME, SUPPORTED_PROFILES, - CryptoLivePoolBacktestRunner, + build_backtest_runner, ) +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = ( (date(2023, 6, 1), date(2024, 5, 31)), @@ -23,6 +25,10 @@ PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { PROFILE_NAME: {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7}, + CRYPTO_EQUITY_COMBO_PROFILE: { + "min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, + "combo_mode": "dynamic", + }, } @@ -45,6 +51,8 @@ def run_walk_forward( windows: tuple[tuple[date, date], ...] = DEFAULT_WINDOWS, synthetic_days: int = 1600, store_root: Path | None = None, + panel: Any = None, + market_history: Any = None, ) -> dict[str, Any]: from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore @@ -53,7 +61,12 @@ def run_walk_forward( 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) + runner = build_backtest_runner( + profile, + panel=panel, + market_history=market_history, + 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) diff --git a/src/crypto_strategies/backtest/combo_simulator.py b/src/crypto_strategies/backtest/combo_simulator.py new file mode 100644 index 0000000..8fece3c --- /dev/null +++ b/src/crypto_strategies/backtest/combo_simulator.py @@ -0,0 +1,175 @@ +"""Simplified crypto equity combo backtest for orchestrator integration.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Literal + +import numpy as np +import pandas as pd + +from crypto_strategies.backtest.live_pool_simulator import LivePoolBacktestResult, _performance_metrics +from crypto_strategies.strategies.crypto_equity_combo import ( + DEFAULT_BTC_WEIGHT, + DEFAULT_TREND_WEIGHT, + DYNAMIC_REGIME_OFF_CUT, +) + +ComboMode = Literal["static", "dynamic"] +BTC_SYMBOL = "BTCUSDT" +ETH_SYMBOL = "ETHUSDT" +ALTS = ("ETH", "SOL", "AVAX", "MATIC", "DOT") +VOL_MULTIPLIERS = {"ETH": 1.0, "SOL": 1.8, "AVAX": 2.2, "MATIC": 2.0, "DOT": 1.6} +SMA_SHORT = 20 +SMA_LONG = 60 +BTC_SMA_REGIME = 200 + + +@dataclass(frozen=True) +class CryptoComboBacktestConfig: + btc_weight: float = DEFAULT_BTC_WEIGHT + trend_weight: float = DEFAULT_TREND_WEIGHT + combo_mode: ComboMode = "dynamic" + min_history_days: int = 260 + dca_amount_usd: float = 100.0 + dynamic_trend_cut: float = DYNAMIC_REGIME_OFF_CUT + + +def build_close_matrix( + market_history: pd.DataFrame, + *, + symbols: tuple[str, ...] = (BTC_SYMBOL, ETH_SYMBOL), +) -> pd.DataFrame: + frame = market_history.copy() + frame["date"] = pd.to_datetime(frame["date"], utc=False).dt.tz_localize(None).dt.normalize() + close = ( + frame.pivot_table(index="date", columns="symbol", values="close", aggfunc="last") + .sort_index() + .reindex(columns=list(symbols)) + ) + return close.astype(float) + + +def _simulate_alt_returns(eth_returns: pd.Series, *, seed: int = 42) -> pd.DataFrame: + rng = np.random.default_rng(seed) + simulated: dict[str, pd.Series] = {} + for alt in ALTS: + mult = VOL_MULTIPLIERS.get(alt, 1.0) + noise = rng.normal(0, 0.005, size=len(eth_returns)) + raw = np.clip(eth_returns.values * mult + noise, -0.25, 0.25) + simulated[alt] = pd.Series(raw, index=eth_returns.index) + return pd.DataFrame(simulated) + + +def _compute_sma(series: pd.Series, window: int) -> pd.Series: + return series.rolling(window=window, min_periods=window).mean() + + +def _combo_daily_returns( + close: pd.DataFrame, + *, + combo_config: CryptoComboBacktestConfig, +) -> pd.Series: + btc_col = BTC_SYMBOL if BTC_SYMBOL in close.columns else close.columns[0] + eth_col = ETH_SYMBOL if ETH_SYMBOL in close.columns else close.columns[min(1, len(close.columns) - 1)] + + btc_close = close[btc_col].dropna() + eth_close = close[eth_col].dropna() + idx = btc_close.index.intersection(eth_close.index).sort_values() + if len(idx) < combo_config.min_history_days: + return pd.Series(dtype=float) + + eth_returns = eth_close.pct_change().dropna() + alt_returns = _simulate_alt_returns(eth_returns.reindex(idx).fillna(0.0)) + alt_prices: dict[str, pd.Series] = {} + for alt in ALTS: + cum = (1.0 + alt_returns[alt]).cumprod() + start_price = float(eth_close.reindex(cum.index).iloc[0] or 1.0) + alt_prices[alt] = start_price * cum / cum.iloc[0] + + btc_sma200 = _compute_sma(btc_close, BTC_SMA_REGIME) + btc_below_sma200 = btc_close < btc_sma200 + + alt_dfs: dict[str, pd.DataFrame] = {} + for alt in ALTS: + ap = alt_prices[alt].reindex(idx) + alt_dfs[alt] = pd.DataFrame( + { + "close": ap, + "sma_short": _compute_sma(ap, SMA_SHORT), + "sma_long": _compute_sma(ap, SMA_LONG), + }, + index=idx, + ) + + portfolio_values: list[float] = [] + alt_positions: dict[str, float] = {} + btc_units = 0.0 + cash_held = 0.0 + dynamic = combo_config.combo_mode == "dynamic" + + for date in idx: + btc_p = float(btc_close.loc[date]) + trend_weight = combo_config.trend_weight + extra_btc_alloc = 0.0 + if dynamic and bool(btc_below_sma200.loc[date]): + trend_weight *= 1.0 - combo_config.dynamic_trend_cut + extra_btc_alloc = combo_config.dca_amount_usd * combo_config.trend_weight * combo_config.dynamic_trend_cut + + btc_alloc = combo_config.dca_amount_usd * combo_config.btc_weight + trend_alloc = combo_config.dca_amount_usd * trend_weight + btc_units += (btc_alloc + extra_btc_alloc) / btc_p + + alt_prices_today: dict[str, float] = {} + alt_candidates: list[str] = [] + for alt in ALTS: + row = alt_dfs[alt].loc[date] + alt_price = float(row["close"]) + alt_prices_today[alt] = alt_price + short_sma = row["sma_short"] + long_sma = row["sma_long"] + if not np.isnan(short_sma) and not np.isnan(long_sma) and short_sma > long_sma: + alt_candidates.append(alt) + + if alt_candidates and trend_alloc > 0: + per_alt = trend_alloc / len(alt_candidates) + for alt in alt_candidates: + alt_positions[alt] = alt_positions.get(alt, 0.0) + per_alt / alt_prices_today[alt] + else: + cash_held += trend_alloc + + btc_value = btc_units * btc_p + alt_value = sum( + alt_positions.get(alt, 0.0) * alt_prices_today.get(alt, 0.0) + for alt in ALTS + ) + portfolio_values.append(btc_value + alt_value + cash_held) + + equity = pd.Series(portfolio_values, index=idx) + return equity.pct_change().fillna(0.0) + + +def run_combo_backtest( + market_history: pd.DataFrame, + *, + combo_config: CryptoComboBacktestConfig | None = None, + universe_symbols: Any = None, +) -> LivePoolBacktestResult: + combo = combo_config or CryptoComboBacktestConfig() + symbols = tuple(universe_symbols or (BTC_SYMBOL, ETH_SYMBOL)) + close = build_close_matrix(market_history, symbols=symbols) + if len(close) < int(combo.min_history_days): + raise ValueError( + f"market_history requires at least {int(combo.min_history_days)} overlapping trading days" + ) + returns = _combo_daily_returns(close, combo_config=combo) + return LivePoolBacktestResult(metrics=_performance_metrics(returns), returns=returns) + + +__all__ = [ + "BTC_SYMBOL", + "ComboMode", + "CryptoComboBacktestConfig", + "build_close_matrix", + "run_combo_backtest", +] diff --git a/src/crypto_strategies/backtest/orchestrator_research.py b/src/crypto_strategies/backtest/orchestrator_research.py new file mode 100644 index 0000000..984eda9 --- /dev/null +++ b/src/crypto_strategies/backtest/orchestrator_research.py @@ -0,0 +1,99 @@ +"""Shared helpers for crypto research scripts calling BacktestOrchestrator adapters.""" + +from __future__ import annotations + +from datetime import date +from typing import Any, Mapping + +import pandas as pd + +from crypto_strategies.backtest.orchestrator_runner import ( + COMBO_DEFAULT_MIN_HISTORY_DAYS, + CryptoEquityComboBacktestRunner, + CryptoLivePoolBacktestRunner, + DEFAULT_MIN_HISTORY_DAYS, + PROFILE_NAME as LIVE_POOL_PROFILE, +) +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE + + +def _result_to_metrics(result: Any) -> dict[str, Any]: + return { + "sharpe_ratio": result.sharpe_ratio, + "max_drawdown": result.max_drawdown, + "annual_return": result.cagr, + "total_return": result.total_return, + "annual_volatility": result.volatility, + "days": result.observation_count, + } + + +def run_live_pool_profile_backtest( + profile: str, + *, + panel: pd.DataFrame | None = None, + synthetic_days: int = 1600, + start_date: date | None = None, + end_date: date | None = None, + params: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Run a single-window live-pool rotation backtest through CryptoLivePoolBacktestRunner.""" + if profile != LIVE_POOL_PROFILE: + raise ValueError(f"unsupported profile={profile!r}") + + runner = CryptoLivePoolBacktestRunner(panel=panel, synthetic_days=synthetic_days) + merged_params = { + "min_history_days": DEFAULT_MIN_HISTORY_DAYS, + "top_n": 2, + "rebalance_every": 7, + } + if params: + merged_params.update(dict(params)) + result = runner.run(profile, merged_params, start_date=start_date, end_date=end_date) + return { + "profile": profile, + "params": merged_params, + "start_date": result.start_date.isoformat() if result.start_date else None, + "end_date": result.end_date.isoformat() if result.end_date else None, + "metrics": _result_to_metrics(result), + "source": "CryptoLivePoolBacktestRunner", + "run_id": getattr(result, "run_id", None), + } + + +def run_combo_profile_backtest( + profile: str, + *, + market_history: pd.DataFrame | None = None, + synthetic_days: int = 1600, + start_date: date | None = None, + end_date: date | None = None, + params: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Run a single-window crypto equity combo backtest through CryptoEquityComboBacktestRunner.""" + if profile != CRYPTO_EQUITY_COMBO_PROFILE: + raise ValueError(f"unsupported profile={profile!r}") + + runner = CryptoEquityComboBacktestRunner( + market_history=market_history, + synthetic_days=synthetic_days, + ) + merged_params = { + "min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, + "combo_mode": "dynamic", + } + if params: + merged_params.update(dict(params)) + result = runner.run(profile, merged_params, start_date=start_date, end_date=end_date) + return { + "profile": profile, + "params": merged_params, + "start_date": result.start_date.isoformat() if result.start_date else None, + "end_date": result.end_date.isoformat() if result.end_date else None, + "metrics": _result_to_metrics(result), + "source": "CryptoEquityComboBacktestRunner", + "run_id": getattr(result, "run_id", None), + } + + +__all__ = ["run_combo_profile_backtest", "run_live_pool_profile_backtest"] diff --git a/src/crypto_strategies/backtest/orchestrator_runner.py b/src/crypto_strategies/backtest/orchestrator_runner.py index cdb1f06..2ae9e5a 100644 --- a/src/crypto_strategies/backtest/orchestrator_runner.py +++ b/src/crypto_strategies/backtest/orchestrator_runner.py @@ -1,14 +1,16 @@ -"""BacktestRunner adapter for crypto live pool rotation.""" +"""BacktestRunner adapter for crypto live pool rotation and equity combo.""" from __future__ import annotations from datetime import date, datetime, timezone -from typing import Any, Mapping +from typing import Any, Mapping, cast import numpy as np import pandas as pd +from crypto_strategies.backtest.combo_simulator import ComboMode, CryptoComboBacktestConfig, run_combo_backtest from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE try: from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult @@ -18,7 +20,8 @@ PROFILE_NAME = "crypto_live_pool_rotation" DEFAULT_MIN_HISTORY_DAYS = 120 -SUPPORTED_PROFILES = frozenset({PROFILE_NAME}) +COMBO_DEFAULT_MIN_HISTORY_DAYS = 260 +SUPPORTED_PROFILES = frozenset({PROFILE_NAME, CRYPTO_EQUITY_COMBO_PROFILE}) def _synthetic_panel(*, days: int = 1500, symbols: tuple[str, ...] = ("BTCUSDT", "ETHUSDT", "SOLUSDT")) -> pd.DataFrame: @@ -42,6 +45,21 @@ def _synthetic_panel(*, days: int = 1500, symbols: tuple[str, ...] = ("BTCUSDT", return panel.sort_index() +def _synthetic_market_history(*, days: int = 1500, start: str = "2020-01-01") -> pd.DataFrame: + dates = pd.date_range(start, periods=days, freq="D") + symbols = ("BTCUSDT", "ETHUSDT") + rates = {"BTCUSDT": 1.0012, "ETHUSDT": 1.0015} + rows: list[dict[str, object]] = [] + for symbol in symbols: + price = 20000.0 if symbol == "BTCUSDT" else 1500.0 + rate = rates[symbol] + for idx, day in enumerate(dates): + price *= rate + close = price * (1.0 + 0.03 * ((idx % 13) - 6) / 13) + rows.append({"date": day, "symbol": symbol, "close": close}) + return pd.DataFrame(rows) + + 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 @@ -53,6 +71,23 @@ def _slice_panel(panel: pd.DataFrame, *, start_date: date | None, end_date: date return frame.sort_index() +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: + effective_start = pd.Timestamp(start_date) - pd.Timedelta(days=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 _metrics_to_result( *, strategy_profile: str, @@ -78,6 +113,7 @@ def _metrics_to_result( cagr=cagr, volatility=float(metrics.get("Annualized Volatility") or 0.0), win_rate=float(metrics.get("Win Rate") or 0.0), + total_return=float(metrics.get("total_return") or 0.0), start_date=start_date, end_date=end_date, observation_count=int(metrics.get("Trading Days") or 0), @@ -106,6 +142,11 @@ def run( f"Unsupported strategy_profile={strategy_profile!r}; " f"supported={sorted(SUPPORTED_PROFILES)}" ) + if strategy_profile != PROFILE_NAME: + raise ValueError( + f"Unsupported strategy_profile={strategy_profile!r}; " + f"use CryptoEquityComboBacktestRunner for {CRYPTO_EQUITY_COMBO_PROFILE!r}" + ) panel = self._panel if panel is None: @@ -132,4 +173,96 @@ def run( ) -__all__ = ["PROFILE_NAME", "SUPPORTED_PROFILES", "CryptoLivePoolBacktestRunner"] +class CryptoEquityComboBacktestRunner: + """Protocol-compatible BacktestRunner for crypto_equity_combo research.""" + + def __init__( + self, + *, + market_history: pd.DataFrame | None = None, + synthetic_days: int = 1600, + ) -> None: + self._market_history = market_history + 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 != CRYPTO_EQUITY_COMBO_PROFILE: + raise ValueError( + f"Unsupported strategy_profile={strategy_profile!r}; " + f"supported={CRYPTO_EQUITY_COMBO_PROFILE!r}" + ) + + min_history_days = int(params.get("min_history_days", COMBO_DEFAULT_MIN_HISTORY_DAYS)) + combo_mode = str(params.get("combo_mode", "dynamic")) + if combo_mode not in {"static", "dynamic"}: + raise ValueError("combo_mode must be 'static' or 'dynamic'") + + history = self._market_history + if history is None: + history = _synthetic_market_history( + days=max(self._synthetic_days, min_history_days + 400), + ) + 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_combo_backtest( + sliced, + combo_config=CryptoComboBacktestConfig( + combo_mode=cast(ComboMode, combo_mode), + 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)] + return _metrics_to_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), + run_duration_seconds=elapsed, + ) + + +def build_backtest_runner( + strategy_profile: str, + *, + panel: pd.DataFrame | None = None, + market_history: pd.DataFrame | None = None, + synthetic_days: int = 1600, +) -> CryptoLivePoolBacktestRunner | CryptoEquityComboBacktestRunner: + if strategy_profile == CRYPTO_EQUITY_COMBO_PROFILE: + return CryptoEquityComboBacktestRunner( + market_history=market_history, + synthetic_days=synthetic_days, + ) + return CryptoLivePoolBacktestRunner( + panel=panel, + synthetic_days=synthetic_days, + ) + + +__all__ = [ + "COMBO_DEFAULT_MIN_HISTORY_DAYS", + "DEFAULT_MIN_HISTORY_DAYS", + "PROFILE_NAME", + "SUPPORTED_PROFILES", + "CryptoEquityComboBacktestRunner", + "CryptoLivePoolBacktestRunner", + "build_backtest_runner", +] diff --git a/tests/test_combo_orchestrator.py b/tests/test_combo_orchestrator.py new file mode 100644 index 0000000..3f111ba --- /dev/null +++ b/tests/test_combo_orchestrator.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import unittest + +import pandas as pd + +from crypto_strategies.backtest.combo_simulator import ( + CryptoComboBacktestConfig, + run_combo_backtest, +) +from crypto_strategies.backtest.orchestrator_research import run_combo_profile_backtest +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE + + +def _fixture_history(*, days: int = 400) -> pd.DataFrame: + rows = [] + for day in pd.date_range("2022-01-01", periods=days, freq="D"): + rows.append({"date": day, "symbol": "BTCUSDT", "close": 30000.0 + hash(day) % 1000}) + rows.append({"date": day, "symbol": "ETHUSDT", "close": 2000.0 + hash(day) % 100}) + return pd.DataFrame(rows) + + +class ComboSimulatorTests(unittest.TestCase): + def test_run_combo_backtest_static_mode(self) -> None: + result = run_combo_backtest( + _fixture_history(), + combo_config=CryptoComboBacktestConfig(combo_mode="static", min_history_days=260), + ) + self.assertGreater(result.metrics["Trading Days"], 0) + self.assertIn("Sharpe", result.metrics) + + def test_run_combo_backtest_dynamic_mode(self) -> None: + result = run_combo_backtest( + _fixture_history(), + combo_config=CryptoComboBacktestConfig(combo_mode="dynamic", min_history_days=260), + ) + self.assertGreater(result.metrics["Trading Days"], 0) + + +class ComboOrchestratorResearchTests(unittest.TestCase): + def test_run_combo_profile_backtest_with_fixture_history(self) -> None: + payload = run_combo_profile_backtest( + CRYPTO_EQUITY_COMBO_PROFILE, + market_history=_fixture_history(), + params={"min_history_days": 260, "combo_mode": "static"}, + ) + self.assertEqual(payload["profile"], CRYPTO_EQUITY_COMBO_PROFILE) + self.assertEqual(payload["source"], "CryptoEquityComboBacktestRunner") + self.assertGreater(payload["metrics"]["days"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py index c21999d..ab8e16b 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -5,19 +5,29 @@ import tempfile import unittest from datetime import date -from pathlib import Path from crypto_strategies.backtest.orchestrator_runner import ( + COMBO_DEFAULT_MIN_HISTORY_DAYS, PROFILE_NAME, SUPPORTED_PROFILES, + CryptoEquityComboBacktestRunner, CryptoLivePoolBacktestRunner, + build_backtest_runner, ) +from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE class CryptoOrchestratorRunnerTests(unittest.TestCase): def test_supported_profile(self) -> None: self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES) + def test_supported_profile_includes_equity_combo(self) -> None: + self.assertIn(CRYPTO_EQUITY_COMBO_PROFILE, SUPPORTED_PROFILES) + + def test_build_backtest_runner_dispatches_combo(self) -> None: + runner = build_backtest_runner(CRYPTO_EQUITY_COMBO_PROFILE, synthetic_days=1600) + self.assertIsInstance(runner, CryptoEquityComboBacktestRunner) + def test_run_returns_backtest_result(self) -> None: runner = CryptoLivePoolBacktestRunner(synthetic_days=1600) result = runner.run( @@ -31,6 +41,7 @@ def test_run_returns_backtest_result(self) -> None: self.assertGreater(result.observation_count, 0) def test_walk_forward_produces_one_result_per_window(self) -> None: + from pathlib import Path from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore @@ -51,5 +62,52 @@ def test_walk_forward_produces_one_result_per_window(self) -> None: self.assertEqual(len(results), 2) +class CryptoEquityComboBacktestRunnerTests(unittest.TestCase): + def test_run_returns_backtest_result(self) -> None: + runner = CryptoEquityComboBacktestRunner(synthetic_days=1600) + result = runner.run( + CRYPTO_EQUITY_COMBO_PROFILE, + {"min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, "combo_mode": "dynamic"}, + start_date=date(2023, 6, 1), + end_date=date(2024, 6, 1), + ) + self.assertEqual(result.strategy_profile, CRYPTO_EQUITY_COMBO_PROFILE) + self.assertEqual(result.domain, "crypto") + self.assertGreater(result.observation_count, 0) + + def test_invalid_combo_mode_raises(self) -> None: + runner = CryptoEquityComboBacktestRunner(synthetic_days=1600) + with self.assertRaises(ValueError): + runner.run( + CRYPTO_EQUITY_COMBO_PROFILE, + {"min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, "combo_mode": "invalid"}, + ) + + def test_walk_forward_combo_profile(self) -> None: + from pathlib import Path + 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", + CryptoEquityComboBacktestRunner(synthetic_days=1600), + ) + windows = ( + (date(2023, 6, 1), date(2023, 12, 31)), + (date(2024, 1, 1), date(2024, 6, 30)), + ) + results = orchestrator.walk_forward( + CRYPTO_EQUITY_COMBO_PROFILE, + domain="crypto", + params={"min_history_days": COMBO_DEFAULT_MIN_HISTORY_DAYS, "combo_mode": "dynamic"}, + windows=windows, + ) + self.assertEqual(len(results), 2) + self.assertTrue(all(item.strategy_profile == CRYPTO_EQUITY_COMBO_PROFILE for item in results)) + + if __name__ == "__main__": unittest.main()