Интеграция Kafka для обработки потоковых тиковых данных на Python

Суть: Построение отказоустойчивого конвейера (pipeline) доставки тиковых данных субмиллисекундной задержки с криптобирж в локальное хранилище или торговый движок с использованием Apache Kafka и библиотеки confluent-kafka на Python.

Содержание

Исходный код

Для достижения максимальной пропускной способности мы используем клиент confluent-kafka, построенный поверх производительной C-библиотеки librdkafka. В отличие от чистого Python-клиента, это решение минимизирует накладные расходы на сериализацию и сетевой ввод-вывод. Ниже представлен готовый к продакшену код асинхронного продюсера, который собирает сделки (trades) с Binance WebSocket и отправляет их в Kafka, а также отказоустойчивого консьюмера.

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

import asyncio
import json
import sys
from confluent_kafka import Producer, Consumer, KafkaError
import websockets

KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'
KAFKA_TOPIC = 'crypto-trades'

# Настройки Producer для High-Throughput / Low-Latency
producer_conf = {
    'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
    'client.id': 'binance-producer',
    'queue.buffering.max.messages': 1000000,
    'queue.buffering.max.ms': 10,
    'linger.ms': 10,
    'compression.type': 'lz4',
    'acks': 1
}

# Настройки Consumer
consumer_conf = {
    'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
    'group.id': 'quant-processors',
    'auto.offset.reset': 'latest',
    'enable.auto.commit': False
}

def delivery_report(err, msg):
    if err is not None:
        print(f'Ошибка доставки сообщения: {err}')

async def start_producer():
    producer = Producer(producer_conf)
    uri = 'wss://stream.binance.com:9443/ws/btcusdt@trade'

    print(f'Подключение к Binance WebSocket: {uri}')
    async with websockets.connect(uri) as websocket:
        while True:
            try:
                message = await websocket.recv()
                data = json.loads(message)

                payload = {
                    'timestamp': data['E'],
                    'symbol': data['s'],
                    'price': float(data['p']),
                    'quantity': float(data['q']),
                    'is_buyer_maker': data['m']
                }

                producer.produce(
                    topic=KAFKA_TOPIC,
                    key=payload['symbol'].encode('utf-8'),
                    value=json.dumps(payload).encode('utf-8'),
                    callback=delivery_report
                )
                producer.poll(0)

            except websockets.ConnectionClosed:
                print('Соединение с WebSocket закрыто. Попытка переподключения...')
                await asyncio.sleep(5)
                break
            except Exception as e:
                print(f'Ошибка в работе продюсера: {e}')
                await asyncio.sleep(1)

def start_consumer():
    consumer = Consumer(consumer_conf)
    consumer.subscribe([KAFKA_TOPIC])
    print(f'Консьюмер запущен. Ожидание сообщений из топика {KAFKA_TOPIC}...')

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    print(f'Ошибка консьюмера: {msg.error()}')
                    break

            data = json.loads(msg.value().decode('utf-8'))
            volume_usd = data['price'] * data['quantity']
            if volume_usd > 100000:
                print(f'[WHALE ALERT] {data["symbol"]} | {data["price"]} | Vol: {data["quantity"]} ({volume_usd:.2f} USD)')

            consumer.commit(asynchronous=True)

    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()

if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == 'consumer':
        start_consumer()
    else:
        asyncio.run(start_producer())

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

queue.buffering.max.ms: Время ожидания (в миллисекундах) для накопления батча сообщений перед отправкой в брокер. Значение 10 обеспечивает минимальную задержку (latency) при сохранении преимуществ пакетной отправки.
compression.type: Алгоритм сжатия данных. Мы используем lz4, так как он обеспечивает наилучший баланс между скоростью сжатия и степенью компрессии, что критично для высокочастотных тиковых данных.
acks: Уровень подтверждения записи брокером. Значение 1 означает, что продюсер ждет подтверждения только от лидера партиции. Это гарантирует высокую скорость работы при минимальном риске потери данных.
enable.auto.commit: Отключение автоматического коммита смещений (False). Позволяет реализовать семантику доставки At-Least-Once, фиксируя смещение только после того, как тиковые данные были успешно обработаны или сохранены в БД.

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

Шаг 1. Разверните локальный брокер Kafka. Проще всего сделать это через Docker Compose:

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on:
      - zookeeper
    ports:
      - '9092:9092'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

Шаг 2. Установите необходимые Python-зависимости в ваше виртуальное окружение: