File size: 7,272 Bytes
bdc6050 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | """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())
|