ramseshk
0446443d36
New strategies: - Cross-Sectional Momentum: long top-N, short bottom-N across HL universe - Spot-Perp Basis Arbitrage: delta-neutral spot vs perp price gap trading - Regime-Switching Ensemble: dynamically allocates strategies by market regime - Portfolio Construction: risk parity, vol targeting, correlation penalty Infrastructure: - DuckDBDataProvider: real tick/candle data for backtests (replaces synthetic) - Walk-Forward Validation: systematic IS/OOS across all 12 strategies - 3 Jupyter research notebooks (EDA, strategy research, portfolio) Pipeline integration: - deploy.py registry, sweep_runner, vbt_runner all updated - fee_tiers support for new strategies - All modules syntax-validated and import-tested
239 lines
9.3 KiB
Python
239 lines
9.3 KiB
Python
"""
|
||
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"},
|
||
"cross_sectional": {"coins": ["BTC","ETH","SOL","HYPE","ARB","OP"], "fee_model": "taker"},
|
||
"spot_perp_basis": {"coins": ["BTC"], "fee_model": "taker"},
|
||
"regime_ensemble": {"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,
|
||
)
|