Market Manipulation Detection: Protecting AI Agents from Wash Trading and Pumps

import numpy as np
from scipy import stats
from collections import deque
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Deque
from enum import Enum
import time

class ManipType(Enum):
    WASH_TRADE = "wash_trade"
    PUMP_DUMP = "pump_dump"
    SPOOFING = "spoofing"
    STOP_HUNT = "stop_hunt"

@dataclass
class ManipAlert:
    manip_type: ManipType
    symbol: str
    severity: str       # 'low', 'medium', 'high', 'critical'
    confidence: float   # 0-1
    description: str
    action: str
    evidence: dict
    timestamp: float = field(default_factory=time.time)

@dataclass
class TradeEvent:
    price: float
    size: float
    side: str        # 'buy' or 'sell'
    timestamp: float
    order_id: Optional[str] = None

class WashTradeDetector:
    """
    Detects wash trading using three independent signals:
    1. Price-volume correlation (low = wash)
    2. Trade size uniformity (high regularity = wash)
    3. Autocorrelation of trade direction (negative = paired trades)
    """

    def __init__(self, window_minutes: int = 30):
        self.window_secs = window_minutes * 60
        self.trades: Deque[TradeEvent] = deque()

    def add_trade(self, trade: TradeEvent):
        self.trades.append(trade)
        cutoff = time.time() - self.window_secs
        while self.trades and self.trades[0].timestamp < cutoff:
            self.trades.popleft()

    def analyze(self, symbol: str) -> Optional[ManipAlert]:
        if len(self.trades) < 50: return None

        trades = list(self.trades)
        prices = np.array([t.price for t in trades])
        sizes = np.array([t.size for t in trades])
        sides = np.array([1 if t.side == 'buy' else -1 for t in trades])

        # Signal 1: Price-volume correlation
        # For 5-min windows, compute |price_change| vs volume
        price_changes = np.abs(np.diff(prices))
        vol_paired = sizes[1:]
        pv_corr, pv_pval = stats.pearsonr(price_changes, vol_paired)

        # Signal 2: Trade size uniformity (coefficient of variation)
        # Low CV = artificially uniform sizes (wash trading signature)
        size_cv = sizes.std() / (sizes.mean() + 1e-8)

        # Signal 3: Trade direction autocorrelation
        # Wash trading: each buy is immediately followed by a sell
        # -> strong negative lag-1 autocorrelation
        dir_autocorr = np.corrcoef(sides[:-1], sides[1:])[0, 1]

        # Score: weighted combination of signals
        wash_score = 0.0
        evidence = {}

        if pv_corr < 0.15 and pv_pval < 0.05:
            wash_score += (0.15 - pv_corr) / 0.15 * 0.4
            evidence['pv_correlation'] = round(pv_corr, 4)

        if size_cv < 0.3:
            wash_score += (0.3 - size_cv) / 0.3 * 0.3
            evidence['size_cv'] = round(size_cv, 4)

        if dir_autocorr < -0.3:
            wash_score += min(abs(dir_autocorr + 0.3) / 0.7, 1.0) * 0.3
            evidence['direction_autocorr'] = round(dir_autocorr, 4)

        if wash_score < 0.35: return None

        severity = 'critical' if wash_score > 0.7 else (
            'high' if wash_score > 0.5 else 'medium'
        )

        return ManipAlert(
            manip_type=ManipType.WASH_TRADE,
            symbol=symbol,
            severity=severity,
            confidence=round(wash_score, 4),
            description=f"Wash trading detected. Score: {wash_score:.2%}. "
                        f"Low price-volume correlation ({pv_corr:.3f}), "
                        f"uniform trade sizes (CV={size_cv:.3f}).",
            action="Treat reported volume as unreliable. Use tick count instead of volume.",
            evidence=evidence,
        )


