""" Adversarial ML spoof detection — recognize market manipulation patterns in L3 (order-by-order) data. Detects: 1. Spoofing: large orders placed far from mid, cancelled before execution 2. Layering: multiple orders at different price levels on one side, all cancelled simultaneously when price moves 3. Quote stuffing: rapid order submission and cancellation to slow competitors 4. Momentum ignition: small aggressive trades followed by large passive orders Uses lightweight feature engineering (no deep learning required): - Order lifetime before cancellation - Distance from mid price - Size relative to typical trade size - Correlation between cancel events and price moves - Pattern matching on order sequences Output feeds into ToxicityFilter for pre-trade gating. """ from __future__ import annotations from collections import deque from typing import Optional class SpoofDetector: """Detect spoofing patterns in order book event streams. Maintains a rolling window of order events (place, cancel, modify) and classifies each order as legitimate or suspicious. Usage: detector = SpoofDetector() detector.record_place(order_id, side, price, size, mid, timestamp) detector.record_cancel(order_id, mid, timestamp) score = detector.spoof_probability() # 0-1 if score > 0.5: # increase toxicity filter, reduce quote sizes """ def __init__( self, window_seconds: float = 60.0, max_orders: int = 1000, spoof_cancel_threshold: float = 0.5, # % lifetime below mid-distance to flag size_multiple: float = 3.0, # order size / avg trade size > this = large price_ticks_threshold: int = 5, # cancel when price moves within N ticks of order ): self._window = window_seconds self._max_orders = max_orders self._cancel_threshold = spoof_cancel_threshold self._size_multiple = size_multiple self._ticks_threshold = price_ticks_threshold self._orders: dict[str, dict] = {} # order_id → {side, px, sz, mid_at_place, time} self._cancel_events: deque = deque(maxlen=max_orders) self._fill_events: deque = deque(maxlen=max_orders // 2) self._mid_prices: deque[float] = deque(maxlen=500) self._trade_sizes: deque[float] = deque(maxlen=500) self._spoof_count: int = 0 self._total_orders: int = 0 self._total_cancels: int = 0 # ── Event recording ────────────────────────────────────── def record_place( self, order_id: str, side: str, price: float, size: float, mid: float, timestamp: float ): """Record a new limit order placement.""" self._orders[order_id] = { "side": side, "px": price, "sz": size, "mid_at_place": mid, "time": timestamp, } self._total_orders += 1 self._mid_prices.append(mid) self._trade_sizes.append(size) # Cleanup old orders if len(self._orders) > self._max_orders: cutoff = timestamp - self._window stale = [oid for oid, o in self._orders.items() if o["time"] < cutoff] for oid in stale: del self._orders[oid] def record_cancel(self, order_id: str, mid: float, timestamp: float): """Record a cancellation. Returns True if classified as spoof.""" self._total_cancels += 1 order = self._orders.pop(order_id, None) if not order: self._cancel_events.append({"spoof": False, "time": timestamp}) return False lifetime = timestamp - order["time"] dist_bps = abs(order["px"] - order["mid_at_place"]) / order["mid_at_place"] * 10000 \ if order["mid_at_place"] > 0 else 0 # Spoof classification rules is_spoof = False reasons = [] # Rule 1: Large order far from mid, cancelled quickly avg_size = sum(self._trade_sizes) / max(len(self._trade_sizes), 1) if order["sz"] > avg_size * self._size_multiple and dist_bps > 20: if lifetime < self._cancel_threshold * dist_bps: # proportional to distance is_spoof = True reasons.append("large_far_quick_cancel") # Rule 2: Cancel right before price approaches (within N ticks) price_moved = abs(mid - order["mid_at_place"]) / order["mid_at_place"] * 10000 \ if order["mid_at_place"] > 0 else 0 if price_moved > 0 and dist_bps > 0: approach_ratio = price_moved / dist_bps if approach_ratio < 0.3 and lifetime > 0.5: is_spoof = True reasons.append("cancel_before_price_approach") # Rule 3: Order size much larger than typical, never fills if order["sz"] > avg_size * 5 and lifetime < 2.0: is_spoof = True reasons.append("oversized_short_lived") if is_spoof: self._spoof_count += 1 self._cancel_events.append({ "spoof": is_spoof, "time": timestamp, "lifetime": round(lifetime, 3), "dist_bps": round(dist_bps, 1), "reasons": reasons, }) return is_spoof def record_fill(self, order_id: str, timestamp: float): """Record a fill — removes order from tracking, not a spoof.""" self._orders.pop(order_id, None) # ── Metrics ────────────────────────────────────────────── def spoof_probability(self) -> float: """Probability that the current market is being spoofed (0-1). Based on recent cancel event ratio and pattern clustering. """ recent = [e for e in self._cancel_events if e["time"] > (self._cancel_events[-1]["time"] if self._cancel_events else 0) - self._window] if not recent: return 0.0 spoof_recent = sum(1 for e in recent if e["spoof"]) ratio = spoof_recent / len(recent) return min(1.0, ratio * 2.0) # amplify: 50% spoof rate = 100% probability def cancel_to_fill_ratio(self) -> float: """Ratio of cancellations to fills. High ratio = suspicious.""" total_fills = len(self._fill_events) if total_fills == 0: return 1.0 if self._total_cancels > 0 else 0.0 return self._total_cancels / total_fills def spoof_count(self) -> int: return self._spoof_count def summary(self) -> dict: return { "spoof_probability": round(self.spoof_probability(), 4), "spoof_count": self._spoof_count, "total_orders": self._total_orders, "total_cancels": self._total_cancels, "cancel_fill_ratio": round(self.cancel_to_fill_ratio(), 2), "active_orders": len(self._orders), }