diff --git a/quant/sim_validate.py b/quant/sim_validate.py new file mode 100644 index 0000000..8daff0a --- /dev/null +++ b/quant/sim_validate.py @@ -0,0 +1,230 @@ +""" +Queue simulation validation — compare VBT candle backtest PnL +against realistic event-driven simulation with queue position, +partial fills, adverse selection, and cancel latency. + +This is the critical gating item before any live deployment. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +import numpy as np + +from sim.engine import SimulationEngine, SimConfig +from sim.maker import MakerConfig, Quote +from sim.fills import FillModelConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class SimValidationResult: + """Comparison between VBT (optimistic) and simulator (realistic) PnL.""" + + strategy: str + interval: str + vbt_sharpe: float + vbt_return_pct: float + vbt_trades: int + sim_sharpe: float + sim_return_pct: float + sim_fill_rate: float # fraction of VBT entry signals that filled + sim_adverse_pct: float # % of fills flagged as adverse + sim_partial_fills: int + sim_total_fills: int + sim_cancels: int + degradation_pct: float # (vbt_return - sim_return) / abs(vbt_return) + verdict: str = "" + + def summary(self) -> dict: + return { + "strategy": self.strategy, + "vbt_sharpe": round(self.vbt_sharpe, 2), + "sim_sharpe": round(self.sim_sharpe, 2), + "vbt_return_pct": round(self.vbt_return_pct, 2), + "sim_return_pct": round(self.sim_return_pct, 2), + "vbt_trades": self.vbt_trades, + "sim_fill_rate": round(self.sim_fill_rate, 2), + "sim_adverse_pct": round(self.sim_adverse_pct, 2), + "degradation_pct": round(self.degradation_pct, 2), + "verdict": self.verdict, + } + + +class SimulationValidator: + """Run a strategy through the event-driven market-making simulator + and compare against VBT candle backtest results. + + Usage: + validator = SimulationValidator() + result = validator.validate('grid_mm', '1h', limit=720) + print(result.summary()) + """ + + def __init__( + self, + partial_fill_prob: float = 0.25, + adverse_selection_prob: float = 0.10, + adverse_move_bps: float = 3.0, + cancel_latency_ms: float = 50.0, + maker_fee_pct: float = 0.00015, + taker_fee_pct: float = 0.00045, + max_inventory: float = 0.005, + base_size: float = 0.001, + ): + self._fills_config = FillModelConfig( + partial_fill_prob=partial_fill_prob, + adverse_selection_prob=adverse_selection_prob, + adverse_move_bps=adverse_move_bps, + cancel_latency_ms=cancel_latency_ms, + ) + + self._config = SimConfig( + maker=MakerConfig(base_size=base_size, max_inventory=max_inventory), + fills=self._fills_config, + max_inventory=max_inventory, + maker_fee_pct=maker_fee_pct, + taker_fee_pct=taker_fee_pct, + cancel_after_ms=5000, + quote_refresh_ms=2000, + seed=42, + ) + + def validate( + self, + strategy: str = "grid_mm", + interval: str = "1h", + limit: int = 720, + params: dict | None = None, + ) -> SimValidationResult: + """Run VBT backtest + event-driven simulation, compare results.""" + + # Step 1: Run VBT backtest + from backtests.vbt_runner import VBTBacktestRunner + runner = VBTBacktestRunner() + vbt_result = runner.run_strategy(strategy=strategy, interval=interval, limit=limit, params=params) + + if not vbt_result or not vbt_result.get("trades"): + return SimValidationResult( + strategy=strategy, interval=interval, + vbt_sharpe=0, vbt_return_pct=0, vbt_trades=0, + sim_sharpe=0, sim_return_pct=0, sim_fill_rate=0, + sim_adverse_pct=0, sim_partial_fills=0, sim_total_fills=0, + sim_cancels=0, degradation_pct=0, verdict="NO_TRADES", + ) + + vbt_sharpe = vbt_result.get("sharpe", 0) + vbt_return = vbt_result.get("total_return_pct", 0) + vbt_trades = len(vbt_result.get("trades", [])) + ec = vbt_result.get("equity_curve", []) + + if not ec: + return SimValidationResult( + strategy=strategy, interval=interval, + vbt_sharpe=vbt_sharpe, vbt_return_pct=vbt_return, vbt_trades=vbt_trades, + sim_sharpe=0, sim_return_pct=0, sim_fill_rate=0, + sim_adverse_pct=0, sim_partial_fills=0, sim_total_fills=0, + sim_cancels=0, degradation_pct=0, verdict="NO_EQUITY_CURVE", + ) + + # Step 2: Build event stream from equity curve + # Convert equity curve points into simulated L2 + trade events + events = [] + for i, pt in enumerate(ec): + px = pt.get("v", 10000) + mid = px * 0.9999 if i > 0 else px # synthetic mid from equity + + # Simulate L2 snapshot at this point + bids = {round(mid * 0.9995, 2): 1.0, round(mid * 0.999, 2): 2.0} + asks = {round(mid * 1.0005, 2): 1.0, round(mid * 1.001, 2): 2.0} + + events.append({ + "type": "l2", + "data": {"bids": bids, "asks": asks}, + "time": float(i) * 0.5, # simulate half-second between events + "coin": "BTC", + }) + + # Simulate occasional trades + if i % 3 == 0 and mid > 0: + events.append({ + "type": "trade", + "data": {"px": mid, "sz": 0.01, "side": "B"}, + "time": float(i) * 0.5 + 0.1, + "coin": "BTC", + }) + + # Step 3: Run event-driven simulator + engine = SimulationEngine(config=self._config, seed=42) + engine.run(events) + + # Step 4: Collect simulation stats + sim_stats = engine.stats() + sim_breakdown = engine.breakdown() + + sim_return = (sim_breakdown.net_pnl / self._config.initial_equity) * 100 + sim_fills = sim_stats.bid_fills + sim_stats.ask_fills + + fill_rate = min(1.0, sim_fills / vbt_trades) if vbt_trades > 0 else 0 + adverse_pct = sim_stats.adverse_rate * 100 + + degradation = (vbt_return - sim_return) / max(abs(vbt_return), 0.01) * 100 if vbt_return != 0 else 0 + + # Verdict + if sim_return > 0 and sim_fills > 3 and degradation < 50: + verdict = "SIMULATE" # simulator says profitable → proceed to live + elif sim_return > 0 and sim_fills > 3: + verdict = "CAUTION" # profitable but heavily degraded + else: + verdict = "DISCARD" # simulator says unprofitable + + return SimValidationResult( + strategy=strategy, interval=interval, + vbt_sharpe=round(vbt_sharpe, 3), + vbt_return_pct=round(vbt_return, 2), + vbt_trades=vbt_trades, + sim_sharpe=round(sim_stats.sharpe, 3), + sim_return_pct=round(sim_return, 2), + sim_fill_rate=round(fill_rate, 3), + sim_adverse_pct=round(adverse_pct, 1), + sim_partial_fills=sim_stats.partial_fills if hasattr(sim_stats, 'partial_fills') else 0, + sim_total_fills=sim_fills, + sim_cancels=sim_stats.cancels, + degradation_pct=round(degradation, 1), + verdict=verdict, + ) + + +class BatchSimValidator: + """Run simulation validation across multiple strategies and params.""" + + def __init__(self, **kwargs): + self._validator_kwargs = kwargs + + def validate_grid_mm_variants(self, interval: str = "1h") -> list[SimValidationResult]: + """Run grid_mm through simulator with multiple param combos.""" + results = [] + configs = [ + {"grid_levels": 5, "spacing_bps": 1, "rebalance_every": 5}, + {"grid_levels": 5, "spacing_bps": 5, "rebalance_every": 10}, + {"grid_levels": 10, "spacing_bps": 10, "rebalance_every": 20}, + ] + + for params in configs: + validator = SimulationValidator(**self._validator_kwargs) + result = validator.validate("grid_mm", interval, params=params) + results.append(result) + + return results + + def print_report(self, results: list[SimValidationResult]): + print(f'{"Params":<30} {"VBT S":>7} {"Sim S":>7} {"VBT ret":>8} {"Sim ret":>8} {"Fill%":>6} {"Adv%":>6} {"Verdict"}') + print("-" * 95) + for r in results: + print(f'{r.interval:<30} {r.vbt_sharpe:>7.2f} {r.sim_sharpe:>7.2f} {r.vbt_return_pct:>7.1f}% {r.sim_return_pct:>7.1f}% {r.sim_fill_rate:>6.1%} {r.sim_adverse_pct:>5.1f}% {r.verdict}')