Как тестировать код против имитаций брокеров (Python)

Инструкция. Цель: проверить своего бота без брокера, сети и денег. В пакете есть модуль brokapi.testing с имитациями брокеров. Каждый пример этой книги идет именно на них.

Имитацию можно запустить и обычным скриптом через asyncio.run, не только тестом: так начинается Быстрый старт.

Что такое имитация

Это настоящий сервер на 127.0.0.1 со случайным портом, который говорит на протоколе брокера: HTTP и WebSocket у БКС, gRPC у Т-Банка и Финама. Клиент brokapi.Client подключается к нему так же, как к брокеру, поэтому тест идет по тому же коду, что и боевой запуск, а не по подмене. Сервер хранит состояние: поставленные заявки видны в active_orders, снимаются, исполняются.

Чего имитация не доказывает: что настоящий брокер отвечает так же. Имитации написаны по документации брокеров.

Минимальный тест

import brokapi
from brokapi.testing import mock_bcs


async def test_the_bot_reads_a_book():
    async with await mock_bcs() as mock:       # сервер стартовал
        client = await mock.connect()          # клиент к нему подключен
        sber = await client.instrument("SBER")
        book = await client.order_book(sber, depth=5)
        assert book.bids[0].price < book.asks[0].price
    # блок кончился: клиенты закрыты, сервер остановлен

Тесты асинхронные. Подойдет pytest-asyncio (asyncio_mode = "auto") или anyio, в самом пакете тесты на первом.

У всех есть synthetic_candles=True: candles тогда отвечает рядом под заданный период, по бару на шаг таймфрейма, и отвергает период длиннее предела брокера, как настоящий. Выключено, отвечает тремя фиксированными барами.

mock.connect(**kw) принимает те же аргументы, что brokapi.connect: name, limiter, account_id и другие. Так делают два клиента на один счет или второй счет, как в инфраструктуре.

Исполнить заявку

Статусы и исполнения приходят событиями. Чтобы проверить, что бот делает при исполнении, имитацию просят исполнить заявку:

mock.fill(cid, qty=params.min_qty, price=Decimal("100.00"))

У БКС, Финама, АЛОРа и Bybit нужны qty и price (на БКС qty меньше объема заявки дает частичное исполнение). Т-Банк исполняет заявку целиком по ее цене и эти аргументы не читает. Т-Банку и Финаму можно указать account.

Ждать события, а не время

В тестах нет sleep. Ждите событие или ответ, а время не отсчитывайте: пауза либо лишняя, либо коротка.

async def wait_event(client, kind, **match):
    async for envelope in client.events:
        if isinstance(envelope.payload, kind) and all(
            getattr(envelope.payload, k) == v for k, v in match.items()
        ):
            return envelope

Оберните чтение в asyncio.wait_for(..., 10): предел нужен, чтобы тест, у которого событие не пришло, упал, а не повис. Порядок событий у разных брокеров разный: в имитациях БКС и Финама сначала приходит событие заявки, а у Т-Банка исполнение пришло раньше статуса Filled. Не рассчитывайте на порядок между двумя видами событий, собирайте их, пока не набрались оба.

Гейтвей в тесте

Бот, который работает через гейтвей, тоже проверяется без брокера и сети. mock.start_gateway() поднимает в вашем процессе настоящий гейтвей на имитации, а gateway.connect(bot="my-bot") дает клиента, которого делает и brokapi.connect_gateway:

async def test_the_bot_behind_a_gateway():
    async with await mock_bcs() as mock:
        gateway = await mock.start_gateway("bcs-main")
        client = await gateway.connect(bot="my-bot")      # тот же Client, что у mock.connect()
        ...
        await gateway.restart()   # гейтвей умер, на его месте новый процесс: бот получит brokapi.Gap

restart() - тот сбой, который бот за гейтвеем обязан пережить. Клиенты остаются подключены, находят новый процесс, и в их потоке появляется Gap с причиной new_epoch. Заявки, стоящие у имитации, никуда не деваются: новый гейтвей сверяет с ними свой журнал до первой команды. Гейтвей останавливается вместе с имитацией. Пример с перезапуском: Подключить бота.

Как устроены тесты самого пакета

В crates/python/tests/: чтение и заявки на каждом брокере (test_brokers.py), общие правила (test_bcs.py), общие счетчики лимитов, слияние потоков и история (test_infrastructure.py), точные числа и ошибки (test_codec.py, test_errors.py, test_numbers.py), примеры книги (test_examples.py, test_rules.py, test_guides.py). Тесты лежат в клоне репозитория (git clone https://github.com/runinfra/brokapi), в пакет pip install brokapi они не входят. Запуск:

cd crates/python
maturin develop        # собрать расширение в текущее окружение (нужен Rust)
pytest tests

Rust не нужен, если поставить из PyPI пакет той же версии, что у клона: pip install brokapi pytest pytest-asyncio, и сразу pytest tests.