#!/usr/bin/env python3 """ Polymarket API client — covers the full public + authenticated surface. APIs wrapped here (see https://docs.polymarket.com/api-reference/introduction): - Gamma (https://gamma-api.polymarket.com) — markets, events, tags, series, comments, sports, search, public profiles. Public, no auth. - Data (https://data-api.polymarket.com) — positions, trades, activity, holders, value, leaderboard, open interest. Public, no auth. - CLOB (https://clob.polymarket.com) — order book, prices, midpoints, spreads, price history (public) + orders/trades (authenticated). - Bridge (https://bridge.polymarket.com) — deposits/withdrawals. - Relayer(https://relayer-v2.polymarket.com) — proxy wallet tx submission. Read endpoints are plain HTTP (requests). Authenticated TRADING (signing orders, creating/deriving API keys, posting/cancelling orders, L2 HMAC headers) is delegated to the official `py-clob-client`, which the Polymarket docs explicitly recommend over hand-rolled EIP-712 signing. See TradingClient below. CLI: python polymarket.py [--json '{...}'] [--key value ...] python polymarket.py list-commands """ from __future__ import annotations import json import os import sys from typing import Any, Optional try: import requests except ImportError: sys.stderr.write("Missing dependency: pip install requests\n") raise GAMMA = "https://gamma-api.polymarket.com" DATA = "https://data-api.polymarket.com" CLOB = "https://clob.polymarket.com" BRIDGE = "https://bridge.polymarket.com" RELAYER = "https://relayer-v2.polymarket.com" DEFAULT_TIMEOUT = 30 class PolymarketError(RuntimeError): pass def _clean(params: Optional[dict]) -> Optional[dict]: """Drop None values; leave False/0 intact.""" if not params: return None return {k: v for k, v in params.items() if v is not None} def _request(method: str, url: str, *, params=None, json_body=None, headers=None, timeout=DEFAULT_TIMEOUT) -> Any: resp = requests.request(method, url, params=_clean(params), json=json_body, headers=headers, timeout=timeout) if resp.status_code >= 400: raise PolymarketError(f"{method} {resp.url} -> {resp.status_code}: {resp.text[:500]}") ctype = resp.headers.get("content-type", "") if "application/json" in ctype: return resp.json() return resp.text # --------------------------------------------------------------------------- # # Gamma API — markets, events, tags, series, comments, search, sports, profiles # --------------------------------------------------------------------------- # class GammaClient: def __init__(self, base=GAMMA): self.base = base def _get(self, path, **params): return _request("GET", f"{self.base}{path}", params=params) # Markets def list_markets(self, **params): return self._get("/markets", **params) def get_market(self, market_id): return self._get(f"/markets/{market_id}") def get_market_tags(self, market_id): return self._get(f"/markets/{market_id}/tags") # Events def list_events(self, **params): return self._get("/events", **params) def list_events_paginated(self, **params): return self._get("/events/pagination", **params) def get_event(self, event_id): return self._get(f"/events/{event_id}") def get_event_by_slug(self, slug): return self._get(f"/events/slug/{slug}") def get_event_tags(self, event_id): return self._get(f"/events/{event_id}/tags") # Tags def list_tags(self, **params): return self._get("/tags", **params) def get_tag(self, tag_id): return self._get(f"/tags/{tag_id}") def get_tag_by_slug(self, slug): return self._get(f"/tags/slug/{slug}") def get_related_tags_by_id(self, tag_id): return self._get(f"/tags/{tag_id}/related-tags") def get_related_tags_by_slug(self, slug): return self._get(f"/tags/slug/{slug}/related-tags") def get_tags_related_to_id(self, tag_id): return self._get(f"/tags/{tag_id}/related-tags/tags") def get_tags_related_to_slug(self, slug): return self._get(f"/tags/slug/{slug}/related-tags/tags") # Series def list_series(self, **params): return self._get("/series", **params) def get_series(self, series_id): return self._get(f"/series/{series_id}") # Comments def list_comments(self, **params): return self._get("/comments", **params) def get_comments_by_id(self, comment_id, **params): return self._get(f"/comments/{comment_id}", **params) def get_comments_by_user(self, address, **params): return self._get(f"/comments/user_address/{address}", **params) # Search def search(self, q, **params): return self._get("/public-search", q=q, **params) # Sports def get_sports_metadata(self): return self._get("/sports") def get_sports_market_types(self): return self._get("/sports/market-types") def list_teams(self, **params): return self._get("/teams", **params) # Public profiles def get_public_profile(self, address): return self._get(f"/public-profile/{address}") # --------------------------------------------------------------------------- # # Data API — positions, trades, activity, holders, value, leaderboard, OI # --------------------------------------------------------------------------- # class DataClient: def __init__(self, base=DATA): self.base = base def _get(self, path, **params): return _request("GET", f"{self.base}{path}", params=params) def positions(self, user, **params): return self._get("/positions", user=user, **params) def closed_positions(self, user, **params): return self._get("/closed-positions", user=user, **params) def market_positions(self, market, **params): return self._get("/positions", market=market, **params) def value(self, user, **params): return self._get("/value", user=user, **params) def activity(self, user, **params): return self._get("/activity", user=user, **params) def combo_activity(self, user, **params): return self._get("/combo-activity", user=user, **params) def combo_positions(self, user, **params): return self._get("/combo-positions", user=user, **params) def trades(self, **params): return self._get("/trades", **params) def holders(self, market, **params): return self._get("/holders", market=market, **params) def leaderboard(self, **params): # category, timePeriod, orderBy(PNL/VOLUME), limit(≤50), offset, user, userName return self._get("/v1/leaderboard", **params) def traded(self, user): return self._get("/traded", user=user) def open_interest(self, **params): return self._get("/oi", **params) def live_volume_event(self, event_id): return self._get("/live-volume", id=event_id) # Builder analytics (public) def builder_leaderboard(self, **params): # timePeriod(DAY/WEEK/MONTH/ALL), limit(≤50), offset return self._get("/v1/builders/leaderboard", **params) def builder_volume(self, **params): # timePeriod(DAY/WEEK/MONTH/ALL) return self._get("/v1/builders/volume", **params) def accounting_snapshot(self, user, out_path=None): """Download the accounting snapshot ZIP (positions.csv + equity.csv). Returns the saved file path. ZIP is binary, so this bypasses the JSON helper and streams to disk. """ url = f"{self.base}/v1/accounting/snapshot" resp = requests.get(url, params={"user": user}, timeout=DEFAULT_TIMEOUT) if resp.status_code >= 400: raise PolymarketError(f"GET {resp.url} -> {resp.status_code}: {resp.text[:300]}") out_path = out_path or f"polymarket_accounting_{user}.zip" with open(out_path, "wb") as f: f.write(resp.content) return {"saved": out_path, "bytes": len(resp.content)} # --------------------------------------------------------------------------- # # Combo markets (RFQ) — separate host # --------------------------------------------------------------------------- # class ComboClient: def __init__(self, base="https://combos-rfq-api.polymarket.com"): self.base = base def combo_markets(self, **params): # limit(1-100), cursor, exclude (comma-separated condition ids) return _request("GET", f"{self.base}/v1/rfq/combo-markets", params=params) # --------------------------------------------------------------------------- # # CLOB API — PUBLIC market data (no auth) # --------------------------------------------------------------------------- # class ClobPublicClient: def __init__(self, base=CLOB): self.base = base def _get(self, path, **params): return _request("GET", f"{self.base}{path}", params=params) def _post(self, path, body): return _request("POST", f"{self.base}{path}", json_body=body) # Pricing / book def order_book(self, token_id): return self._get("/book", token_id=token_id) def order_books(self, token_ids): return self._post("/books", [{"token_id": t} for t in token_ids]) def price(self, token_id, side): return self._get("/price", token_id=token_id, side=side) def prices(self, params_list): # params_list: [{"token_id":..,"side":"BUY"|"SELL"}, ...] return self._post("/prices", params_list) def midpoint(self, token_id): return self._get("/midpoint", token_id=token_id) def midpoints(self, token_ids): return self._post("/midpoints", [{"token_id": t} for t in token_ids]) def spread(self, token_id): return self._get("/spread", token_id=token_id) def spreads(self, token_ids): return self._post("/spreads", [{"token_id": t} for t in token_ids]) def last_trade_price(self, token_id): return self._get("/last-trade-price", token_id=token_id) def last_trade_prices(self, token_ids): return self._post("/last-trades-prices", [{"token_id": t} for t in token_ids]) def tick_size(self, token_id): return self._get("/tick-size", token_id=token_id) def fee_rate_bps(self, **params): return self._get("/fee-rate-bps", **params) # Price history def prices_history(self, market, **params): # market = clob token id. interval/startTs/endTs/fidelity return self._get("/prices-history", market=market, **params) # Market discovery (CLOB-side) def get_market(self, condition_id): return self._get(f"/markets/{condition_id}") def list_markets(self, next_cursor=""): return self._get("/markets", next_cursor=next_cursor) def simplified_markets(self, next_cursor=""): return self._get("/simplified-markets", next_cursor=next_cursor) def sampling_markets(self, next_cursor=""): return self._get("/sampling-markets", next_cursor=next_cursor) def sampling_simplified_markets(self, next_cursor=""): return self._get("/sampling-simplified-markets", next_cursor=next_cursor) def server_time(self): return self._get("/time") # Rewards — market configs (public, no auth) def rewards_markets_current(self, sponsored=False, next_cursor=""): return self._get("/rewards/markets/current", sponsored=sponsored, next_cursor=next_cursor) def rewards_markets_multi(self, **params): # q, tag_slug, event_id, order_by, position, min/max_volume_24hr, # min/max_spread, min/max_price, next_cursor, page_size return self._get("/rewards/markets/multi", **params) def rewards_market(self, condition_id, **params): return self._get(f"/rewards/markets/{condition_id}", **params) # Builder-attributed trades (public; needs builder_code) def builder_trades(self, builder_code, **params): # id, market, asset_id, before, after, next_cursor return self._get("/builder/trades", builder_code=builder_code, **params) # --------------------------------------------------------------------------- # # Bridge API — deposits / withdrawals # --------------------------------------------------------------------------- # class BridgeClient: def __init__(self, base=BRIDGE): self.base = base def supported_assets(self): return _request("GET", f"{self.base}/assets") def get_quote(self, body): return _request("POST", f"{self.base}/quote", json_body=body) def create_bridge_addresses(self, body): return _request("POST", f"{self.base}/bridge-addresses", json_body=body) def create_withdrawal_addresses(self, body): return _request("POST", f"{self.base}/withdrawal-addresses", json_body=body) def transaction_status(self, tx_id): return _request("GET", f"{self.base}/transactions/{tx_id}") # --------------------------------------------------------------------------- # # Relayer API — proxy wallet helpers # --------------------------------------------------------------------------- # class RelayerClient: def __init__(self, base=RELAYER): self.base = base def is_deployed(self, address): return _request("GET", f"{self.base}/deployed", params={"address": address}) def nonce(self, address): return _request("GET", f"{self.base}/nonce", params={"address": address}) def relayer_address(self, address): return _request("GET", f"{self.base}/relayer-address", params={"address": address}) def transaction(self, tx_id): return _request("GET", f"{self.base}/transaction", params={"id": tx_id}) def recent_transactions(self, address): return _request("GET", f"{self.base}/transactions", params={"address": address}) def submit(self, body): return _request("POST", f"{self.base}/submit", json_body=body) # --------------------------------------------------------------------------- # # Authenticated TRADING — delegated to official py-clob-client # --------------------------------------------------------------------------- # # The Polymarket docs recommend the official v2 client for EIP-712 L1 signing, # API-key derivation, and L2 HMAC headers. Reimplementing order signing by hand # risks losing real funds, so we wrap the maintained library: # # pip install py_clob_client_v2 # # Auth model: # L1 (wallet private key, EIP-712) -> create/derive API credentials, sign orders # L2 (apiKey/secret/passphrase, HMAC-SHA256) -> post/cancel/query orders, and the # user-scoped rewards/rebates GETs (implemented here via _l2_get). # # Env vars expected: # PK wallet private key (0x...) # POLY_PROXY_ADDRESS your Polymarket proxy/funder address (for proxy wallets) # POLY_SIGNATURE_TYPE 0=EOA, 1=POLY_PROXY, 2=POLY_GNOSIS_SAFE (optional) # CLOB_API_KEY / CLOB_SECRET / CLOB_PASSPHRASE (optional; else auto-derived) class TradingClient: def __init__(self, private_key=None, chain_id=137, signature_type=None, funder=None, host=CLOB, creds=None): try: from py_clob_client_v2 import ClobClient, ApiCreds except ImportError as e: raise PolymarketError( "Authenticated trading needs the official v2 client: " "pip install py_clob_client_v2" ) from e private_key = private_key or os.environ.get("PK") if not private_key: raise PolymarketError("Set PK env var (wallet private key) or pass private_key=") funder = funder or os.environ.get("POLY_PROXY_ADDRESS") if signature_type is None and os.environ.get("POLY_SIGNATURE_TYPE"): signature_type = int(os.environ["POLY_SIGNATURE_TYPE"]) self.host = host kwargs = dict(host=host, key=private_key, chain_id=chain_id) if signature_type is not None: kwargs["signature_type"] = signature_type if funder: kwargs["funder"] = funder self._ApiCreds = ApiCreds self.client = ClobClient(**kwargs) # Attach L2 creds: explicit -> env -> derive. self.creds = creds or self._creds_from_env() or self.create_or_derive_api_key() if hasattr(self.client, "set_api_creds"): self.client.set_api_creds(self.creds) def _creds_from_env(self): k = os.environ.get("CLOB_API_KEY") s = os.environ.get("CLOB_SECRET") p = os.environ.get("CLOB_PASSPHRASE") if k and s and p: return self._ApiCreds(api_key=k, api_secret=s, api_passphrase=p) return None # --- API key lifecycle (L1) --- def create_or_derive_api_key(self, nonce=None): # Derive first: existing accounts already have creds, so trying to CREATE # first just logs a noisy 400 ("Could not create api key"). Fall back to # create only if derive genuinely fails (brand-new account). try: return self.client.derive_api_key(nonce=nonce) except Exception: return self.client.create_api_key(nonce=nonce) # --- Orders (L1 sign + L2 post) --- def create_and_post_order(self, token_id, price, size, side, order_type="GTC"): from py_clob_client_v2 import OrderArgs, OrderType args = OrderArgs( token_id=token_id, price=float(price), size=float(size), side="BUY" if str(side).upper() == "BUY" else "SELL", # SDK accepts str ) return self.client.create_and_post_order(args, order_type=getattr(OrderType, order_type)) def create_and_post_market_order(self, token_id, amount, side, order_type="FOK"): from py_clob_client_v2 import MarketOrderArgs, OrderType args = MarketOrderArgs( token_id=token_id, amount=float(amount), side="BUY" if str(side).upper() == "BUY" else "SELL", ) return self.client.create_and_post_market_order(args, order_type=getattr(OrderType, order_type)) # --- Cancel --- def cancel_order(self, order_id): # Accept a raw order id/hash string or an OrderPayload. from py_clob_client_v2 import OrderPayload payload = order_id if not isinstance(order_id, str) else OrderPayload(orderID=order_id) return self.client.cancel_order(payload) def cancel_orders(self, order_hashes): return self.client.cancel_orders(order_hashes) def cancel_all(self): return self.client.cancel_all() def cancel_market_orders(self, payload): return self.client.cancel_market_orders(payload) # --- Query (L2) --- # These accept plain filter kwargs and build the SDK's param objects for you. # Pass a ready-made `params` object to bypass that. Note: a freshly placed # order can take a moment to show up here (indexing lag) even though it's live. def get_order(self, order_id): return self.client.get_order(order_id) def get_open_orders(self, market=None, asset_id=None, id=None, params=None, only_first_page=False): from py_clob_client_v2 import OpenOrderParams if params is None and (market or asset_id or id): params = OpenOrderParams(market=market, asset_id=asset_id, id=id) return self.client.get_open_orders(params, only_first_page=only_first_page) def get_trades(self, market=None, asset_id=None, maker_address=None, id=None, before=None, after=None, params=None): from py_clob_client_v2 import TradeParams if params is None and any([market, asset_id, maker_address, id, before, after]): params = TradeParams(market=market, asset_id=asset_id, maker_address=maker_address, id=id, before=before, after=after) return self.client.get_trades(params) def is_order_scoring(self, order_id=None, params=None): from py_clob_client_v2 import OrderScoringParams if params is None and order_id: params = OrderScoringParams(orderId=order_id) return self.client.is_order_scoring(params) def are_orders_scoring(self, order_ids=None, params=None): from py_clob_client_v2 import OrdersScoringParams if params is None and order_ids: params = OrdersScoringParams(orderIds=order_ids) return self.client.are_orders_scoring(params) def _balance_params(self, asset_type="COLLATERAL", token_id=None): # COLLATERAL = your USDC; CONDITIONAL = a specific outcome token (needs token_id). from py_clob_client_v2 import BalanceAllowanceParams, AssetType at = AssetType.CONDITIONAL if str(asset_type).upper() == "CONDITIONAL" else AssetType.COLLATERAL return BalanceAllowanceParams(asset_type=at, token_id=token_id) def get_balance_allowance(self, asset_type="COLLATERAL", token_id=None): return self.client.get_balance_allowance(self._balance_params(asset_type, token_id)) def update_balance_allowance(self, asset_type="COLLATERAL", token_id=None): return self.client.update_balance_allowance(self._balance_params(asset_type, token_id)) # --- L2-authenticated rewards / rebates GETs ---------------------------- # # These read-only endpoints require L2 HMAC headers but the SDK does not # surface them, so we sign the GET ourselves with the derived creds. This is # the documented, deterministic L2 scheme (no fund movement, no EIP-712). def _l2_get(self, path, params=None): import base64, hashlib, hmac as _hmac, time as _time from urllib.parse import urlencode c = self.creds api_key = getattr(c, "api_key", None) or c["api_key"] secret = getattr(c, "api_secret", None) or c["api_secret"] passphrase = getattr(c, "api_passphrase", None) or c["api_passphrase"] address = self.client.get_address() if hasattr(self.client, "get_address") else os.environ.get("POLY_ADDRESS", "") request_path = path qs = urlencode(_clean(params) or {}) if qs: request_path += "?" + qs ts = str(int(_time.time())) # L2 HMAC: base64url(HMAC_SHA256(b64url_decode(secret), ts+method+requestPath)) message = ts + "GET" + request_path digest = _hmac.new(base64.urlsafe_b64decode(secret), message.encode(), hashlib.sha256).digest() sig = base64.urlsafe_b64encode(digest).decode() headers = { "POLY_ADDRESS": address, "POLY_SIGNATURE": sig, "POLY_TIMESTAMP": ts, "POLY_API_KEY": api_key, "POLY_PASSPHRASE": passphrase, } return _request("GET", f"{self.host}{request_path}", headers=headers) def rewards_user_earnings(self, date, **params): return self._l2_get("/rewards/user", {"date": date, **params}) def rewards_user_total(self, date, **params): return self._l2_get("/rewards/user/total", {"date": date, **params}) def rewards_user_percentages(self, **params): return self._l2_get("/rewards/user/percentages", params) def rewards_user_markets(self, **params): return self._l2_get("/rewards/user/markets", params) def rebates_current(self, date, maker_address, **params): return self._l2_get("/rebates/current", {"date": date, "maker_address": maker_address, **params}) # --------------------------------------------------------------------------- # # CLI # --------------------------------------------------------------------------- # def _parse_cli(argv): """argv after command name -> (kwargs). Supports --key value and --json '{...}'.""" kwargs = {} i = 0 while i < len(argv): tok = argv[i] if tok.startswith("--"): key = tok[2:] if key == "json": kwargs.update(json.loads(argv[i + 1])) i += 2 continue val = argv[i + 1] if i + 1 < len(argv) else "true" # light coercion if val.lower() in ("true", "false"): val = val.lower() == "true" i += 2 kwargs[key] = val else: i += 1 return kwargs # command -> (client_factory, method_name) def _registry(): g, d, c, b, r = GammaClient(), DataClient(), ClobPublicClient(), BridgeClient(), RelayerClient() combo = ComboClient() reg = {} def add(prefix, obj): for name in dir(obj): if name.startswith("_"): continue attr = getattr(obj, name) if callable(attr): reg[f"{prefix}.{name}"] = attr add("gamma", g) add("data", d) add("clob", c) add("bridge", b) add("relayer", r) add("combo", combo) return reg def main(argv): if not argv or argv[0] in ("-h", "--help", "help"): print(__doc__) return 0 cmd = argv[0] reg = _registry() if cmd == "list-commands": for k in sorted(reg): print(k) print("\nTrading (needs py-clob-client + PK env): use TradingClient in Python, e.g.") print(" python -c 'from polymarket import TradingClient; " "print(TradingClient().create_and_post_order(TOKEN, 0.5, 10, \"BUY\"))'") return 0 if cmd.startswith("trade.") or cmd.startswith("trading."): sys.stderr.write( "Trading commands run in Python (they need your private key in env), " "not via this flat CLI. See references/trading.md.\n") return 2 if cmd not in reg: sys.stderr.write(f"Unknown command: {cmd}\nTry: python polymarket.py list-commands\n") return 2 kwargs = _parse_cli(argv[1:]) result = reg[cmd](**kwargs) print(json.dumps(result, indent=2, ensure_ascii=False) if not isinstance(result, str) else result) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))