GRPC на Python для передачи котировок между торговыми серверами

Суть: Использование gRPC вместо классических REST API или WebSockets позволяет снизить задержку (latency) при передаче рыночных данных между торговыми серверами до субмиллисекундного уровня. Это достигается за счет бинарной сериализации Protocol Buffers и мультиплексирования запросов в рамках одного постоянного HTTP/2 соединения.

Содержание

Почему gRPC незаменим в алготрейдинге

При построении распределенных торговых систем скорость обмена данными между модулями (например, коннектором к бирже и модулем принятия решений) является критическим фактором. Традиционный JSON-over-HTTP создает огромный оверхед на парсинг текста и установку TCP-соединений. Даже WebSockets не всегда обеспечивают нужную производительность из-за отсутствия строгой типизации и эффективного сжатия.

gRPC решает эти проблемы, используя бинарный протокол Protobuf и транспорт HTTP/2. Полученные по gRPC котировки можно использовать для сложных расчетов в реальном времени, например, для расчета Taker Buy/Sell Ratio с Binance AggTrades на Python.

Высокая скорость доставки рыночных данных критически важна для маркетмейкинга и систем риск-менеджмента, таких как автоматическое дельта-хеджирование на Deribit с Python, где задержка в несколько миллисекунд может привести к проскальзыванию ордеров.

Если вы агрегируете данные из внешних источников, помните о защите веб-ресурсов. Например, вам может потребоваться настроить обход Cloudflare на DexScreener перед тем, как транслировать очищенные котировки во внутреннюю gRPC-сеть вашей торговой инфраструктуры.

Исходный код

Ниже представлена реализация высокопроизводительного стриминга котировок. Сначала создается файл описания интерфейса quotes.proto, а затем серверный и клиентский скрипты на Python.

Шаг 1: Определение структуры данных в файле quotes.proto

syntax = "proto3";

package quotes;

service QuoteStreamer {
  rpc StreamQuotes (SubscriptionRequest) returns (stream QuoteResponse);
}

message SubscriptionRequest {
  string symbol = 1;
}

message QuoteResponse {
  string symbol = 1;
  double bid = 2;
  double ask = 3;
  double volume = 4;
  uint64 timestamp = 5;
}

Шаг 2: Реализация gRPC-сервера (server.py), генерирующего поток котировок

import time
import random
import grpc
from concurrent import futures
import quotes_pb2
import quotes_pb2_grpc

class QuoteStreamerServicer(quotes_pb2_grpc.QuoteStreamerServicer):
    def StreamQuotes(self, request, context):
        print(f"Получена подписка на инструмент: {request.symbol}")
        bid = 50000.0
        
        while context.is_active():
            bid += random.uniform(-2.5, 2.5)
            ask = bid + random.uniform(0.5, 2.0)
            
            quote = quotes_pb2.QuoteResponse(
                symbol=request.symbol,
                bid=round(bid, 2),
                ask=round(ask, 2),
                volume=round(random.uniform(0.1, 5.0), 4),
                timestamp=int(time.time() * 1000)
            )
            
            yield quote
            time.sleep(0.001)  # Имитация высокой частоты обновлений (1000 tick/sec)

def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=5),
        options=[
            ('grpc.keepalive_time_ms', 10000),
            ('grpc.keepalive_timeout_ms', 5000)
        ]
    )
    quotes_pb2_grpc.add_QuoteStreamerServicer_to_server(QuoteStreamerServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC сервер котировок запущен на порту 50051...")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

Шаг 3: Реализация gRPC-клиента (client.py) для приема потока данных

import grpc
import quotes_pb2
import quotes_pb2_grpc

def run():
    # Настройка канала с оптимизацией под низкую задержку
    options = [
        ('grpc.tcp_nodelay', 1),
        ('grpc.keepalive_time_ms', 10000)
    ]
    
    with grpc.insecure_channel('localhost:50051', options=options) as channel:
        stub = quotes_pb2_grpc.QuoteStreamerStub(channel)
        request = quotes_pb2.SubscriptionRequest(symbol="BTCUSDT")
        
        try:
            quote_stream = stub.StreamQuotes(request)
            for quote in quote_stream:
                print(f"[{quote.timestamp}] {quote.symbol} | BID: {quote.bid} | ASK: {quote.ask} | VOL: {quote.volume}")
        except grpc.RpcError as e:
            print(f"Соединение разорвано: {e.code()} - {e.details()}")

if __name__ == '__main__':
    run()

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

stream QuoteResponse: Ключевое слово stream указывает на то, что сервер будет отдавать непрерывный поток сообщений в рамках одного RPC-вызова, исключая накладные расходы на повторные запросы.
grpc.tcp_nodelay: Параметр сокета, отключающий алгоритм Нагла. Он заставляет TCP отправлять пакеты немедленно, без буферизации, что критически важно для снижения latency котировок.
grpc.keepalive_time_ms: Интервал отправки пинг-пакетов для поддержания соединения активным через промежуточные сетевые экраны и роутеры.
insecure_channel: Создает канал без шифрования TLS. Внутри локального контура торговых серверов это позволяет сэкономить процессорное время на шифровании данных.

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

Для запуска инфраструктуры выполните следующие шаги в терминале:

1. Установите необходимые библиотеки компиляции и рантайма gRPC:

pip install grpcio grpcio-tools

2. Скомпилируйте .proto файл для генерации Python-классов:

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. quotes.proto

3. Запустите сервер котировок в первом терминале:

python server.py

4. Запустите клиент во втором терминале для получения котировок в реальном времени:

python client.py

gRPC на Python для передачи котировок между торговыми серверами

Почему gRPC незаменим в алготрейдинге

Исходный код

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

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