Исходный код
Для создания торговых роботов и проведения качественного количественного анализа (HFT) критически важно иметь актуальный стакан ордеров с минимальной задержкой. В отличие от простого чтения веб-сокета, профессиональный подход требует синхронизации локального состояния с REST-снимком (snapshot) и последующего пошагового применения дифференциальных обновлений (diff depth stream).
Ниже представлен готовый к продакшену класс BinanceOrderBook, который реализует этот алгоритм. Он автоматически буферизует события, запрашивает снимок стакана, валидирует последовательность обновлений по U и u параметрам, а также выводит лучшие цены покупки (bids) и продажи (asks).
Этот подход к обработке потоковых данных во многом схож с тем, как мы анализируем транзакции в блокчейне. Например, вы можете изучить Мониторинг мемпула Ethereum на Python: Скрипт Whale Alerts через Web3.py или погрузиться в Декодирование транзакций Uniswap V3 Mint/Burn из мемпула на Python для понимания работы с асинхронными потоками данных в DeFi.
import asyncio
import aiohttp
import json
from typing import Dict, List, Optional
class BinanceOrderBook:
def __init__(self, symbol: str):
self.symbol = symbol.lower()
self.symbol_upper = symbol.upper()
self.bids: Dict[float, float] = {}
self.asks: Dict[float, float] = {}
self.last_update_id: int = 0
self.buffer: List[dict] = []
self.is_synced: bool = False
self.ws_url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth"
self.snapshot_url = f"https://api.binance.com/api/v3/depth?symbol={self.symbol_upper}&limit=1000"
async def fetch_snapshot(self, session: aiohttp.ClientSession) -> Optional[dict]:
try:
async with session.get(self.snapshot_url) as response:
if response.status == 200:
return await response.json()
else:
print(f"Ошибка получения снимка: {response.status}")
return None
except Exception as e:
print(f"Исключение при получении снимка: {e}")
return None
def apply_update(self, update: dict):
for bid in update.get("b", []):
price, qty = float(bid[0]), float(bid[1])
if qty == 0.0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
for ask in update.get("a", []):
price, qty = float(ask[0]), float(ask[1])
if qty == 0.0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
def process_buffer(self):
self.buffer.sort(key=lambda x: x["u"])
for event in self.buffer:
u = event["u"]
U = event["U"]
if u <= self.last_update_id:
continue
if U <= self.last_update_id + 1 <= u:
self.apply_update(event)
self.last_update_id = u
self.is_synced = True
elif self.is_synced and U == self.last_update_id + 1:
self.apply_update(event)
self.last_update_id = u
elif self.is_synced:
print("Обнаружен пропуск пакетов! Требуется ресинхронизация.")
self.is_synced = False
self.buffer.clear()
break
def print_best_levels(self):
if not self.is_synced:
return
sorted_bids = sorted(self.bids.items(), key=lambda x: x[0], reverse=True)
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])
best_bid = sorted_bids[0] if sorted_bids else (0.0, 0.0)
best_ask = sorted_asks[0] if sorted_asks else (0.0, 0.0)
print(f"[{self.symbol_upper}] Bid: {best_bid[0]:.2f} ({best_bid[1]:.4f}) | Ask: {best_ask[0]:.2f} ({best_ask[1]:.4f}) | Spread: {best_ask[0]-best_bid[0]:.2f}")
async def start(self):
async with aiohttp.ClientSession() as session:
async with session.ws_connect(self.ws_url) as ws:
print("WebSocket соединение установлено. Начинаем сбор буфера...")
# Запускаем задачу получения снимка стакана
snapshot_task = asyncio.create_task(self.fetch_snapshot(session))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
self.buffer.append(data)
if snapshot_task.done() and not self.is_synced:
snapshot = snapshot_task.result()
if snapshot:
self.last_update_id = snapshot["lastUpdateId"]
self.bids = {float(price): float(qty) for price, qty in snapshot["bids"]}
self.asks = {float(price): float(qty) for price, qty in snapshot["asks"]}
print(f"Снимок получен. LastUpdateId: {self.last_update_id}. Синхронизируем...")
self.process_buffer()
else:
# Если снимок не удался, пробуем еще раз
snapshot_task = asyncio.create_task(self.fetch_snapshot(session))
elif self.is_synced:
self.process_buffer()
self.print_best_levels()
# Очищаем обработанную часть буфера
self.buffer = [e for e in self.buffer if e["u"] > self.last_update_id]
if __name__ == "__main__":
ob = BinanceOrderBook("BTCUSDT")
try:
asyncio.run(ob.start())
except KeyboardInterrupt:
print("Работа завершена.")
Разбор параметров
Для корректной работы со стаканом ордеров Binance важно понимать структуру получаемых данных и логику их синхронизации:
symbol: Торговая пара (например,BTCUSDT). В WebSocket-урле используется нижний регистр, в REST API — верхний.lastUpdateId: Уникальный идентификатор последнего обновления в полученном снимке стакана. Служит отправной точкой для синхронизации.U(First update ID): Идентификатор первого изменения в пакете обновления от WebSocket.u(Final update ID): Идентификатор последнего изменения в пакете обновления. Новое событие должно иметьU == previous_u + 1.bidsиasks: Словари, хранящие цену в качестве ключа и объем в качестве значения. Если пришедший объем равен0.0, уровень удаляется из стакана.
Как запустить
Для запуска скрипта вам понадобится Python версии 3.8 или выше и библиотека aiohttp.
1. Установите необходимую зависимость с помощью pip:
pip install aiohttp2. Сохраните приведенный выше код в файл binance_order_book.py и запустите его:
python binance_order_book.pyПосле запуска скрипт подключится к WebSocket-серверу Binance, скачает актуальный снимок стакана через REST API, синхронизирует данные и начнет выводить в консоль лучшие цены покупки, продажи и текущий спред в реальном времени.
Полученные данные стакана можно использовать для построения маркет-мейкинг алгоритмов. О том, как отправлять лимитные ордера с минимальной комиссией, читайте в нашей статье Пинг-понг стратегия на Bybit V5: Post-Only ордера на Python.
