""" Polymarket CLOB client wrapper. Thin layer around the official py-clob-client SDK, adapted for the LuckySt Trading Terminal interface. Provides synchronous helpers (the SDK itself is synchronous) that our async traders call via asyncio.to_thread() when needed. Requires: pip install py-clob-client """ import time import json import requests from typing import Optional, List, Dict, Any from dataclasses import dataclass, field # --------------------------------------------------------------------------- # Polymarket SDK imports # --------------------------------------------------------------------------- from py_clob_client.client import ClobClient from py_clob_client.clob_types import ( OrderArgs, MarketOrderArgs, OrderType, BookParams, OpenOrderParams, ) from py_clob_client.order_builder.constants import BUY, SELL # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- POLY_HOST = "https://clob.polymarket.com" GAMMA_HOST = "https://gamma-api.polymarket.com" POLYGON_CHAIN_ID = 137 # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class PolyOrderBookLevel: price: float # 0.00 – 1.00 size: float @dataclass class PolyOrderBook: bids: List[PolyOrderBookLevel] = field(default_factory=list) asks: List[PolyOrderBookLevel] = field(default_factory=list) tick_size: str = "0.01" neg_risk: bool = False @dataclass class PolyMarket: """Subset of Gamma market fields we care about.""" condition_id: str question: str slug: str outcomes: List[str] # ["Yes", "No"] token_ids: List[str] # [yes_token_id, no_token_id] outcome_prices: List[str] # ["0.55", "0.45"] end_date: str = "" active: bool = True neg_risk: bool = False tick_size: str = "0.01" # --------------------------------------------------------------------------- # Client wrapper # --------------------------------------------------------------------------- class PolyClient: """Wrapper around py-clob-client for LuckySt terminal integration.""" def __init__( self, private_key: str, *, signature_type: int = 0, # 0=EOA, 2=POLY_GNOSIS_SAFE funder: Optional[str] = None, ): # Normalise private key pk = private_key.strip() if not pk.startswith("0x"): pk = "0x" + pk self.client = ClobClient( host=POLY_HOST, key=pk, chain_id=POLYGON_CHAIN_ID, signature_type=signature_type, funder=funder, ) # Derive / create L2 API credentials creds = self.client.create_or_derive_api_creds() self.client.set_api_creds(creds) self.api_key = creds.api_key # Derive wallet address from key from eth_account import Account self.address = Account.from_key(pk).address # ------------------------------------------------------------------ # Market discovery # ------------------------------------------------------------------ def search_markets( self, slug_contains: Optional[str] = None, active: bool = True, limit: int = 50, ) -> List[PolyMarket]: """Search Gamma API for markets.""" params: Dict[str, Any] = {"limit": limit} if active: params["active"] = True params["closed"] = False if slug_contains: params["slug"] = slug_contains resp = requests.get(f"{GAMMA_HOST}/markets", params=params, timeout=10) resp.raise_for_status() results = [] for m in resp.json(): try: token_ids = json.loads(m.get("clobTokenIds", "[]")) outcome_prices = json.loads(m.get("outcomePrices", "[]")) outcomes = m.get("outcomes", '["Yes","No"]') if isinstance(outcomes, str): outcomes = json.loads(outcomes) results.append(PolyMarket( condition_id=m.get("conditionId", ""), question=m.get("question", ""), slug=m.get("slug", ""), outcomes=outcomes, token_ids=token_ids, outcome_prices=outcome_prices, end_date=m.get("endDate", ""), active=m.get("active", True), neg_risk=m.get("negRisk", False), tick_size=str(m.get("minimum_tick_size", "0.01")), )) except Exception: continue return results def find_interval_markets(self, asset: str = "btc", interval: str = "15m") -> List[PolyMarket]: """Find active 5m/15m crypto interval markets.""" slug_pattern = f"{asset.lower()}-updown-{interval}" return self.search_markets(slug_contains=slug_pattern, active=True) def get_market_by_condition(self, condition_id: str) -> Optional[PolyMarket]: """Get a single market by condition_id.""" try: resp = requests.get( f"{GAMMA_HOST}/markets", params={"id": condition_id, "limit": 1}, timeout=10, ) resp.raise_for_status() markets = resp.json() if markets: m = markets[0] token_ids = json.loads(m.get("clobTokenIds", "[]")) outcome_prices = json.loads(m.get("outcomePrices", "[]")) outcomes = m.get("outcomes", '["Yes","No"]') if isinstance(outcomes, str): outcomes = json.loads(outcomes) return PolyMarket( condition_id=m.get("conditionId", ""), question=m.get("question", ""), slug=m.get("slug", ""), outcomes=outcomes, token_ids=token_ids, outcome_prices=outcome_prices, end_date=m.get("endDate", ""), active=m.get("active", True), neg_risk=m.get("negRisk", False), tick_size=str(m.get("minimum_tick_size", "0.01")), ) except Exception: pass return None # ------------------------------------------------------------------ # Order book # ------------------------------------------------------------------ def get_orderbook(self, token_id: str) -> PolyOrderBook: """Get order book for a token_id (YES or NO side).""" try: book = self.client.get_order_book(token_id) bids = [PolyOrderBookLevel(float(b.price), float(b.size)) for b in (book.bids or [])] asks = [PolyOrderBookLevel(float(a.price), float(a.size)) for a in (book.asks or [])] # Sort: bids descending, asks ascending bids.sort(key=lambda x: x.price, reverse=True) asks.sort(key=lambda x: x.price) return PolyOrderBook( bids=bids, asks=asks, tick_size=str(getattr(book, "tick_size", "0.01")), neg_risk=getattr(book, "neg_risk", False), ) except Exception: return PolyOrderBook() def get_midpoint(self, token_id: str) -> Optional[float]: try: mid = self.client.get_midpoint(token_id) return float(mid) if mid else None except Exception: return None # ------------------------------------------------------------------ # Order placement # ------------------------------------------------------------------ def place_limit_order( self, token_id: str, price: float, size: float, side: str = "BUY", tick_size: str = "0.01", neg_risk: bool = False, ) -> Optional[str]: """Place a GTC limit order. Returns order_id or None.""" try: order_args = OrderArgs( token_id=token_id, price=price, size=size, side=BUY if side.upper() == "BUY" else SELL, fee_rate_bps=0, nonce=0, expiration=0, ) signed = self.client.create_order(order_args) resp = self.client.post_order(signed, OrderType.GTC) return resp.get("orderID") except Exception as e: print(f"❌ Polymarket order failed: {e}") return None def cancel_order(self, order_id: str) -> bool: try: self.client.cancel(order_id) return True except Exception: return False def cancel_all(self) -> bool: try: self.client.cancel_all() return True except Exception: return False def cancel_market_orders(self, token_id: str) -> bool: """Cancel all orders for a specific token.""" try: self.client.cancel_market_orders(market=token_id) return True except Exception: return False # ------------------------------------------------------------------ # Orders & trades # ------------------------------------------------------------------ def get_open_orders(self, token_id: Optional[str] = None) -> List[Dict]: """Get open orders, optionally filtered by token_id.""" try: params = OpenOrderParams() if token_id: params.asset_id = token_id orders = self.client.get_orders(params) return orders if isinstance(orders, list) else [] except Exception: return [] def get_trades(self) -> List[Dict]: try: return self.client.get_trades() or [] except Exception: return [] # ------------------------------------------------------------------ # Utility # ------------------------------------------------------------------ def close(self): """Clean up (no persistent connections in SDK).""" pass