Парсинг данных order book с бирж в реальном времени
Order book данные нужны в трёх сценариях: построение торгового бота, создание агрегатора ликвидности, или мониторинг рынка. В каждом сценарии есть своя специфика, но общая проблема одна — стабильное получение high-frequency данных без потерь и с минимальной латентностью.
WebSocket vs REST polling
REST polling (GET /api/v3/depth?symbol=BTCUSDT) — неправильный выбор для real-time order book. На активных рынках стакан обновляется 10–100 раз в секунду. Поллинг раз в секунду даст устаревшие данные и нагрузит API лимиты. Правильный подход — WebSocket потоки с инкрементальными обновлениями.
Большинство крупных CEX (Binance, Bybit, OKX) следуют одной схеме:
- Получить snapshot через REST (полный стакан на текущий момент)
- Подписаться на WebSocket поток обновлений
- Применять обновления к snapshot, поддерживая локальную копию стакана
import asyncio, json, aiohttp
from sortedcontainers import SortedDict
class OrderBook:
def __init__(self):
self.bids = SortedDict(lambda x: -x) # убывающий порядок
self.asks = SortedDict()
self.last_update_id = 0
def apply_update(self, bids: list, asks: list, update_id: int):
if update_id <= self.last_update_id:
return # устаревшее обновление, игнорируем
for price, qty in bids:
price, qty = float(price), float(qty)
if qty == 0:
self.bids.pop(price, None) # удалить уровень
else:
self.bids[price] = qty
for price, qty in asks:
price, qty = float(price), float(qty)
if qty == 0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
self.last_update_id = update_id
@property
def best_bid(self) -> tuple[float, float] | None:
if self.bids:
price = self.bids.keys()[0]
return price, self.bids[price]
return None
@property
def best_ask(self) -> tuple[float, float] | None:
if self.asks:
price = self.asks.keys()[0]
return price, self.asks[price]
return None
Binance depth stream: детали протокола
Binance — самый частый запрос. У них два stream варианта:
-
btcusdt@depth— обновления каждые 100ms или 1000ms (параметр@depth@100ms) -
btcusdt@depth20— топ-20 уровней каждые 100ms (без инкрементальных обновлений, всегда полный)
Для полного стакана с применением патчей:
async def maintain_binance_orderbook(symbol: str):
ob = OrderBook()
buffer = [] # буфер обновлений до получения snapshot
async def handle_ws_message(msg):
data = json.loads(msg)
# Накапливаем обновления ПОКА не получим snapshot
if ob.last_update_id == 0:
buffer.append(data)
return
# Binance: обновление валидно если U <= lastUpdateId+1 <= u
if data['U'] <= ob.last_update_id + 1 <= data['u']:
ob.apply_update(data['b'], data['a'], data['u'])
# Запускаем WS
ws_task = asyncio.create_task(connect_ws(
f"wss://stream.binance.com:9443/ws/{symbol.lower()}@depth@100ms",
handle_ws_message
))
# Получаем snapshot (немного ждём чтобы буфер накопился)
await asyncio.sleep(0.5)
async with aiohttp.ClientSession() as session:
async with session.get(
f"https://api.binance.com/api/v3/depth",
params={"symbol": symbol.upper(), "limit": 1000}
) as resp:
snapshot = await resp.json()
# Инициализируем стакан из snapshot
for price, qty in snapshot['bids']:
ob.bids[float(price)] = float(qty)
for price, qty in snapshot['asks']:
ob.asks[float(price)] = float(qty)
ob.last_update_id = snapshot['lastUpdateId']
# Применяем буферизованные обновления
for update in buffer:
if update['u'] > ob.last_update_id:
ob.apply_update(update['b'], update['a'], update['u'])
await ws_task
Критический момент: если пропущен update (gap в U → u последовательности) — стакан рассинхронизирован. Нужна логика ресинхронизации: детектировать gap и переинициализировать с нового snapshot.
Агрегация нескольких бирж
Для агрегатора ликвидности или кросс-биржевого арбитража — поддерживать стаканы нескольких бирж параллельно:
EXCHANGES = {
"binance": BinanceOrderBook,
"bybit": BybitOrderBook,
"okx": OKXOrderBook,
}
async def run_aggregator(symbol: str):
books = {name: cls(symbol) for name, cls in EXCHANGES.items()}
tasks = [book.run() for book in books.values()]
await asyncio.gather(*tasks)
def get_best_price_across_exchanges(books: dict[str, OrderBook]) -> dict:
best_bids = [(name, *ob.best_bid) for name, ob in books.items() if ob.best_bid]
best_asks = [(name, *ob.best_ask) for name, ob in books.items() if ob.best_ask]
best_bids.sort(key=lambda x: x[1], reverse=True)
best_asks.sort(key=lambda x: x[1])
return {
"best_bid": {"exchange": best_bids[0][0], "price": best_bids[0][1], "qty": best_bids[0][2]},
"best_ask": {"exchange": best_asks[0][0], "price": best_asks[0][1], "qty": best_asks[0][2]},
"spread": best_asks[0][1] - best_bids[0][1]
}
Хранение и воспроизведение
Для backtesting или аудита — хранение потока обновлений, а не только снапшотов. L2 order book updates — это большой объём: для BTC/USDT на Binance ~100MB/час несжатых данных.
Эффективное хранение:
# Запись в бинарный формат через msgpack
import msgpack, lz4.frame
def serialize_update(update: dict) -> bytes:
packed = msgpack.packb(update, use_bin_type=True)
return lz4.frame.compress(packed)
# TimescaleDB для time-series хранения
# Гипертаблица автоматически партиционирует по времени
CREATE TABLE ob_updates (
time TIMESTAMPTZ NOT NULL,
exchange TEXT NOT NULL,
symbol TEXT NOT NULL,
side CHAR(1) NOT NULL, -- 'b' или 'a'
price NUMERIC NOT NULL,
quantity NUMERIC NOT NULL
);
SELECT create_hypertable('ob_updates', 'time');
Rate limits и управление соединениями
Binance: максимум 5 WebSocket соединений на stream при публичном доступе, 1024 при использовании API ключа. При отключении — exponential backoff:
async def connect_ws_with_retry(url: str, handler, max_retries=10):
for attempt in range(max_retries):
try:
async with websockets.connect(url, ping_interval=20) as ws:
async for message in ws:
await handler(message)
except (websockets.exceptions.ConnectionClosed, Exception) as e:
wait = min(2 ** attempt, 60) # max 60 секунд
logging.warning(f"WS disconnected: {e}, retry in {wait}s")
await asyncio.sleep(wait)
Полная система агрегации 3–5 бирж с хранением в TimescaleDB и REST API для запросов: 3–4 недели.