Polymarket API: данные, заявки и торговые боты

Архитектура: что и откуда брать

Установка клиента и первый запрос

pip install py-clob-client requests

# --- 1. Справочные данные: список активных рынков (без авторизации) ---
import requests

GAMMA = "https://gamma-api.polymarket.com"

resp = requests.get(f"{GAMMA}/markets", params={
    "active": "true",
    "closed": "false",
    "limit": 50,
    "order": "volume24hr",
    "ascending": "false",
}, timeout=15)
resp.raise_for_status()

for m in resp.json():
    print(f'{m["question"][:60]:62} vol24h={m.get("volume24hr", 0):>12,.0f}')

Как получить идентификатор стороны рынка

Каждая сторона рынка — «Да» и «Нет» — это отдельный токен со своим числовым идентификатором. Он приходит в поле clobTokenIds справочного ответа, поэтому вписывать идентификаторы руками не нужно: их всегда можно получить запросом.

import json, requests

GAMMA = "https://gamma-api.polymarket.com"


def token_ids(market_slug):
    """Возвращает пару идентификаторов (yes, no) по слагу рынка."""
    r = requests.get(f"{GAMMA}/markets", params={"slug": market_slug}, timeout=15)
    r.raise_for_status()
    market = r.json()[0]
    yes_id, no_id = json.loads(market["clobTokenIds"])
    return yes_id, no_id


def top_market():
    """Слаг самого ликвидного открытого рынка за последние сутки."""
    r = requests.get(f"{GAMMA}/markets", params={
        "active": "true", "closed": "false", "limit": 1,
        "order": "volume24hr", "ascending": "false",
    }, timeout=15)
    r.raise_for_status()
    return r.json()[0]["slug"]


slug = top_market()
yes_id, no_id = token_ids(slug)
print(slug, yes_id, no_id, sep="\n")

Получение стакана и лучшей цены

from py_clob_client.client import ClobClient

HOST = "https://clob.polymarket.com"
client = ClobClient(HOST, chain_id=137)          # 137 = Polygon mainnet

yes_id, no_id = token_ids(top_market())

book = client.get_order_book(yes_id)
best_bid = max(float(o.price) for o in book.bids)
best_ask = min(float(o.price) for o in book.asks)

spread = best_ask - best_bid
print(f"bid={best_bid:.3f}  ask={best_ask:.3f}  spread={spread:.3f} "
      f"({spread / best_ask * 100:.2f}%)")

Аутентификация для торговых методов

Торговые операции подписываются приватным ключом кошелька. Ключ никогда не передаётся на сервер — подпись формируется локально, а на эндпоинт уходит уже подписанное сообщение.

import os
from py_clob_client.client import ClobClient
from py_clob_client.clob_types import OrderArgs, OrderType
from py_clob_client.order_builder.constants import BUY

client = ClobClient(
    "https://clob.polymarket.com",
    key=os.environ["PM_PRIVATE_KEY"],     # ключ только из переменных окружения
    chain_id=137,
)
client.set_api_creds(client.create_or_derive_api_creds())

yes_id, _ = token_ids("will-the-fed-cut-rates-in-september")

order = client.create_order(OrderArgs(
    token_id=yes_id,
    price=0.64,          # лимитная цена — вы мейкер, комиссия не взимается
    size=100,            # количество долей
    side=BUY,
))
result = client.post_order(order, OrderType.GTC)
print(result)

WebSocket: поток обновлений в реальном времени

Опрос REST-эндпоинта в цикле — плохая идея: вы упрётесь в лимиты и получите задержку. Для торговых сценариев используйте потоковую подписку.

import json, asyncio, websockets

WS = "wss://ws-subscriptions-clob.polymarket.com/ws/market"

async def stream(token_ids):
    async with websockets.connect(WS, ping_interval=20) as ws:
        await ws.send(json.dumps({"assets_ids": token_ids, "type": "market"}))
        async for raw in ws:
            for msg in json.loads(raw):
                if msg.get("event_type") == "book":
                    bid = max((float(b["price"]) for b in msg.get("bids", [])), default=0)
                    ask = min((float(a["price"]) for a in msg.get("asks", [])), default=1)
                    print(f'{msg["asset_id"][:10]}  {bid:.3f} / {ask:.3f}')

