File size: 4,303 Bytes
d8dfb8b | 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 | """Run the institutional eval on a strategy.
Usage:
python run_eval.py --data /path/to/curated --symbol BTCUSDT --freq 1h \
--strategy funding_carry --cpcv-blocks 6 --cpcv-k 2
Pipeline:
1. load curated data, assert schema and no-lookahead invariants
2. CPCV (N=6, K=2 default) with purge=label horizon, embargo=8 bars
3. for each combination: train tuning-free strategy, test OOS
4. reconstruct paths, report PSR/DSR/PBO/MinBTL + raw risk metrics
5. record the source-hash manifest with every run
"""
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.strategies import STRATEGIES
from crypto_eval.cv import make_cpcv, make_walkforward
from crypto_eval.stats import eval_backtest, eval_trial_set
PERIODS_PER_YEAR = { # bars per year by freq
"1h": 24 * 365, "4h": 6 * 365, "1d": 365, "1m": 60 * 24 * 365,
}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--data", required=True)
ap.add_argument("--symbol", default="BTCUSDT")
ap.add_argument("--freq", default="1h")
ap.add_argument("--strategy", default="funding_carry")
ap.add_argument("--horizon", type=int, default=8,
help="label horizon in bars (default 8 = one funding interval)")
ap.add_argument("--embargo", type=int, default=8)
ap.add_argument("--cpcv-blocks", type=int, default=6)
ap.add_argument("--cpcv-k", type=int, default=2)
ap.add_argument("--out", default="eval_report.json")
args = ap.parse_args()
data = load_local(args.data, args.symbol, args.freq)
manifest = load_manifest(args.data)
strategy_fn = STRATEGIES[args.strategy]
n = len(data.bars)
print(f"[eval] {args.symbol} {args.freq}: {n:,} bars, "
f"{data.check_funding_coverage(args.freq):,} funding ticks in range")
splitter, combos, audit = make_cpcv(
data.bars.index, horizon=args.horizon, embargo=args.embargo,
freq=args.freq, n_blocks=args.cpcv_blocks, k=args.cpcv_k,
)
print(f"[eval] CPCV: {len(combos)} combinations, "
f"{len(combos) * args.cpcv_blocks // args.cpcv_k} paths, audit OK")
# strategies here are tuning-free; run OOS per combination
positions = strategy_fn(data.bars, data.funding)
bt = run_backtest(data.bars, positions, data.funding)
oos_returns = []
for j, (train_idx, test_idx) in enumerate(combos):
# with a tuning-free strategy, train is unused; the test slice is
# the OOS segment. With tuned strategies, tune on train ONLY.
rets = bt["net_return"].to_numpy()[test_idx]
oos_returns.append(rets)
ppy = PERIODS_PER_YEAR[args.freq]
trial_stats = eval_trial_set(oos_returns)
per_combo = []
for j, rets in enumerate(oos_returns):
s = eval_backtest(
rets, n_trials=1, trial_sr_variance=0.0,
periods_per_year=ppy,
)
per_combo.append(s)
report = {
"symbol": args.symbol,
"freq": args.freq,
"strategy": args.strategy,
"n_bars": n,
"cpcv": {"n_blocks": args.cpcv_blocks, "k": args.cpcv_k,
"n_combinations": len(combos)},
"purge_horizon": args.horizon,
"embargo": args.embargo,
"audit": {"ok": audit.ok, "violations": audit.violations},
"data_manifest_sha256_inputs": sorted(set(manifest["outputs"].keys())),
"costs": {
"total_fees": bt["total_fees"],
"total_slippage": bt["total_slippage"],
"total_funding": bt["total_funding"],
"net_total_pnl": bt["net_total_pnl"],
},
"trials": trial_stats,
"per_combination": per_combo,
"decision_rule": "accept only if deflated_sharpe > 0.95 and PBO <= 0.5",
}
Path(args.out).write_text(json.dumps(report, indent=2, default=str))
print(f"[eval] wrote {args.out}")
print(f"[eval] PBO={trial_stats['probability_of_backtest_overfitting']:.3f} "
f"DSR(mean combo)="
f"{np.mean([c['deflated_sharpe'] for c in per_combo if c['deflated_sharpe'] is not None]):.3f}")
return 0
if __name__ == "__main__":
sys.exit(main())
|