1d4d3a22f4
Public market data via direct HTTP; authenticated CLOB trading via the official py-clob-client-v2 SDK. Includes README, endpoint reference, and trading/auth guide. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
673 lines
26 KiB
Python
Executable file
673 lines
26 KiB
Python
Executable file
#!/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 <command> [--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:]))
|