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