""" clob_helpers.py — shared CLOB client helpers. Drop this in your bot's project root or as a sibling import. It encodes the hard-won lessons from the source material: * the funder/signature_type mismatch trap * the EOA token-allowance requirement * the three-phase retry on FOK orders * the wei-conversion bug in get_balance_allowance * deterministic API-cred derivation and caching Tested against py-clob-client v0.34.6 on Polygon mainnet (chain_id=137). """ from __future__ import annotations import asyncio import os import time from dataclasses import dataclass from typing import Optional from py_clob_client.client import ClobClient from py_clob_client.clob_types import ( ApiCreds, AssetType, BalanceAllowanceParams, OrderArgs, MarketOrderArgs, OrderType, ) from py_clob_client.order_builder.constants import BUY, SELL # Polymarket constants HOST = "https://clob.polymarket.com" CHAIN_ID = 137 # Polygon mainnet USDC_DECIMALS = 6 # USDC.e on Polygon # --------------------------------------------------------------------------- # Client construction # --------------------------------------------------------------------------- def make_client( private_key: str, funder: str, signature_type: int = 2, creds: Optional[ApiCreds] = None, ) -> ClobClient: """ Construct a fully-authenticated ClobClient. signature_type: 0 = EOA (private key directly controls funds; needs allowances) 1 = Email/Magic wallet (proxy) 2 = Browser/Gnosis Safe proxy (the most common case for users who signed up via the Polymarket UI) funder is the address that holds the USDC. For signature_type 1 and 2, this is the proxy address visible in the Polymarket UI's deposit screen — NOT the EOA address derived from `private_key`. Mixing these up is the single most common bug. """ if signature_type not in (0, 1, 2): raise ValueError(f"signature_type must be 0/1/2, got {signature_type}") client = ClobClient( host=HOST, chain_id=CHAIN_ID, key=private_key, signature_type=signature_type, funder=funder, ) if creds is None: creds = client.create_or_derive_api_creds() client.set_api_creds(creds) return client def healthcheck(client: ClobClient) -> dict: """Run on bot startup. Returns a dict of (passing, value) tuples.""" out = {} try: ok = client.get_ok() out["clob_ok"] = (True, ok) except Exception as e: out["clob_ok"] = (False, str(e)) try: t = client.get_server_time() drift = abs(time.time() - t) out["time_drift_s"] = (drift < 60, drift) except Exception as e: out["time_drift_s"] = (False, str(e)) try: bal = client.get_balance_allowance( BalanceAllowanceParams(asset_type=AssetType.COLLATERAL) ) # Note: balance is returned in wei (USDC has 6 decimals) usdc_balance = int(bal["balance"]) / (10 ** USDC_DECIMALS) out["usdc_balance"] = (usdc_balance > 0, usdc_balance) except Exception as e: out["usdc_balance"] = (False, str(e)) return out # --------------------------------------------------------------------------- # Order placement with three-phase retry # --------------------------------------------------------------------------- @dataclass class OrderResult: success: bool order_id: Optional[str] filled_size: float avg_price: Optional[float] error: Optional[str] phase: int # 1 / 2 / 3 (which retry phase succeeded) async def place_fok_with_retry( client: ClobClient, token_id: str, side: str, # BUY or SELL target_price: float, size_shares: float, max_slippage: float = 0.03, ) -> OrderResult: """ Three-phase retry for FOK orders. Used in arbitrage and copy trading. Phase 1: try at target_price Phase 2: try at target_price ± 1.5% Phase 3: try at target_price ± max_slippage Each phase is one shot; FOK either fills entirely or cancels. Returns OrderResult; check .success and .phase. """ side_const = BUY if side.upper() == "BUY" else SELL direction = 1 if side_const == BUY else -1 phases = [ target_price, target_price + direction * 0.015, target_price + direction * max_slippage, ] last_err = None for phase_idx, attempt_price in enumerate(phases, start=1): # Clamp to [0.01, 0.99] — Polymarket prices are bounded attempt_price = max(0.01, min(0.99, round(attempt_price, 4))) try: args = OrderArgs( token_id=token_id, price=attempt_price, size=size_shares, side=side_const, ) signed = client.create_order(args) resp = client.post_order(signed, OrderType.FOK) if resp.get("success"): return OrderResult( success=True, order_id=resp.get("orderID"), filled_size=float(resp.get("makingAmount", 0)), avg_price=attempt_price, error=None, phase=phase_idx, ) last_err = resp.get("errorMsg") or str(resp) except Exception as e: last_err = str(e) await asyncio.sleep(0.1) # tiny gap before next phase return OrderResult( success=False, order_id=None, filled_size=0.0, avg_price=None, error=last_err, phase=3, ) def place_gtc( client: ClobClient, token_id: str, side: str, price: float, size_shares: float, ) -> OrderResult: """Resting limit order. Used in market making, end-game sweep, AI bot.""" side_const = BUY if side.upper() == "BUY" else SELL try: args = OrderArgs( token_id=token_id, price=round(max(0.01, min(0.99, price)), 4), size=size_shares, side=side_const, ) signed = client.create_order(args) resp = client.post_order(signed, OrderType.GTC) if resp.get("success"): return OrderResult( success=True, order_id=resp.get("orderID"), filled_size=0.0, # GTC = resting, not filled yet avg_price=price, error=None, phase=1, ) return OrderResult( success=False, order_id=None, filled_size=0.0, avg_price=None, error=resp.get("errorMsg"), phase=1, ) except Exception as e: return OrderResult( success=False, order_id=None, filled_size=0.0, avg_price=None, error=str(e), phase=1, ) def cancel_all(client: ClobClient) -> int: """Cancel everything and return the count cancelled. Use on shutdown.""" try: resp = client.cancel_all() return int(resp.get("count", 0)) except Exception: return 0 # --------------------------------------------------------------------------- # Balance / position helpers # --------------------------------------------------------------------------- def get_usdc_balance(client: ClobClient) -> float: """USDC.e balance in actual dollars (not wei).""" bal = client.get_balance_allowance( BalanceAllowanceParams(asset_type=AssetType.COLLATERAL) ) return int(bal["balance"]) / (10 ** USDC_DECIMALS) def get_token_position(client: ClobClient, token_id: str) -> float: """Position size in shares for a specific outcome token.""" bal = client.get_balance_allowance( BalanceAllowanceParams(asset_type=AssetType.CONDITIONAL, token_id=token_id) ) # Conditional token balances are also in wei (decimal=6) return int(bal["balance"]) / (10 ** USDC_DECIMALS) # --------------------------------------------------------------------------- # EOA allowance setup (signature_type=0 only) # --------------------------------------------------------------------------- def set_allowances_eoa(private_key: str, rpc_url: str): """ Approve the CLOB Exchange and NegRisk Adapter to move USDC.e and CTF tokens. Run ONCE per EOA wallet before trading. Costs ~$0.05 in POL gas. For signature_type=1 and 2 wallets, allowances are handled automatically; skip this function entirely. Required: web3==7.12.1 (pip install web3==7.12.1) See nautilus_trader/adapters/polymarket/scripts/set_allowances.py for the most up-to-date version; this is a minimal implementation. """ from web3 import Web3 from web3.middleware import geth_poa_middleware w3 = Web3(Web3.HTTPProvider(rpc_url)) if not w3.is_connected(): raise RuntimeError(f"Cannot connect to RPC {rpc_url}") acct = w3.eth.account.from_key(private_key) print(f"Setting allowances for EOA {acct.address}") USDC = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" CTF = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045" CLOB_EXCHANGE = "0x4bFb41d5B3570DeFd03C39a9A4D8dE6Bd8B8982E" NEGRISK_ADAPTER = "0xC5d563A36AE78145C45a50134d48A1215220f80a" MAX_UINT = 2 ** 256 - 1 erc20_abi = [ {"name": "approve", "type": "function", "inputs": [{"name": "spender", "type": "address"}, {"name": "amount", "type": "uint256"}], "outputs": [{"name": "", "type": "bool"}], "stateMutability": "nonpayable"} ] ctf_abi = [ {"name": "setApprovalForAll", "type": "function", "inputs": [{"name": "operator", "type": "address"}, {"name": "approved", "type": "bool"}], "outputs": [], "stateMutability": "nonpayable"} ] txs = [] nonce = w3.eth.get_transaction_count(acct.address) usdc = w3.eth.contract(address=Web3.to_checksum_address(USDC), abi=erc20_abi) ctf = w3.eth.contract(address=Web3.to_checksum_address(CTF), abi=ctf_abi) pairs = [ ("USDC -> CLOB", usdc.functions.approve( Web3.to_checksum_address(CLOB_EXCHANGE), MAX_UINT)), ("USDC -> NegRisk", usdc.functions.approve( Web3.to_checksum_address(NEGRISK_ADAPTER), MAX_UINT)), ("CTF -> CLOB", ctf.functions.setApprovalForAll( Web3.to_checksum_address(CLOB_EXCHANGE), True)), ("CTF -> NegRisk", ctf.functions.setApprovalForAll( Web3.to_checksum_address(NEGRISK_ADAPTER), True)), ] for label, fn in pairs: tx = fn.build_transaction({ "from": acct.address, "nonce": nonce, "maxPriorityFeePerGas": w3.to_wei(30, "gwei"), "maxFeePerGas": w3.to_wei(60, "gwei"), "chainId": CHAIN_ID, }) signed = acct.sign_transaction(tx) h = w3.eth.send_raw_transaction(signed.rawTransaction) receipt = w3.eth.wait_for_transaction_receipt(h, timeout=120) print(f" {label}: {receipt.status} (tx {h.hex()})") nonce += 1 txs.append(h.hex()) return txs # --------------------------------------------------------------------------- # Market metadata helpers # --------------------------------------------------------------------------- def get_market_book(client: ClobClient, token_id: str): """Wrapper that returns a clean dict instead of the raw OrderBookSummary.""" book = client.get_order_book(token_id) return { "market": book.market, "asset_id": book.asset_id, "best_bid": float(book.bids[0].price) if book.bids else None, "best_ask": float(book.asks[0].price) if book.asks else None, "bids": [(float(o.price), float(o.size)) for o in book.bids], "asks": [(float(o.price), float(o.size)) for o in book.asks], } def book_depth_at_price(book_side: list[tuple[float, float]], target_price: float, side: str) -> float: """ Total size available at or better than target_price. For BUY (asks): sum sizes where price <= target_price. For SELL (bids): sum sizes where price >= target_price. """ if side.upper() == "BUY": return sum(s for p, s in book_side if p <= target_price) return sum(s for p, s in book_side if p >= target_price)