"""Strategy sweep: run the full config grid through the eval harness. Every configuration is a COUNTED TRIAL. The report ranks all configs by per-bar net Sharpe, computes PBO across the whole grid (CSCV, 8 splits), and applies the deflated Sharpe gate with n_trials = grid size to the best config, plus full CPCV OOS evaluation of the top 5. Usage: python sweep.py --data /path/curated --out sweep_report.json """ from __future__ import annotations import argparse import json import sys from pathlib import Path import numpy as np from crypto_eval.data import load_local, load_manifest from crypto_eval.engine import run_backtest from crypto_eval.grid import ( carry, donchian, funding_extreme, meanrev, tsmom, ) from crypto_eval.cv import make_cpcv from crypto_eval.stats import eval_backtest, eval_trial_set def build_grid() -> dict: """Config grid grounded in the 2026-10 literature survey. TSMOM is the primary family (strongest surviving evidence), carry and meanrev are expected losers included as controls so the PBO table has known-negative entries. """ g: dict = {} lookbacks = [24, 120, 240, 480, 1560, 3600, 4800] # 1d..200d for lb in lookbacks: for lo in (False, True): for vt in (None, 0.20, 0.30, 0.50): for band in (None, 0.20): name = (f"tsmom_lb{lb}_{'lo' if lo else 'ls'}" f"_vt{vt if vt is not None else 'none'}" f"_b{band if band is not None else 'none'}") g[name] = (tsmom, dict( lookback=lb, long_only=lo, vol_target=vt, band=band)) for w in (480, 1320): # 20d, 55d for lo in (False, True): for vt in (None, 0.30): g[f"donchian_w{w}_{'lo' if lo else 'ls'}_vt{vt}"] = ( donchian, dict(window=w, long_only=lo, vol_target=vt)) for tm in (1.0, 2.0, 3.0): g[f"carry_tm{tm}_taker"] = (carry, dict(threshold_mult=tm)) g[f"carry_tm{tm}_vt30"] = (carry, dict(threshold_mult=tm, vol_target=0.30)) for k in (3.0, 5.0): for hb in (24, 72): g[f"funding_extreme_k{k}_h{hb}"] = ( funding_extreme, dict(k=k, hold_bars=hb)) for w in (24, 72, 168): for k in (2.0, 3.0): g[f"meanrev_w{w}_k{k}"] = (meanrev, dict(window=w, k=k)) return g def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--data", required=True) ap.add_argument("--out", default="sweep_report.json") ap.add_argument("--top-k", type=int, default=5) ap.add_argument("--pbo-splits", type=int, default=8) ap.add_argument("--max-configs", type=int, default=0, help="cap grid size for smoke tests (0 = all)") args = ap.parse_args() data = load_local(args.data) manifest = load_manifest(args.data) bars, funding = data.bars, data.funding print(f"[sweep] {len(bars):,} bars, " f"{data.check_funding_coverage('1h'):,} funding ticks", flush=True) grid = build_grid() if args.max_configs: grid = dict(list(grid.items())[: args.max_configs]) n_trials = len(grid) print(f"[sweep] {n_trials} configurations (counted trials)", flush=True) # --- full-sample net returns per config ----------------------------- returns: dict[str, np.ndarray] = {} failures: dict[str, str] = {} for i, (name, (fn, kw)) in enumerate(grid.items()): try: pos = fn(bars, funding, **kw) bt = run_backtest(bars, pos, funding) returns[name] = bt["net_return"].to_numpy() except Exception as e: # one bad config must not discard the sweep failures[name] = repr(e) print(f"[sweep] CONFIG FAILED {name}: {e}", flush=True) if (i + 1) % 25 == 0: print(f"[sweep] {i+1}/{n_trials} backtests done", flush=True) if failures: print(f"[sweep] {len(failures)} configs failed", flush=True) n_trials = len(returns) # counted trials = configs actually evaluated 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"[sweep] best per-bar SR: {ranked[0]} = {sr[ranked[0]]:.5f}", flush=True) # --- PBO across the whole grid (CSCV) ------------------------------- print("[sweep] computing PBO (slow) ...", flush=True) trials = eval_trial_set([returns[n] for n in ranked]) print(f"[sweep] PBO={trials['probability_of_backtest_overfitting']:.3f} " f"n_eff={trials['effective_n_trials']}", flush=True) # --- DSR gate on the best config, n_trials = grid size --------------- best = ranked[0] trial_sr_variance = trials["trial_sr_variance"] best_stats = eval_backtest( returns[best], n_trials=n_trials, trial_sr_variance=trial_sr_variance, periods_per_year=24 * 365, ) print(f"[sweep] best-config DSR={best_stats['deflated_sharpe']:.4f}", flush=True) # --- full CPCV OOS for the top-K ------------------------------------ combos = None audit = None oos_top: dict = {} for name in ranked[: args.top_k]: fn, kw = grid[name] pos = fn(bars, funding, **kw) bt = run_backtest(bars, pos, funding) if combos is None: _, combos, audit = make_cpcv( bars.index, horizon=8, embargo=8, freq="1h", n_blocks=6, k=2) print(f"[sweep] CPCV: {len(combos)} combinations, audit " f"{'OK' if audit.ok else 'FAILED'}", flush=True) per_combo = [ eval_backtest(bt["net_return"].to_numpy()[test_idx], n_trials=1, trial_sr_variance=0.0, periods_per_year=24 * 365) for _, test_idx in combos ] sr_list = [c["sharpe_annualized"] for c in per_combo] oos_top[name] = { "oos_mean_sharpe_annualized": float(np.mean(sr_list)), "oos_min_sharpe_annualized": float(np.min(sr_list)), "oos_max_sharpe_annualized": float(np.max(sr_list)), "n_combos_positive": int(sum(1 for s in sr_list if s > 0)), "net_total_pnl": bt["net_total_pnl"], "total_fees": bt["total_fees"], "total_slippage": bt["total_slippage"], "total_funding": bt["total_funding"], } print(f"[sweep] OOS {name}: mean SR {oos_top[name]['oos_mean_sharpe_annualized']:.2f}", flush=True) all_sr = {n: {"sharpe_per_bar": sr[n], "sharpe_annualized": sr[n] * np.sqrt(24 * 365)} for n in ranked} report = { "data": "Avi-L/crypto-quant-eval-data", "n_bars": len(bars), "n_trials": n_trials, "n_configs_failed": len(failures), "config_failures": failures, "pbo_splits": args.pbo_splits, "grid": trials, "best_config": best, "best_config_stats": best_stats, "decision_rule": "accept only if DSR > 0.95 (n_trials=grid size) " "and PBO <= 0.5 and leakage audit OK", "top_k_oos": oos_top, "all_configs_ranked": all_sr, "data_manifest_outputs": sorted(manifest["outputs"].keys()), } Path(args.out).write_text(json.dumps(report, indent=2, default=str)) print(f"[sweep] wrote {args.out}", flush=True) 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"[sweep] 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())