如何构建 Polymarket 交易机器人

pip install polymarket-client
# or: uv add polymarket-client / poetry add polymarket-client

import asyncio
from polymarket import AsyncPublicClient


async def main() -> None:
    async with AsyncPublicClient() as client:
        pages = client.list_markets(closed=False, page_size=50)

        # Iterate items directly when page boundaries do not matter.
        async for market in pages.iter_items():
            if "updown" in (market.slug or ""):
                print(market.slug, market.condition_id)


asyncio.run(main())

import os
from polymarket import AsyncSecureClient

client = await AsyncSecureClient.create(
    private_key=os.environ["POLYMARKET_PRIVATE_KEY"],
    wallet=os.environ["POLYMARKET_WALLET_ADDRESS"],
)

from polymarket.streams import MarketSpec

async with await client.subscribe([MarketSpec(token_ids=[token_id])]) as stream:
    async for event in stream:
        # MarketBookEvent | MarketPriceChangeEvent | MarketLastTradePriceEvent
        # | MarketTickSizeChangeEvent | MarketBestBidAskEvent
        # | NewMarketEvent | MarketResolvedEvent
        handle(event)

market = await client.get_market(slug="<market-slug>")
token_id = market.outcomes.yes.token_id
assert token_id is not None

response = await client.place_market_order(
    token_id=token_id,
    side="BUY",
    amount="10",          # spend up to 10 pUSD
)

if not response.ok:
    raise RuntimeError(response.message)

# Settlement is on-chain and asynchronous — this waits for it.
hashes = await client.wait_for_order_fill_settlement(response)

TOK=<token_id_of_a_resolved_market>
START=<epoch seconds the window opened>
END=$((START + 300))

curl -s "https://clob.polymarket.com/book?token_id=$TOK"
# {"error":"No orderbook exists for the requested token id"}   HTTP 404

curl -s "https://clob.polymarket.com/prices-history?market=$TOK&startTs=$START&endTs=$END&fidelity=1"
# about 6 points for a 5-minute market; interval=max returns [] for older ones

import os, requests

BASE = "https://api.polytest.io/api/v1"
H = {"X-API-Key": os.environ["POLYTEST_KEY"]}

markets = requests.get(f"{BASE}/markets",
                       params={"coin": "btc", "market_type": "5m", "limit": 50},
                       headers=H).json()["markets"]

snaps = requests.get(f"{BASE}/markets/{markets[0]['id']}/snapshots",
                     params={"include_orderbook": "true"},
                     headers=H).json()["snapshots"]


def simulate_fill(levels, qty):
    """Walk the ladder. Mid-price is not a price you can trade at."""
    filled = cost = 0.0
    for lvl in levels:
        take = min(lvl["size"], qty - filled)
        cost += take * lvl["price"]
        filled += take
        if filled >= qty:
            break
    return cost / filled if filled >= qty else None


print(simulate_fill(snaps[0]["orderbook"]["asks"], 500))