brokapi.record

The event log: write the events of a run to a file, read them back, play them again.

This is brokapi_runtime::record of the Rust library, and the file is the same: a header line ({"brokapi_log":1,"envelope_schema":1}) and one JSON line per envelope. A log recorded here reads in Rust and the other way round. Prices, quantities and money are written as decimal strings, so they come back exactly.

Need	Item
Append envelopes to a file	`EnvelopeWriter`, or `record_events()` around a stream
Read them back in order	`EnvelopeReader`
Feed them to the code that reads `client.events`, fast or at the recorded pace	`replay()`, `ReplaySpeed`

What the file keeps and what it does not:

The monotonic clock of the receive time (rx_mono in Rust) is not stored, and Python does not show it at all. Use Timestamps.rx_ns and broker_ns to order and measure replayed data.
A line that cannot be read stops the read with a ~brokapi.LogError that names the line. A reader never skips a bad line silently.
An envelope with a schema newer than this build understands is refused (LogError of kind schema).
It is a log of a stream, not a catalog: there is no index by instrument or time, so filter while you read.
Only an envelope that came from a connector or from a log can be written: it carries the line it was made from. An Envelope that you build by hand cannot.

>>> import asyncio, os, tempfile
>>> from brokapi import record
>>> from brokapi.testing import mock_bcs
>>> async def demo():
...     path = os.path.join(tempfile.mkdtemp(), "run.jsonl")
...     async with await mock_bcs() as mock:
...         client = await mock.connect()
...         with record.EnvelopeWriter(path) as log:
...             log.write(await client.events.get(timeout=10))
...         await client.close()
...     return len([envelope async for envelope in record.replay(path)])
>>> asyncio.run(demo())
1

class EnvelopeReader:

Reads the envelopes of a log back, in the order they were written: for envelope in EnvelopeReader(path).

Opening the file reads and checks the header: a file that is not a log, an empty one, or one written by a newer version raises ~brokapi.LogError at once. After the first bad line the iteration raises it and ends: what follows a bad line cannot be trusted to be in order.

EnvelopeReader(path: str | os.PathLike[str])

class EnvelopeWriter:

Appends envelopes to a new log file, one line each. Use it as with: the file is flushed and closed at the end.

The file is created and the header written when the writer is made. An existing file is an error (LogError of kind io) unless overwrite=True: a log is a record of one run, and a second run must not silently replace the first. Writes are buffered: flush() or close() after the last one, so the tail is not left in a buffer when the program ends. After a failed write do not append to the same file, start a new one.

EnvelopeWriter(path: str | os.PathLike[str], *, overwrite: bool = False)

def write(self, envelope: brokapi.Envelope) -> None:

Appends one envelope as one line. Raises ~brokapi.LogError when the write fails.

written: int

How many envelopes were written (the header is not counted).

def flush(self) -> None:

Flushes the buffer to the file.

def close(self) -> None:

Flushes and closes the file. Safe to call twice.

LOG_FORMAT = 1

The version of the file format. It is in the header line of every file.

@dataclass(frozen=True, slots=True)

class ReplaySpeed:

How fast replay() plays. Build it with instant() or paced().

ReplaySpeed( factor: float | None = None, max_gap: datetime.timedelta = datetime.timedelta(seconds=1))

factor: float | None

None is as fast as the consumer reads. Else the speed-up: 1.0 is real time, 10.0 ten times faster.

max_gap: datetime.timedelta

The longest single pause, so that a night between two sessions does not stop the replay.

@classmethod

def instant(cls) -> ReplaySpeed:

As fast as the consumer reads. For backtests and tests. The default.

@classmethod

def paced( cls, factor: float = 1.0, max_gap: datetime.timedelta = datetime.timedelta(seconds=1)) -> ReplaySpeed:

At the recorded pace divided by factor. The pace is the difference of rx_ts of neighbouring envelopes, a pause is never longer than max_gap.

async def record_events( events: AsyncIterator[brokapi.Envelope], path: str | os.PathLike[str], *, overwrite: bool = False) -> AsyncIterator[brokapi.Envelope]:

Passes the events of a stream on and writes each one to a log first: async for envelope in record_events(client.events, "run.jsonl").

It records what your code reads, which is the stream as the bot saw it. Each envelope is flushed to the file as it is written, so a program that stops in the middle of a run (a crash, break) leaves a log that is complete up to the last envelope it read. The file is closed when the stream ends or the generator is closed (contextlib.aclosing), not at the break itself.

A ~brokapi.Gap (the stream of a client made by ~brokapi.connect_gateway()) is passed on and not written: it is not an event of a connector, and the file has the format of those.

def replay( path: str | os.PathLike[str], speed: ReplaySpeed | None = None, *, buffer: int = 256) -> AsyncIterator[brokapi.Envelope]:

Plays a log as a stream of envelopes, in order, like client.events: async for envelope in replay(path).

The envelopes keep their source and sub, so the code that reads them routes them as it did live. The difference from a live connector is the one that matters for research: a replay waits for the reader, so a slow reader slows it down and no event is lost. A live connector cannot wait and applies the loss policy of the subscription instead. Code that handles Overflow therefore needs its own test: a replay never produces one.

buffer is how many envelopes are read ahead. A bad line raises ~brokapi.LogError after the envelopes before it. The file is opened (and its header checked) at the call, not at the first await.