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