ftdt-quant-lab/microstructure/spoof_detector.py

"""
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),
        }