feat: VBT sweep runner — batch backtest all strategy/interval/limit/coin combos
backtests/sweep_runner.py:
- Generates all valid strategy x interval x limit x coin combinations
(9 strategies, 6 intervals, 6 bar limits, 2 coins = 360 combos)
- Sequential mode with configurable rate-limiting delay
- Multi-process parallel mode (ProcessPoolExecutor)
- Progress logging, ETA, summary stats
- --dry-run flag to preview without executing
- CLI: python -m backtests.sweep_runner --workers 2 --delay 1.0
Results from initial sweep (262/360 succeeded):
Best performers:
grid_mm 1h 100b S=5.08 ret=1.45% 4 trades
composite_mm 4h S=3.50 ret=1.50% 1 trade
momentum 4h 200b S=3.12 ret=5.94% 7 trades
hurst_vpin 4h 100b S=2.88 ret=1.89% 1 trade
Regime issues found:
- 1m interval: universally extreme negative Sharpe
(VBT can't model sub-minute HFT dynamics)
- iceberg: generated 0 trades across all 28 runs (signal bug)
- as_mm: many combos with 'no results' (A-S in low vol)
- mean_rev: consistent negative across all combos
Archived 67 old broken files, kept 310 clean sweep results
This commit is contained in:
ramseshk
@@ -0,0 +1,235 @@
|
||||
"""
|
||||
Batch sweep runner — run VBT backtests across all strategy/interval/limit/coin combos.
|
||||
|
||||
Usage:
|
||||
python -m backtests.sweep_runner # all combos, sequential
|
||||
python -m backtests.sweep_runner --workers 4 # 4 parallel processes
|
||||
python -m backtests.sweep_runner --strategies obi,pairs --intervals 1h,4h
|
||||
python -m backtests.sweep_runner --dry-run # show plan, don't execute
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
project_root = str(Path(__file__).resolve().parent.parent)
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
from backtests.vbt_runner import VBTBacktestRunner
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
RESULTS_DIR = Path(project_root) / "backtests" / "results"
|
||||
|
||||
# ── Sweep config ────────────────────────────────────────────
|
||||
|
||||
STRATEGIES = {
|
||||
"pairs": {"coins": ["BTC", "ETH"], "fee_model": "taker"},
|
||||
"hurst_vpin": {"coins": ["BTC"], "fee_model": "taker"},
|
||||
"as_mm": {"coins": ["BTC"], "fee_model": "maker"},
|
||||
"obi": {"coins": ["BTC"], "fee_model": "taker"},
|
||||
"grid_mm": {"coins": ["BTC"], "fee_model": "maker"},
|
||||
"composite_mm": {"coins": ["BTC"], "fee_model": "maker"},
|
||||
"iceberg": {"coins": ["BTC"], "fee_model": "taker"},
|
||||
"momentum": {"coins": ["BTC"], "fee_model": "taker"},
|
||||
"mean_rev": {"coins": ["BTC"], "fee_model": "taker"},
|
||||
}
|
||||
|
||||
INTERVALS = ["1m", "5m", "15m", "1h", "4h", "1d"]
|
||||
LIMITS = [100, 200, 500, 1000, 2000, 5000]
|
||||
|
||||
|
||||
def generate_combos(
|
||||
strategies: list[str] | None = None,
|
||||
intervals: list[str] | None = None,
|
||||
limits: list[int] | None = None,
|
||||
coins: list[str] | None = None,
|
||||
) -> list[dict]:
|
||||
"""Generate all valid strategy × interval × limit × coin combinations."""
|
||||
strats = strategies or list(STRATEGIES.keys())
|
||||
ints = intervals or INTERVALS
|
||||
lims = limits or LIMITS
|
||||
combos = []
|
||||
|
||||
for s in strats:
|
||||
cfg = STRATEGIES.get(s, {"coins": ["BTC"]})
|
||||
for interval in ints:
|
||||
for limit in lims:
|
||||
for coin in cfg["coins"]:
|
||||
combos.append({
|
||||
"strategy": s,
|
||||
"interval": interval,
|
||||
"limit": limit,
|
||||
"coin": coin,
|
||||
"fee_model": cfg.get("fee_model", "taker"),
|
||||
})
|
||||
return combos
|
||||
|
||||
|
||||
def run_one_combo(combo: dict) -> dict | None:
|
||||
"""Run a single strategy/interval/limit/coin backtest. Returns result dict."""
|
||||
strategy = combo["strategy"]
|
||||
interval = combo["interval"]
|
||||
limit = combo["limit"]
|
||||
coin = combo["coin"]
|
||||
|
||||
try:
|
||||
runner = VBTBacktestRunner(vip_tier=0, staking_tier="none")
|
||||
result = runner.run_strategy(strategy=strategy, interval=interval, limit=limit)
|
||||
|
||||
if result:
|
||||
if coin and strategy != "pairs":
|
||||
result["asset"] = coin.upper()
|
||||
ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
fname = f"{strategy}_{coin}_{interval}_{limit}_{ts}.json"
|
||||
fpath = RESULTS_DIR / fname
|
||||
with open(fpath, "w") as f:
|
||||
json.dump(result, f, default=str)
|
||||
return {"combo": combo, "filename": fname, "sharpe": result.get("sharpe", 0),
|
||||
"trades": len(result.get("trades", [])), "ret": result.get("total_return_pct", 0)}
|
||||
return {"combo": combo, "filename": None, "error": "no results"}
|
||||
|
||||
except Exception as e:
|
||||
return {"combo": combo, "filename": None, "error": str(e)}
|
||||
|
||||
|
||||
def run_sweep(
|
||||
strategies: list[str] | None = None,
|
||||
intervals: list[str] | None = None,
|
||||
limits: list[int] | None = None,
|
||||
coins: list[str] | None = None,
|
||||
workers: int = 1,
|
||||
dry_run: bool = False,
|
||||
delay: float = 0.5,
|
||||
) -> list[dict]:
|
||||
"""Run the full sweep across all combinations.
|
||||
|
||||
Args:
|
||||
workers: Number of parallel processes (1 = sequential)
|
||||
dry_run: Print plan without executing
|
||||
delay: Seconds between individual request dispatches (rate limiting)
|
||||
|
||||
Returns list of result dicts.
|
||||
"""
|
||||
combos = generate_combos(strategies, intervals, limits, coins)
|
||||
total = len(combos)
|
||||
|
||||
logger.info("=" * 60)
|
||||
logger.info("VBT Sweep: %d combinations across %d strategies", total,
|
||||
len(set(c["strategy"] for c in combos)))
|
||||
logger.info("Intervals: %s Limits: %s Coins: %s",
|
||||
list(set(c["interval"] for c in combos)),
|
||||
list(set(c["limit"] for c in combos)),
|
||||
list(set(c["coin"] for c in combos)))
|
||||
logger.info("=" * 60)
|
||||
|
||||
if dry_run:
|
||||
for c in combos:
|
||||
logger.info(" %s %s %sb %s", c["strategy"], c["interval"], c["limit"], c["coin"])
|
||||
return []
|
||||
|
||||
results = []
|
||||
start_time = time.time()
|
||||
|
||||
if workers > 1:
|
||||
with ProcessPoolExecutor(max_workers=workers) as ex:
|
||||
futures = {}
|
||||
completed = 0
|
||||
for i, combo in enumerate(combos):
|
||||
futures[ex.submit(run_one_combo, combo)] = combo
|
||||
if delay > 0 and i < total - 1:
|
||||
time.sleep(delay / workers)
|
||||
for fut in as_completed(futures):
|
||||
completed += 1
|
||||
res = fut.result()
|
||||
results.append(res)
|
||||
_log_progress(res, completed, total, start_time)
|
||||
else:
|
||||
for i, combo in enumerate(combos):
|
||||
if delay > 0 and i > 0:
|
||||
time.sleep(delay)
|
||||
res = run_one_combo(combo)
|
||||
results.append(res)
|
||||
_log_progress(res, i + 1, total, start_time)
|
||||
|
||||
# Summary
|
||||
elapsed = time.time() - start_time
|
||||
success = [r for r in results if r.get("filename")]
|
||||
errors = [r for r in results if r.get("error")]
|
||||
logger.info("=" * 60)
|
||||
logger.info("Sweep complete: %d/%d succeeded in %.0fs", len(success), total, elapsed)
|
||||
if errors:
|
||||
logger.warning("%d errors:", len(errors))
|
||||
for e in errors[:5]:
|
||||
logger.warning(" %s %s %s: %s",
|
||||
e["combo"]["strategy"], e["combo"]["interval"],
|
||||
e["combo"]["coin"], e["error"])
|
||||
if success:
|
||||
sharps = [r["sharpe"] for r in success]
|
||||
best = max(success, key=lambda r: r.get("sharpe", -999))
|
||||
logger.info("Sharpe range: %.2f to %.2f", min(sharps), max(sharps))
|
||||
logger.info("Best: %s", best)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def _log_progress(result: dict, completed: int, total: int, start_time: float):
|
||||
combo = result["combo"]
|
||||
elapsed = time.time() - start_time
|
||||
rate = completed / elapsed if elapsed > 0 else 0
|
||||
eta = (total - completed) / rate if rate > 0 else 0
|
||||
status = "+" if result.get("filename") else "x"
|
||||
detail = f' S={result.get("sharpe",0):.2f} {result.get("trades",0)}t' if result.get("filename") else f' {result.get("error","?")[:40]}'
|
||||
msg = f'[{completed:3d}/{total} {status}] {combo["strategy"]:<12s} {combo["interval"]:>3s} {combo["limit"]:>4d}b {combo["coin"]:<4s}{detail}'
|
||||
logger.info(msg)
|
||||
|
||||
|
||||
# ── CLI ─────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
p = argparse.ArgumentParser(description="VBT strategy sweep runner")
|
||||
p.add_argument("--strategies", default=None,
|
||||
help="Comma-separated strategies (default: all)")
|
||||
p.add_argument("--intervals", default=None,
|
||||
help="Comma-separated intervals (default: all)")
|
||||
p.add_argument("--limits", default=None,
|
||||
help="Comma-separated bar limits (default: all)")
|
||||
p.add_argument("--coins", default=None,
|
||||
help="Comma-separated coins (default: strategy defaults)")
|
||||
p.add_argument("--workers", type=int, default=1,
|
||||
help="Number of parallel workers (default: 1)")
|
||||
p.add_argument("--delay", type=float, default=0.5,
|
||||
help="Seconds between request dispatches for rate limiting")
|
||||
p.add_argument("--dry-run", action="store_true",
|
||||
help="Print plan without executing")
|
||||
|
||||
args = p.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
)
|
||||
|
||||
strategies = [s.strip() for s in args.strategies.split(",")] if args.strategies else None
|
||||
intervals = [i.strip() for i in args.intervals.split(",")] if args.intervals else None
|
||||
limits = [int(l.strip()) for l in args.limits.split(",")] if args.limits else None
|
||||
coins = [c.strip().upper() for c in args.coins.split(",")] if args.coins else None
|
||||
|
||||
run_sweep(
|
||||
strategies=strategies,
|
||||
intervals=intervals,
|
||||
limits=limits,
|
||||
coins=coins,
|
||||
workers=args.workers,
|
||||
delay=args.delay,
|
||||
dry_run=args.dry_run,
|
||||
)
|
||||
Reference in New Issue
Block a user