sweep.py

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