class PumpDumpDetector:
    """
    Detects pump-and-dump using:
    1. Price velocity (z-score of recent move vs historical)
    2. Volume surge (current vs rolling average)
    3. Breadth (is this an isolated asset or market-wide move?)
    """

    def __init__(self, lookback: int = 200):
        self.prices: Deque[float] = deque(maxlen=lookback)
        self.volumes: Deque[float] = deque(maxlen=lookback)
        self.lookback = lookback

    def add_candle(self, close: float, volume: float):
        self.prices.append(close)
        self.volumes.append(volume)

    def analyze(self, symbol: str, market_return: float = 0.0) -> Optional[ManipAlert]:
        if len(self.prices) < 60: return None

        prices = np.array(self.prices)
        volumes = np.array(self.volumes)

        # Recent 10-candle return (typically 10 min for 1m candles)
        recent_return = (prices[-1] - prices[-10]) / prices[-10]

        # Idiosyncratic return (remove market component)
        idiosync_return = recent_return - market_return

        # Historical 10-candle returns distribution
        hist_returns = np.array([
            (prices[i] - prices[i-10]) / prices[i-10]
            for i in range(10, len(prices) - 10)
        ])
        if hist_returns.std() == 0: return None

        return_z = (idiosync_return - hist_returns.mean()) / hist_returns.std()

        # Volume surge: recent 10 vs historical average
        recent_vol = volumes[-10:].mean()
        hist_vol = volumes[:-10].mean()
        vol_ratio = recent_vol / (hist_vol + 1e-8)

        # Flag: large positive z-score + volume surge
        # We only flag pumps, not normal rallies
        pump_score = 0.0
        evidence = {
            'return_z': round(return_z, 3),
            'recent_return_pct': round(idiosync_return * 100, 2),
            'volume_ratio': round(vol_ratio, 2),
        }

        if return_z > 3.0 and vol_ratio > 3.0:
            # Classic pump signature: explosive move + volume
            pump_score = min((return_z - 3.0) / 4.0, 0.5) + min((vol_ratio - 3.0) / 5.0, 0.5)
        elif return_z > 5.0:
            # Extreme move even without volume surge is suspicious
            pump_score = 0.5

        if pump_score < 0.3: return None

        return ManipAlert(
            manip_type=ManipType.PUMP_DUMP,
            symbol=symbol,
            severity='critical' if pump_score > 0.7 else 'high',
            confidence=round(min(pump_score, 1.0), 4),
            description=f"Pump-and-dump signature: {idiosync_return:.1%} idiosyncratic "
                        f"move ({return_z:.1f} sigma) with {vol_ratio:.1f}x volume surge.",
            action="DO NOT chase. Initiate short position with tight stop above recent high.",
            evidence=evidence,
        )


class SpoofingDetector:
    """
    Detects spoofing using order-to-trade ratio and order book depth changes.
    Requires access to order-level data (not just trades).
    """

    def __init__(self, window_seconds: int = 300):
        self.orders_placed: int = 0
        self.orders_cancelled: int = 0
        self.orders_filled: int = 0
        self.large_cancel_events: List[dict] = []
        self.window = window_seconds

    def record_order(self, action: str, size: float, price: float, mid: float):
        if action == 'place':
            self.orders_placed += 1
        elif action == 'cancel':
            self.orders_cancelled += 1
            if size > 0:
                distance_bps = abs(price - mid) / mid * 10000
                self.large_cancel_events.append({
                    'size': size,
                    'distance_bps': distance_bps,
                    'time': time.time()
                })
        elif action == 'fill':
            self.orders_filled += 1

    def analyze(self, symbol: str) -> Optional[ManipAlert]:
        if self.orders_placed < 20: return None

        # Order-to-trade ratio: orders_cancelled / orders_filled
        if self.orders_filled == 0:
            otr = float('inf')
        else:
            otr = self.orders_cancelled / self.orders_filled

        cancel_rate = self.orders_cancelled / self.orders_placed

        if otr < 10 or cancel_rate < 0.85: return None

        confidence = min((otr - 10) / 20, 1.0)

        return ManipAlert(
            manip_type=ManipType.SPOOFING,
            symbol=symbol,
            severity='high' if confidence > 0.6 else 'medium',
            confidence=round(confidence, 4),
            description=f"Spoofing detected. OTR={otr:.1f}:1, cancel rate={cancel_rate:.1%}.",
            action="Ignore displayed book depth. Use only recent fills for direction.",
            evidence={'otr': otr, 'cancel_rate': cancel_rate, 'orders_placed': self.orders_placed},
        )


class ManipulationScanner:
    """
    Orchestrates all detectors for a given symbol.
    Returns prioritized alert list on each analysis cycle.
    """

    def __init__(self, symbol: str):
        self.symbol = symbol
        self.wash = WashTradeDetector(window_minutes=30)
        self.pump = PumpDumpDetector(lookback=200)
        self.spoof = SpoofingDetector(window_seconds=300)

    def scan(self) -> List[ManipAlert]:
        alerts = []
        for detector, args in [
            (self.wash, {}),
            (self.pump, {}),
            (self.spoof, {}),
        ]:
            result = detector.analyze(symbol=self.symbol, **args)
            if result: alerts.append(result)

        # Sort by confidence descending
        return sorted(alerts, key=lambda a: a.confidence, reverse=True)