""" Tick-level backtest runner — replays real Parquet L2/trade events through the event-driven simulation engine. This is the foundation for HFT strategy validation. Unlike the VBT runner which uses candle data, this replays every L2 snapshot, trade tick, and mark price update sequentially through the queue-position-aware simulator. Usage: python backtests/tick_runner.py --coin BTC --start 2026-08-01 --end 2026-08-07 python backtests/tick_runner.py --coin ETH --days 3 --maker as_mm --gamma 0.15 python backtests/tick_runner.py --coin BTC --maker vpin_as_mm --vpin-threshold 0.3 """ from __future__ import annotations import argparse import json import logging import math import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from sim.engine import SimulationEngine, SimConfig from sim.maker import AvellanedaStoikovMaker, MakerConfig from sim.fills import FillModelConfig from sim.scenario import ScenarioConfig from quant.significance import QuantVerdict logger = logging.getLogger(__name__) RESULTS_DIR = Path(__file__).resolve().parent / "results" / "tick" RESULTS_DIR.mkdir(parents=True, exist_ok=True) class VPINGatedASMaker(AvellanedaStoikovMaker): """A-S market maker with VPIN toxicity gating and inventory skew. Inherits the base A-S quoting logic and adds: - VPIN gating: stop quoting when VPIN exceeds threshold - Inventory skew: bias quotes toward reducing inventory - Dynamic spread: widen spread when VPIN is elevated but below alarm """ def __init__( self, config: MakerConfig | None = None, vpin_threshold: float = 0.30, vpin_alarm: float = 0.50, vpin_window: int = 50, ): super().__init__(config) self._vpin_threshold = vpin_threshold self._vpin_alarm = vpin_alarm self._current_vpin: float = 0.0 self._buy_vol: list[float] = [] self._sell_vol: list[float] = [] self._vpin_window = vpin_window self._vpins: list[float] = [] def update_vpin(self, buy_vol: float, sell_vol: float): self._buy_vol.append(buy_vol) self._sell_vol.append(sell_vol) if len(self._buy_vol) > self._vpin_window * 20: self._buy_vol = self._buy_vol[-self._vpin_window * 20:] self._sell_vol = self._sell_vol[-self._vpin_window * 20:] self._recompute_vpin() def _recompute_vpin(self): from microstructure.toxicity import compute_vpin result = compute_vpin( list(self._buy_vol), list(self._sell_vol), n_buckets=self._vpin_window, ) self._current_vpin = result.get("vpin_value", 0.0) self._vpins.append(self._current_vpin) if len(self._vpins) > 200: self._vpins = self._vpins[-200:] @property def vpin(self) -> float: return self._current_vpin def allowed_to_quote(self) -> tuple[bool, float]: if self._current_vpin >= self._vpin_alarm: return False, 0.0 if self._current_vpin >= self._vpin_threshold: reduction = (self._current_vpin - self._vpin_threshold) / ( self._vpin_alarm - self._vpin_threshold ) return True, max(0.0, 1.0 - reduction) return True, 1.0 def quote( self, mid_price: float, inventory: float, elapsed_hours: float, ) -> Optional["Quote"]: from sim.maker import Quote allowed, size_mult = self.allowed_to_quote() if not allowed or size_mult <= 0: return None base = super().quote(mid_price, inventory, elapsed_hours) inv_ratio = inventory / max(self._cfg.max_inventory, 0.0001) skew = inv_ratio * self._cfg.skew_factor * base.spread_bps / 10000 * mid_price spread_mult = 1.0 if self._current_vpin >= self._vpin_threshold * 0.7: spread_mult = 1.0 + (self._current_vpin - self._vpin_threshold * 0.7) / ( self._vpin_alarm - self._vpin_threshold * 0.7 ) bid = base.bid - skew ask = base.ask - skew half_spread = base.spread_bps / 10000 * mid_price * spread_mult / 2 bid = mid_price + (bid - mid_price) - half_spread * (spread_mult - 1) ask = mid_price + (ask - mid_price) + half_spread * (spread_mult - 1) if inventory > 0: bid_size = base.bid_size * size_mult * (1.0 - inv_ratio * self._cfg.skew_factor) ask_size = base.ask_size * size_mult * (1.0 + inv_ratio * self._cfg.skew_factor) else: bid_size = base.bid_size * size_mult * (1.0 + abs(inv_ratio) * self._cfg.skew_factor) ask_size = base.ask_size * size_mult * (1.0 - abs(inv_ratio) * self._cfg.skew_factor) bid = max(bid, 1.0) ask = max(ask, bid + mid_price * self._cfg.min_spread_bps / 10000) bid_size = max(bid_size, self._cfg.base_size * 0.1) ask_size = max(ask_size, self._cfg.base_size * 0.1) new_spread_bps = (ask - bid) / mid_price * 10000 if mid_price > 0 else 0 return Quote( bid=round(bid, 2), ask=round(ask, 2), bid_size=round(bid_size, 6), ask_size=round(ask_size, 6), reservation=round(base.reservation, 2), spread_bps=round(new_spread_bps, 2), ) class TickBacktestRunner: """Replays stored Parquet L2/trade events through the simulation engine. The engine processes events sequentially: 1. L2 update → update book, maybe re-quote 2. Trade → check fills, update PnL 3. Periodic → funding tick, re-quote, cancel stale orders """ def __init__( self, data_dir: str = "data/raw", maker_type: str = "as_mm", maker_config: MakerConfig | None = None, vpin_threshold: float = 0.30, vpin_alarm: float = 0.50, gamma: float = 0.1, base_size: float = 0.001, max_inventory: float = 0.005, skew_factor: float = 0.5, maker_fee_pct: float = 0.0002, taker_fee_pct: float = 0.0005, adverse_selection_prob: float = 0.15, cancel_after_ms: float = 5000.0, quote_refresh_ms: float = 2000.0, seed: int | None = 42, ): self._data_dir = data_dir self._maker_type = maker_type self._vpin_threshold = vpin_threshold self._vpin_alarm = vpin_alarm self._maker_fee_pct = maker_fee_pct self._taker_fee_pct = taker_fee_pct maker_cfg = maker_config or MakerConfig( gamma=gamma, base_size=base_size, max_inventory=max_inventory, skew_factor=skew_factor, ) if maker_type == "vpin_as_mm": self._maker = VPINGatedASMaker( config=maker_cfg, vpin_threshold=vpin_threshold, vpin_alarm=vpin_alarm, ) else: self._maker = AvellanedaStoikovMaker(maker_cfg) self._sim_config = SimConfig( maker=maker_cfg, fills=FillModelConfig(adverse_selection_prob=adverse_selection_prob), maker_fee_pct=maker_fee_pct, taker_fee_pct=taker_fee_pct, cancel_after_ms=cancel_after_ms, quote_refresh_ms=quote_refresh_ms, seed=seed, ) self._engine: Optional[SimulationEngine] = None self._seed = seed def load_events( self, coin: str, start_date: str, end_date: str, ) -> list[dict]: """Load L2 and trade events from Parquet store and merge into a single timeline. Returns a list of events sorted by timestamp, each with: {"type": "l2"|"trade", "data": {...}, "time": float, "coin": str} """ from data.store import read_range logger.info("Loading L2 data for %s from %s to %s...", coin, start_date, end_date) l2_msgs = read_range(self._data_dir, channel="l2book", coin=coin.upper(), start_date=start_date, end_date=end_date) logger.info("Loaded %d L2 messages", len(l2_msgs)) logger.info("Loading trade data for %s from %s to %s...", coin, start_date, end_date) trade_msgs = read_range(self._data_dir, channel="trades", coin=coin.upper(), start_date=start_date, end_date=end_date) logger.info("Loaded %d trade messages", len(trade_msgs)) events = [] t0 = None for msg in l2_msgs: payload = msg.get("payload", {}) local_ts = msg.get("local_ts", 0) if not t0: t0 = local_ts bids = {} asks = {} levels = payload.get("levels", []) msg_type = payload.get("type", "snapshot") if msg_type == "snapshot" and isinstance(levels, list): if len(levels) >= 1: for bid in levels[0]: sz = float(bid.get("sz", 0)) if sz > 0: bids[float(bid["px"])] = sz if len(levels) >= 2: for ask in levels[1]: sz = float(ask.get("sz", 0)) if sz > 0: asks[float(ask["px"])] = sz elif msg_type == "delta": delta = payload.get("delta", {}) if delta: px = float(delta.get("px", 0)) sz = float(delta.get("sz", 0)) side = delta.get("side", "B") if sz <= 0: continue bids = {px: sz} if side == "B" else {} asks = {px: sz} if side == "A" else {} if bids or asks: events.append({ "type": "l2", "data": {"bids": bids, "asks": asks}, "time": local_ts - t0 if t0 else local_ts, "coin": coin.upper(), }) for msg in trade_msgs: payload = msg.get("payload", {}) local_ts = msg.get("local_ts", 0) events.append({ "type": "trade", "data": { "px": payload.get("px", "0"), "sz": payload.get("sz", "0"), "side": payload.get("side", "B"), }, "time": local_ts - t0 if t0 else local_ts, "coin": coin.upper(), }) events.sort(key=lambda e: e["time"]) logger.info("Total events: %d (%.1f hours)", len(events), (events[-1]["time"] - events[0]["time"]) / 3600 if events else 0) return events def run( self, coin: str, start_date: str, end_date: str, ) -> dict: """Load events and run the simulation engine. Returns result dict.""" events = self.load_events(coin, start_date, end_date) if not events or len(events) < 2: logger.error("Not enough events to run backtest") return self._empty_result(coin, start_date, end_date) engine = SimulationEngine( config=self._sim_config, maker=self._maker, seed=self._seed, ) engine.run(events) self._engine = engine stats = engine.stats() breakdown = engine.breakdown() equity_curve = engine.reporter.equity_curve total_trades = stats.total_trades duration_hours = (events[-1]["time"] - events[0]["time"]) / 3600 if events else 0 returns = [] eq_vals = [p["v"] for p in equity_curve] for i in range(1, len(eq_vals)): if eq_vals[i - 1] > 0: returns.append(math.log(eq_vals[i] / eq_vals[i - 1])) wf_consistency = 0.5 if stats.sharpe > 0 else 0.0 verdict = QuantVerdict( observed_sharpe=stats.sharpe, wf_consistency=wf_consistency, n_trials=4, n_periods=max(total_trades, 1), positive_regimes=1 if stats.sharpe > 0 else 0, ).evaluate() vpin_curve = [] if isinstance(self._maker, VPINGatedASMaker): vpin_curve = self._maker._vpins[-200:] result = { "strategy": self._maker_type, "coin": coin.upper(), "start_date": start_date, "end_date": end_date, "duration_hours": round(duration_hours, 2), "n_events": len(events), "total_trades": total_trades, "bid_fills": stats.bid_fills, "ask_fills": stats.ask_fills, "cancels": stats.cancels, "toxic_fills": stats.toxic_fills, "adverse_rate": stats.adverse_rate, "avg_spread_bps": stats.avg_spread_bps, "max_inventory": stats.max_inventory, "max_drawdown_pct": stats.max_drawdown, "sharpe": stats.sharpe, "sortino": stats.sortino, "uptime_pct": stats.uptime_pct, "pnl_breakdown": { "spread_capture": breakdown.spread_capture, "inventory_pnl": breakdown.inventory_pnl, "maker_fees": breakdown.maker_fees, "taker_fees": breakdown.taker_fees, "funding_pnl": breakdown.funding_pnl, "adverse_selection_cost": breakdown.adverse_selection_cost, "gross_pnl": breakdown.gross_pnl, "net_pnl": breakdown.net_pnl, }, "equity_curve": equity_curve, "vpin_curve": vpin_curve, "verdict": verdict["verdict"], "dsr": verdict["deflated_sharpe"], "psr": verdict["psr"], "haircut_sharpe": verdict["haircut_sharpe"], "maker_fee_pct": self._maker_fee_pct, "taker_fee_pct": self._taker_fee_pct, "maker_params": { "gamma": self._maker.config.gamma, "base_size": self._maker.config.base_size, "max_inventory": self._maker.config.max_inventory, "skew_factor": self._maker.config.skew_factor, "vpin_threshold": self._vpin_threshold if self._maker_type == "vpin_as_mm" else None, "vpin_alarm": self._vpin_alarm if self._maker_type == "vpin_as_mm" else None, }, "generated_at": datetime.now(timezone.utc).isoformat(), } self._save_result(result) return result def _empty_result(self, coin: str, start_date: str, end_date: str) -> dict: return { "strategy": self._maker_type, "coin": coin.upper(), "start_date": start_date, "end_date": end_date, "duration_hours": 0, "n_events": 0, "total_trades": 0, "bid_fills": 0, "ask_fills": 0, "cancels": 0, "toxic_fills": 0, "adverse_rate": 0, "max_drawdown_pct": 0, "sharpe": 0, "sortino": 0, "pnl_breakdown": {}, "equity_curve": [], "vpin_curve": [], "verdict": "INSUFFICIENT_DATA", "error": "No events available", "generated_at": datetime.now(timezone.utc).isoformat(), } def _save_result(self, result: dict): coin = result["coin"] strategy = result["strategy"] start = result["start_date"] end = result["end_date"] fname = f"{strategy}_{coin}_{start}_{end}_{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}.json" fpath = RESULTS_DIR / fname with open(fpath, "w") as f: json.dump(result, f, default=str) logger.info("Saved result to %s", fpath) def print_summary(self, result: dict): print("\n" + "=" * 60) print(f" Tick Backtest — {result['strategy']} on {result['coin']}") print(f" Period: {result['start_date']} → {result['end_date']}") print(f" Duration: {result['duration_hours']}h | Events: {result['n_events']:,}") print(f" Verdict: {result['verdict']}") print("=" * 60) print(f" Trades: {result['total_trades']} " f"(bids: {result['bid_fills']}, asks: {result['ask_fills']})") print(f" Cancels: {result['cancels']}") print(f" Toxic fills: {result['toxic_fills']} ({result['adverse_rate']:.1%})") print(f" Avg spread: {result['avg_spread_bps']:.1f} bps") print(f" Max inventory: {result['max_inventory']:.6f}") print(f" Max drawdown: {result['max_drawdown_pct']:.2f}%") print(f" Sharpe: {result['sharpe']:.3f} Sortino: {result['sortino']:.3f}") print(f" DSR: {result['dsr']:.4f} PSR: {result['psr']:.4f} " f"Haircut: {result['haircut_sharpe']:.4f}") print(f"\n PnL Breakdown:") bd = result["pnl_breakdown"] print(f" Spread capture: ${bd.get('spread_capture', 0):>10.4f}") print(f" Inventory PnL: ${bd.get('inventory_pnl', 0):>10.4f}") print(f" Maker fees: ${bd.get('maker_fees', 0):>10.4f}") print(f" Taker fees: ${bd.get('taker_fees', 0):>10.4f}") print(f" Funding PnL: ${bd.get('funding_pnl', 0):>10.4f}") print(f" Adverse selection: ${bd.get('adverse_selection_cost', 0):>10.4f}") print(f" " + "─" * 35) print(f" Gross PnL: ${bd.get('gross_pnl', 0):>10.4f}") print(f" Net PnL: ${bd.get('net_pnl', 0):>10.4f}") print(f"\n Fee model: maker={result['maker_fee_pct']*100:.2f}% " f"taker={result['taker_fee_pct']*100:.2f}%") def cmd_tick_backtest(args): runner = TickBacktestRunner( data_dir=args.data_dir, maker_type=args.maker, gamma=args.gamma, base_size=args.base_size, max_inventory=args.max_inventory, skew_factor=args.skew_factor, vpin_threshold=args.vpin_threshold, vpin_alarm=args.vpin_alarm, maker_fee_pct=args.maker_fee / 100, taker_fee_pct=args.taker_fee / 100, adverse_selection_prob=args.adverse_prob, cancel_after_ms=args.cancel_after_ms, quote_refresh_ms=args.quote_refresh_ms, seed=args.seed, ) result = runner.run( coin=args.coin, start_date=args.start_date, end_date=args.end_date, ) runner.print_summary(result) return result if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S") p = argparse.ArgumentParser(description="Tick-level backtest runner") p.add_argument("--coin", default="BTC") p.add_argument("--data-dir", default="data/raw") p.add_argument("--start-date", default="2026-08-01") p.add_argument("--end-date", default="2026-08-07") p.add_argument("--maker", default="as_mm", choices=["as_mm", "vpin_as_mm"]) p.add_argument("--gamma", type=float, default=0.1) p.add_argument("--base-size", type=float, default=0.001) p.add_argument("--max-inventory", type=float, default=0.005) p.add_argument("--skew-factor", type=float, default=0.5) p.add_argument("--vpin-threshold", type=float, default=0.30) p.add_argument("--vpin-alarm", type=float, default=0.50) p.add_argument("--maker-fee", type=float, default=0.02, help="Maker fee in % (0.02 = 2bps)") p.add_argument("--taker-fee", type=float, default=0.05, help="Taker fee in % (0.05 = 5bps)") p.add_argument("--adverse-prob", type=float, default=0.15) p.add_argument("--cancel-after-ms", type=float, default=5000.0) p.add_argument("--quote-refresh-ms", type=float, default=2000.0) p.add_argument("--seed", type=int, default=42) args = p.parse_args() cmd_tick_backtest(args)