Збір даних про угоди (trades) з криптобірж

Збір даних про угоди (trades) з криптобірж

Нещодавно клієнт втратив три дні, намагаючись зібрати trades з Binance через REST — пропустив 15% угод через ліміти. При пікових навантаженнях 1200 запитів на хвилину не вистачало для покриття 20 торгових пар, і дані надходили із затримкою понад секунду. Ми перевели його на WebSocket і отримали затримку 2 мс, повнота даних — 99.99%. Такі кейси — норма: обмеження API, розриви з'єднань, різнобій форматів. Збір trades у реальному часі — інженерне завдання, яке ми вирішуємо під ключ. Обробляємо до 100 000 угод/сек на одному VPS. Зв'яжіться з нами для оцінки вашого проекту.

Чому WebSocket — єдиний варіант для real-time?

REST polling додає 500–2000 мс затримки та пропускає угоди при піковому навантаженні. WebSocket дає потокову передачу із затримкою 1–50 мс. Binance WebSocket documentation рекомендує до 300 стрімів на одне з'єднання. Ми використовуємо кілька з'єднань для покриття всіх пар.

Управління множиною бірж

CCXT Pro надає єдиний інтерфейс watch_trades для 30+ бірж із автореконектом. Код нижче підключається до будь-якої CEX за кілька рядків:

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 в 10 разів швидший у розробці, ніж власні WebSocket-конектори під кожну біржу. Для масштабування ми використовуємо кластеризацію: кілька інстансів розподіляють навантаження по різних групах торгових пар. Це дозволяє обробляти тисячі пар без втрати продуктивності.

Як збирати угоди з DEX?

На 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 хвилин від появи в блоці. Прямий RPC моніторинг — підписка на Swap-події через eth_subscribe. Кроки:

1. Підключіться до RPC вузла через WebSocket.
2. Підпишіться на подію Swap для пулу.
3. Декодуйте sqrtPriceX96 в ціну.
4. Обробіть та збережіть дані.

Приклад на TypeScript:

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; } 

Порівняння методів збору DEX-даних

Порівняння CEX та DEX за збором trades

Як обходити rate limits та блокування?

Binance: 1200 запитів/хв на IP для REST, WebSocket — до 300 стрімів на з'єднання. Використовуємо кілька з'єднань по 300 пар.

Bybit та OKX мають аналогічні обмеження. Bybit відключає WebSocket при відсутності активності — ping кожні 20 с. Ротація IP працює для REST, але не для WebSocket. Для high-frequency використовуємо кілька VPS в різних датацентрах.

Також налаштовуємо стиснення та пули з'єднань для зниження навантаження. Досвід нашої команди (50+ інтеграцій) гарантує стабільність навіть при пікових обсягах.

# Налаштування CCXT Pro з контролем лімітів exchange = ccxtpro.binance({ 'enableRateLimit': True, 'rateLimit': 1000, 'options': { 'tradesLimit': 1000, 'watchTrades': {'limit': 100}, }, }) 

CREATE TABLE trades ( id BIGSERIAL PRIMARY KEY, exchange VARCHAR(50) NOT NULL, symbol VARCHAR(30) NOT NULL, trade_id VARCHAR(100), price NUMERIC(30, 10) NOT NULL, quantity NUMERIC(30, 10) NOT NULL, side CHAR(4) NOT NULL, ts TIMESTAMPTZ NOT NULL, received_at TIMESTAMPTZ DEFAULT NOW() ) PARTITION BY RANGE (ts); CREATE INDEX ON trades (exchange, symbol, ts DESC); CREATE INDEX ON trades (symbol, ts DESC);