Запись событий и проигрыш, Python

Инструкция. Цель: записать поток событий в файл и проигрывать его из файла столько раз, сколько нужно для разбора, одним и тем же кодом разбора для живого потока и для файла. Формат и правила общие для Rust и Python: файл один и тот же, журнал, записанный в Rust, читается в Python и наоборот (Как собрать исследовательский движок). Код ниже - куски программы record.py, она запускается тестом против имитации БКС.

Модуль называется brokapi.record.

Файл: строка-заголовок {"brokapi_log":1,"envelope_schema":1} и по строке JSON на событие (конверт). Цены, количества и деньги записаны строками, поэтому возвращаются точно (250.25 остается 250.25).

Один разбор на живой поток и на файл

replay отдает Envelope так же, как client.events, поэтому функция разбора читает любой из них.

@dataclass
class Stats:
    kinds: Counter[str] = field(default_factory=Counter)
    traded: Decimal = Decimal(0)  # the sum of price * qty of your fills


async def analyze(events: AsyncIterator[brokapi.Envelope], stop_after: int | None = None) -> Stats:
    """Reads events and counts them. It does not know whether they are live or from a file."""
    stats = Stats()
    async for envelope in events:
        payload = envelope.payload
        stats.kinds[type(payload).__name__] += 1
        if isinstance(payload, brokapi.Execution):
            stats.traded += payload.price * payload.qty
        if stop_after is not None and sum(stats.kinds.values()) >= stop_after:
            break
    return stats

Записать

async def record_run(client: brokapi.Client, path: Path, count: int) -> Stats:
    """The live stream, written to `path` as it is read, and analysed."""
    async with contextlib.aclosing(record.record_events(client.events, path)) as events:
        return await analyze(events, stop_after=count)

record_events пропускает события дальше и перед этим пишет каждое в файл. Каждое событие сбрасывается на диск сразу, поэтому программа, которая остановилась посреди работы (сбой, break), оставляет журнал, полный до последнего прочитанного события. Он записывает то, что прочитал ваш код, то есть поток таким, каким его видел бот. Файл закрывается, когда поток кончился или генератор закрыт (contextlib.aclosing).

Новый файл создается при создании писателя. Если такой файл есть, это LogError (kind == "io"): запись одного запуска не должна молча затираться вторым. Нужно заменить файл - overwrite=True.

Писать можно только конверт, который пришел от коннектора или из журнала: он помнит строку, из которой сделан. Envelope, собранный руками, записать нельзя (TypeError).

Проиграть

async def analyze_file(path: Path) -> Stats:
    """The same analysis over the file. A replay waits for the reader, so nothing is lost."""
    return await analyze(record.replay(path))

Скорость: ReplaySpeed.instant() (насколько быстро читает потребитель, по умолчанию) или ReplaySpeed.paced(factor, max_gap): темп записи, деленный на factor. Темп - разница rx_ts соседних конвертов, пауза не длиннее max_gap, чтобы ночь между двумя сессиями не останавливала проигрыш.

Отличие от живого потока. Проигрыш ждет потребителя, поэтому медленный потребитель замедляет проигрыш и ничего не теряется. Живой коннектор ждать не может и применяет режим подписки. Значит, проигрыш никогда не даст Overflow: код, который разбирает Overflow, нужно проверять отдельно.

Прочитать без asyncio

def sources(path: Path) -> set[str]:
    """Reading a log without asyncio: an iterator of Envelope. A bad line raises brokapi.LogError and ends it."""
    return {envelope.source for envelope in record.EnvelopeReader(path)}

Что файл хранит и чего нет

• Монотонные часы получения (rx_mono в Rust) не хранятся, а в Python их нет совсем. Для порядка и замеров используйте Timestamps.rx_ns и broker_ns.
• Поврежденная строка останавливает чтение: brokapi.LogError с kind == "format" и номером строки (заголовок - строка 1). Пропустить строку молча нельзя. Проигрыш отдает конверты до нее и потом поднимает эту ошибку.
• Файл более новой версии формата или с более новыми конвертами читать нельзя: LogError с kind == "schema" (found, supported). Сбой диска или пути - kind == "io". Текст ошибки побуквенно равен тексту LogError в Rust.
• Это журнал потока, а не каталог: индекса по инструменту и времени нет, фильтруйте при чтении. Для лет данных нужен каталог (в Rust-части страницы есть сравнение с NautilusTrader).

Как проверить

Для этой команды нужен клон репозитория (git clone https://github.com/runinfra/brokapi, затем pip install brokapi pytest pytest-asyncio): тесты и примеры в пакет pip install brokapi не входят.

pytest crates/python/tests/test_record.py

Главная проверка: записанные события, прочитанные из файла и проигранные из файла, дают ту же последовательность, что видел бот, на имитациях всех трех брокеров. Еще проверено: журнал в формате Rust читается и записывается обратно байт в байт, поврежденная строка и новая версия отвергаются с номером строки, второй запуск не затирает первый журнал, медленный читатель проигрыша ничего не теряет.