daily panel sweep runner (66 trials, per-config CPCV)

sweep2.py

@@ -0,0 +1,171 @@
+"""Panel sweep v2: daily multi-asset strategies through the eval gates.
+
+Every configuration is a counted trial. Reports PBO across the grid,
+DSR (n_trials = grid size) on the best config, and CPCV OOS for the top-K.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from crypto_eval.engine2 import run_panel_backtest
+from crypto_eval.grid2 import (
+    buy_and_hold_btc, donchian_panel, xs_momentum,
+    xs_momentum_tsmom_filter, tsmom_panel,
+)
+from crypto_eval.cv import make_cpcv
+from crypto_eval.stats import eval_backtest, eval_trial_set
+
+PANEL_META_DEFAULT_ADV = {
+    "BTC": 20e9, "ETH": 10e9, "LTC": 1.0e9, "XRP": 1.5e9,
+    "ADA": 0.5e9, "DOGE": 1.0e9, "ETC": 0.3e9, "DASH": 0.1e9,
+}
+
+
+def build_grid() -> dict:
+    """Returns {name: (fn, kwargs, subset_cols or None)}. subset restricts
+    the panel to one asset (single-asset configs count as trials too)."""
+    g: dict = {}
+    for lb in (10, 20, 50, 100, 200, 400):
+        for lo in (False, True):
+            for vt in (None, 0.30):
+                g[f"tsmom_lb{lb}_{'lo' if lo else 'ls'}_vt{vt}"] = (
+                    tsmom_panel, dict(lookback=lb, long_only=lo, vol_target=vt), None)
+    for w in (20, 55):
+        for lo in (False, True):
+            for vt in (None, 0.30):
+                g[f"donchian_w{w}_{'lo' if lo else 'ls'}_vt{vt}"] = (
+                    donchian_panel, dict(window=w, long_only=lo, vol_target=vt), None)
+    for lb in (30, 90, 180):
+        for k in (2, 3):
+            for sb in (False, True):
+                g[f"xs_lb{lb}_k{k}_{'sb' if sb else 'lo'}"] = (
+                    xs_momentum, dict(lookback=lb, top_k=k, short_bottom=sb), None)
+            g[f"xs_lb{lb}_k{k}_tfilt"] = (
+                xs_momentum_tsmom_filter, dict(lookback=lb, top_k=k), None)
+    # single-asset long-only trend (era-stability family)
+    for sym in ("BTC", "ETH"):
+        for lb in (20, 50, 100, 200):
+            for fam, fn in (("donch", donchian_panel), ("tsmom", tsmom_panel)):
+                kw = ({"window": lb} if fam == "donch" else {"lookback": lb})
+                kw["long_only"] = True
+                g[f"{sym}_{fam}_lb{lb}_lo"] = (fn, kw, [sym])
+    return g
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--panel", required=True, help="panel_daily_close.parquet")
+    ap.add_argument("--out", default="sweep2_report.json")
+    ap.add_argument("--top-k", type=int, default=5)
+    ap.add_argument("--max-configs", type=int, default=0)
+    args = ap.parse_args()
+
+    panel = pd.read_parquet(args.panel)
+    panel.index = pd.to_datetime(panel.index, utc=True)
+    panel = panel.sort_index()
+    start = "2014-01-01"  # >= 4 assets listed
+    panel = panel[panel.index >= start]
+    print(f"[sweep2] panel {panel.shape[0]:,} days x {panel.shape[1]} assets "
+          f"from {panel.index[0].date()}", flush=True)
+
+    grid = build_grid()
+    if args.max_configs:
+        grid = dict(list(grid.items())[: args.max_configs])
+    print(f"[sweep2] {len(grid)} configurations", flush=True)
+
+    returns, failures = {}, {}
+    for i, (name, (fn, kw, subset)) in enumerate(grid.items()):
+        try:
+            p = panel[subset] if subset else panel
+            adv = {s: PANEL_META_DEFAULT_ADV[s] for s in p.columns}
+            pos = fn(p, **kw)
+            bt = run_panel_backtest(p, pos, adv)
+            returns[name] = bt["portfolio_return"].to_numpy()
+        except Exception as e:
+            failures[name] = repr(e)
+            print(f"[sweep2] CONFIG FAILED {name}: {e}", flush=True)
+        if (i + 1) % 10 == 0:
+            print(f"[sweep2] {i+1}/{len(grid)} done", flush=True)
+    if failures:
+        print(f"[sweep2] {len(failures)} configs failed", flush=True)
+    n_trials = len(returns)
+
+    sr = {n: float(np.nanmean(r) / (np.nanstd(r, ddof=1) + 1e-12))
+          for n, r in returns.items()}
+    ranked = sorted(sr, key=sr.get, reverse=True)
+    print(f"[sweep2] best per-day SR: {ranked[0]} = {sr[ranked[0]]:.5f} "
+          f"(ann {sr[ranked[0]]*np.sqrt(365):.2f})", flush=True)
+
+    print("[sweep2] computing PBO ...", flush=True)
+    trials = eval_trial_set([returns[n] for n in ranked])
+    best = ranked[0]
+    best_stats = eval_backtest(
+        returns[best], n_trials=n_trials,
+        trial_sr_variance=trials["trial_sr_variance"],
+        periods_per_year=365,
+    )
+    print(f"[sweep2] PBO={trials['probability_of_backtest_overfitting']:.3f} "
+          f"DSR={best_stats['deflated_sharpe']:.4f}", flush=True)
+
+    # CPCV OOS for top-K (daily bars, horizon 1, embargo 5 days)
+    # CPCV audit on the full panel (per-config combos built on demand)
+    _, _, audit = make_cpcv(panel.index, horizon=1, embargo=5, freq="1D")
+    print(f"[sweep2] CPCV audit {'OK' if audit.ok else 'FAILED'}", flush=True)
+    oos_top = {}
+    for name in ranked[: args.top_k]:
+        fn, kw, subset = grid[name]
+        p = panel[subset] if subset else panel
+        adv = {s: PANEL_META_DEFAULT_ADV[s] for s in p.columns}
+        pos = fn(p, **kw)
+        bt = run_panel_backtest(p, pos, adv)
+        ret_series = bt["portfolio_return"]
+        rets = ret_series.to_numpy()
+        # CPCV on the config's OWN history (single-asset configs start at
+        # the asset's listing date, shorter than the panel)
+        _, combos_k, audit_k = make_cpcv(ret_series.index, horizon=1,
+                                         embargo=5, freq="1D")
+        if not audit_k.ok:
+            raise RuntimeError(f"OOS audit failed for {name}")
+        per = [eval_backtest(rets[t], n_trials=1, trial_sr_variance=0.0,
+                             periods_per_year=365) for _, t in combos_k]
+        srs = [c["sharpe_annualized"] for c in per]
+        oos_top[name] = {
+            "oos_mean_sharpe_annualized": float(np.mean(srs)),
+            "n_combos_positive": int(sum(1 for s in srs if s > 0)),
+            "net_total_pnl": float(np.nansum(rets) * 1_000_000),
+        }
+        print(f"[sweep2] OOS {name}: mean SR {np.mean(srs):.2f}", flush=True)
+
+    report = {
+        "panel_start": str(panel.index[0].date()),
+        "n_days": len(panel), "n_assets": panel.shape[1],
+        "n_trials": n_trials, "n_configs_failed": len(failures),
+        "config_failures": failures,
+        "grid": trials,
+        "best_config": best,
+        "best_config_stats": best_stats,
+        "top_k_oos": oos_top,
+        "all_configs_ranked": {n: {"sharpe_per_day": sr[n],
+                                   "sharpe_annualized": sr[n] * np.sqrt(365)}
+                               for n in ranked},
+        "decision_rule": "accept only if DSR > 0.95 (n_trials=grid) and "
+                         "PBO <= 0.5 and leakage audit OK",
+    }
+    Path(args.out).write_text(json.dumps(report, indent=2, default=str))
+    passed = (best_stats["deflated_sharpe"] is not None
+              and best_stats["deflated_sharpe"] > 0.95
+              and trials["probability_of_backtest_overfitting"] <= 0.5)
+    print(f"[sweep2] VERDICT: {'PASS' if passed else 'FAIL'} "
+          f"(DSR={best_stats['deflated_sharpe']:.4f}, "
+          f"PBO={trials['probability_of_backtest_overfitting']:.3f})", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())