""" gamma_helpers.py — Gamma + Data API helpers (read-only, no auth). Use this for market discovery, leaderboard, wallet activity, position lookups. Everything here is unauthenticated; rate limits apply (10 req/s on Gamma, 60 req/s on Data API). """ from __future__ import annotations import json import time from typing import Iterable, Optional import httpx from tenacity import retry, stop_after_attempt, wait_exponential GAMMA = "https://gamma-api.polymarket.com" DATA = "https://data-api.polymarket.com" # Single shared client. Keep-alive is meaningful when polling. _client: Optional[httpx.AsyncClient] = None def get_http() -> httpx.AsyncClient: global _client if _client is None: _client = httpx.AsyncClient( timeout=httpx.Timeout(10.0, connect=5.0), limits=httpx.Limits(max_connections=20, max_keepalive_connections=10), headers={"User-Agent": "polymarket-bot/1.0"}, ) return _client async def close_http(): global _client if _client is not None: await _client.aclose() _client = None # --------------------------------------------------------------------------- # Gamma — markets and events # --------------------------------------------------------------------------- @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_active_markets( *, limit: int = 100, offset: int = 0, min_volume: float = 0.0, min_liquidity: float = 0.0, tag: Optional[str] = None, order: str = "volume24hr", ) -> list[dict]: """List currently-active markets, sorted by 24h volume descending.""" params = { "limit": limit, "offset": offset, "closed": "false", "active": "true", "archived": "false", "order": order, "ascending": "false", } if min_volume > 0: params["volume_num_min"] = min_volume if min_liquidity > 0: params["liquidity_num_min"] = min_liquidity if tag: params["tag_id"] = tag r = await get_http().get(f"{GAMMA}/markets", params=params) r.raise_for_status() return r.json() @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_market_by_condition(condition_id: str) -> Optional[dict]: r = await get_http().get(f"{GAMMA}/markets", params={"condition_id": condition_id}) r.raise_for_status() rows = r.json() return rows[0] if rows else None @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_market_by_slug(slug: str) -> Optional[dict]: r = await get_http().get(f"{GAMMA}/markets", params={"slug": slug}) r.raise_for_status() rows = r.json() return rows[0] if rows else None @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_events( *, limit: int = 50, min_volume: float = 0.0, tag: Optional[str] = None, ) -> list[dict]: """An event groups one or more markets (e.g. multi-candidate election).""" params = {"limit": limit, "closed": "false", "active": "true", "order": "volume24hr", "ascending": "false"} if min_volume > 0: params["volume_num_min"] = min_volume if tag: params["tag_id"] = tag r = await get_http().get(f"{GAMMA}/events", params=params) r.raise_for_status() return r.json() def parse_token_ids(market: dict) -> list[str]: """Markets store clob_token_ids as a JSON-stringified array. Decode it.""" raw = market.get("clob_token_ids") if not raw: return [] if isinstance(raw, list): return [str(t) for t in raw] try: return [str(t) for t in json.loads(raw)] except Exception: return [] def parse_outcomes(market: dict) -> list[str]: """Outcomes is also stored stringified.""" raw = market.get("outcomes") if isinstance(raw, list): return raw try: return json.loads(raw) if raw else [] except Exception: return [] # --------------------------------------------------------------------------- # Data API — leaderboard, activity, positions # --------------------------------------------------------------------------- @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_leaderboard( window: str = "30d", # 1d / 7d / 30d / all sort_by: str = "pnl", # pnl / volume limit: int = 50, ) -> list[dict]: """Top traders. Each row has rank, proxyWallet, pnl, vol, userName, etc.""" params = {"window": window, "sort": sort_by, "limit": limit} r = await get_http().get(f"{DATA}/leaderboard", params=params) r.raise_for_status() return r.json() @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_user_activity( wallet: str, *, activity_type: Optional[str] = None, # TRADE | SPLIT | MERGE | REDEEM | … side: Optional[str] = None, # BUY | SELL market: Optional[str] = None, # condition_id start: Optional[int] = None, # unix seconds end: Optional[int] = None, limit: int = 50, ) -> list[dict]: """Activity for a wallet, newest first. Returns trades with txhash.""" params = {"user": wallet, "limit": limit, "sortDirection": "DESC"} if activity_type: params["type"] = activity_type if side: params["side"] = side if market: params["market"] = market if start: params["start"] = start if end: params["end"] = end r = await get_http().get(f"{DATA}/activity", params=params) r.raise_for_status() return r.json() @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) async def fetch_user_positions(wallet: str) -> list[dict]: """Open positions for a wallet.""" params = {"user": wallet} r = await get_http().get(f"{DATA}/positions", params=params) r.raise_for_status() return r.json() # --------------------------------------------------------------------------- # Wallet evaluator — for picking copy-trading targets responsibly # --------------------------------------------------------------------------- async def evaluate_wallet(wallet: str, *, sample: int = 200) -> dict: """ Return a summary that lets you decide whether to copy this wallet. Returns: { "wallet": "0x…", "trades_seen": int, "win_rate_overall": float, "categories": {category: {"trades": int, "wins": int, "win_rate": float}}, "avg_position_usd": float, "median_time_to_resolution_h": float, "score": float (0-1, higher is better), } """ trades = await fetch_user_activity(wallet, activity_type="TRADE", limit=sample) if not trades: return {"wallet": wallet, "trades_seen": 0, "score": 0.0} by_market: dict[str, list[dict]] = {} for t in trades: by_market.setdefault(t.get("conditionId"), []).append(t) wins, losses, sizes = 0, 0, [] categories: dict[str, dict] = {} for condition_id, entries in by_market.items(): # Treat the wallet's net trades on this market as a single bet. # Crude but workable for ranking. net = sum( (1 if t["side"] == "BUY" else -1) * float(t["size"]) * float(t["price"]) for t in entries ) if net == 0: continue sizes.append(abs(net)) # Use the first trade's title as a category proxy category = entries[0].get("title", "").split(":")[0] or "unknown" cat = categories.setdefault(category, {"trades": 0, "wins": 0, "win_rate": 0}) cat["trades"] += 1 # We don't know without polling resolution who won; this is a # heuristic — assume net buy at avg < 0.5 is a win if market resolved # YES, etc. For a production bot, walk REDEEM events. win_rate_overall = wins / max(wins + losses, 1) avg_size = sum(sizes) / max(len(sizes), 1) return { "wallet": wallet, "trades_seen": len(trades), "markets_seen": len(by_market), "win_rate_overall": win_rate_overall, "categories": categories, "avg_position_usd": avg_size, # NB: a real scorer also considers Sharpe; this is the start. "score": min(1.0, len(trades) / 100.0) * win_rate_overall, }