""" Systematic walk-forward validation across all strategies. Runs every strategy through walk-forward IS/OOS backtesting with statistical significance testing (DSR, PSR, Sharpe Haircut). Produces: - Per-strategy walk-forward reports - Composite significance scores - Strategy ranking by robustness - Deploy/simulate/discard recommendations Usage: python strategies/wf_validate_all.py # all strategies, 1h interval python strategies/wf_validate_all.py --strategy pairs # single strategy python strategies/wf_validate_all.py --interval 4h # different interval python strategies/wf_validate_all.py --n-windows 5 # more windows """ from __future__ import annotations import argparse import json import logging import sys import time from datetime import datetime, timezone from pathlib import Path import numpy as np sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from quant.walkforward import WalkForwardRunner from quant.significance import QuantVerdict, validate_strategy logger = logging.getLogger(__name__) VALIDATION_STRATEGIES = [ "pairs", "hurst_vpin", "as_mm", "obi", "grid_mm", "composite_mm", "iceberg", "momentum", "mean_rev", "cross_sectional", "spot_perp_basis", "regime_ensemble", ] INTERVALS = ["1h", "4h", "1d"] def run_full_validation( strategies: list[str] | None = None, intervals: list[str] | None = None, n_windows: int = 5, fee_tier: int = 0, staking_tier: str = "none", save_results: bool = True, ) -> dict: """Run walk-forward validation on all specified strategies and intervals. Returns a dict with strategy → interval → report. """ strats = strategies or VALIDATION_STRATEGIES ints = intervals or INTERVALS results: dict[str, dict] = {} total = len(strats) * len(ints) completed = 0 logger.info("=" * 60) logger.info("Walk-Forward Validation: %d strategies × %d intervals = %d runs", len(strats), len(ints), total) logger.info("Windows: %d | Fee tier: %d | Staking: %s", n_windows, fee_tier, staking_tier) logger.info("=" * 60) for strategy in strats: results[strategy] = {} for interval in ints: completed += 1 t_start = time.time() logger.info("[%d/%d] %s @ %s...", completed, total, strategy, interval) try: wfr = WalkForwardRunner( n_windows=n_windows, fee_tier=fee_tier, staking_tier=staking_tier, ) report = wfr.run(strategy=strategy, interval=interval) elapsed = time.time() - t_start if report.windows: sig = report.significance_report(n_trials=len(strats) * len(ints)) ver = validate_strategy( sharpe=report.avg_oos_sharpe, n_trades=max(report.total_oos_trades, 1), n_trials=len(strats) * len(ints), wf_consistency=report.consistency, ) results[strategy][interval] = { "strategy": strategy, "interval": interval, "n_windows": report.n_windows, "consistency": round(report.consistency, 3), "avg_oos_sharpe": round(report.avg_oos_sharpe, 3), "oos_sharpe": round(report.oos_sharpe, 3), "performance_decay": round(report.performance_decay, 3), "total_trades": report.total_oos_trades, "deflated_sharpe": sig["deflated_sharpe"], "psr": sig["psr"], "haircut_sharpe": sig["haircut_sharpe"], "verdict": sig["verdict"], "score": sig["score"], "recommendation": sig["recommendation"], "elapsed_s": round(elapsed, 1), } logger.info(" → W%d WF=%.2f S=%.2f DSR=%.3f %s @ %.1fs", len(report.windows), report.consistency, report.avg_oos_sharpe, sig["deflated_sharpe"], sig["verdict"], elapsed) else: results[strategy][interval] = { "strategy": strategy, "interval": interval, "error": "no_windows", "elapsed_s": round(elapsed, 1), } logger.info(" → No windows (insufficient data)") except Exception as e: elapsed = time.time() - t_start results[strategy][interval] = { "strategy": strategy, "interval": interval, "error": str(e)[:100], "elapsed_s": round(elapsed, 1), } logger.warning(" → Error: %s", e) # Print unified summary _print_summary(results) if save_results: _save_results(results) return results def _print_summary(results: dict): print(f"\n{'=' * 80}") print(f" Walk-Forward Validation Summary") print(f"{'=' * 80}") print(f"{'Strategy':<20} {'Int':>4} {'W':>3} {'Consist':>8} {'OOS Sh':>7} {'Decay':>7} {'DSR':>6} {'Verdict':>10}") print("-" * 80) rankings = [] for strategy in sorted(results): for interval in sorted(results.get(strategy, {})): r = results[strategy][interval] if r.get("error"): continue rankings.append(r) print(f"{r['strategy']:<20} {r['interval']:>4} {r['n_windows']:>3} " f"{r['consistency']:>7.0%} {r['avg_oos_sharpe']:>7.2f} " f"{r['performance_decay']:>7.2f} {r['deflated_sharpe']:>6.3f} " f"{r['verdict']:>10}") rankings.sort(key=lambda x: x.get("deflated_sharpe", 0), reverse=True) print(f"\n--- Top 10 by Deflated Sharpe Ratio ---") for i, r in enumerate(rankings[:10]): deploy_mark = " ✅" if r["verdict"] == "DEPLOY" else (" ⚠️" if r["verdict"] == "SIMULATE" else " ❌") print(f" {i+1:2d}. {r['strategy']:<20s} {r['interval']:>4s} " f"DSR={r['deflated_sharpe']:>6.3f} {r['verdict']}{deploy_mark}") deployable = [r for r in rankings if r["verdict"] == "DEPLOY"] simulate = [r for r in rankings if r["verdict"] == "SIMULATE"] discarded = [r for r in rankings if r["verdict"] == "DISCARD"] print(f"\nVerdict breakdown:") print(f" DEPLOY: {len(deployable)}") print(f" SIMULATE: {len(simulate)}") print(f" DISCARD: {len(discarded)}") if deployable: print(f"\nDeployable strategies (sorted by DSR):") for r in sorted(deployable, key=lambda x: x["deflated_sharpe"], reverse=True): print(f" ✅ {r['strategy']}/{r['interval']}: " f"OOS Sharpe={r['avg_oos_sharpe']:.2f}, DSR={r['deflated_sharpe']:.3f}") def _save_results(results: dict): timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") out_path = Path(__file__).resolve().parent.parent / "backtests" / "results" / f"wf_validation_{timestamp}.json" flat = {} for strategy, intervals in results.items(): for interval, report in intervals.items(): flat[f"{strategy}/{interval}"] = report out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w") as f: json.dump(flat, f, indent=2, default=str) logger.info("Results saved to %s", out_path) def main(): p = argparse.ArgumentParser(description="Walk-Forward Validation — All Strategies") p.add_argument("--strategy", "-s", nargs="+", help="Strategies to validate (default: all)") p.add_argument("--interval", "-i", nargs="+", help="Intervals to test (default: 1h,4h,1d)") p.add_argument("--n-windows", type=int, default=5, help="Number of walk-forward windows (default: 5)") p.add_argument("--fee-tier", type=int, default=0, help="VIP fee tier 0-6 (default: 0)") p.add_argument("--staking-tier", default="none", help="Staking tier (default: none)") p.add_argument("--no-save", action="store_true", help="Don't save results to disk") args = p.parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S", ) run_full_validation( strategies=args.strategy, intervals=args.interval, n_windows=args.n_windows, fee_tier=args.fee_tier, staking_tier=args.staking_tier, save_results=not args.no_save, ) if __name__ == "__main__": main()