Граббинг данных о сделках (trades) с бирж

Парсинг данных о сделках (trades) с бирж

Задача звучит просто, пока не столкнёшься с ограничениями API: Binance банит IP после 1200 запросов в минуту, у Bybit WebSocket disconnects каждые 5 минут, у DEX данные децентрализованы по десяткам нод. Надёжный сбор данных о trades в production — это не "вызвать REST endpoint", это управление соединениями, rate limits, нормализация форматов и гарантированная доставка данных без потерь.

CEX: REST vs WebSocket

Для реального времени — только WebSocket. REST polling добавляет задержку в величину интервала опроса (500мс-2сек), пропускает сделки при высокой нагрузке, и быстро упирается в rate limits.

Binance WebSocket Streams:

import asyncio
import websockets
import json

async def subscribe_binance_trades(symbols: list[str]):
    streams = "/".join(f"{s.lower()}@trade" for s in symbols)
    url = f"wss://stream.binance.com:9443/stream?streams={streams}"
    
    async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
        async for message in ws:
            data = json.loads(message)
            trade = data['data']
            yield {
                'exchange': 'binance',
                'symbol': trade['s'],
                'trade_id': trade['t'],
                'price': float(trade['p']),
                'qty': float(trade['q']),
                'timestamp_ms': trade['T'],
                'side': 'sell' if trade['m'] else 'buy',  # m=True если maker=buyer (значит taker=seller)
            }

Поле m (is_buyer_maker) в Binance — это инверсия интуиции. m=true означает buyer был maker, значит агрессор — seller. Нормализуйте правильно.

Управление множеством бирж через CCXT:

import ccxt.pro as ccxtpro
import asyncio

async def collect_trades(exchange_id: str, symbols: list[str], queue: asyncio.Queue):
    exchange = getattr(ccxtpro, exchange_id)({
        'enableRateLimit': True,
        'options': {'tradesLimit': 1000},
    })
    
    try:
        while True:
            try:
                trades = await exchange.watch_trades_for_symbols(symbols)
                for trade in trades:
                    await queue.put({
                        'exchange': exchange_id,
                        'symbol': trade['symbol'],
                        'id': trade['id'],
                        'price': trade['price'],
                        'amount': trade['amount'],
                        'side': trade['side'],
                        'timestamp': trade['timestamp'],
                    })
            except Exception as e:
                print(f"Error {exchange_id}: {e}, reconnecting...")
                await asyncio.sleep(1)
    finally:
        await exchange.close()

CCXT Pro поддерживает 30+ бирж с единым интерфейсом watch_trades. Внутри — WebSocket с автореконнектом.

DEX: on-chain события

На DEX сделки — это события смарт-контрактов. Три метода получения:

The Graph subgraph — готовые данные через GraphQL. Для Uniswap V3:

{
  swaps(
    first: 100
    orderBy: timestamp
    orderDirection: desc
    where: { pool: "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8" }
  ) {
    id
    timestamp
    amount0
    amount1
    sqrtPriceX96
    tick
    transaction { id }
  }
}

Задержка — 1-5 минут от on-chain события.

Прямой RPC мониторинг — eth_subscribe("logs", filter) на Uniswap V3 Swap event:

import { createPublicClient, webSocket, parseAbiItem } from 'viem';

const SWAP_EVENT = parseAbiItem(
  'event Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)'
);

client.watchContractEvent({
  address: UNISWAP_V3_POOL,
  event: SWAP_EVENT,
  onLogs: (logs) => {
    for (const log of logs) {
      const { amount0, amount1, sqrtPriceX96 } = log.args;
      const price = sqrtPriceX96ToPrice(sqrtPriceX96, token0Decimals, token1Decimals);
      processSwap({ price, amount0, amount1, txHash: log.transactionHash });
    }
  }
});

Цена в Uniswap V3 хранится как sqrtPriceX96 (Q64.96 fixed point). Декодирование:

function sqrtPriceX96ToPrice(sqrtPriceX96: bigint, d0: number, d1: number): number {
  const price = Number(sqrtPriceX96 ** 2n * BigInt(10 ** d0)) / 
                Number(BigInt(2 ** 192) * BigInt(10 ** d1));
  return price;
}

Rate limits и обход блокировок

Binance: 1200 requests/min на IP для REST, WebSocket лимит — 300 streams на соединение, до 5 соединений с одного IP. При работе с большим количеством пар — несколько соединений, каждое по 300 пар.

Bybit, OKX: аналогичные ограничения. Bybit отключает WebSocket каждые 5 минут при отсутствии активности — нужен ping каждые 20 сек.

Ротация IP через прокси работает для REST, но не для WebSocket (долгосрочные соединения). Для high-frequency сбора данных с нескольких бирж — несколько VPS в разных датацентрах.

Хранение и нормализация

Нормализованная схема для cross-exchange trades:

CREATE TABLE trades (
    id           BIGSERIAL PRIMARY KEY,
    exchange     VARCHAR(50) NOT NULL,
    symbol       VARCHAR(30) NOT NULL,    -- нормализованный: 'BTC/USDT'
    trade_id     VARCHAR(100),            -- original ID с биржи
    price        NUMERIC(30, 10) NOT NULL,
    quantity     NUMERIC(30, 10) NOT NULL,
    side         CHAR(4) NOT NULL,        -- 'buy' / 'sell'
    ts           TIMESTAMPTZ NOT NULL,
    received_at  TIMESTAMPTZ DEFAULT NOW()
) PARTITION BY RANGE (ts);

-- Партиция по дням для управляемого размера
CREATE TABLE trades_2024_11 PARTITION OF trades
    FOR VALUES FROM ('2024-11-01') TO ('2024-12-01');

CREATE INDEX ON trades (exchange, symbol, ts DESC);
CREATE INDEX ON trades (symbol, ts DESC);

TimescaleDB hypertable делает это автоматически и добавляет time_bucket агрегации для OHLCV.

Deduplification. При reconnect WebSocket часто переотправляет последние N трейдов. Unique constraint на (exchange, trade_id) предотвращает дубли.