Граббінг даних про угоди (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) запобігає дублям.