"""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())