Интеграция Redis для кэширования стакана ордеров L2 на Python

Суть: Организация сверхбыстрого межпроцессного взаимодействия (IPC) в Python-HFT системах через кэширование L2 стакана ордеров в Redis с использованием бинарного протокола Msgpack. Это позволяет снизить latency чтения/записи до субмиллисекундного уровня (< 1 мс).

Содержание

Исходный код

В высокочастотном трейдинге критически важно разделять поток получения рыночных данных (Market Data Feed) и поток исполнения ордеров (Execution Bot). Для этого используется Redis как быстрое in-memory хранилище. Использование стандартного JSON для сериализации создает бутылочное горлышко, поэтому мы применяем высокопроизводительный бинарный формат msgpack.

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

import asyncio
import time
import msgpack
import redis.asyncio as aioredis

class RedisOrderBookCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_url = redis_url
        self.redis = None

    async def connect(self):
        """Установка асинхронного соединения с Redis"""
        self.redis = await aioredis.from_url(
            self.redis_url, 
            encoding="utf-8", 
            decode_responses=False  # Важно для работы с бинарным msgpack
        )

    async def set_order_book(self, symbol: str, bids: list, asks: list):
        """Сериализация и запись стакана L2 в Redis"""
        payload = {
            "symbol": symbol,
            "bids": bids,  # Формат: [[price, qty], [price, qty], ...]
            "asks": asks,
            "timestamp_ns": time.time_ns()
        }
        serialized_data = msgpack.packb(payload)
        # Используем SET для перезаписи актуального состояния стакана
        await self.redis.set(f"orderbook:{symbol}", serialized_data)

    async def get_order_book(self, symbol: str) -> dict:
        """Чтение и десериализация стакана из Redis"""
        raw_data = await self.redis.get(f"orderbook:{symbol}")
        if not raw_data:
            return {}
        
        # Распаковываем бинарные данные обратно в Python dict
        data = msgpack.unpackb(raw_data, strict_map_key=False)
        return data

    async def close(self):
        """Закрытие пула соединений"""
        if self.redis:
            await self.redis.close()

async def main():
    # Инициализируем кэш
    cache = RedisOrderBookCache()
    await cache.connect()

    # Имитируем L2 стакан (10 уровней bids и asks)
    mock_bids = [[100.0 - i * 0.1, 1.5 + i] for i in range(10)]
    mock_asks = [[100.1 + i * 0.1, 2.0 + i] for i in range(10)]
    symbol = "BTCUSDT"

    # Запись стакана в Redis
    start_write = time.time_ns()
    await cache.set_order_book(symbol, mock_bids, mock_asks)
    write_latency = (time.time_ns() - start_write) / 1_000_000
    print(f"Время записи стакана в Redis: {write_latency:.3f} мс")

    # Чтение стакана из Redis
    start_read = time.time_ns()
    cached_book = await cache.get_order_book(symbol)
    read_latency = (time.time_ns() - start_read) / 1_000_000
    print(f"Время чтения стакана из Redis: {read_latency:.3f} мс")

    # Вычисляем сквозную задержку (End-to-End Latency)
    current_ns = time.time_ns()
    e2e_latency = (current_ns - cached_book[b'timestamp_ns']) / 1_000_000
    print(f"Сквозная задержка передачи данных: {e2e_latency:.3f} мс")
    print(f"Топ бид: {cached_book[b'bids'][0]}, Топ аск: {cached_book[b'asks'][0]}")

    await cache.close()

if __name__ == "__main__":
    asyncio.run(main())

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

decode_responses=False: Этот параметр отключает автоматическое декодирование строк в UTF-8 на уровне клиента Redis. Это критически важно, так как msgpack упаковывает данные в бинарный формат, который нельзя декодировать как обычный текст.
msgpack.packb(): Функция сериализации Python-объектов в компактный бинарный формат. Работает в 3-5 раз быстрее стандартного модуля json и создает значительно меньшую нагрузку на сеть и память.
time.time_ns(): Возвращает текущее время в наносекундах. Используется для точного замера задержки (latency) внутри распределенной торговой системы.
orderbook:{symbol}: Ключ в Redis, использующий неймспейс двоеточия. Позволяет структурировать данные и быстро находить нужные стаканы в консоли Redis.

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

Для запуска скрипта вам понадобится установленный сервер Redis и несколько Python библиотек. Выполните следующие шаги:

1. Установите зависимости через терминал:

pip install redis msgpack

2. Запустите локальный сервер Redis. Если у вас установлен Docker, это можно сделать одной командой:

docker run -d -p 6379:6379 redis:alpine

3. Сохраните приведенный выше Python-код в файл redis_ob_cache.py и запустите его:

python redis_ob_cache.py

После успешного запуска вы увидите метрики задержки чтения и записи. Обычно они составляют менее 0.5 миллисекунды на локальной машине.

Полученные из Redis данные стакана можно использовать для принятия решений в торговых стратегиях, например, таких как Пинг-понг стратегия на Bybit V5: Post-Only ордера на Python. В случае возникновения критических ошибок или резкого роста задержек, вы можете настроить алерты, используя руководство Интеграция Python скрипта с Telegram Bot API для отправки PnL отчета CCXT.