feat(backtest): crypto_equity_combo orchestrator migration (wave 3)

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] = {}

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)

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)

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",
+]