asyncio.run(stream(list(token_ids(top_market()))))

Лимиты и правила хорошего тона

• Всегда обрабатывайте код ответа 429 с экспоненциальной задержкой.
• Ставьте таймаут на каждый запрос — зависший вызов блокирует торговый цикл.
• Логируйте каждую заявку с идентификатором: без этого разбор инцидента невозможен.
• Держите независимый аварийный выключатель, снимающий все заявки одной командой.

Каркас торгового бота с контролем риска

Ниже минимальный, но правильно устроенный скелет: разделены получение данных, принятие решения и контроль риска. Именно этой структуры не хватает большинству самописных ботов.

import time, logging, statistics
from collections import deque, defaultdict
from dataclasses import dataclass, field

from py_clob_client.clob_types import OrderArgs, OrderType
from py_clob_client.order_builder.constants import BUY

log = logging.getLogger("bot")


@dataclass
class RiskLimits:
    max_position_usd: float = 250.0     # максимум на один рынок
    max_total_usd: float = 2000.0       # максимум суммарной экспозиции
    max_daily_loss: float = 150.0       # дневной стоп
    min_liquidity_usd: float = 100_000  # не торгуем тонкие рынки
    min_edge: float = 0.05              # порог преимущества — 5 п.п.
    order_usd: float = 50.0             # размер одной заявки


@dataclass
class Bot:
    client: object
    limits: RiskLimits
    dry_run: bool = True                # боевые заявки только при dry_run=False
    day_pnl: float = 0.0
    exposure: dict = field(default_factory=lambda: defaultdict(float))
    history: dict = field(default_factory=lambda: defaultdict(lambda: deque(maxlen=60)))

    def fair_value(self, token_id, mid):
        """Базовая модель: сглаженная середина стакана за последние 60 наблюдений.
        Замените её собственной оценкой вероятности — здесь и живёт ваше преимущество."""
        window = self.history[token_id]
        window.append(mid)
        return statistics.fmean(window)

    def can_trade(self, market, size_usd):
        if self.day_pnl <= -self.limits.max_daily_loss:
            log.warning("дневной стоп достигнут — торговля остановлена")
            return False
        if float(market["volume24hr"]) < self.limits.min_liquidity_usd:
            return False
        if self.exposure[market["id"]] + size_usd > self.limits.max_position_usd:
            return False
        return sum(self.exposure.values()) + size_usd <= self.limits.max_total_usd

    def step(self, market, token_id):
        book = self.client.get_order_book(token_id)
        bid = max(float(o.price) for o in book.bids)
        ask = min(float(o.price) for o in book.asks)
        edge = self.fair_value(token_id, (bid + ask) / 2) - ask
        if edge < self.limits.min_edge or not self.can_trade(market, self.limits.order_usd):
            return

        price = round(bid + 0.01, 2)        # лимит внутри спреда: мы мейкер, комиссии нет
        size = round(self.limits.order_usd / price, 2)
        log.info("вход: %s edge=%.3f price=%.2f size=%.2f",
                 market["question"][:40], edge, price, size)

        if not self.dry_run:
            order = self.client.create_order(
                OrderArgs(token_id=token_id, price=price, size=size, side=BUY))
            self.client.post_order(order, OrderType.GTC)
        self.exposure[market["id"]] += self.limits.order_usd

    def run(self, targets, pause=5):
        """targets — список пар (market, token_id)."""
        while True:
            for market, token_id in targets:
                try:
                    self.step(market, token_id)
                except Exception:
                    log.exception("ошибка в цикле по рынку %s", market["id"])
            time.sleep(pause)

Типовые сценарии автоматизации

Частые вопросы

📅 Опубликовано: 14 ноября 2024🔄 Последняя редактура: 6 августа 2026📚 Проверок фактов: 14

Читайте дальше