"""Anti-spoofing depth checker for micro-scanner. Implements AntiSpoofChecker — a 3-sample depth consistency check that runs at watchlist PROMOTION time (SIMP-2 fix: moved from entry gate to pre-filter). Architecture: - 3 depth snapshots taken over 2 seconds (antispoof_window_ms / 2 apart) - Fails if bid-side depth drops by > 50% between any two consecutive samples - Runs as async background task (max 3 concurrent via asyncio.Semaphore) - Symbols failing anti-spoof are blacklisted in micro_coin_candidates for 5 min Signal Design §2.1 (PRE-FILTER LAYER), SIMP-2, R-5 (non-blocking via async tasks). """ from __future__ import annotations import asyncio import psycopg from datetime import datetime, timezone, timedelta from typing import Any import structlog log = structlog.get_logger(__name__) # Maximum concurrent anti-spoof checks (R-5: non-blocking, bounded concurrency) _MAX_CONCURRENT_CHECKS: int = 3 # Depth drop threshold: fail if bid depth drops by more than this fraction _SPOOF_DROP_THRESHOLD: float = 0.50 class AntiSpoofChecker: """Checks order-book depth consistency to detect spoofed bid walls. Usage: checker = AntiSpoofChecker(conn) is_genuine = await checker.check_depth_consistency(exchange, "BTCUSDT") The semaphore limits concurrency to 3 simultaneous checks so the event loop is not saturated during batch watchlist promotions (R-5 fix). """ def __init__(self, conn: psycopg.Connection) -> None: self._conn = conn self._semaphore = asyncio.Semaphore(_MAX_CONCURRENT_CHECKS) async def check_depth_consistency( self, exchange: Any, symbol: str, samples: int = 3, window_ms: int = 2000, blacklist_minutes: int = 5, ) -> bool: """Take *samples* depth snapshots over *window_ms* milliseconds. Returns True if bid-side depth appears genuine (stable across samples). Returns False if suspected spoof (depth drops > 50% between any two samples). On detection of spoofing, the symbol is blacklisted in micro_coin_candidates for *blacklist_minutes* minutes. Runs under a semaphore — at most _MAX_CONCURRENT_CHECKS checks execute concurrently. """ async with self._semaphore: return await self._run_check( exchange, symbol, samples=samples, window_ms=window_ms, blacklist_minutes=blacklist_minutes, ) async def _run_check( self, exchange: Any, symbol: str, samples: int, window_ms: int, blacklist_minutes: int, ) -> bool: """Internal: collect depth samples and evaluate consistency.""" if samples < 2: # Cannot compare — treat as genuine return True interval_s = (window_ms / 1000.0) / (samples - 1) bid_depths: list[float] = [] for i in range(samples): if i > 0: await asyncio.sleep(interval_s) try: snapshot = await _fetch_depth(exchange, symbol) except Exception as exc: log.warning( "antispoof_depth_fetch_error", symbol=symbol, sample=i, error=str(exc), ) # Treat fetch failure as suspect — cannot confirm genuine depth _blacklist_symbol(self._conn, symbol, blacklist_minutes) return False bid_depth = _sum_bid_depth(snapshot) bid_depths.append(bid_depth) log.debug( "antispoof_sample_taken", symbol=symbol, sample=i, bid_depth=bid_depth, ) # Evaluate consistency across consecutive samples is_genuine = _evaluate_depth_consistency(bid_depths) if not is_genuine: log.warning( "antispoof_spoof_detected", symbol=symbol, bid_depths=bid_depths, drop_threshold_pct=_SPOOF_DROP_THRESHOLD * 100, ) _blacklist_symbol(self._conn, symbol, blacklist_minutes) else: log.debug( "antispoof_depth_genuine", symbol=symbol, bid_depths=bid_depths, ) return is_genuine # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- async def _fetch_depth(exchange: Any, symbol: str) -> dict: """Fetch order book depth for *symbol* from the exchange. Expects exchange to be a ccxt async exchange instance with fetch_order_book(). Returns a dict with 'bids' and 'asks' lists of [price, amount] pairs. """ order_book = await exchange.fetch_order_book(symbol, limit=20) return order_book def _sum_bid_depth(order_book: dict) -> float: """Sum the total bid-side quantity (notional) from order book snapshot.""" bids = order_book.get("bids", []) if not bids: return 0.0 # Sum qty (index 1) across all bid levels return sum(float(level[1]) for level in bids if len(level) >= 2) def _evaluate_depth_consistency(bid_depths: list[float]) -> bool: """Return True if bid depth is stable across all consecutive sample pairs. Fails (returns False) if depth drops by more than _SPOOF_DROP_THRESHOLD between any two consecutive samples — this signals a spoof wall being pulled. """ for i in range(1, len(bid_depths)): prev = bid_depths[i - 1] curr = bid_depths[i] if prev <= 0.0: # Cannot compute relative drop — treat as suspect return False drop_fraction = (prev - curr) / prev if drop_fraction > _SPOOF_DROP_THRESHOLD: return False return True def _blacklist_symbol( conn: psycopg.Connection, symbol: str, blacklist_minutes: int, ) -> None: """Update micro_coin_candidates status to 'blacklisted' with expiry time.""" blacklist_until = datetime.now(timezone.utc) + timedelta(minutes=blacklist_minutes) try: conn.execute( """ UPDATE micro_coin_candidates SET status = 'blacklisted', blacklisted_until = %s, updated_at = CURRENT_TIMESTAMP WHERE symbol = %s """, (blacklist_until.isoformat(), symbol), ) conn.commit() log.info( "antispoof_symbol_blacklisted", symbol=symbol, blacklist_minutes=blacklist_minutes, until=blacklist_until.isoformat(), ) except psycopg.Error as exc: log.error( "antispoof_blacklist_db_error", symbol=symbol, error=str(exc), )