Repository Public
1"""Run the research pipeline on every PERP strategy version found under a directory (META['kind'] = 'perp').23 uv run python -m spq.perp_pipeline --strategies strategies --data-dir data --out results/2026-11-10-perps \4 --rebalance-threshold 0.05 --decision-every har_rv_perp=2456Each strategy runs on its own META['universe'], on the timestamps common to EVERY series used by the run (so all7the reports share the same bars and the same train / validation / test dates). One version-2 report per strategy8(<out>/<name>-<version>.json). All the trials of all the strategies are merged in <out>/trials.jsonl and the9deflated Sharpe ratio of every report is recomputed with that total: choosing among several strategies is itself10a selection. <out>/summary.md and summary.json hold the tables, with the benchmarks of the whole universe.11"""12from __future__ import annotations1314import argparse15import json16import sys17from concurrent.futures import ProcessPoolExecutor18from pathlib import Path1920from .data.hyperliquid import SOURCE, load_meta, load_perp_universe, load_spot_reference21from .dsr import deflated_sharpe_from_returns22from .metrics import compute_metrics23from .perp_backtest import add_cost_arguments, make_config24from .perp_metrics import compute_perp_metrics25from .perp_report import build_perp_report, returns_from_report, write_perp_report26from .perp_research import PerpExperiment27from .pipeline import _fmt, find_strategies28from .report import _unique_trials29from .research import read_ledger30from .robustness import bootstrap_sharpe_ci31from .strategy import is_spot_leg, load_strategy323334def _load(job: dict, symbols: list[str]):35 """Data of `symbols` restricted to the bars common to the whole run."""36 data, quality, fq, raw = load_perp_universe(job["all_symbols"], job["interval"], job["start"], job["end"],37 root=job["data_dir"], spot_source=job["spot_source"])38 index = next(iter(data.values())).index39 perps = [s for s in job["all_symbols"] if not is_spot_leg(s)]40 spot, spot_q = load_spot_reference(perps, job["interval"], index, root=job["data_dir"], source=job["spot_source"])41 sub = {s: data[s] for s in symbols}42 keep = [s for s in symbols if not is_spot_leg(s)]43 return (sub, {s: quality[s] for s in symbols}, {s: fq[s] for s in keep}, {s: raw[s] for s in keep},44 {s: spot[s] for s in keep if s in spot}, spot_q)454647def _config(job: dict, symbols, decision_every: int = 1):48 return make_config(argparse.Namespace(**job["costs"]), job["interval"], load_meta(job["data_dir"]), symbols,49 decision_every)505152def _one(job: dict) -> dict:53 strategy = load_strategy(job["dir"])54 data, quality, fq, raw, spot, spot_q = _load(job, strategy.universe)55 config = _config(job, strategy.universe, job["decision_every"].get(strategy.name, 1))56 notes = job["notes"] + [f"spot benchmark from {job['spot_source']}: {s} {q.n_bars} bars, {q.missing_bars} missing"57 for s, q in spot_q.items() if s in spot]58 report, _ = build_perp_report(strategy, data, config, quality=quality, funding_quality=fq, funding_raw=raw,59 spot_data=spot, ledger=job["ledger"], source=SOURCE, notes=notes,60 n_resamples=job["resamples"], seed=job["seed"])61 return report626364def universe_benchmarks(job: dict) -> dict:65 """perp_long_1x and spot buy-and-hold of the whole perpetual universe, on each split."""66 perps = [s for s in job["all_symbols"] if not is_spot_leg(s)]67 data, _, _, _, spot, _ = _load(job, perps)68 exp = PerpExperiment(data, _config(job, perps), spot_data=spot)69 out = {"symbols": perps, "buy_and_hold_is_spot": None, "splits": exp.splits.to_report()}70 for split in ("train", "validation", "test"):71 res = exp.benchmarks(split)72 out[split] = {}73 for name in ("perp_long_1x", "buy_and_hold"):74 r = res[name]75 m = compute_perp_metrics(r) if name == "perp_long_1x" else (76 compute_metrics(r) if exp.buy_and_hold_is_spot else compute_perp_metrics(r))77 m["sharpe_ci"] = bootstrap_sharpe_ci(r.returns.to_numpy(), job["interval"], n_samples=job["resamples"],78 seed=job["seed"])["ci"]79 out[split][name] = m80 out["buy_and_hold_is_spot"] = bool(exp.buy_and_hold_is_spot)81 return out828384ROW_KEYS = ("total_return", "cagr", "volatility", "sharpe", "sortino", "max_drawdown", "worst_intrabar_drawdown",85 "calmar", "turnover", "trades", "win_rate", "profit_factor", "fees_paid", "slippage_paid", "funding_pnl",86 "exposure", "long_exposure", "short_exposure", "gross_leverage_mean", "gross_leverage_max",87 "net_leverage_mean", "time_in_market", "avg_holding_hours", "liquidations", "wiped_out",88 "margin_utilisation_max")899091def summary_rows(reports: list[dict], bench: dict) -> list[dict]:92 rows = []93 for rep in reports:94 m, rob = rep["metrics"], rep["robustness"]95 rows.append({96 "strategy": f"{rep['strategy']['name']}/{rep['strategy']['version']}", "sha256": rep["strategy_sha256"],97 "symbols": rep["config"]["symbols"],98 **{k: m["test"][k] for k in ROW_KEYS},99 "sharpe_ci": rob["bootstrap_sharpe_ci"],100 "train": {k: m["train"][k] for k in ("total_return", "sharpe", "max_drawdown", "liquidations",101 "funding_pnl", "fees_paid", "trades")},102 "validation": {k: m["validation"][k] for k in ("total_return", "sharpe", "max_drawdown", "liquidations",103 "funding_pnl", "fees_paid", "trades")},104 "own_trials": rob["deflated_sharpe_detail"].get("own_trials"),105 "trials_tested": rob["trials_tested"], "deflated_sharpe": rob["deflated_sharpe"],106 "own_benchmarks": {n: {k: b[k] for k in ("total_return", "sharpe", "max_drawdown")}107 for n, b in rep["benchmarks"].items() if n in ("perp_long_1x", "buy_and_hold")},108 "checks": {k: rep["checks"][k] for k in ("lookahead", "data_quality", "ledger", "funding_alignment")},109 })110 for name in ("perp_long_1x", "buy_and_hold"):111 b = bench["test"][name]112 label = {"perp_long_1x": "benchmark: perp long 1x incl. funding",113 "buy_and_hold": "benchmark: buy and hold spot" if bench["buy_and_hold_is_spot"]114 else "benchmark: perp price held, NO funding (no spot data)"}[name]115 rows.append({"strategy": label, "symbols": bench["symbols"], **{k: b.get(k) for k in ROW_KEYS},116 "sharpe_ci": b["sharpe_ci"],117 "train": {k: bench["train"][name].get(k) for k in ("total_return", "sharpe", "max_drawdown")},118 "validation": {k: bench["validation"][name].get(k)119 for k in ("total_return", "sharpe", "max_drawdown")}})120 return rows121122123def _ci(ci):124 return f"[{_fmt(ci[0])}, {_fmt(ci[1])}]" if ci and ci[0] is not None else "n/a"125126127def _usd(x):128 return "n/a" if x is None else f"{x:+.2f}"129130131def summary_markdown(rows: list[dict], reports: list[dict], bench: dict, depth: dict | None = None) -> str:132 c = reports[0]["config"]133 s = bench["splits"]134 out = [135 f"# Perpetual-futures strategies on {', '.join(bench['symbols'])} at {c['interval']} ({c['source']})",136 "",137 f"Data {c['start']} to {c['end']} ({c['bars']} common closed bars). Train {s['train'][0][:10]} to "138 f"{s['train'][1][:10]}, validation {s['validation'][0][:10]} to {s['validation'][1][:10]}, "139 f"**test {s['test'][0][:10]} to {s['test'][1][:10]}**.",140 f"Costs: taker fee {c['taker_fee_bps']} bps + slippage {c['slippage_bps']} bps on traded notional "141 f"(fills pay the {c['liquidity']} fee); spot legs and the spot benchmark {c['spot_fee_bps']} + "142 f"{c['spot_slippage_bps']} bps; liquidation penalty {c['liquidation_penalty_bps']} bps of the liquidated "143 f"notional; initial collateral {c['initial_cash']} USDC; minimum order {c['min_order']} USD; rebalance "144 f"threshold {c['rebalance_threshold']}.",145 "Parameters are the ones written in each META before any backtest; nothing was selected on the test split.",146 "",147 "## Test split",148 "",149 "| Strategy | Return | Sharpe | Sharpe 95 % CI | Max DD | Worst intrabar DD | Trades | Fees + slippage (USD) | "150 "Funding PnL (USD) | Gross lev. mean / max | Net lev. mean | Time in market | Avg holding (h) | "151 "Liquidations | Trials | Deflated Sharpe |",152 "|---|---:|---:|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|",153 ]154 for r in rows:155 costs = (r["fees_paid"] or 0) + (r["slippage_paid"] or 0)156 liq = "n/a" if r.get("liquidations") is None else f"{r['liquidations']}{' (wiped out)' if r.get('wiped_out') else ''}"157 out.append(158 f"| {r['strategy']} | {_fmt(r['total_return'], True)} | {_fmt(r['sharpe'])} | {_ci(r.get('sharpe_ci'))} | "159 f"{_fmt(-r['max_drawdown'] if r['max_drawdown'] is not None else None, True)} | "160 f"{_fmt(-r['worst_intrabar_drawdown'] if r.get('worst_intrabar_drawdown') is not None else None, True)} | "161 f"{r['trades']} | {costs:.2f} | {_usd(r.get('funding_pnl'))} | "162 f"{_fmt(r.get('gross_leverage_mean'))} / {_fmt(r.get('gross_leverage_max'))} | "163 f"{_fmt(r.get('net_leverage_mean'))} | {_fmt(r.get('time_in_market'))} | "164 f"{_fmt(r.get('avg_holding_hours'), nd=0)} | {liq} | {r.get('trials_tested', '')} | "165 f"{_fmt(r.get('deflated_sharpe'), nd=3) if 'deflated_sharpe' in r else ''} |")166 out += [167 "",168 "## Train, validation and test side by side (each split backtested from cash)",169 "",170 "| Strategy | Train return | Train Sharpe | Train liq. | Validation return | Validation Sharpe | "171 "Validation liq. | Test return | Test Sharpe | Test liq. |",172 "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|",173 ]174 for r in rows:175 t, v = r["train"], r["validation"]176 out.append(f"| {r['strategy']} | {_fmt(t['total_return'], True)} | {_fmt(t['sharpe'])} | "177 f"{t.get('liquidations', 'n/a')} | {_fmt(v['total_return'], True)} | {_fmt(v['sharpe'])} | "178 f"{v.get('liquidations', 'n/a')} | {_fmt(r['total_return'], True)} | {_fmt(r['sharpe'])} | "179 f"{r.get('liquidations') if r.get('liquidations') is not None else 'n/a'} |")180 out += [181 "",182 "Return and drawdowns in percent; Sharpe annualised (8 760 bars per year at 1h, risk-free 0); the interval is a "183 "95 % stationary-block-bootstrap interval of the test Sharpe ratio. Max DD uses bar closes; worst intrabar DD "184 "values every long at the bar's low and every short at its high against the previous closing peak. Funding "185 "PnL < 0 means the strategy paid funding on balance. Leverage = notional / equity at bar closes. Trials = "186 "distinct (strategy, parameters) configurations run on train + validation across the whole run; the deflated "187 "Sharpe ratio (Bailey & López de Prado 2014) is the probability that the true Sharpe ratio is positive given "188 "that many trials. Below 0.95, 'no skill' is not rejected.",189 "",190 ]191 if depth:192 out += ["## History served by the venue (measured)", "",193 "| Perpetual | 1h candles from | 4h candles from | 1d candles from | Funding from |", "|---|---|---|---|---|"]194 for sym, d in depth.items():195 cd = d.get("candles", {})196 cells = [f"{cd[iv]['first'][:10]} ({cd[iv]['days']:.0f} d)" if iv in cd else "n/a"197 for iv in ("1h", "4h", "1d")]198 cells.append(f"{d['funding']['first'][:10]} ({d['funding']['days']:.0f} d)" if "funding" in d else "n/a")199 out.append(f"| {sym} | " + " | ".join(cells) + " |")200 out.append("")201 return "\n".join(out)202203204def main(argv=None) -> int:205 p = argparse.ArgumentParser(prog="python -m spq.perp_pipeline", description=__doc__.split("\n")[0])206 p.add_argument("--strategies", default="strategies")207 p.add_argument("--interval", default="1h")208 p.add_argument("--start", default=None)209 p.add_argument("--end", default=None)210 p.add_argument("--data-dir", required=True)211 p.add_argument("--spot-source", default="coinbase")212 add_cost_arguments(p)213 p.add_argument("--decision-every", action="append", default=[], metavar="NAME=N")214 p.add_argument("--resamples", type=int, default=2000)215 p.add_argument("--seed", type=int, default=0)216 p.add_argument("--workers", type=int, default=4)217 p.add_argument("--depth", default=None, help="JSON file {symbol: measure_depth(...)} to include in the summary")218 p.add_argument("--out", required=True)219 a = p.parse_args(argv)220221 out = Path(a.out)222 (out / "ledger").mkdir(parents=True, exist_ok=True)223 dirs = [d for d in find_strategies(a.strategies) if load_strategy(d).kind == "perp"]224 if not dirs:225 print("no perp strategy found", file=sys.stderr)226 return 2227 all_symbols: list[str] = []228 for d in dirs:229 for s in load_strategy(d).universe:230 if s not in all_symbols:231 all_symbols.append(s)232 costs = {k: getattr(a, k) for k in (233 "taker_fee_bps", "maker_fee_bps", "liquidity", "slippage_bps", "liquidation_penalty_bps",234 "maintenance_fraction", "leverage", "spot_fee_bps", "spot_slippage_bps", "initial_cash", "min_order",235 "rebalance_threshold")}236 every = {k: int(v) for k, v in (x.split("=") for x in a.decision_every)}237 base = {"all_symbols": all_symbols, "interval": a.interval, "start": a.start, "end": a.end,238 "data_dir": a.data_dir, "spot_source": a.spot_source, "costs": costs, "decision_every": every,239 "resamples": a.resamples, "seed": a.seed,240 "notes": [f"bars common to every series of the run: {', '.join(all_symbols)}"]}241 jobs = []242 for d in dirs:243 ledger = out / "ledger" / f"{d.parent.name}-{d.name}.jsonl"244 ledger.unlink(missing_ok=True)245 jobs.append({**base, "dir": str(d), "ledger": str(ledger)})246 with ProcessPoolExecutor(max_workers=a.workers) as pool:247 reports = list(pool.map(_one, jobs))248 bench = universe_benchmarks(base)249250 records = [r for j in jobs for r in read_ledger(j["ledger"])]251 (out / "trials.jsonl").write_text("".join(json.dumps(r) + "\n" for r in records))252 trials = _unique_trials(records, a.interval)253 sharpes = [t.get("validation_sharpe_per_bar") for t in trials]254 for rep in reports:255 own = rep["robustness"]["deflated_sharpe_detail"]256 dsr = deflated_sharpe_from_returns(returns_from_report(rep), sharpes)257 dsr.update({"n_trials": len(trials), "sharpe_split": "test", "trial_sharpes_split": "validation",258 "scope": "all perp strategies and parameter sets of this pipeline run",259 "own_trials": own["n_trials"], "own_deflated_sharpe": own["deflated_sharpe"]})260 rep["robustness"].update({"trials_tested": len(trials), "deflated_sharpe": dsr["deflated_sharpe"],261 "deflated_sharpe_detail": dsr})262 write_perp_report(rep, out / f"{rep['strategy']['name']}-{rep['strategy']['version']}.json")263 rows = summary_rows(reports, bench)264 depth = json.loads(Path(a.depth).read_text()) if a.depth else None265 (out / "summary.json").write_text(json.dumps({"rows": rows, "benchmarks": bench, "history_depth": depth},266 indent=1) + "\n")267 md = summary_markdown(rows, reports, bench, depth)268 (out / "summary.md").write_text(md)269 print(md)270 return 0271272273if __name__ == "__main__":274 sys.exit(main())275