Repository Public
1"""Backtest one strategy version and write the JSON report.23 uv run python -m spq.backtest --strategy strategies/ma_trend/1 --symbols ETH,LINK --interval 1h \4 --start 2024-11-10 --end 2026-11-10 --fee-bps 10 --slippage-bps 10 --out report.json56Data come from --data-dir (local files written by spq.data.fetch), else SPQ_DATA_DIR, else DATABASE_URL.7The report is validated against schema/backtest-report.schema.json before it is written. Exit code 0 means a8report was written; it says nothing about whether the strategy is any good.9"""10from __future__ import annotations1112import argparse13import json14import sys1516from .data import DataError, load_universe17from .engine import BacktestConfig18from .report import build_report, write_report19from .research import cartesian_grid20from .strategy import StrategyError, load_strategy212223def parser() -> argparse.ArgumentParser:24 p = argparse.ArgumentParser(prog="python -m spq.backtest", description=__doc__.split("\n")[0])25 p.add_argument("--strategy", required=True, help="strategy version directory (contains strategy.py)")26 p.add_argument("--sha256", default=None, help="refuse to run if strategy.py has another hash")27 p.add_argument("--symbols", default=None, help="comma-separated; default: META['universe'] (required when it is ['*'])")28 p.add_argument("--interval", default=None, help="default: META['interval']")29 p.add_argument("--start", default=None, help="UTC, inclusive")30 p.add_argument("--end", default=None, help="UTC, exclusive")31 p.add_argument("--fee-bps", type=float, default=10.0)32 p.add_argument("--slippage-bps", type=float, default=10.0)33 p.add_argument("--initial-cash", type=float, default=1000.0)34 p.add_argument("--min-order", type=float, default=0.0, help="USD; smaller orders are skipped")35 p.add_argument("--rebalance-threshold", type=float, default=0.05,36 help="weight distance under which a position is left alone (default 0.05; 0 = rebalance to "37 "the exact target at every bar, which trades at every bar)")38 p.add_argument("--decision-every", type=int, default=1, help="ask the strategy every N bars")39 p.add_argument("--split", default="0.6,0.2,0.2", help="train,validation,test fractions (chronological)")40 p.add_argument("--grid", default=None,41 help='JSON object parameter -> list of values for the sensitivity grid, e.g. \'{"fast":[12,24]}\'; '42 "default: each numeric parameter halved and doubled, one at a time")43 p.add_argument("--folds", type=int, default=4, help="walk-forward folds")44 p.add_argument("--resamples", type=int, default=2000, help="bootstrap / Monte Carlo draws")45 p.add_argument("--seed", type=int, default=0)46 p.add_argument("--ledger", default=None,47 help="JSONL file recording every trial and test evaluation across runs; the deflated Sharpe "48 "ratio then counts all the trials in it")49 p.add_argument("--job-ledger", default=None,50 help="v7: the trial ledger of the research job that produced this strategy (the runner's copy); "51 "its trials are added to the trial count of the deflated Sharpe ratio")52 p.add_argument("--data-dir", default=None, help="root of local candle files (<root>/<source>/<SYM>_<interval>)")53 p.add_argument("--source", default=None, help="candle source (default: coinbase for files)")54 p.add_argument("--out", required=True, help="report path (JSON)")55 return p565758def main(argv=None) -> int:59 a = parser().parse_args(argv)60 try:61 strategy = load_strategy(a.strategy, a.sha256)62 if strategy.kind != "spot":63 raise StrategyError(f"{strategy.name} is a {strategy.kind} strategy: use python -m spq.perp_backtest")64 given = [s.strip().upper() for s in a.symbols.split(",") if s.strip()] if a.symbols else None65 symbols = strategy.symbols_for(given) # v5: a generic strategy takes --symbols (validated); a fixed one is unchanged66 interval = a.interval or strategy.meta["interval"]67 config = BacktestConfig(interval=interval, fee_bps=a.fee_bps, slippage_bps=a.slippage_bps,68 initial_cash=a.initial_cash, min_order=a.min_order,69 rebalance_threshold=a.rebalance_threshold, decision_every=a.decision_every)70 fractions = tuple(float(x) for x in a.split.split(","))71 grid = cartesian_grid(strategy.params, json.loads(a.grid)) if a.grid else None72 kwargs = {"root": a.data_dir} if a.data_dir else {}73 data, quality = load_universe(symbols, interval, a.start, a.end, source=a.source, **kwargs)74 raw_bars = {s: q.n_bars for s, q in quality.items()}75 aligned = len(next(iter(data.values())))76 notes = []77 if any(n != aligned for n in raw_bars.values()):78 notes.append(f"bars per symbol before alignment {raw_bars}, common bars kept: {aligned}")79 source = next(iter(quality.values())).source80 report, _ = build_report(strategy, data, config, quality=quality, fractions=fractions, grid=grid,81 ledger=a.ledger, job_ledger=a.job_ledger, n_folds=a.folds, n_resamples=a.resamples, seed=a.seed,82 source=source, notes=notes)83 write_report(report, a.out)84 except (StrategyError, DataError, ValueError) as e:85 print(f"spq.backtest: {type(e).__name__}: {e}", file=sys.stderr)86 return 287 t, b = report["metrics"]["test"], report["benchmarks"]["buy_and_hold"]88 print(json.dumps({89 "report": a.out, "strategy_sha256": report["strategy_sha256"],90 "test": {k: t[k] for k in ("total_return", "sharpe", "max_drawdown", "trades", "fees_paid", "exposure")},91 "buy_and_hold_test": {k: b[k] for k in ("total_return", "sharpe", "max_drawdown")},92 "trials_tested": report["robustness"]["trials_tested"],93 "deflated_sharpe": report["robustness"]["deflated_sharpe"],94 "checks": {k: report["checks"][k] for k in ("lookahead", "data_quality")},95 }))96 return 0979899if __name__ == "__main__":100 sys.exit(main())101