Интеграция бота-маркетмейкера на Python с Redis для обновления котировок

Суть: Использование Redis в качестве ультра-низкозадержечного (ultra-low latency) кэша в оперативной памяти позволяет разделить процесс получения рыночных данных (WebSocket-фид) и логику маркет-мейкинга. Это исключает блокировку основного потока вычислений и снижает задержку (latency) обработки котировок до субмиллисекундного уровня.

Содержание

Исходный код

Для реализации сверхбыстрого обновления котировок мы применим асинхронный стек на Python с использованием библиотеки redis-py (с поддержкой asyncio). Архитектура разделена на два независимых компонента: коннектор к бирже (имитирует получение тиков) и торговый модуль маркет-мейкера, который мгновенно забирает актуальный стакан из Redis.

Если вам также требуется надежное хранилище для исторического анализа этих тиков, рекомендуем изучить статью про интеграцию TimescaleDB для хранения тиков с Binance на Python.

import asyncio
import json
import random
import time
import redis.asyncio as aioredis

REDIS_URL = "redis://localhost:6379/0"
SYMBOL = "BTCUSDT"

async def market_data_producer():
    """
    Имитатор быстрого WebSocket-соединения с биржей.
    Записывает лучшие цены Bid/Ask в Redis Hash с минимальной задержкой.
    """
    r = aioredis.from_url(REDIS_URL)
    print("[Producer] Подключение к Redis установлено. Начинаем трансляцию котировок...")
    
    # Имитируем базовую цену
    mid_price = 65000.0
    
    try:
        while True:
            # Генерируем случайное изменение цены
            mid_price += random.uniform(-2.0, 2.0)
            spread = random.uniform(0.5, 1.5)
            bid = round(mid_price - (spread / 2), 2)
            ask = round(mid_price + (spread / 2), 2)
            
            # Формируем пакет данных
            quote_data = {
                "bid": str(bid),
                "ask": str(ask),
                "timestamp": str(time.time())
            }
            
            # Записываем в Redis Hash. HSET выполняется за O(1)
            await r.hset(f"ticker:{SYMBOL}", mapping=quote_data)
            
            # Имитируем высокую частоту обновлений (например, 100 тиков в секунду)
            await asyncio.sleep(0.01)
            
    except asyncio.CancelledError:
        print("[Producer] Остановка стриминга котировок.")
    finally:
        await r.close()

async def market_maker_bot():
    """
    Бот маркет-мейкер. Считывает котировки из Redis,
    рассчитывает уровни лимитных ордеров и имитирует их выставление.
    """
    r = aioredis.from_url(REDIS_URL)
    print("[MM Bot] Запуск торгового модуля...")
    
    last_update_time = 0.0
    
    try:
        while True:
            # Читаем данные из Redis за O(1)
            quote = await r.hgetall(f"ticker:{SYMBOL}")
            
            if quote:
                # Декодируем байты из Redis
                bid = float(quote[b'bid'])
                ask = float(quote[b'ask'])
                ts = float(quote[b'timestamp'])
                
                # Проверяем, обновились ли котировки с момента последнего цикла
                if ts > last_update_time:
                    last_update_time = ts
                    latency = (time.time() - ts) * 1000 # Задержка в миллисекундах
                    
                    # Расчет собственных котировок маркет-мейкера (спред 0.1%)
                    my_bid = round(bid * 0.9995, 2)
                    my_ask = round(ask * 1.0005, 2)
                    
                    print(f"[MM Bot] Считан тик: Bid={bid} | Ask={ask} | Latency={latency:.3f}ms")
                    print(f"[MM Bot] Выставляем лимиты: Buy={my_bid} | Sell={my_ask}")
                    
                    # В реальном боте здесь будет отправка ордеров на биржу.
                    # Для авторизации на Bybit используйте:
                    # https://finfluct.com/python-api/generatsiya-podpisi-hmac-sha256-dlya-bybit-api-v5-na-python/
            
            # Высокая частота опроса локального кэша Redis
            await asyncio.sleep(0.005)
            
    except asyncio.CancelledError:
        print("[MM Bot] Остановка торгового модуля.")
    finally:
        await r.close()

async def main():
    # Запуск параллельных задач
    producer_task = asyncio.create_task(market_data_producer())
    bot_task = asyncio.create_task(market_maker_bot())
    
    # Даем поработать 10 секунд для демонстрации
    await asyncio.sleep(10)
    
    producer_task.cancel()
    bot_task.cancel()
    await asyncio.gather(producer_task, bot_task, return_exceptions=True)

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

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

REDIS_URL: Строка подключения к локальному или удаленному серверу Redis. Использование базы данных 0 по умолчанию.
hset / hgetall: Команды для работы с хэш-таблицами Redis. Позволяют атомарно обновлять и считывать несколько полей (цена, объем, таймстемп) за константное время O(1).
latency: Метрика задержки доставки котировки от продюсера до торгового бота через Redis. Обычно составляет менее 0.5 миллисекунды.
asyncio.sleep(0.005): Интервал опроса Redis торговым ботом (200 раз в секунду), что гарантирует моментальную реакцию на изменение спреда.

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

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

pip install redis

Убедитесь, что ваш Redis-сервер запущен локально на порту 6379. После этого запустите скрипт:

python mm_redis_integration.py

Вы увидите логи высокой частоты обновлений с расчетом задержки (latency), которая в локальной сети стремится к нулю. Для масштабирования этой архитектуры на несколько торговых пар вы можете использовать скрипт автоматического ребалансирования криптовалютного портфеля на Python через CCXT, интегрировав его с единой шиной данных Redis.