Python 56.2%
TeX 43.7%
1"""Polymarket API client for Gamma (metadata) and CLOB (prices/order books)."""23from __future__ import annotations45import asyncio6from typing import Any78import httpx9from tenacity import retry, stop_after_attempt, wait_exponential1011from src.config import (12 CLOB_API,13 ELECTION_2024_EVENT_ID,14 ELECTION_2024_MARKETS,15 FED_JUNE_2026_EVENT_ID,16 FED_JUNE_2026_MARKETS,17 GAMMA_API,18 DATA_RAW_POLY,19)20from src.utils import save_json, setup_logging2122logger = setup_logging("polymarket")2324# Shared client limits25_LIMITS = httpx.Limits(max_connections=5)262728# ---------------------------------------------------------------------------29# Low-level helpers30# ---------------------------------------------------------------------------3132@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))33async def _get(client: httpx.AsyncClient, url: str, params: dict | None = None) -> Any:34 """GET with retry and JSON decode."""35 resp = await client.get(url, params=params)36 resp.raise_for_status()37 return resp.json()383940# ---------------------------------------------------------------------------41# Gamma API (market metadata / discovery)42# ---------------------------------------------------------------------------4344async def fetch_market(market_id: str) -> dict:45 """Fetch market metadata from Gamma API."""46 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:47 data = await _get(client, f"{GAMMA_API}/markets/{market_id}")48 save_json(data, DATA_RAW_POLY / f"market_{market_id}.json")49 logger.info("Fetched market %s", market_id)50 return data515253async def fetch_event(event_id: int) -> dict:54 """Fetch event with all its markets."""55 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:56 data = await _get(client, f"{GAMMA_API}/events/{event_id}")57 save_json(data, DATA_RAW_POLY / f"event_{event_id}.json")58 logger.info("Fetched event %s", event_id)59 return data606162async def discover_markets(volume_min: int = 500_000, limit: int = 100) -> list[dict]:63 """Discover high-volume markets from Gamma API."""64 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:65 data = await _get(66 client,67 f"{GAMMA_API}/markets",68 params={69 "volume_num_min": volume_min,70 "limit": limit,71 "active": "true",72 "order": "volume",73 "ascending": "false",74 },75 )76 save_json(data, DATA_RAW_POLY / "discovered_markets.json")77 logger.info("Discovered %d high-volume markets", len(data) if isinstance(data, list) else 0)78 return data if isinstance(data, list) else []798081# ---------------------------------------------------------------------------82# CLOB API (prices, order books)83# ---------------------------------------------------------------------------8485async def fetch_price_history(86 token_id: str,87 interval: str = "max",88 fidelity: int = 1440,89) -> list[dict]:90 """Fetch price history from CLOB API.9192 Returns list of {"t": unix_timestamp, "p": price_string}.93 For closed markets: interval="max", fidelity=1440 (daily).94 For active markets: interval="all", fidelity=1 (minutely) or fidelity=60 (hourly).95 """96 async with httpx.AsyncClient(limits=_LIMITS, timeout=60) as client:97 data = await _get(98 client,99 f"{CLOB_API}/prices-history",100 params={"market": token_id, "interval": interval, "fidelity": fidelity},101 )102 history = data.get("history", data) if isinstance(data, dict) else data103 short_id = token_id[:16]104 save_json(history, DATA_RAW_POLY / f"price_history_{short_id}.json")105 logger.info("Fetched price history for token %s... (%d points)", short_id, len(history) if isinstance(history, list) else 0)106 return history if isinstance(history, list) else []107108109async def fetch_order_book(token_id: str) -> dict:110 """Fetch current order book snapshot."""111 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:112 data = await _get(client, f"{CLOB_API}/book", params={"token_id": token_id})113 short_id = token_id[:16]114 save_json(data, DATA_RAW_POLY / f"order_book_{short_id}.json")115 logger.info("Fetched order book for token %s...", short_id)116 return data117118119async def fetch_midpoint(token_id: str) -> float:120 """Fetch current midpoint price."""121 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:122 data = await _get(client, f"{CLOB_API}/midpoint", params={"token_id": token_id})123 mid = float(data.get("mid", 0.0)) if isinstance(data, dict) else 0.0124 logger.info("Midpoint for %s...: %.4f", token_id[:16], mid)125 return mid126127128async def fetch_spread(token_id: str) -> float:129 """Fetch current spread."""130 async with httpx.AsyncClient(limits=_LIMITS, timeout=30) as client:131 data = await _get(client, f"{CLOB_API}/spread", params={"token_id": token_id})132 spread = float(data.get("spread", 0.0)) if isinstance(data, dict) else 0.0133 logger.info("Spread for %s...: %.4f", token_id[:16], spread)134 return spread135136137# ---------------------------------------------------------------------------138# Master fetch139# ---------------------------------------------------------------------------140141def _extract_tokens_from_event(event_data: dict) -> list[dict]:142 """Extract market info with correct tokens from event response."""143 results = []144 for m in event_data.get("markets", []):145 import json as _json146 tokens = _json.loads(m.get("clobTokenIds", "[]"))147 if tokens:148 results.append({149 "id": m["id"],150 "name": m.get("groupItemTitle", m.get("question", "")[:40]),151 "question": m.get("question", ""),152 "yes_token": tokens[0],153 "no_token": tokens[1] if len(tokens) > 1 else "",154 "closed": m.get("closed", False),155 "prices": m.get("outcomePrices", ""),156 "volume": m.get("volumeNum", m.get("volume", 0)),157 })158 return results159160161async def _fetch_event_with_histories(162 event_id: int,163 label: str,164 interval: str = "max",165 fidelity: int = 1440,166) -> list[dict]:167 """Fetch an event, then fetch price history for every market's YES token."""168 logger.info("--- %s (event %d) ---", label, event_id)169 try:170 event_data = await fetch_event(event_id)171 except Exception as exc:172 logger.warning("Could not fetch event %d: %s", event_id, exc)173 return []174175 markets = _extract_tokens_from_event(event_data)176 for m in markets:177 try:178 hist = await fetch_price_history(m["yes_token"], interval=interval, fidelity=fidelity)179 logger.info(" %s: %d price points", m["name"], len(hist))180 except Exception as exc:181 logger.warning(" Could not fetch history for %s: %s", m["name"], exc)182 await asyncio.sleep(0.2)183184 return markets185186187async def fetch_all_data() -> None:188 """Master function: fetch all configured markets and save to data/raw/polymarket/."""189 DATA_RAW_POLY.mkdir(parents=True, exist_ok=True)190 logger.info("=== Starting Polymarket data acquisition ===")191192 # 1. Fed June 2026 event — Arrow-Debreu complete state space193 fed_markets = await _fetch_event_with_histories(194 FED_JUNE_2026_EVENT_ID, "Fed Decision June 2026"195 )196197 # 2. Election 2024 event — highest-volume prediction market ever198 election_markets = await _fetch_event_with_histories(199 ELECTION_2024_EVENT_ID, "Presidential Election 2024"200 )201202 # 3. Order book snapshots for active Fed markets203 logger.info("--- Order book snapshots ---")204 for m in fed_markets:205 if not m["closed"]:206 try:207 await fetch_order_book(m["yes_token"])208 except Exception as exc:209 logger.warning("Could not fetch order book for %s: %s", m["name"], exc)210211 # 4. Discover high-volume markets212 logger.info("--- Market discovery ---")213 try:214 await discover_markets(volume_min=500_000, limit=100)215 except Exception as exc:216 logger.warning("Could not discover markets: %s", exc)217218 logger.info("=== Polymarket data acquisition complete ===")219