"""Polymarket API client for Gamma (metadata) and CLOB (prices/order books).""" from __future__ import annotations import asyncio from typing import Any import httpx from tenacity import retry, stop_after_attempt, wait_exponential from src.config import ( CLOB_API, ELECTION_2024_EVENT_ID, ELECTION_2024_MARKETS, FED_JUNE_2026_EVENT_ID, FED_JUNE_2026_MARKETS, GAMMA_API, DATA_RAW_POLY, ) from src.utils import save_json, setup_logging logger = setup_logging("polymarket") # Shared client limits _LIMITS = httpx.Limits(max_connections=5) # --------------------------------------------------------------------------- # Low-level helpers # --------------------------------------------------------------------------- @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def _get(client: httpx.AsyncClient, url: str, params: dict | None = None) -> Any: """GET with retry and JSON decode.""" resp = await client.get(url, params=params) resp.raise_for_status() return resp.json() # --------------------------------------------------------------------------- # Gamma API (market metadata / discovery) # --------------------------------------------------------------------------- async def fetch_market(market_id: str) -> dict: """Fetch market metadata from Gamma API.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get(client, f"{GAMMA_API}/markets/{market_id}") save_json(data, DATA_RAW_POLY / f"market_{market_id}.json") logger.info("Fetched market %s", market_id) return data async def fetch_event(event_id: int) -> dict: """Fetch event with all its markets.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get(client, f"{GAMMA_API}/events/{event_id}") save_json(data, DATA_RAW_POLY / f"event_{event_id}.json") logger.info("Fetched event %s", event_id) return data async def discover_markets(volume_min: int = 500_000, limit: int = 100) -> list[dict]: """Discover high-volume markets from Gamma API.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get( client, f"{GAMMA_API}/markets", params={ "volume_num_min": volume_min, "limit": limit, "active": "true", "order": "volume", "ascending": "false", }, ) save_json(data, DATA_RAW_POLY / "discovered_markets.json") logger.info("Discovered %d high-volume markets", len(data) if isinstance(data, list) else 0) return data if isinstance(data, list) else [] # --------------------------------------------------------------------------- # CLOB API (prices, order books) # --------------------------------------------------------------------------- async def fetch_price_history( token_id: str, interval: str = "max", fidelity: int = 1440, ) -> list[dict]: """Fetch price history from CLOB API. Returns list of {"t": unix_timestamp, "p": price_string}. For closed markets: interval="max", fidelity=1440 (daily). For active markets: interval="all", fidelity=1 (minutely) or fidelity=60 (hourly). """ async with httpx.AsyncClient(limits=_LIMITS, timeout=60) as client: data = await _get( client, f"{CLOB_API}/prices-history", params={"market": token_id, "interval": interval, "fidelity": fidelity}, ) history = data.get("history", data) if isinstance(data, dict) else data short_id = token_id[:16] save_json(history, DATA_RAW_POLY / f"price_history_{short_id}.json") logger.info("Fetched price history for token %s... (%d points)", short_id, len(history) if isinstance(history, list) else 0) return history if isinstance(history, list) else [] async def fetch_order_book(token_id: str) -> dict: """Fetch current order book snapshot.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get(client, f"{CLOB_API}/book", params={"token_id": token_id}) short_id = token_id[:16] save_json(data, DATA_RAW_POLY / f"order_book_{short_id}.json") logger.info("Fetched order book for token %s...", short_id) return data async def fetch_midpoint(token_id: str) -> float: """Fetch current midpoint price.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get(client, f"{CLOB_API}/midpoint", params={"token_id": token_id}) mid = float(data.get("mid", 0.0)) if isinstance(data, dict) else 0.0 logger.info("Midpoint for %s...: %.4f", token_id[:16], mid) return mid async def fetch_spread(token_id: str) -> float: """Fetch current spread.""" async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client: data = await _get(client, f"{CLOB_API}/spread", params={"token_id": token_id}) spread = float(data.get("spread", 0.0)) if isinstance(data, dict) else 0.0 logger.info("Spread for %s...: %.4f", token_id[:16], spread) return spread # --------------------------------------------------------------------------- # Master fetch # --------------------------------------------------------------------------- def _extract_tokens_from_event(event_data: dict) -> list[dict]: """Extract market info with correct tokens from event response.""" results = [] for m in event_data.get("markets", []): import json as _json tokens = _json.loads(m.get("clobTokenIds", "[]")) if tokens: results.append({ "id": m["id"], "name": m.get("groupItemTitle", m.get("question", "")[:40]), "question": m.get("question", ""), "yes_token": tokens[0], "no_token": tokens[1] if len(tokens) > 1 else "", "closed": m.get("closed", False), "prices": m.get("outcomePrices", ""), "volume": m.get("volumeNum", m.get("volume", 0)), }) return results async def _fetch_event_with_histories( event_id: int, label: str, interval: str = "max", fidelity: int = 1440, ) -> list[dict]: """Fetch an event, then fetch price history for every market's YES token.""" logger.info("--- %s (event %d) ---", label, event_id) try: event_data = await fetch_event(event_id) except Exception as exc: logger.warning("Could not fetch event %d: %s", event_id, exc) return [] markets = _extract_tokens_from_event(event_data) for m in markets: try: hist = await fetch_price_history(m["yes_token"], interval=interval, fidelity=fidelity) logger.info(" %s: %d price points", m["name"], len(hist)) except Exception as exc: logger.warning(" Could not fetch history for %s: %s", m["name"], exc) await asyncio.sleep(0.2) return markets async def fetch_all_data() -> None: """Master function: fetch all configured markets and save to data/raw/polymarket/.""" DATA_RAW_POLY.mkdir(parents=True, exist_ok=True) logger.info("=== Starting Polymarket data acquisition ===") # 1. Fed June 2026 event — Arrow-Debreu complete state space fed_markets = await _fetch_event_with_histories( FED_JUNE_2026_EVENT_ID, "Fed Decision June 2026" ) # 2. Election 2024 event — highest-volume prediction market ever election_markets = await _fetch_event_with_histories( ELECTION_2024_EVENT_ID, "Presidential Election 2024" ) # 3. Order book snapshots for active Fed markets logger.info("--- Order book snapshots ---") for m in fed_markets: if not m["closed"]: try: await fetch_order_book(m["yes_token"]) except Exception as exc: logger.warning("Could not fetch order book for %s: %s", m["name"], exc) # 4. Discover high-volume markets logger.info("--- Market discovery ---") try: await discover_markets(volume_min=500_000, limit=100) except Exception as exc: logger.warning("Could not discover markets: %s", exc) logger.info("=== Polymarket data acquisition complete ===")