""" Slippage Calibration — replace 0.1bps-per-level heuristic with VST-calibrated model. The current CWM uses: expected_slippage_bps = levels_consumed * 0.1 This is a constant heuristic. Real slippage is CONDITIONAL on: - Book depth at fill time - Order size relative to book depth - Market regime (trending vs choppy) - Asset-specific depth profile (alpha from power-law decay) Calibration process: 1. Collect VST fill data: (levels_consumed, slippage_bps, order_usd, book_depth_usd) 2. Fit per-asset model: slippage_bps = alpha * levels + beta * (order_usd / book_depth_usd) + gamma 3. Store calibrated params in AssetBehavior 4. CWM uses calibrated params instead of 0.1bps constant The key insight from the OB study: - BTC alpha=0.70: deep book → ~0.05 bps/level - ETH alpha=0.75: ~0.08 bps/level - SOL alpha=0.85: ~0.15 bps/level - DOGE alpha=1.00: thin book → ~0.3 bps/level VST tape format (from Flight7): Each row: timestamp, symbol, side, fill_price, fill_qty, best_bid, best_ask, bid_depth_levels, ask_depth_levels, bid_depth_usd, ask_depth_usd, spread_bps, order_latency_ms """ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple import math @dataclass(frozen=True, slots=True) class SlippageCalibration: """Per-asset calibrated slippage model. Model: slippage_bps = alpha * levels_consumed + beta * (order_usd / book_depth_usd) + intercept """ alpha: float = 0.1 # bps per level consumed (default: 0.1) beta: float = 0.5 # bps per unit of order/book depth ratio intercept: float = 0.0 # base slippage (bps) n_samples: int = 0 # how many fills used for calibration r_squared: float = 0.0 # model fit quality def expected_slippage_bps( self, levels_consumed: int, order_usd: float = 0.0, book_depth_usd: float = 1.0, ) -> float: """Predict slippage based on fill parameters.""" depth_ratio = order_usd / max(book_depth_usd, 1.0) return self.alpha * levels_consumed + self.beta * depth_ratio + self.intercept @dataclass class SlippageCalibrator: """Calibrate slippage model from VST fill data. Usage: calibrator = SlippageCalibrator() for fill in vst_fills: calibrator.record(fill) model = calibrator.calibrate() """ _records: List[Dict[str, float]] = field(default_factory=list) def record( self, levels_consumed: int, slippage_bps: float, order_usd: float, book_depth_usd: float, is_maker: bool = False, spread_bps: float = 0.0, ) -> None: """Record a VST fill for calibration.""" self._records.append({ "levels": levels_consumed, "slippage": slippage_bps, "order_usd": order_usd, "book_depth_usd": book_depth_usd, "is_maker": float(is_maker), "spread_bps": spread_bps, }) def calibrate(self) -> SlippageCalibration: """Fit slippage model using least-squares regression. Model: slippage_bps = alpha * levels + beta * (order_usd / book_depth_usd) + intercept """ if len(self._records) < 10: return SlippageCalibration() # Build feature matrix: [levels, depth_ratio] X = [] y = [] for r in self._records: depth_ratio = r["order_usd"] / max(r["book_depth_usd"], 1.0) X.append([r["levels"], depth_ratio]) y.append(r["slippage"]) n = len(X) # Simple least-squares: y = alpha*x1 + beta*x2 + intercept # Using normal equations: (X^T X)^-1 X^T y sum_x1 = sum(row[0] for row in X) sum_x2 = sum(row[1] for row in X) sum_y = sum(y) sum_x1sq = sum(row[0]**2 for row in X) sum_x2sq = sum(row[1]**2 for row in X) sum_x1x2 = sum(row[0]*row[1] for row in X) sum_x1y = sum(row[0]*y[i] for i, row in enumerate(X)) sum_x2y = sum(row[1]*y[i] for i, row in enumerate(X)) # Normal equations matrix det = (n * sum_x1sq * sum_x2sq + 2 * sum_x1 * sum_x2 * sum_x1x2 - sum_x1sq * sum_x2**2 - sum_x2sq * sum_x1**2 - n * sum_x1x2**2) if abs(det) < 1e-12: return SlippageCalibration() alpha = (sum_x2sq * sum_x1y - sum_x1x2 * sum_x2y + sum_x1x2 * sum_y - sum_x1 * sum_x2 * sum_x1y / n) / det * n beta = (sum_x1sq * sum_x2y - sum_x1x2 * sum_x1y + sum_x1x2 * sum_y - sum_x2 * sum_x1 * sum_x1y / n) / det * n intercept = (sum_y - alpha * sum_x1 - beta * sum_x2) / n # R-squared y_mean = sum_y / n ss_res = sum((y[i] - (alpha * X[i][0] + beta * X[i][1] + intercept))**2 for i in range(n)) ss_tot = sum((y[i] - y_mean)**2 for i in range(n)) r_squared = 1.0 - ss_res / max(ss_tot, 1e-12) return SlippageCalibration( alpha=alpha, beta=beta, intercept=intercept, n_samples=n, r_squared=r_squared, ) @staticmethod def default_per_asset() -> Dict[str, SlippageCalibration]: """Default calibration from OB study research (Bouchaud/Cont/Stoikov). These are PRIOR values from the power-law depth model: alpha = estimated bps per level consumed """ return { "BTCUSDT": SlippageCalibration(alpha=0.05, beta=0.3, intercept=0.02, n_samples=0), "ETHUSDT": SlippageCalibration(alpha=0.08, beta=0.4, intercept=0.03, n_samples=0), "SOLUSDT": SlippageCalibration(alpha=0.15, beta=0.6, intercept=0.05, n_samples=0), "DOGEUSDT": SlippageCalibration(alpha=0.30, beta=1.0, intercept=0.10, n_samples=0), "ADAUSDT": SlippageCalibration(alpha=0.25, beta=0.8, intercept=0.08, n_samples=0), "AVAXUSDT": SlippageCalibration(alpha=0.20, beta=0.7, intercept=0.06, n_samples=0), "UNIUSDT": SlippageCalibration(alpha=0.35, beta=1.2, intercept=0.12, n_samples=0), "LINKUSDT": SlippageCalibration(alpha=0.18, beta=0.6, intercept=0.05, n_samples=0), "BNBUSDT": SlippageCalibration(alpha=0.06, beta=0.35, intercept=0.02, n_samples=0), "MATICUSDT": SlippageCalibration(alpha=0.22, beta=0.75, intercept=0.07, n_samples=0), "AAVEUSDT": SlippageCalibration(alpha=0.30, beta=1.0, intercept=0.10, n_samples=0), "DOTUSDT": SlippageCalibration(alpha=0.18, beta=0.6, intercept=0.05, n_samples=0), "ATOMUSDT": SlippageCalibration(alpha=0.25, beta=0.8, intercept=0.08, n_samples=0), } # Global registry SLIPPAGE_MODELS: Dict[str, SlippageCalibration] = SlippageCalibrator.default_per_asset() def get_slippage_model(symbol: str) -> SlippageCalibration: """Get calibrated slippage model for a symbol.""" return SLIPPAGE_MODELS.get(symbol, SlippageCalibration(alpha=0.1, beta=0.5)) def expected_slippage_bps( symbol: str, levels_consumed: int, order_usd: float = 0.0, book_depth_usd: float = 1.0, ) -> float: """Predict slippage for a given fill parameters, using calibrated model.""" model = get_slippage_model(symbol) return model.expected_slippage_bps(levels_consumed, order_usd, book_depth_usd)