Парсинг даних order book з бірж у реальному часі
Order book дані потрібні у трьох сценаріях: побудова торгового бота, створення агрегатора ліквідності, або мониторинг ринку. У кожному сценарії своя специфіка, але спільна проблема одна — стабільне отримання high-frequency даних без втрат та з мінімальною затримкою.
WebSocket vs REST polling
REST polling (GET /api/v3/depth?symbol=BTCUSDT) — неправильний вибір для real-time order book. На активних ринках стакан оновлюється 10–100 разів у секунду. Polling раз у секунду дає застарілі дані та перевантажує 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
Критичний момент: якщо пропущено оновлення (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 тижні.