From af5bbd0bce1bcd2b9749be0a8ad18fb586223f94 Mon Sep 17 00:00:00 2001 From: SII-leiyu <115807123+chaoleiyv@users.noreply.github.com> Date: Wed, 6 May 2026 12:06:45 +0800 Subject: [PATCH] Replace HTTP polling with RTDS WebSocket for real-time trade monitoring - Add rtds_client.py: persistent WebSocket connection to wss://ws-live-data.polymarket.com with auto-reconnect, heartbeat, and trade message parsing - Rewrite trade_monitor.py: single WebSocket receives ALL trades in real-time, replacing per-market HTTP polling loops (zero missed trades, sub-second latency) - Remove QPS rate limiter and per-market polling infrastructure (no longer needed) - Update default LLM model to gemini-3.1-pro-preview - Add websockets and python-socks dependencies Co-Authored-By: Claude Opus 4.6 (1M context) --- pyproject.toml | 2 + src/config/settings.py | 4 +- src/main.py | 15 +- src/services/rtds_client.py | 229 ++++++++++ src/services/trade_monitor.py | 762 ++++++++++++---------------------- src/utils/logger.py | 8 +- 6 files changed, 503 insertions(+), 517 deletions(-) create mode 100644 src/services/rtds_client.py diff --git a/pyproject.toml b/pyproject.toml index 62f06ad..ba16ee4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,8 @@ requires-python = ">=3.10" license = {text = "MIT"} dependencies = [ "httpx>=0.27.0", + "websockets>=13.0", + "python-socks[asyncio]>=2.0.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "python-dotenv>=1.0.0", diff --git a/src/config/settings.py b/src/config/settings.py index 026c16f..d01dd1b 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -13,7 +13,7 @@ class Settings(BaseSettings): # LLM API (OpenAI-compatible proxy) gemini_api_key: str = Field(default="", alias="GEMINI_API_KEY") - llm_base_url: str = Field(default="http://apicz.boyuerichdata.com/v1/", alias="LLM_BASE_URL") + llm_base_url: str = Field(default="https://generativelanguage.googleapis.com/v1beta/openai/", alias="LLM_BASE_URL") # Twitter API (for social sentiment search) twitter_api_key: str = Field(default="", alias="TWITTER_API_KEY") @@ -68,7 +68,7 @@ class Settings(BaseSettings): tier3_poll_interval: int = Field(default=300, alias="TIER3_POLL_INTERVAL") # LLM Settings - llm_model: str = Field(default="gemini-3-flash-preview", alias="LLM_MODEL") + llm_model: str = Field(default="gemini-3.1-pro-preview", alias="LLM_MODEL") llm_temperature: float = Field(default=0.0, alias="LLM_TEMPERATURE") diff --git a/src/main.py b/src/main.py index 123858e..da86a40 100644 --- a/src/main.py +++ b/src/main.py @@ -1,13 +1,13 @@ """ Polymarket Whale Watcher - Main Entry Point -This bot monitors trending Polymarket markets for large (whale) trades -and generates AI-powered analysis reports to assist user decision-making. +This bot monitors Polymarket markets for large (whale) trades via +real-time WebSocket (RTDS) and generates AI-powered analysis reports. Flow: -1. Fetch trending markets (by 24hr volume, excluding sports) -2. Monitor these markets for trades -3. Detect anomalous trades ($1,000+, price 0.2-0.8) +1. Fetch market list (for enrichment metadata) +2. RTDS WebSocket receives ALL trades in real-time (zero missed trades) +3. Filter for whale trades (size, price range, conviction) 4. Generate analysis reports using LLM 5. Output reports for user review (no automatic trading) """ @@ -312,15 +312,14 @@ class WhaleWatcher: # Log startup logger.monitoring_started( market_count=len(self.trade_monitor._monitored_markets), - interval=self.settings.fetch_interval_seconds, min_trade_size=self.settings.min_trade_size_usd, min_price=self.settings.min_price, max_price=self.settings.max_price, ) # Start monitoring tasks: - # 1. Trade monitor - watches top markets for whale trades - # 2. Market refresh - refreshes the market list periodically + # 1. Trade monitor - RTDS WebSocket real-time trade stream + # 2. Market refresh - refreshes market metadata periodically # 3. Daily briefing - generates daily summary at midnight # 4. Resolution check - checks if markets with signals have resolved # NOTE: Price volatility monitor is temporarily disabled diff --git a/src/services/rtds_client.py b/src/services/rtds_client.py new file mode 100644 index 0000000..1c72ff4 --- /dev/null +++ b/src/services/rtds_client.py @@ -0,0 +1,229 @@ +""" +RTDS (Real-Time Data Socket) client for Polymarket. + +Connects to wss://ws-live-data.polymarket.com and subscribes to +activity/trades for real-time trade data across ALL markets. + +Replaces the per-market HTTP polling approach with a single persistent +WebSocket connection — zero missed trades, sub-second latency. +""" +import asyncio +import json +import logging +import time +from typing import Awaitable, Callable, Optional + +import websockets +from websockets.asyncio.client import ClientConnection + +from src.models.trade import TradeActivity + +logger = logging.getLogger(__name__) + +RTDS_URI = "wss://ws-live-data.polymarket.com" +HEARTBEAT_INTERVAL = 5 # seconds +RECONNECT_DELAYS = [1, 2, 5, 10, 30, 60] # backoff schedule + + +class RTDSClient: + """ + Persistent WebSocket client for Polymarket RTDS trade stream. + + Features: + - Auto-reconnect with exponential backoff + - Heartbeat (PING every 5s) + - Parses raw messages into TradeActivity objects + - Fires an async callback for each trade + """ + + def __init__( + self, + on_trade: Optional[Callable[[TradeActivity], Awaitable[None]]] = None, + ): + self._on_trade = on_trade + self._running = False + self._ws: Optional[ClientConnection] = None + self._trade_count = 0 + self._connect_count = 0 + + # ================================================================ + # Message parsing + # ================================================================ + + @staticmethod + def _parse_trade(payload: dict) -> Optional[TradeActivity]: + """Convert an RTDS trade payload into a TradeActivity.""" + try: + side = (payload.get("side") or "").upper() + size = float(payload.get("size", 0) or 0) + price = float(payload.get("price", 0) or 0) + usdc_size = size * price + + outcome = payload.get("outcome", "Yes") + outcome_index = int(payload.get("outcomeIndex", 0 if outcome == "Yes" else 1)) + + ts = int(payload.get("timestamp", 0) or 0) + if ts == 0: + ts = int(time.time()) + + return TradeActivity( + transaction_hash=payload.get("transactionHash", ""), + timestamp=ts, + condition_id=payload.get("conditionId", ""), + asset=payload.get("asset", ""), + side=side, + size=size, + usdc_size=usdc_size, + price=price, + outcome=outcome, + outcome_index=outcome_index, + title=payload.get("title", ""), + slug=payload.get("slug"), + event_slug=payload.get("eventSlug"), + proxy_wallet=payload.get("proxyWallet"), + name=payload.get("name") or payload.get("pseudonym"), + ) + except Exception as e: + logger.debug(f"Failed to parse RTDS trade: {e}") + return None + + # ================================================================ + # Connection lifecycle + # ================================================================ + + async def _heartbeat(self, ws: ClientConnection) -> None: + """Send PING every HEARTBEAT_INTERVAL seconds.""" + try: + while True: + await asyncio.sleep(HEARTBEAT_INTERVAL) + await ws.send("PING") + except (asyncio.CancelledError, websockets.ConnectionClosed): + pass + + async def _subscribe(self, ws: ClientConnection) -> None: + """Subscribe to the activity/trades stream.""" + msg = { + "action": "subscribe", + "subscriptions": [ + {"topic": "activity", "type": "trades", "filters": ""} + ], + } + await ws.send(json.dumps(msg)) + logger.info("Subscribed to RTDS activity/trades") + + async def _consume(self, ws: ClientConnection) -> None: + """Read messages from the WebSocket and dispatch trades.""" + async for raw in ws: + if not self._running: + break + + if raw == "PONG" or not raw.strip(): + continue + + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + + if msg.get("topic") != "activity" or msg.get("type") != "trades": + continue + + payload = msg.get("payload") + if not payload: + continue + + activity = self._parse_trade(payload) + if not activity: + continue + + self._trade_count += 1 + + if self._on_trade: + try: + await self._on_trade(activity) + except Exception as e: + logger.error(f"Error in trade callback: {e}") + + async def _connect_and_run(self) -> None: + """Single connection attempt: connect → subscribe → consume.""" + self._connect_count += 1 + logger.info( + f"Connecting to RTDS ({self._connect_count})... " + f"(total trades so far: {self._trade_count})" + ) + + async with websockets.connect(RTDS_URI, ping_interval=None) as ws: + self._ws = ws + logger.info("RTDS connected") + + await self._subscribe(ws) + + hb_task = asyncio.create_task(self._heartbeat(ws)) + try: + await self._consume(ws) + finally: + hb_task.cancel() + self._ws = None + + # ================================================================ + # Public API + # ================================================================ + + async def run(self) -> None: + """ + Start the RTDS client with auto-reconnect. + + Runs forever until stop() is called. + """ + self._running = True + consecutive_failures = 0 + + while self._running: + try: + await self._connect_and_run() + # Clean disconnect (stop() called) — exit + if not self._running: + break + # Unexpected clean close — reconnect immediately + consecutive_failures = 0 + except ( + websockets.ConnectionClosed, + websockets.InvalidURI, + websockets.InvalidHandshake, + OSError, + ConnectionError, + ) as e: + if not self._running: + break + delay_idx = min(consecutive_failures, len(RECONNECT_DELAYS) - 1) + delay = RECONNECT_DELAYS[delay_idx] + consecutive_failures += 1 + logger.warning( + f"RTDS disconnected: {type(e).__name__}: {e}. " + f"Reconnecting in {delay}s (attempt {consecutive_failures})" + ) + await asyncio.sleep(delay) + except Exception as e: + if not self._running: + break + logger.error(f"Unexpected RTDS error: {e}. Reconnecting in 10s") + await asyncio.sleep(10) + + logger.info(f"RTDS client stopped (total trades received: {self._trade_count})") + + def stop(self) -> None: + """Stop the RTDS client.""" + self._running = False + if self._ws: + asyncio.ensure_future(self._ws.close()) + logger.info("RTDS client stopping...") + + @property + def trade_count(self) -> int: + """Total number of trades received since start.""" + return self._trade_count + + @property + def is_connected(self) -> bool: + """Whether the WebSocket is currently connected.""" + return self._ws is not None and self._ws.state.name == "OPEN" diff --git a/src/services/trade_monitor.py b/src/services/trade_monitor.py index 23c4df0..f7d456a 100644 --- a/src/services/trade_monitor.py +++ b/src/services/trade_monitor.py @@ -1,18 +1,22 @@ """ -Trade monitoring service - per-market parallel architecture. +Trade monitoring service — RTDS WebSocket architecture. -Each market runs its own independent async task that: -1. Polls the official Polymarket data-api for new trades -2. Detects whale trades -3. Fetches trader ranking + history in parallel -4. Fires the whale callback (LLM report generation) without blocking other markets +Single WebSocket connection receives ALL trades in real-time from +Polymarket RTDS (wss://ws-live-data.polymarket.com). -Modeled after paper_trading/paper_trading.py's _market_loop pattern. +For each incoming trade: +1. Record for cluster detection (anomaly detector) +2. Dedup by transaction hash +3. Filter: whale pre-filter (price range, size, conviction) +4. Enrich: trader ranking + history → anomaly score +5. If score passes threshold → full enrichment + LLM callback + +Replaces the previous per-market HTTP polling architecture. """ import asyncio import json import logging -import random +import math import time as _time from datetime import datetime from pathlib import Path @@ -27,24 +31,22 @@ from src.models.trade import ( EventPosition, MarketTopTrader, ) from src.services.anomaly_detector import AnomalyDetector +from src.services.rtds_client import RTDSClient logger = logging.getLogger(__name__) # Gamma API for fetching latest market prices GAMMA_API_URL = "https://gamma-api.polymarket.com/markets" -# Official Polymarket data-api for trade data -# URL and key loaded from settings (.env) - # File to persist processed transaction hashes PROCESSED_TXNS_FILE = Path(__file__).parent.parent.parent / "data" / "processed_transactions.json" class TradeMonitor: """ - Monitors Polymarket markets for large trades. + Monitors Polymarket markets for large trades via RTDS WebSocket. - Architecture: one asyncio.Task per market, fully parallel. + Architecture: single WebSocket connection → filter → enrich → callback. """ def __init__( @@ -53,10 +55,13 @@ class TradeMonitor: ): self.settings = get_settings() - # Official Polymarket data-api + # RTDS WebSocket client (created in run()) + self._rtds: Optional[RTDSClient] = None + + # HTTP client for enrichment API calls (trader ranking, history, etc.) self.data_api_url = "https://data-api.polymarket.com" - self.trades_endpoint = f"{self.data_api_url}/trades" self.leaderboard_endpoint = f"{self.data_api_url}/v1/leaderboard" + self.trades_endpoint = f"{self.data_api_url}/trades" self._client = httpx.AsyncClient( timeout=httpx.Timeout(30.0, pool=120.0), limits=httpx.Limits( @@ -66,20 +71,14 @@ class TradeMonitor: ), ) - # Per-market last-fetch timestamps for incremental polling - self._market_last_ts: Dict[str, int] = {} - - # Rate limiter: Lock + Semaphore created lazily in run() to avoid "attached to different loop" error - self._api_lock: Optional[asyncio.Lock] = None - self._api_sem: Optional[asyncio.Semaphore] = None # concurrency limiter - self._api_last_request: float = 0.0 - self._api_global_interval: float = 0.2 # min 0.2s between requests = 5 QPS - # Cache for trader rankings to avoid repeated API calls self._trader_ranking_cache: Dict[str, TraderRanking] = {} - # Markets being monitored: market_id -> Market + # Markets being monitored: condition_id -> Market + # Used for enrichment (market question, description, etc.) self._monitored_markets: Dict[str, Market] = {} + # condition_id -> market_id mapping + self._condition_to_market_id: Dict[str, str] = {} # Track processed transactions to avoid duplicates self._processed_txns: Set[str] = set() @@ -91,12 +90,12 @@ class TradeMonitor: # Callback for whale detection self._on_whale_detected = on_whale_detected - # Control flag and per-market tasks + # Control flag self._running = False - self._market_tasks: Dict[str, asyncio.Task] = {} - # Flag to track if initial scan is complete (ignore historical trades) - self._initial_scan_complete = False + # Flag to suppress alerts during initial warmup + self._warmup_complete = False + self._warmup_seconds = 10 # seconds to collect baseline before alerting # ================================================================ # Persistence @@ -147,194 +146,216 @@ class TradeMonitor: def set_monitored_markets(self, markets: List[TrendingMarket]): """Update the list of markets to monitor.""" self._monitored_markets = {} + self._condition_to_market_id = {} for tm in markets: - if tm.market.id: - self._monitored_markets[tm.market.id] = tm.market + m = tm.market + if m.id and m.condition_id: + self._monitored_markets[m.condition_id] = m + self._condition_to_market_id[m.condition_id] = m.id logger.info(f"Now monitoring {len(self._monitored_markets)} markets") def set_tiered_markets(self, tiers: dict[str, list]) -> None: - """ - Set markets with per-tier poll intervals. - - Stores poll_interval per market_id in _market_poll_intervals dict. - """ + """Set markets from tiered scan (same interface as before).""" self._monitored_markets = {} - self._market_poll_intervals: dict[str, int] = {} - - tier_intervals = { - "tier1": self.settings.tier1_poll_interval, - "tier2": self.settings.tier2_poll_interval, - "tier3": self.settings.tier3_poll_interval, - } + self._condition_to_market_id = {} for tier_name, markets in tiers.items(): - interval = tier_intervals.get(tier_name, self.settings.fetch_interval_seconds) for tm in markets: - if tm.market.id: - self._monitored_markets[tm.market.id] = tm.market - self._market_poll_intervals[tm.market.id] = interval + m = tm.market + if m.id and m.condition_id: + self._monitored_markets[m.condition_id] = m + self._condition_to_market_id[m.condition_id] = m.id + total = len(self._monitored_markets) tier_counts = {k: len(v) for k, v in tiers.items()} - logger.info( - f"Tiered monitoring: {tier_counts} " - f"(intervals: {tier_intervals}s), total={len(self._monitored_markets)}" - ) + logger.info(f"Tiered monitoring: {tier_counts}, total={total}") # ================================================================ - # Trade fetching + # RTDS trade handler (core of the new architecture) # ================================================================ - _MAX_RETRIES = 4 - _RETRY_BACKOFF = [2, 5, 10, 20] # seconds between retries (with jitter) - - # ================================================================ - # Official Polymarket data-api: fetch trades - # ================================================================ - - async def fetch_market_trades(self, market_id: str) -> List[TradeActivity]: + async def _on_rtds_trade(self, activity: TradeActivity) -> None: """ - Fetch recent trades using the official Polymarket data-api /trades endpoint. + Called for every trade received from RTDS WebSocket. - The official API returns trades with fields: - - id, taker_order_id, market, asset, side, size, price, status - - match_time, transaction_hash, outcome, bucket_index, owner, type + This replaces the per-market polling loop. + """ + condition_id = activity.condition_id + + # Look up market info (enrichment data) + market = self._monitored_markets.get(condition_id) + market_id = self._condition_to_market_id.get(condition_id, "") + + # Record every trade for cluster detection (even unmonitored markets) + if market_id: + self._anomaly_detector.record_trade(activity, market_id) + + # Dedup by transaction hash + outcome (same tx can have multiple fills) + dedup_key = f"{activity.transaction_hash}_{activity.outcome}_{activity.size}" + if dedup_key in self._processed_txns: + return + self._processed_txns.add(dedup_key) + + # Skip unmonitored markets + if not market: + return + + # Skip during warmup period (avoid alerting on historical trades) + if not self._warmup_complete: + return + + # Only track BUY trades (new positions) + if activity.side != "BUY": + return + + # Whale pre-filter + if not self._is_whale_trade(activity, market=market): + return + + # Handle whale (enrich + score + callback) + asyncio.create_task(self._handle_whale(activity, market_id, market)) + + # ================================================================ + # Whale detection (unchanged from original) + # ================================================================ + + def _is_whale_trade(self, activity: TradeActivity, market: Optional[Market] = None) -> bool: + """ + Multi-layer pre-filter mirroring options flow SignalFilter._check_signal. + + Filter chain (early rejection): + 1. Price range — like moneyness filter (OTM/ITM range) + 2. Direction — BUY only (like enabled direction_filters) + 3. Resolution window — like DTE filter (3-60 days sweet spot) + 4. Size — like premium filter ($250K+ minimum) + 5. Dynamic size — like dynamic_premium (base × √(vol / baseline)) + 6. Signal strength — like ask_ratio filter (conviction check) + """ + # --- 1. Price range --- + if not (self.settings.min_price <= activity.price <= self.settings.max_price): + return False + + # --- 2. Direction: BUY only --- + # Already enforced upstream + + # --- 3. Resolution window --- + if market and market.end_date: + try: + end_dt = datetime.fromisoformat(market.end_date.replace("Z", "+00:00")) + now_dt = datetime.utcnow().replace(tzinfo=end_dt.tzinfo) if end_dt.tzinfo else datetime.utcnow() + hours_to_resolution = max(0, (end_dt - now_dt).total_seconds() / 3600) + if hours_to_resolution < 3: + return False + if hours_to_resolution > 180 * 24: + return False + except (ValueError, TypeError): + pass + + # --- 4. Size --- + if activity.usdc_size < 3_000: + return False + + # --- 5. Dynamic size --- + base_size = 5_000.0 + baseline_volume = 1_000_000.0 + + if market and market.volume > 0: + threshold = base_size * math.sqrt(market.volume / baseline_volume) + threshold = max(3_000.0, min(threshold, 50_000.0)) + else: + threshold = base_size + + if activity.usdc_size < threshold: + return False + + # --- 6. Signal strength --- + if market and market.outcome_prices: + if activity.outcome == "Yes": + market_mid = market.outcome_prices[0] + elif len(market.outcome_prices) > 1: + market_mid = market.outcome_prices[1] + else: + market_mid = 1.0 - market.outcome_prices[0] + + if activity.price < market_mid + 0.01: + return False + + return True + + async def _handle_whale(self, activity: TradeActivity, market_id: str, market: Market): + """ + Handle a single whale trade: + 1. Fetch trader info (ranking + history) for anomaly scoring + 2. Compute multi-dimensional anomaly score as pre-filter + 3. If score passes threshold, fetch full enrichment data and fire LLM callback """ try: - market = self._monitored_markets.get(market_id) - if not market: - return [] - - # The official /trades endpoint uses condition_id as the "market" param - condition_id = market.condition_id - if not condition_id: - return [] - - last_ts = self._market_last_ts.get(market_id) - - params: Dict[str, object] = { - "market": condition_id, - "limit": 50, - } - - sem = self._api_sem or asyncio.Semaphore(20) - last_err: Optional[Exception] = None - async with sem: - for attempt in range(self._MAX_RETRIES): - try: - async with self._api_lock: - now = _time.monotonic() - wait = self._api_global_interval - (now - self._api_last_request) - if wait > 0: - await asyncio.sleep(wait) - self._api_last_request = _time.monotonic() - - response = await self._client.get( - f"{self.data_api_url}/trades", params=params, - ) - response.raise_for_status() - break - except httpx.HTTPStatusError as e: - if e.response.status_code in (502, 503, 504) and attempt < self._MAX_RETRIES - 1: - delay = self._RETRY_BACKOFF[attempt] - logger.debug( - f"Official API {e.response.status_code} for {market_id} " - f"(attempt {attempt + 1}/{self._MAX_RETRIES}), " - f"retrying in {delay}s" - ) - await asyncio.sleep(delay) - continue - raise - except httpx.HTTPError as e: - last_err = e - if attempt < self._MAX_RETRIES - 1: - delay = self._RETRY_BACKOFF[attempt] + random.uniform(0, 2) - logger.debug( - f"Official API retry for {market_id} " - f"(attempt {attempt + 1}/{self._MAX_RETRIES}): " - f"{type(e).__name__}, retrying in {delay:.1f}s" - ) - await asyncio.sleep(delay) - else: - logger.warning( - f"Official API connection error for {market_id} " - f"(attempt {attempt + 1}/{self._MAX_RETRIES}, giving up): " - f"{type(e).__name__}: {e}" - ) - return [] - else: - return [] - - data = response.json() - if not data: - return [] - - activities = [] - max_ts = last_ts or 0 - - for item in data: - try: - side = item.get("side", "").upper() - - # Only track BUY trades (new positions) - if side != "BUY": - continue - - size = float(item.get("size", 0) or 0) - price = float(item.get("price", 0) or 0) - usdc_size = size * price # Official API: USDC value = tokens * price - - outcome = item.get("outcome", "Yes") - outcome_index = int(item.get("outcomeIndex", 0 if outcome == "Yes" else 1)) - - # Timestamp is epoch seconds in the official API - ts = int(item.get("timestamp", 0) or 0) - if ts == 0: - ts = int(_time.time()) - - if ts > max_ts: - max_ts = ts - - tx_hash = item.get("transactionHash", "") - - activity = TradeActivity( - transaction_hash=tx_hash, - timestamp=ts, - condition_id=item.get("conditionId", condition_id), - asset=item.get("asset", ""), - side="BUY", - size=size, - usdc_size=usdc_size, - price=price, - outcome=outcome, - outcome_index=outcome_index, - title=item.get("title", ""), - slug=item.get("slug"), - event_slug=item.get("eventSlug"), - proxy_wallet=item.get("proxyWallet"), - name=item.get("name") or item.get("pseudonym"), - ) - activities.append(activity) - except Exception as e: - logger.debug(f"Failed to parse official API trade: {e}") - continue - - if max_ts > 0: - self._market_last_ts[market_id] = max_ts - - return activities - - except httpx.HTTPStatusError as e: - logger.warning( - f"Official trades API HTTP {e.response.status_code} for {market_id}: " - f"{e.response.text[:200]}" + # Phase 1: Quick fetch — ranking + history for anomaly scoring + trader_ranking, trader_history = await asyncio.gather( + self.fetch_trader_ranking(activity.proxy_wallet), + self.fetch_trader_history(activity.proxy_wallet), ) - return [] + + # Phase 2: Anomaly scoring + should_analyze, score, breakdown = self._anomaly_detector.should_analyze( + activity, market=market, trader_history=trader_history, + market_id=market_id, + ) + + rank_str = f"(Rank #{trader_ranking.rank})" if trader_ranking and trader_ranking.rank else "(Unranked)" + breakdown_short = " | ".join(f"{k}={v:.2f}" for k, v in breakdown.items()) + + if not should_analyze: + logger.info( + f"⚪ Whale below threshold: ${activity.usdc_size:,.2f} " + f"BUY {activity.outcome} @ {activity.price:.4f} {rank_str} " + f"score={score:.2f} [{breakdown_short}] — skipped LLM" + ) + return + + logger.info( + f"🐋 Whale trade detected! ${activity.usdc_size:,.2f} " + f"BUY {activity.outcome} @ {activity.price:.4f} {rank_str} " + f"score={score:.2f} [{breakdown_short}] on '{market.question[:50]}...'" + ) + + # Phase 3: Full enrichment + event_positions, (top_buyers, top_sellers) = await asyncio.gather( + self.fetch_whale_event_positions( + activity.proxy_wallet, + activity.event_slug, + market.condition_id or "", + ), + self.fetch_market_top_traders( + market_id, condition_id=market.condition_id or "", + outcome_prices=market.outcome_prices, + ), + ) + + whale_trade = WhaleTrade( + id=f"{market_id}_{activity.transaction_hash}", + trade=activity, + market_id=market_id, + market_question=market.question, + market_description=market.description, + market_outcomes=market.outcomes, + market_outcome_prices=market.outcome_prices, + trader_ranking=trader_ranking, + trader_history=trader_history, + whale_event_positions=event_positions, + market_top_buyers=top_buyers, + market_top_sellers=top_sellers, + ) + + # Fire callback (LLM report generation) + if self._on_whale_detected: + await self._on_whale_detected(whale_trade) + except Exception as e: - logger.warning(f"Error fetching official trades for {market_id}: {type(e).__name__}: {e}") - return [] + logger.error(f"Error handling whale trade in {market_id}: {e}") # ================================================================ - # Official API: trader info (ranking + history) + # Enrichment API calls (unchanged — still uses HTTP) # ================================================================ async def fetch_trader_ranking(self, wallet_address: str) -> Optional[TraderRanking]: @@ -444,23 +465,13 @@ class TradeMonitor: logger.debug(f"Error fetching history for {wallet_address}: {e}") return None - # ================================================================ - # Event positions & market top traders - # ================================================================ - async def fetch_whale_event_positions( self, wallet_address: str, event_slug: str, current_condition_id: str, ) -> List[EventPosition]: - """ - Fetch the whale's current positions across all markets in the same event. - - Uses the Polymarket data API positions endpoint directly: - GET https://data-api.polymarket.com/positions?user= - Then filters by event_slug to find related holdings. - """ + """Fetch the whale's positions across all markets in the same event.""" if not wallet_address or not event_slug: return [] @@ -474,7 +485,6 @@ class TradeMonitor: if not all_positions: return [] - # Filter positions belonging to the same event, excluding current market result = [] for pos in all_positions: pos_event_slug = pos.get("eventSlug", "") @@ -487,7 +497,7 @@ class TradeMonitor: size = float(pos.get("size", 0) or 0) if size == 0: - continue # skip empty positions + continue outcome = pos.get("outcome", "Yes") avg_price = float(pos.get("avgPrice", 0) or 0) @@ -497,7 +507,6 @@ class TradeMonitor: cash_pnl = float(pos.get("cashPnl", 0) or 0) title = pos.get("title", "") - # Build human-readable summary if outcome == "Yes": side_summary = f"Holding Yes {size:,.0f} tokens @ avg {avg_price:.2%}, current {cur_price:.2%}" else: @@ -516,7 +525,6 @@ class TradeMonitor: side_summary=side_summary, )) - # Sort by position value descending result.sort(key=lambda x: x.current_value, reverse=True) logger.debug( f"Found {len(result)} event positions for {wallet_address} " @@ -532,21 +540,10 @@ class TradeMonitor: self, market_id: str, condition_id: str = "", outcome_prices: Optional[List[float]] = None, top_n: int = 5, ) -> tuple[List[MarketTopTrader], List[MarketTopTrader]]: - """ - Fetch top holders (bulls and bears) for a market. - - Uses the official Polymarket data-api /holders endpoint which returns - the top position holders for each outcome token, sorted by amount. - - Returns: - (top_buyers, top_sellers) — each up to top_n entries. - top_buyers = top Yes token holders (bullish). - top_sellers = top No token holders (bearish). - """ + """Fetch top holders (bulls and bears) for a market.""" if not condition_id: return [], [] - # outcome_prices: [yes_price, no_price] yes_price = outcome_prices[0] if outcome_prices and len(outcome_prices) > 0 else 0.5 no_price = outcome_prices[1] if outcome_prices and len(outcome_prices) > 1 else 0.5 @@ -568,7 +565,6 @@ class TradeMonitor: if not holders: continue - # outcomeIndex: 0 = Yes (bulls), 1 = No (bears) outcome_index = holders[0].get("outcomeIndex", 0) token_price = yes_price if outcome_index == 0 else no_price @@ -576,7 +572,6 @@ class TradeMonitor: wallet = h.get("proxyWallet", "") name = h.get("name") or h.get("pseudonym") or None amount = float(h.get("amount", 0) or 0) - # Convert token amount to USD value usd_value = amount * token_price trader = MarketTopTrader( @@ -591,7 +586,7 @@ class TradeMonitor: else: top_sellers.append(trader) - # Fetch rankings for top traders in parallel + # Fetch rankings in parallel ranking_tasks = [] trader_refs = [] for t in top_buyers + top_sellers: @@ -617,309 +612,70 @@ class TradeMonitor: logger.warning(f"Error fetching top holders for {market_id}: {e}") return [], [] - # ================================================================ - # Whale detection - # ================================================================ - - def _is_whale_trade(self, activity: TradeActivity, market: Optional[Market] = None) -> bool: - """ - Multi-layer pre-filter mirroring options flow SignalFilter._check_signal. - - Filter chain (early rejection, same order as options flow): - 1. Price range — like moneyness filter (OTM/ITM range) - 2. Direction — BUY only (like enabled direction_filters) - 3. Resolution window — like DTE filter (3-60 days sweet spot) - 4. Size — like premium filter ($250K+ minimum) - 5. Dynamic size — like dynamic_premium (base × √(vol / baseline)) - 6. Signal strength — like ask_ratio filter (conviction check) - """ - import math - from datetime import datetime as _dt - - # --- 1. Price range (like moneyness: OTM 0-20%) --- - # Price 0.2-0.8 = uncertain outcome = tradeable - # Price < 0.2 or > 0.8 = near-consensus = no edge - if not (self.settings.min_price <= activity.price <= self.settings.max_price): - return False - - # --- 2. Direction: BUY only (like direction_filters.enabled) --- - # Already enforced upstream (only BUY trades reach here) - - # --- 3. Resolution window (like DTE min=3, max=60) --- - # Markets resolving < 6 hours = price already settled (like DTE < 3) - # Markets resolving > 90 days = too far out, edge diluted (like DTE > 60) - if market and market.end_date: - try: - end_dt = _dt.fromisoformat(market.end_date.replace("Z", "+00:00")) - now_dt = _dt.utcnow().replace(tzinfo=end_dt.tzinfo) if end_dt.tzinfo else _dt.utcnow() - hours_to_resolution = max(0, (end_dt - now_dt).total_seconds() / 3600) - if hours_to_resolution < 3: - return False # too close, like DTE < 3 - if hours_to_resolution > 180 * 24: - return False # too far, like DTE > 60 - except (ValueError, TypeError): - pass # unknown end date, don't reject - - # --- 4. Size (like premium min=$250K) --- - # Base minimum: $5,000 (Polymarket scale vs options $250K) - if activity.usdc_size < 3_000: - return False - - # --- 5. Dynamic size (like dynamic_premium = base × √(mcap / baseline)) --- - # Larger markets require proportionally larger trades to be meaningful - base_size = 5_000.0 - baseline_volume = 1_000_000.0 - - if market and market.volume > 0: - threshold = base_size * math.sqrt(market.volume / baseline_volume) - threshold = max(3_000.0, min(threshold, 50_000.0)) # floor $3K, cap $50K - else: - threshold = base_size - - if activity.usdc_size < threshold: - return False - - # --- 6. Signal strength (like ask_ratio > 70%) --- - # In Polymarket: buyer paying above market mid = conviction - # Reject trades at or below market mid (no conviction, possibly hedging) - if market and market.outcome_prices: - if activity.outcome == "Yes": - market_mid = market.outcome_prices[0] - elif len(market.outcome_prices) > 1: - market_mid = market.outcome_prices[1] - else: - market_mid = 1.0 - market.outcome_prices[0] - - # Must pay above market mid (no discount buys = no conviction) - if activity.price < market_mid + 0.01: - return False - - return True - - async def _handle_whale(self, activity: TradeActivity, market_id: str, market: Market): - """ - Handle a single whale trade: - 1. Fetch trader info (ranking + history) for anomaly scoring - 2. Compute multi-dimensional anomaly score as pre-filter - 3. If score passes threshold, fetch full enrichment data and fire LLM callback - """ - try: - # Phase 1: Quick fetch — only ranking + history (needed for anomaly scoring) - trader_ranking, trader_history = await asyncio.gather( - self.fetch_trader_ranking(activity.proxy_wallet), - self.fetch_trader_history(activity.proxy_wallet), - ) - - # Phase 2: Multi-dimensional anomaly scoring (pre-filter before LLM) - should_analyze, score, breakdown = self._anomaly_detector.should_analyze( - activity, market=market, trader_history=trader_history, - market_id=market_id, - ) - - rank_str = f"(Rank #{trader_ranking.rank})" if trader_ranking and trader_ranking.rank else "(Unranked)" - breakdown_short = " | ".join(f"{k}={v:.2f}" for k, v in breakdown.items()) - - if not should_analyze: - logger.info( - f"⚪ Whale below threshold: ${activity.usdc_size:,.2f} " - f"BUY {activity.outcome} @ {activity.price:.4f} {rank_str} " - f"score={score:.2f} [{breakdown_short}] — skipped LLM" - ) - return - - logger.info( - f"🐋 Whale trade detected! ${activity.usdc_size:,.2f} " - f"BUY {activity.outcome} @ {activity.price:.4f} {rank_str} " - f"score={score:.2f} [{breakdown_short}] on '{market.question[:50]}...'" - ) - - # Phase 3: Full enrichment (only for trades that pass pre-filter) - event_positions, (top_buyers, top_sellers) = await asyncio.gather( - self.fetch_whale_event_positions( - activity.proxy_wallet, - activity.event_slug, - market.condition_id or "", - ), - self.fetch_market_top_traders( - market_id, condition_id=market.condition_id or "", - outcome_prices=market.outcome_prices, - ), - ) - - whale_trade = WhaleTrade( - id=f"{market_id}_{activity.transaction_hash}", - trade=activity, - market_id=market_id, - market_question=market.question, - market_description=market.description, - market_outcomes=market.outcomes, - market_outcome_prices=market.outcome_prices, - trader_ranking=trader_ranking, - trader_history=trader_history, - whale_event_positions=event_positions, - market_top_buyers=top_buyers, - market_top_sellers=top_sellers, - ) - - # Fire callback (LLM report generation) - if self._on_whale_detected: - await self._on_whale_detected(whale_trade) - - except Exception as e: - logger.error(f"Error handling whale trade in {market_id}: {e}") - - # ================================================================ - # Per-market independent loop - # ================================================================ - - async def _market_loop(self, market_id: str, initial_delay: float): - """ - Independent polling loop for a single market. - - Each market runs this as its own asyncio.Task: - 1. Wait initial_delay (stagger startup to avoid request storm) - 2. First poll: record existing transactions (no alerts) - 3. Subsequent polls: detect whales, handle in parallel - """ - if initial_delay > 0: - await asyncio.sleep(initial_delay) - - market = self._monitored_markets.get(market_id) - if not market: - return - - # Per-market interval (from tiered monitoring) or global default - poll_intervals = getattr(self, '_market_poll_intervals', {}) - poll_interval = poll_intervals.get(market_id, self.settings.fetch_interval_seconds) - # If we already have a last_ts for this market, it means the loop was - # restarted (e.g. after a market list refresh) — skip the silent - # first-poll window to avoid missing trades. - is_first_poll = market_id not in self._market_last_ts - - while self._running: - try: - # Check if market was removed during refresh - market = self._monitored_markets.get(market_id) - if not market: - logger.debug(f"Market {market_id} no longer monitored, stopping loop") - break - - activities = await self.fetch_market_trades(market_id) - - # Collect whale handling tasks for this poll cycle - whale_tasks = [] - - for activity in activities: - # Record every trade for cluster detection - self._anomaly_detector.record_trade(activity, market_id) - - if activity.transaction_hash in self._processed_txns: - continue - self._processed_txns.add(activity.transaction_hash) - - # First poll: only record, don't alert - if is_first_poll: - continue - - if self._is_whale_trade(activity, market=market): - # Launch whale handling as a parallel task - whale_tasks.append( - asyncio.create_task( - self._handle_whale(activity, market_id, market) - ) - ) - - # Wait for all whale handlers in this cycle to complete - if whale_tasks: - await asyncio.gather(*whale_tasks, return_exceptions=True) - - is_first_poll = False - - except asyncio.CancelledError: - break - except Exception as e: - logger.error(f"Error in market loop {market_id}: {e}") - - await asyncio.sleep(poll_interval) - # ================================================================ # Main run loop # ================================================================ async def run(self): """ - Start the parallel monitoring loop. + Start the RTDS-based monitoring loop. - Architecture (modeled after paper_trading._poll_trades): - - Each market gets its own asyncio.Task (_market_loop) - - Startup is staggered to avoid request storms - - Main loop handles: task lifecycle, persistence, new market spawning + Architecture: + - Single RTDS WebSocket receives ALL trades in real-time + - _on_rtds_trade filters and handles each trade + - Periodic persistence of processed transactions """ self._running = True - poll_interval = self.settings.fetch_interval_seconds - # Create lock/semaphore inside event loop (avoids "attached to different loop" error) - self._api_lock = asyncio.Lock() - self._api_sem = asyncio.Semaphore(10) # max 10 concurrent API requests + # Create RTDS client with our trade handler + self._rtds = RTDSClient(on_trade=self._on_rtds_trade) logger.info( - f"Starting parallel trade monitor " - f"({len(self._monitored_markets)} markets, interval: {poll_interval}s)" + f"Starting RTDS trade monitor " + f"({len(self._monitored_markets)} monitored markets)" ) - try: - # Spawn per-market tasks with staggered start - market_ids = list(self._monitored_markets.keys()) - n_markets = len(market_ids) - stagger_window = max(poll_interval, n_markets * 1.0) # ~1s per market + # Start warmup timer — suppress alerts for first N seconds + # to avoid firing on trades already in the RTDS pipeline + async def warmup_timer(): + await asyncio.sleep(self._warmup_seconds) + self._warmup_complete = True + logger.info( + f"Warmup complete ({self._warmup_seconds}s). " + f"Now alerting on new whale trades." + ) - for i, market_id in enumerate(market_ids): - delay = (i / max(n_markets, 1)) * stagger_window - task = asyncio.create_task(self._market_loop(market_id, initial_delay=delay)) - self._market_tasks[market_id] = task - - logger.info(f"Spawned {len(self._market_tasks)} parallel market tasks") - - # Main supervisory loop - save_interval = 60 # save processed txns every 60 seconds - last_save = asyncio.get_event_loop().time() + warmup_task = asyncio.create_task(warmup_timer()) + # Periodic persistence task + async def persistence_loop(): while self._running: - now = asyncio.get_event_loop().time() + await asyncio.sleep(60) + self._save_processed_txns() + # Trim processed txns set to prevent unbounded growth + if len(self._processed_txns) > 100_000: + # Keep only the most recent 50K (approximate — set is unordered, + # but old hashes won't repeat so trimming is safe) + excess = len(self._processed_txns) - 50_000 + for _ in range(excess): + self._processed_txns.pop() + logger.info(f"Trimmed processed txns to {len(self._processed_txns)}") - # Spawn tasks for newly added markets (from set_monitored_markets) - for market_id in self._monitored_markets: - if market_id not in self._market_tasks or self._market_tasks[market_id].done(): - task = asyncio.create_task( - self._market_loop(market_id, initial_delay=0) - ) - self._market_tasks[market_id] = task - logger.info(f"Spawned new task for market {market_id}") - - # Clean up tasks for removed markets - removed = [mid for mid in self._market_tasks if mid not in self._monitored_markets] - for mid in removed: - self._market_tasks[mid].cancel() - del self._market_tasks[mid] - - # Periodic persistence - if now - last_save >= save_interval: - self._save_processed_txns() - last_save = now - - await asyncio.sleep(5.0) + persistence_task = asyncio.create_task(persistence_loop()) + try: + # Run RTDS client (blocks until stop) + await self._rtds.run() finally: - # Cancel all market tasks - for task in self._market_tasks.values(): - task.cancel() - await asyncio.gather(*self._market_tasks.values(), return_exceptions=True) - self._market_tasks.clear() + warmup_task.cancel() + persistence_task.cancel() self._save_processed_txns() def stop(self): """Stop the monitoring loop.""" self._running = False + if self._rtds: + self._rtds.stop() logger.info("Trade monitor stopping...") def clear_processed_transactions(self): diff --git a/src/utils/logger.py b/src/utils/logger.py index 2f86d46..9566fa1 100644 --- a/src/utils/logger.py +++ b/src/utils/logger.py @@ -94,14 +94,14 @@ class WhaleWatcherLogger: f"[bold magenta]{'='*60}[/bold magenta]\n" ) - def monitoring_started(self, market_count: int, interval: int, min_trade_size: float = 1000, min_price: float = 0.2, max_price: float = 0.8) -> None: + def monitoring_started(self, market_count: int, interval: int = 0, min_trade_size: float = 1000, min_price: float = 0.2, max_price: float = 0.8) -> None: """Log monitoring start.""" self.console.print( f"\n[bold green]{'='*60}[/bold green]\n" - f"[bold green]🚀 WHALE WATCHER STARTED[/bold green]\n" + f"[bold green]🚀 WHALE WATCHER STARTED (RTDS WebSocket)[/bold green]\n" f"[bold green]{'='*60}[/bold green]\n" - f"[green]Monitoring:[/green] {market_count} markets\n" - f"[green]Interval:[/green] {interval} seconds\n" + f"[green]Monitored Markets:[/green] {market_count}\n" + f"[green]Mode:[/green] Real-time WebSocket (zero missed trades)\n" f"[green]Min Trade Size:[/green] ${min_trade_size:,.0f} USD\n" f"[green]Price Range:[/green] {min_price} - {max_price}\n" f"[bold green]{'='*60}[/bold green]\n"