""" RMI Prediction Market Intelligence Service =========================================== Multi-source prediction market data aggregation for crypto security intelligence. Data sources (all free, zero auth for read-only): - Polymarket: Gamma API (search/discovery), CLOB API (prices/history), Data API (trades) - Kalshi: REST API /series, /markets, /events, /orderbook (unauthenticated) - Limitless: REST API /markets (Base chain, crypto-native) - Manifold: REST API /markets (play-money, open-source sentiment signals) Open-source reference implementations (GitHub): - homerun (braedonsaunders/homerun): Open-source prediction market platform for Polymarket + Kalshi. Python strategies, backtesting, data sources, live trading. - prediction-market-edge-bot: SX Bet + Polymarket aggregator with smart order routing. - Awesome-Prediction-Market-Tools (aarora4): Curated directory of 50+ tools including Oddpool (cross-venue aggregator), analytics dashboards, trading bots. Architecture: - Direct external API calls (NEVER route through own API - anti-circular-dependency rule) - All 4 sources queried in parallel with individual try/except - Results normalized into unified PredictionMarket dataclass - Redis caching: 30s TTL prices, 5min searches, 1hr digests Integration points: - SENTINEL scanner: cross-reference token risk scores with market probability - Wallet Memory Bank: entity/deployer reputation from prediction market odds - RugMaps: visual correlation between market odds and on-chain wallet clusters - x402 tools: expose as paid security intelligence endpoints Pitfalls: - Polymarket Gamma API double-encodes outcomePrices/clobTokenIds as JSON strings - One source timeout must not kill the entire call - individual try/except per source - Prediction market data is probabilistic, not definitive - always cross-reference with on-chain scanner results - Don't poll every market every tick - use targeted search + category filters """ import asyncio import hashlib import json import logging import os from dataclasses import dataclass, field from datetime import UTC, datetime import httpx logger = logging.getLogger("prediction_market") # ── API Endpoints ──────────────────────────────────────────────── POLYMARKET_GAMMA = "https://gamma-api.polymarket.com" POLYMARKET_CLOB = "https://clob.polymarket.com" POLYMARKET_DATA = "https://data-api.polymarket.com" KALSHI_BASE = "https://external-api.kalshi.com/trade-api/v2" LIMITLESS_BASE = "https://api.limitless.exchange" MANIFOLD_BASE = "https://api.manifold.markets/v0" # ── Category mappings for security relevance ──────────────────── SECURITY_KEYWORDS = [ "hack", "exploit", "rug", "scam", "fraud", "breach", "leak", "drain", "phish", "backdoor", "vulnerability", "zero-day", "sanction", "indict", "arrest", "freeze", "seize", "clampdown", "depeg", "insolvent", "bankrupt", "collapse", "default", "theft", "heist", "compromise", "ransomware", "malware", "SEC", "CFTC", "DOJ", "FBI", "regulatory", "enforcement", ] CRYPTO_KEYWORDS = [ "bitcoin", "ethereum", "solana", "crypto", "defi", "token", "blockchain", "web3", "nft", "stablecoin", "usdt", "usdc", "dai", "exchange", "binance", "coinbase", "uniswap", "aave", "tether", "circle", "layer", "L1", "L2", "rollup", "bridge", "polygon", "arbitrum", "optimism", "avalanche", "fantom", "chainlink", "makerdao", "lido", "eigenlayer", ] # ── Dataclasses ───────────────────────────────────────────────── @dataclass class PredictionMarket: """Unified prediction market result across all sources.""" source: str # "polymarket" | "kalshi" | "limitless" | "manifold" source_id: str # native ID from source (slug, ticker, etc.) question: str slug: str probability_yes: float # 0.0-1.0 probability_no: float # 0.0-1.0 volume_usd: float liquidity_usd: float = 0.0 category: str = "" tags: list[str] = field(default_factory=list) tokens_mentioned: list[str] = field(default_factory=list) is_security_relevant: bool = False is_crypto_relevant: bool = False url: str = "" ends_at: str | None = None updated_at: str = "" def __post_init__(self): """Auto-classify relevance based on keywords in question.""" q_lower = self.question.lower() self.is_security_relevant = any(kw in q_lower for kw in SECURITY_KEYWORDS) self.is_crypto_relevant = any(kw in q_lower for kw in CRYPTO_KEYWORDS) @dataclass class PredictionDigest: """Daily intelligence digest of security-relevant prediction markets.""" generated_at: str total_markets_searched: int security_relevant_count: int crypto_relevant_count: int top_threats: list[PredictionMarket] = field(default_factory=list) token_specific_markets: list[PredictionMarket] = field(default_factory=list) ecosystem_risk_markets: list[PredictionMarket] = field(default_factory=list) regulatory_markets: list[PredictionMarket] = field(default_factory=list) # ── Service ───────────────────────────────────────────────────── class PredictionMarketService: """Multi-source prediction market data with parallel fetching + caching.""" def __init__(self, http_client: httpx.AsyncClient | None = None): self._http = http_client or httpx.AsyncClient(timeout=15.0) self._redis = None # Lazy init via get_redis() def _get_redis(self): """Lazy Redis connection for caching. Returns None if unavailable.""" if self._redis is not None: return self._redis try: import redis.asyncio as aioredis self._redis = aioredis.from_url( os.getenv("REDIS_URL", "redis://localhost:6379/0"), decode_responses=True, socket_connect_timeout=3, ) # Set to False if we couldn't actually connect if self._redis is None: self._redis = False except Exception as e: logger.warning(f"Redis unavailable, caching disabled: {e}") self._redis = False return self._redis if self._redis is not False else None # ── Public API ────────────────────────────────────────── async def search( self, query: str, categories: list[str] | None = None, min_volume: float = 0, security_only: bool = False, ) -> list[PredictionMarket]: """Search all prediction market sources in parallel. Args: query: Search term (token name, event, protocol, etc.) categories: Optional filter by source categories min_volume: Minimum USD volume to include security_only: Only return security-relevant markets """ # Check Redis cache cache_key = f"predmkt:search:{_cache_hash(query, categories, min_volume, security_only)}" redis = self._get_redis() if redis: try: cached = await redis.get(cache_key) if cached: markets_data = json.loads(cached) return [_dict_to_market(d) for d in markets_data] except Exception: pass # Cache miss or Redis error - fall through to live query # Fire all 4 sources in parallel results: list[list[PredictionMarket]] = await asyncio.gather( self._search_polymarket(query), self._search_kalshi(query), self._search_limitless(query), self._search_manifold(query), return_exceptions=True, ) # Flatten and handle exceptions all_markets: list[PredictionMarket] = [] sources = ["polymarket", "kalshi", "limitless", "manifold"] for i, result in enumerate(results): if isinstance(result, Exception): logger.warning(f"{sources[i]} search failed: {result}") continue if isinstance(result, list): all_markets.extend(result) # Filter if min_volume > 0: all_markets = [m for m in all_markets if m.volume_usd >= min_volume] if security_only: all_markets = [m for m in all_markets if m.is_security_relevant] # Sort by volume descending all_markets.sort(key=lambda m: m.volume_usd, reverse=True) # Cache (5 min TTL) if redis: try: await redis.setex(cache_key, 300, json.dumps([_market_to_dict(m) for m in all_markets[:50]])) except Exception as e: logger.warning(f"Redis cache write failed: {e}") return all_markets async def token_markets(self, symbol: str) -> list[PredictionMarket]: """Find all prediction markets mentioning a specific token symbol.""" results = await self.search(f'"{symbol}" token crypto', security_only=False) # Filter to markets actually about this token (mention in question) symbol_lower = symbol.lower() token_markets = [m for m in results if symbol_lower in m.question.lower()] return token_markets async def security_digest(self) -> PredictionDigest: """Generate daily intelligence digest of security-relevant prediction markets. Queries crypto + security keywords across all sources, categorizes results by threat type: top threats, token-specific, ecosystem risk, regulatory. """ cache_key = f"predmkt:digest:{datetime.now(UTC).strftime('%Y-%m-%d')}" redis = self._get_redis() if redis: try: cached = await redis.get(cache_key) if cached: data = json.loads(cached) return _dict_to_digest(data) except Exception: pass # Cache miss or Redis error - fall through to live query # Search for security-relevant crypto markets security_queries = [ "crypto hack exploit scam", "defi rug pull fraud", "exchange insolvent breach", "stablecoin depeg collapse", "crypto regulation SEC enforcement", "blockchain vulnerability zero-day", ] all_markets: list[PredictionMarket] = [] searches = [self.search(q, security_only=False) for q in security_queries] results = await asyncio.gather(*searches, return_exceptions=True) for result in results: if isinstance(result, list): all_markets.extend(result) # De-duplicate by question similarity seen_questions = set() unique_markets = [] for m in all_markets: q_key = m.question.lower().strip()[:80] if q_key not in seen_questions: seen_questions.add(q_key) unique_markets.append(m) # Categorize top_threats = [m for m in unique_markets if m.is_security_relevant and m.volume_usd > 10000] top_threats.sort(key=lambda m: m.volume_usd, reverse=True) token_specific = [ m for m in unique_markets if m.is_crypto_relevant and m.is_security_relevant and len(m.tokens_mentioned) > 0 ] token_specific.sort(key=lambda m: m.volume_usd, reverse=True) ecosystem_risk = [ m for m in unique_markets if m.is_crypto_relevant and not m.is_security_relevant and m.volume_usd > 50000 ] ecosystem_risk.sort(key=lambda m: m.volume_usd, reverse=True) regulatory = [ m for m in unique_markets if m.is_security_relevant and any(kw in m.question.lower() for kw in ["sec", "cftc", "doj", "regulation", "sanction", "ban"]) ] regulatory.sort(key=lambda m: m.volume_usd, reverse=True) digest = PredictionDigest( generated_at=datetime.now(UTC).isoformat(), total_markets_searched=len(unique_markets), security_relevant_count=len([m for m in unique_markets if m.is_security_relevant]), crypto_relevant_count=len([m for m in unique_markets if m.is_crypto_relevant]), top_threats=top_threats[:20], token_specific_markets=token_specific[:20], ecosystem_risk_markets=ecosystem_risk[:10], regulatory_markets=regulatory[:10], ) # Cache (1 hour) if redis: try: await redis.setex(cache_key, 3600, json.dumps(_digest_to_dict(digest))) except Exception as e: logger.warning(f"Redis digest cache write failed: {e}") return digest async def trending(self, limit: int = 20, source: str | None = None) -> list[PredictionMarket]: """Get top trending prediction markets by volume across all sources.""" # Fetch top events from each source in parallel tasks = [] if not source or source == "polymarket": tasks.append(self._trending_polymarket(limit)) else: tasks.append(asyncio.sleep(0)) # placeholder if not source or source == "kalshi": tasks.append(self._trending_kalshi(limit)) else: tasks.append(asyncio.sleep(0)) if not source or source == "limitless": tasks.append(self._trending_limitless(limit)) else: tasks.append(asyncio.sleep(0)) results = await asyncio.gather(*tasks, return_exceptions=True) all_markets = [] for result in results: if isinstance(result, list): all_markets.extend(result) elif isinstance(result, Exception): pass # Individual source failures logged in _trending_* methods all_markets.sort(key=lambda m: m.volume_usd, reverse=True) return all_markets[:limit] async def market_detail(self, source: str, market_id: str) -> PredictionMarket | None: """Get detailed data for a specific market including orderbook.""" if source == "polymarket": return await self._polymarket_detail(market_id) elif source == "kalshi": return await self._kalshi_detail(market_id) # Limitless and Manifold details on demand return None # ── Polymarket ────────────────────────────────────────── async def _search_polymarket(self, query: str) -> list[PredictionMarket]: """Search Polymarket Gamma API.""" try: resp = await self._http.get( f"{POLYMARKET_GAMMA}/public-search", params={"q": query}, timeout=10.0, ) if resp.status_code != 200: logger.warning(f"Polymarket search returned {resp.status_code}") return [] data = resp.json() events = data.get("events", []) markets = [] for event in events[:10]: for m in event.get("markets", [])[:5]: pm = self._parse_polymarket_market(m, event) if pm: markets.append(pm) return markets except Exception as e: logger.warning(f"Polymarket search error: {e}") return [] def _parse_polymarket_market(self, m: dict, event: dict | None = None) -> PredictionMarket | None: """Parse a Polymarket market dict into unified PredictionMarket.""" try: question = m.get("question", "") slug = m.get("slug", "") # Parse double-encoded JSON fields prices = self._parse_json_field(m.get("outcomePrices", "[]")) self._parse_json_field(m.get("outcomes", "[]")) self._parse_json_field(m.get("clobTokenIds", "[]")) if isinstance(prices, list) and len(prices) >= 2: prob_yes = float(prices[0]) prob_no = float(prices[1]) else: prob_yes = 0.5 prob_no = 0.5 volume = float(m.get("volume", 0)) liquidity = float(m.get("liquidity", 0)) # Extract token mentions from question tokens_mentioned = _extract_token_symbols(question) tags = [] if event: tags.extend([t.get("label", "") for t in event.get("tags", [])]) return PredictionMarket( source="polymarket", source_id=slug, question=question, slug=slug, probability_yes=prob_yes, probability_no=prob_no, volume_usd=volume, liquidity_usd=liquidity, category=m.get("category", event.get("category", "") if event else ""), tags=tags, tokens_mentioned=tokens_mentioned, url=f"https://polymarket.com/event/{slug}" if slug else "", ends_at=m.get("endDate", event.get("endDate", "") if event else ""), updated_at=datetime.now(UTC).isoformat(), ) except Exception as e: logger.warning(f"Failed to parse Polymarket market: {e}") return None async def _trending_polymarket(self, limit: int) -> list[PredictionMarket]: """Get trending Polymarket events by volume.""" try: resp = await self._http.get( f"{POLYMARKET_GAMMA}/events", params={ "limit": limit, "active": "true", "closed": "false", "order": "volume", "ascending": "false", }, timeout=10.0, ) if resp.status_code != 200: return [] events = resp.json() markets = [] for event in events[:limit]: for m in event.get("markets", [])[:3]: pm = self._parse_polymarket_market(m, event) if pm: markets.append(pm) return markets except Exception as e: logger.warning(f"Polymarket trending error: {e}") return [] async def _polymarket_detail(self, slug: str) -> PredictionMarket | None: """Get detailed Polymarket market data including CLOB prices.""" try: # Fetch from Gamma resp = await self._http.get( f"{POLYMARKET_GAMMA}/markets", params={"slug": slug}, timeout=10.0, ) if resp.status_code != 200: return None data = resp.json() if not data: return None m = data[0] pm = self._parse_polymarket_market(m) # Also fetch CLOB price for live data if pm: tokens = self._parse_json_field(m.get("clobTokenIds", "[]")) if isinstance(tokens, list) and len(tokens) >= 2: try: price_resp = await self._http.get( f"{POLYMARKET_CLOB}/price", params={"token_id": tokens[0], "side": "buy"}, timeout=5.0, ) if price_resp.status_code == 200: price_data = price_resp.json() live_price = float(price_data.get("price", pm.probability_yes)) pm.probability_yes = live_price pm.probability_no = 1.0 - live_price except Exception: pass # CLOB price is a bonus, Gamma price is fine return pm except Exception as e: logger.warning(f"Polymarket detail error for {slug}: {e}") return None # ── Kalshi ────────────────────────────────────────────── async def _search_kalshi(self, query: str) -> list[PredictionMarket]: """Search Kalshi by scanning events then fetching their markets.""" try: # Step 1: Get open events (organized by category, not sports-dominant) resp = await self._http.get( f"{KALSHI_BASE}/events", params={"status": "open", "limit": 50}, headers={"Accept": "application/json"}, timeout=10.0, ) if resp.status_code != 200: logger.warning(f"Kalshi events returned {resp.status_code}") return [] data = resp.json() events = data.get("events", []) query_lower = query.lower() query_terms = query_lower.split() results = [] # Step 2: Check event titles for matches, then fetch markets for i, event in enumerate(events[:10]): event_title = event.get("title", "").lower() event_ticker = event.get("ticker", "") # Match if query terms appear in event title if not any(term in event_title for term in query_terms): continue # Rate limit: small delay between event fetches if i > 0: await asyncio.sleep(0.3) # Step 3: Fetch markets for this event try: mr = await self._http.get( f"{KALSHI_BASE}/markets", params={"event_ticker": event_ticker, "status": "open", "limit": 10}, headers={"Accept": "application/json"}, timeout=8.0, ) if mr.status_code == 200: markets_data = mr.json() for m in markets_data.get("markets", []): pm = self._parse_kalshi_market(m) if pm: results.append(pm) except Exception: continue return results except Exception as e: logger.warning(f"Kalshi search error: {e}") return [] def _parse_kalshi_market(self, m: dict) -> PredictionMarket | None: """Parse a Kalshi market dict into unified PredictionMarket.""" try: ticker = m.get("ticker", "") title = m.get("title", "") yes_bid = float(m.get("yes_bid_dollars", 0)) volume = float(m.get("volume_fp", 0)) # Kalshi uses fake-penny notation m.get("event_ticker", "") category = m.get("category", "") # Skip multi-outcome markets (sports parlays, etc.) - they have no yes_bid if yes_bid <= 0 or ",yes " in title.lower(): return None prob_yes = yes_bid # Best YES bid approximates probability prob_no = 1.0 - prob_yes if prob_yes else 0.5 tokens_mentioned = _extract_token_symbols(title) return PredictionMarket( source="kalshi", source_id=ticker, question=title, slug=ticker, probability_yes=prob_yes, probability_no=prob_no, volume_usd=volume, category=category, tokens_mentioned=tokens_mentioned, url=f"https://kalshi.com/markets/{ticker}" if ticker else "", updated_at=datetime.now(UTC).isoformat(), ) except Exception as e: logger.warning(f"Failed to parse Kalshi market: {e}") return None async def _trending_kalshi(self, limit: int) -> list[PredictionMarket]: """Get trending Kalshi markets by volume - uses events-first approach.""" try: # Get open events (avoid sports-multi-outcome noise from raw /markets) resp = await self._http.get( f"{KALSHI_BASE}/events", params={"status": "open", "limit": min(limit * 2, 30)}, timeout=10.0, ) if resp.status_code != 200: return [] data = resp.json() events = data.get("events", []) results = [] for event in events[:limit]: event_ticker = event.get("ticker", "") try: mr = await self._http.get( f"{KALSHI_BASE}/markets", params={"event_ticker": event_ticker, "status": "open", "limit": 5}, timeout=8.0, ) if mr.status_code == 200: markets_data = mr.json() for m in markets_data.get("markets", []): pm = self._parse_kalshi_market(m) if pm: results.append(pm) except Exception: continue return results[:limit] except Exception as e: logger.warning(f"Kalshi trending error: {e}") return [] async def _kalshi_detail(self, ticker: str) -> PredictionMarket | None: """Get detailed Kalshi market data including orderbook.""" try: resp = await self._http.get( f"{KALSHI_BASE}/markets/{ticker}/orderbook", timeout=10.0, ) if resp.status_code != 200: return None data = resp.json() orderbook = data.get("orderbook_fp", {}) yes_bids = orderbook.get("yes_dollars", []) best_yes = float(yes_bids[0][0]) if yes_bids else 0.5 # Also get market metadata meta_resp = await self._http.get( f"{KALSHI_BASE}/markets", params={"ticker": ticker}, timeout=10.0, ) if meta_resp.status_code == 200: meta_data = meta_resp.json() markets = meta_data.get("markets", []) if markets: pm = self._parse_kalshi_market(markets[0]) if pm: pm.probability_yes = best_yes pm.probability_no = 1.0 - best_yes return pm return None except Exception as e: logger.warning(f"Kalshi detail error for {ticker}: {e}") return None # ── Limitless ─────────────────────────────────────────── async def _search_limitless(self, query: str) -> list[PredictionMarket]: """Search Limitless Exchange markets by fetching active and filtering.""" try: resp = await self._http.get( f"{LIMITLESS_BASE}/markets/active", params={"limit": 25}, headers={"Accept": "application/json"}, timeout=10.0, ) if resp.status_code != 200: logger.warning(f"Limitless markets returned {resp.status_code}") return [] data = resp.json() all_markets = data.get("data", []) # Filter client-side by query terms query_lower = query.lower() query_terms = query_lower.split() results = [] for m in all_markets: title = m.get("title", "").lower() if any(term in title for term in query_terms): pm = self._parse_limitless_market(m) if pm: results.append(pm) return results except Exception as e: logger.warning(f"Limitless search error: {e}") return [] def _parse_limitless_market(self, m: dict) -> PredictionMarket | None: """Parse a Limitless market dict into unified PredictionMarket.""" try: title = m.get("title", "") slug = m.get("slug", str(m.get("id", ""))) # Prices: [YES%, NO%] - e.g., [42.8, 57.2] prices = m.get("prices", [50, 50]) prob_yes = float(prices[0]) / 100 if isinstance(prices, list) and len(prices) >= 1 else 0.5 prob_no = float(prices[1]) / 100 if isinstance(prices, list) and len(prices) >= 2 else 0.5 # Volume: use volumeFormatted if available, else volume vol_str = m.get("volumeFormatted", str(m.get("volume", 0))) volume = float(vol_str) if vol_str else 0.0 categories = m.get("categories", []) category = categories[0] if categories else "" tags = m.get("tags", []) tokens_mentioned = _extract_token_symbols(title) return PredictionMarket( source="limitless", source_id=str(slug), question=title, slug=str(slug), probability_yes=prob_yes, probability_no=prob_no, volume_usd=volume, category=category, tags=tags, tokens_mentioned=tokens_mentioned, url=f"https://limitless.exchange/markets/{slug}" if slug else "", ends_at=m.get("expirationDate", ""), updated_at=datetime.now(UTC).isoformat(), ) except Exception as e: logger.warning(f"Failed to parse Limitless market: {e}") return None async def _trending_limitless(self, limit: int) -> list[PredictionMarket]: """Get trending Limitless markets.""" try: limit = min(limit, 25) # API max resp = await self._http.get( f"{LIMITLESS_BASE}/markets/active", params={"limit": limit}, headers={"Accept": "application/json"}, timeout=10.0, ) if resp.status_code != 200: return [] data = resp.json() markets = data.get("data", []) return [pm for m in markets[:limit] if (pm := self._parse_limitless_market(m))] except Exception as e: logger.warning(f"Limitless trending error: {e}") return [] # ── Manifold ──────────────────────────────────────────── async def _search_manifold(self, query: str) -> list[PredictionMarket]: """Search Manifold Markets (play-money, sentiment signals). Manifold is pure play-money but useful for: - Forecasting community sentiment - Early signal detection (top forecasters often move before real-money markets) - Broad question coverage (more niche crypto questions than Polymarket) """ try: resp = await self._http.get( f"{MANIFOLD_BASE}/search-markets", params={"term": query, "limit": 20}, timeout=10.0, ) if resp.status_code != 200: logger.warning(f"Manifold search returned {resp.status_code}") return [] data = resp.json() contracts = data if isinstance(data, list) else data.get("contracts", data.get("markets", [])) results = [] for c in contracts[:10]: pm = self._parse_manifold_market(c) if pm: results.append(pm) return results except Exception as e: logger.warning(f"Manifold search error: {e}") return [] def _parse_manifold_market(self, c: dict) -> PredictionMarket | None: """Parse a Manifold contract into unified PredictionMarket.""" try: question = c.get("question", "") slug = c.get("slug", c.get("id", "")) prob = float(c.get("probability", c.get("prob", 0.5))) volume = float(c.get("volume", c.get("volume24Hours", 0))) # Manifold uses "Mana" play money, volume is a signal but lower weight tokens_mentioned = _extract_token_symbols(question) tags = list(c.get("tags", [])) return PredictionMarket( source="manifold", source_id=str(slug), question=question, slug=str(slug), probability_yes=prob, probability_no=1.0 - prob, volume_usd=volume, category="", tags=tags, tokens_mentioned=tokens_mentioned, url=f"https://manifold.markets/{c.get('creatorUsername', '')}/{slug}" if slug else "", updated_at=datetime.now(UTC).isoformat(), ) except Exception as e: logger.warning(f"Failed to parse Manifold market: {e}") return None # ── Helpers ───────────────────────────────────────────── @staticmethod def _parse_json_field(val): """Parse double-encoded JSON fields (Polymarket Gamma API).""" if isinstance(val, str): try: return json.loads(val) except (json.JSONDecodeError, TypeError): return val return val # ── Singleton ──────────────────────────────────────────────────── _service: PredictionMarketService | None = None def get_prediction_market_service() -> PredictionMarketService: """Get or create the singleton PredictionMarketService.""" global _service if _service is None: _service = PredictionMarketService() return _service # ── Helpers: Token Extraction ──────────────────────────────────── # Common token symbols to detect in market questions _COMMON_TOKENS = { "BTC", "ETH", "SOL", "USDT", "USDC", "DAI", "BNB", "XRP", "ADA", "DOGE", "MATIC", "POL", "DOT", "AVAX", "LINK", "UNI", "AAVE", "ARB", "OP", "SUI", "APT", "TIA", "SEI", "STRK", "WLD", "PEPE", "SHIB", "BONK", "WIF", "JUP", "PYTH", "RNDR", "FET", "AGIX", "OCEAN", "IMX", "INJ", "EIGEN", "ENA", "ETHFI", } # Broader project names often referenced in prediction markets _COMMON_PROJECTS = { "polymarket", "kalshi", "manifold", "uniswap", "sushiswap", "aave", "compound", "makerdao", "maker", "lido", "eigenlayer", "chainlink", "arbitrum", "optimism", "polygon", "avalanche", "fantom", "near", "celestia", "worldcoin", "tether", "circle", "coinbase", "binance", "kraken", "ftx", "celcius", "blockfi", "three arrows", "alameda", "jump crypto", "wintermute", "curve", "balancer", "thorchain", "osmosis", "dydx", "gmx", "hyperliquid", "jupiter", "raydium", "orca", "wormhole", "layerzero", "zksync", "starknet", "scroll", "linea", "base", "mantle", "mode", "blast", } def _extract_token_symbols(text: str) -> list[str]: """Extract known token symbols and project names from text.""" found = [] text_upper = text.upper() text_lower = text.lower() # Check token symbols (typically uppercase in text) for token in _COMMON_TOKENS: # Match as word boundary: " BTC " or "BTC's" or "$BTC" if ( f" {token} " in f" {text_upper} " or f"${token}" in text_upper or text_upper.startswith(f"{token} ") or text_upper.endswith(f" {token}") ) and token not in found: found.append(token) # Check project names (case-insensitive) for project in _COMMON_PROJECTS: if project in text_lower and project.upper() not in found: found.append(project) return found # ── Caching Helpers ────────────────────────────────────────────── def _cache_hash(*args) -> str: """Create a short hash for cache keys.""" raw = "|".join(str(a) for a in args) return hashlib.md5(raw.encode()).hexdigest()[:12] def _market_to_dict(m: PredictionMarket) -> dict: """Serialize PredictionMarket to dict for JSON caching.""" return { "source": m.source, "source_id": m.source_id, "question": m.question, "slug": m.slug, "probability_yes": m.probability_yes, "probability_no": m.probability_no, "volume_usd": m.volume_usd, "liquidity_usd": m.liquidity_usd, "category": m.category, "tags": m.tags, "tokens_mentioned": m.tokens_mentioned, "is_security_relevant": m.is_security_relevant, "is_crypto_relevant": m.is_crypto_relevant, "url": m.url, "ends_at": m.ends_at, "updated_at": m.updated_at, } def _dict_to_market(d: dict) -> PredictionMarket: """Deserialize dict back to PredictionMarket.""" return PredictionMarket( source=d.get("source", ""), source_id=d.get("source_id", ""), question=d.get("question", ""), slug=d.get("slug", ""), probability_yes=float(d.get("probability_yes", 0.5)), probability_no=float(d.get("probability_no", 0.5)), volume_usd=float(d.get("volume_usd", 0)), liquidity_usd=float(d.get("liquidity_usd", 0)), category=d.get("category", ""), tags=d.get("tags", []), tokens_mentioned=d.get("tokens_mentioned", []), is_security_relevant=d.get("is_security_relevant", False), is_crypto_relevant=d.get("is_crypto_relevant", False), url=d.get("url", ""), ends_at=d.get("ends_at", ""), updated_at=d.get("updated_at", ""), ) def _digest_to_dict(d: PredictionDigest) -> dict: """Serialize PredictionDigest to dict for JSON caching.""" return { "generated_at": d.generated_at, "total_markets_searched": d.total_markets_searched, "security_relevant_count": d.security_relevant_count, "crypto_relevant_count": d.crypto_relevant_count, "top_threats": [_market_to_dict(m) for m in d.top_threats], "token_specific_markets": [_market_to_dict(m) for m in d.token_specific_markets], "ecosystem_risk_markets": [_market_to_dict(m) for m in d.ecosystem_risk_markets], "regulatory_markets": [_market_to_dict(m) for m in d.regulatory_markets], } def _dict_to_digest(d: dict) -> PredictionDigest: """Deserialize dict back to PredictionDigest.""" return PredictionDigest( generated_at=d.get("generated_at", ""), total_markets_searched=d.get("total_markets_searched", 0), security_relevant_count=d.get("security_relevant_count", 0), crypto_relevant_count=d.get("crypto_relevant_count", 0), top_threats=[_dict_to_market(m) for m in d.get("top_threats", [])], token_specific_markets=[_dict_to_market(m) for m in d.get("token_specific_markets", [])], ecosystem_risk_markets=[_dict_to_market(m) for m in d.get("ecosystem_risk_markets", [])], regulatory_markets=[_dict_to_market(m) for m in d.get("regulatory_markets", [])], )