Мультиплексирование WebSocket Binance Futures на Python

Суть: Мультиплексирование каналов (Multiplexing) в Binance Futures позволяет объединить подписки на десятки торговых пар и типов данных (например, стаканы и сделки) в рамках одного TCP-соединения. Это снижает накладные расходы на сеть, предотвращает блокировки по IP и упрощает архитектуру торгового бота за счет использования единого асинхронного цикла на базе библиотеки websockets.

Содержание

Исходный код

При создании высокочастотных торговых систем (HFT) или систем сбора рыночных данных (Market Data Feed) критически важно минимизировать задержки и эффективно утилизировать сетевые ресурсы. Binance Futures накладывает жесткие лимиты на количество одновременных WebSocket-соединений с одного IP-адреса. Решением этой проблемы является мультиплексирование — объединение нескольких потоков данных в один сокет.

В этой статье мы создадим отказоустойчивый асинхронный клиент на Python с использованием библиотеки websockets, который умеет динамически обрабатывать данные из нескольких каналов, автоматически восстанавливать соединение при обрывах и корректно обрабатывать системные пинги от биржи.

Для построения распределенных систем сбора данных вы можете ознакомиться с руководством Как связать Python и C++ через ZeroMQ для передачи сигналов. При проектировании торговых роботов критически важно защищать систему от рыночных аномалий, используя паттерны вроде Circuit Breaker на Python: Защита Торгового Бота от Flash Crash. Если вас интересует специализированный сбор данных о принудительных закрытиях позиций, изучите статью про Отслеживание ликвидаций на Binance Futures через WebSocket на Python.

import asyncio
import json
import logging
import sys
import websockets

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

class BinanceFuturesMultiplexer:
    def __init__(self, streams: list[str]):
        self.streams = streams
        # Базовый эндпоинт для мультиплексирования на Binance Futures
        self.base_url = "wss://fstream.binance.com/stream?streams="
        self.url = f"{self.base_url}{'/'.join(streams)}"
        self.is_running = False
        self._reconnect_delay = 5

    async def start(self):
        self.is_running = True
        while self.is_running:
            try:
                logging.info(f"Connecting to Binance Futures WS: {self.url}")
                async with websockets.connect(self.url, ping_interval=20, ping_timeout=10) as ws:
                    logging.info("Connection established successfully.")
                    self._reconnect_delay = 5  # Сброс задержки при успешном подключении
                    await self._receive_loop(ws)
            except websockets.exceptions.ConnectionClosed as e:
                logging.warning(f"Connection closed: {e}. Reconnecting...")
                await self._handle_reconnect()
            except Exception as e:
                logging.error(f"Unexpected error: {e}. Reconnecting...")
                await self._handle_reconnect()

    async def _receive_loop(self, ws):
        while self.is_running:
            try:
                message = await ws.recv()
                data = json.loads(message)
                await self._process_message(data)
            except websockets.exceptions.ConnectionClosed:
                break

    async def _process_message(self, data: dict):
        # Мультиплексированные сообщения содержат обертку с именем стрима
        stream_name = data.get("stream")
        payload = data.get("data", {})
        event_type = payload.get("e")
        symbol = payload.get("s")

        if event_type == "aggTrade":
            price = payload.get("p")
            quantity = payload.get("q")
            logging.info(f"[{stream_name}] Trade: {symbol} | Price: {price} | Qty: {quantity}")
        elif event_type == "depthUpdate":
            logging.info(f"[{stream_name}] Depth update received for {symbol}")
        else:
            logging.info(f"[{stream_name}] Event: {event_type} received.")

    async def _handle_reconnect(self):
        logging.info(f"Waiting {self._reconnect_delay} seconds before reconnecting...")
        await asyncio.sleep(self._reconnect_delay)
        self._reconnect_delay = min(self._reconnect_delay * 2, 60)  # Экспоненциальный бэк-офф

    def stop(self):
        logging.info("Stopping multiplexer...")
        self.is_running = False

async def main():
    # Пример подписки на агрегированные сделки и стаканы для BTCUSDT и ETHUSDT
    streams = [
        "btcusdt@aggTrade",
        "btcusdt@depth5@100ms",
        "ethusdt@aggTrade",
        "ethusdt@depth5@100ms"
    ]
    client = BinanceFuturesMultiplexer(streams)
    try:
        await client.start()
    except KeyboardInterrupt:
        client.stop()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Application stopped by user.")

Разбор параметров

streams: Список строковых идентификаторов каналов Binance (например, btcusdt@aggTrade), которые объединяются в один URL-запрос.
base_url: Базовый URL для мультиплексного подключения к Binance Futures (wss://fstream.binance.com/stream?streams=).
ping_interval: Интервал отправки пинг-фреймов библиотекой websockets для поддержания соединения активным и предотвращения сброса со стороны биржи.
_reconnect_delay: Экспоненциальная задержка перед повторным подключением для предотвращения спама запросами при падении сети.

Как запустить

Для запуска скрипта вам понадобится Python версии 3.8 или выше и установленная библиотека websockets. Установите зависимость с помощью pip:

pip install websockets

Сохраните код в файл multiplexer.py и запустите его из терминала:

python multiplexer.py

Клиент автоматически подключится к указанным стримам, начнет выводить лог поступающих событий в реальном времени и восстановит соединение в случае сетевого сбоя.