Repository Public
1"""Run the full research pipeline on every strategy version found under a directory.23 uv run python -m spq.pipeline --strategies strategies --symbols ETH,BTC,LINK,SOL --interval 1h \4 --start 2024-11-10 --end 2026-11-10 --data-dir data --out results/2026-11-10 \5 --rebalance-threshold 0.05 --min-order 5 --decision-every garch_regime=2467For each strategy: one report (<out>/<name>-<version>.json). All the trials of all the strategies are merged8in <out>/trials.jsonl and the deflated Sharpe ratio of every report is then recomputed with that total number of9trials: choosing among eight strategies is itself a selection, so each one is deflated by everything that was tried.10A summary table is written to <out>/summary.md and <out>/summary.json.11"""12from __future__ import annotations1314import argparse15import json16import sys17from concurrent.futures import ProcessPoolExecutor18from pathlib import Path1920from .data import cross_source_deviation, load_universe21from .dsr import deflated_sharpe_from_returns22from .engine import BacktestConfig23from .report import _unique_trials, build_report, test_returns_from_report, write_report24from .research import read_ledger25from .strategy import load_strategy262728def find_strategies(root) -> list[Path]:29 return sorted(p.parent for p in Path(root).glob("*/*/strategy.py"))303132def _one(job: dict) -> dict:33 strategy = load_strategy(job["dir"])34 data, quality = load_universe(job["symbols"], job["interval"], job["start"], job["end"], root=job["data_dir"],35 source=job["source"])36 config = BacktestConfig(interval=job["interval"], fee_bps=job["fee_bps"], slippage_bps=job["slippage_bps"],37 initial_cash=job["initial_cash"], min_order=job["min_order"],38 rebalance_threshold=job["rebalance_threshold"],39 decision_every=job["decision_every"].get(strategy.name, 1))40 report, _ = build_report(strategy, data, config, quality=quality, ledger=job["ledger"], source=job["source"],41 notes=job["notes"], n_resamples=job["resamples"], seed=job["seed"])42 return report434445def _fmt(x, pct=False, nd=2):46 if x is None:47 return "n/a"48 return f"{100 * x:+.1f} %" if pct else f"{x:.{nd}f}"495051def summary_rows(reports: list[dict]) -> list[dict]:52 rows = []53 for rep in reports:54 m, rob = rep["metrics"]["test"], rep["robustness"]55 rows.append({56 "strategy": f"{rep['strategy']['name']}/{rep['strategy']['version']}", "sha256": rep["strategy_sha256"],57 **{k: m[k] for k in ("total_return", "cagr", "volatility", "sharpe", "sortino", "max_drawdown", "calmar",58 "turnover", "trades", "win_rate", "profit_factor", "fees_paid", "slippage_paid",59 "exposure")},60 "sharpe_ci": rob["bootstrap_sharpe_ci"], "own_trials": rob["deflated_sharpe_detail"].get("own_trials"),61 "trials_tested": rob["trials_tested"], "deflated_sharpe": rob["deflated_sharpe"],62 "validation_sharpe": rep["metrics"]["validation"]["sharpe"],63 "train_sharpe": rep["metrics"]["train"]["sharpe"],64 "lookahead": rep["checks"]["lookahead"], "data_quality": rep["checks"]["data_quality"],65 })66 bench = reports[0]["benchmarks"]67 for name in ("buy_and_hold", "equal_weight"):68 b = bench[name]69 rows.append({"strategy": f"benchmark: {name}", **{k: b[k] for k in (70 "total_return", "cagr", "volatility", "sharpe", "sortino", "max_drawdown", "calmar", "turnover", "trades",71 "win_rate", "profit_factor", "fees_paid", "slippage_paid", "exposure")},72 "sharpe_ci": reports[0]["robustness"]["bootstrap"]["buy_and_hold_sharpe_ci"] if name == "buy_and_hold"73 else None})74 return rows757677def summary_markdown(rows: list[dict], reports: list[dict]) -> str:78 c = reports[0]["config"]79 s = reports[0]["splits"]80 out = [81 f"# Library strategies on {', '.join(c['symbols'])} at {c['interval']} ({c['source']})",82 "",83 f"Data {c['start']} to {c['end']} ({c['bars']} common closed bars). Train {s['train'][0][:10]} to "84 f"{s['train'][1][:10]}, validation {s['validation'][0][:10]} to {s['validation'][1][:10]}, "85 f"**test {s['test'][0][:10]} to {s['test'][1][:10]}**.",86 f"Costs: fee {c['fee_bps']} bps + slippage {c['slippage_bps']} bps on traded notional; initial cash "87 f"{c['initial_cash']} USD; minimum order {c['min_order']} USD; rebalance threshold {c['rebalance_threshold']}.",88 "Parameters are the defaults written in each META before any backtest; nothing was selected on the test split.",89 "",90 "Test split:",91 "",92 "| Strategy | Return | CAGR | Vol | Sharpe | Sharpe 95 % CI | Sortino | Max DD | Calmar | Turnover | Trades | "93 "Win rate | Profit factor | Fees + slippage (USD) | Exposure | Trials | Deflated Sharpe |",94 "|---|---:|---:|---:|---:|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|",95 ]96 for r in rows:97 ci = r.get("sharpe_ci")98 ci_s = f"[{_fmt(ci[0])}, {_fmt(ci[1])}]" if ci and ci[0] is not None else "n/a"99 costs = (r["fees_paid"] or 0) + (r["slippage_paid"] or 0)100 out.append(101 f"| {r['strategy']} | {_fmt(r['total_return'], True)} | {_fmt(r['cagr'], True)} | "102 f"{_fmt(r['volatility'], True)} | {_fmt(r['sharpe'])} | {ci_s} | {_fmt(r['sortino'])} | "103 f"{_fmt(-r['max_drawdown'] if r['max_drawdown'] is not None else None, True)} | {_fmt(r['calmar'])} | "104 f"{_fmt(r['turnover'], nd=1)} | {r['trades']} | {_fmt(r['win_rate'])} | {_fmt(r['profit_factor'])} | "105 f"{costs:.2f} | {_fmt(r['exposure'])} | {r.get('trials_tested', '')} | "106 f"{_fmt(r.get('deflated_sharpe'), nd=3) if 'deflated_sharpe' in r else ''} |")107 out += [108 "",109 "Return, CAGR, Vol and Max DD in percent; Sharpe and Sortino annualised (8 760 bars per year, risk-free 0); "110 "turnover = traded notional / average equity over the split; win rate and profit factor on closed lots.",111 "Trials = distinct (strategy, parameters) configurations run on train + validation across the whole library; "112 "the deflated Sharpe ratio (Bailey & López de Prado 2014) is the probability that the true Sharpe ratio is "113 "positive given that many trials. Below 0.95, 'no skill' is not rejected.",114 "",115 ]116 return "\n".join(out)117118119def main(argv=None) -> int:120 p = argparse.ArgumentParser(prog="python -m spq.pipeline", description=__doc__.split("\n")[0])121 p.add_argument("--strategies", default="strategies")122 p.add_argument("--symbols", required=True)123 p.add_argument("--interval", default="1h")124 p.add_argument("--start", default=None)125 p.add_argument("--end", default=None)126 p.add_argument("--data-dir", required=True)127 p.add_argument("--source", default="coinbase")128 p.add_argument("--cross-source", default="kraken", help="second source for a deviation note, if files exist")129 p.add_argument("--fee-bps", type=float, default=10.0)130 p.add_argument("--slippage-bps", type=float, default=10.0)131 p.add_argument("--initial-cash", type=float, default=1000.0)132 p.add_argument("--min-order", type=float, default=0.0)133 p.add_argument("--rebalance-threshold", type=float, default=0.0)134 p.add_argument("--decision-every", action="append", default=[], metavar="NAME=N")135 p.add_argument("--resamples", type=int, default=2000)136 p.add_argument("--seed", type=int, default=0)137 p.add_argument("--workers", type=int, default=4)138 p.add_argument("--out", required=True)139 a = p.parse_args(argv)140141 out = Path(a.out)142 (out / "ledger").mkdir(parents=True, exist_ok=True)143 symbols = [s.strip().upper() for s in a.symbols.split(",") if s.strip()]144 notes = []145 for s in symbols:146 dev = cross_source_deviation(a.data_dir, s, a.interval, a.source, a.cross_source)147 if dev:148 notes.append(f"{s}: close vs {a.cross_source} on {dev['bars']} common bars: median |diff| "149 f"{dev['median_abs_rel_diff']:.5f}, max {dev['max_abs_rel_diff']:.5f} (relative)")150 every = {k: int(v) for k, v in (x.split("=") for x in a.decision_every)}151 dirs = [d for d in find_strategies(a.strategies) if load_strategy(d).kind == "spot"] # perps: spq.perp_pipeline152 jobs = []153 for d in dirs:154 name = d.parent.name155 ledger = out / "ledger" / f"{name}-{d.name}.jsonl"156 ledger.unlink(missing_ok=True)157 jobs.append({"dir": str(d), "symbols": symbols, "interval": a.interval, "start": a.start, "end": a.end,158 "data_dir": a.data_dir, "source": a.source, "fee_bps": a.fee_bps, "slippage_bps": a.slippage_bps,159 "initial_cash": a.initial_cash, "min_order": a.min_order,160 "rebalance_threshold": a.rebalance_threshold, "decision_every": every, "ledger": str(ledger),161 "notes": notes, "resamples": a.resamples, "seed": a.seed})162 with ProcessPoolExecutor(max_workers=a.workers) as pool:163 reports = list(pool.map(_one, jobs))164165 records = [r for j in jobs for r in read_ledger(j["ledger"])]166 (out / "trials.jsonl").write_text("".join(json.dumps(r) + "\n" for r in records))167 trials = _unique_trials(records, a.interval)168 sharpes = [t.get("validation_sharpe_per_bar") for t in trials]169 for rep in reports:170 own = rep["robustness"]["deflated_sharpe_detail"]171 dsr = deflated_sharpe_from_returns(test_returns_from_report(rep), sharpes)172 dsr.update({"n_trials": len(trials), "sharpe_split": "test", "trial_sharpes_split": "validation",173 "scope": "all strategies and parameter sets of this pipeline run",174 "own_trials": own["n_trials"], "own_deflated_sharpe": own["deflated_sharpe"]})175 rep["robustness"].update({"trials_tested": len(trials), "deflated_sharpe": dsr["deflated_sharpe"],176 "deflated_sharpe_detail": dsr})177 write_report(rep, out / f"{rep['strategy']['name']}-{rep['strategy']['version']}.json")178 rows = summary_rows(reports)179 (out / "summary.json").write_text(json.dumps(rows, indent=1) + "\n")180 md = summary_markdown(rows, reports)181 (out / "summary.md").write_text(md)182 print(md)183 return 0184185186if __name__ == "__main__":187 sys.exit(main())188