quantlab.runs.live_predictions.LivePredictionStore

class quantlab.runs.live_predictions.LivePredictionStore(path)

Bases: object

An append-only store of one run’s live predictions, one bar per append.

See the module docstring for the layout. The store is opened on every call and nothing is held, so one object serves a writer and readers.

Parameters:

pathstr or os.PathLike

Directory of the Zarr store; it need not exist yet.

Attributes:

pathPath

The store.

records_pathPath

<store>.rows.json, the per-row records.

Examples

>>> store = LivePredictionStore("live/live_predictions.zarr")
>>> store.exists, store.records_path.name
(False, 'live_predictions.zarr.rows.json')

FORMAT_VERSION = 1

The version of the live attributes and the records file.

RECORDS_SUFFIX = '.rows.json'

Appended to the store’s path to name its records file.

append(row, header, record)

Append one bar’s predictions, then its record.

Parameters:

rowxr.Dataset

The predictions of one bar on (timestamp, symbol), one timestamp, the variables exactly the label names of header.

headermapping

The store’s attributes (header), checked against a store already written (check_header).

recordmapping

JSON-ready; written under the bar’s label with timestamp added.

Raises:

ValueError

If the row is not one bar, its variables are not the labels, the bar is not after the store’s last, or check_header refuses.

Examples

>>> store.append(row, header, {"written_at": "2026-10-08T10:02:11+00:00"})
>>> store.bars()[-1] == row["timestamp"].values[0]
True

attrs()

The store’s attributes (see the module docstring); empty before the first append.

Examples

>>> sorted(LivePredictionStore("live/live_predictions.zarr").attrs())
['checkpoint', 'format_version', 'labels', 'live_format_version', 'run_dir', 'trained_run']

bars()

The bars predicted so far, in order; empty before the first append.

Examples

>>> LivePredictionStore("live/live_predictions.zarr").bars()
DatetimeIndex(['2026-10-06', '2026-10-07'], dtype='datetime64[ns]', name='timestamp', freq=None)

check_header(header)

Refuse a writer whose run, checkpoint or labels are not the store’s.

A store holds the predictions of one run and one checkpoint; a retrained model starts a new store. Nothing is checked before the first append.

Raises:

ValueError

Naming each attribute that differs.

Examples

>>> store.check_header(LivePredictionStore.header(specs, run_dir=other_run, checkpoint=cp))
Traceback (most recent call last):
ValueError: live/live_predictions.zarr holds the predictions of another run: run_dir ...

property exists

Whether the store has been written.

Examples

>>> LivePredictionStore("nowhere.zarr").exists
False

has(timestamp)

Whether the bar timestamp has been predicted.

Examples

>>> LivePredictionStore("live/live_predictions.zarr").has("2026-10-07")
True

static header(labels, *, run_dir, checkpoint, trained_run=None)

The attributes a store of these labels, run and checkpoint carries.

Parameters:

labelssequence of LabelSpec

The run’s label specs, in the order of its prediction variables.

run_dir, checkpoint, trained_runstr or os.PathLike

The backtest run directory, the checkpoint file and its trained unit’s directory; written as absolute paths.

Returns:

dict

JSON-ready attributes.

Examples

>>> LivePredictionStore.header(
...     [LabelSpec("ret_5", "raw", 1, 5)], run_dir="/runs/a", checkpoint="/m/model.joblib"
... )["labels"]
'[{"name": "ret_5", "scale": "raw", "delay": 1, "span": 5}]'

labels()

The specs of the labels the store’s variables predict.

Examples

>>> LivePredictionStore("live/live_predictions.zarr").labels()
(LabelSpec(name='ret_5', scale='raw', delay=1, span=5),)

record(timestamp)

The record of one bar’s row, or None when it has none.

Examples

>>> sorted(LivePredictionStore("live/live_predictions.zarr").record("2026-10-07"))[:3]
['checkpoint', 'data_fingerprint', 'lagging']

records()

Every row’s record, keyed by bar label (see the module docstring).

Examples

>>> list(LivePredictionStore("live/live_predictions.zarr").records())
['2026-10-06', '2026-10-07']

row(timestamp)

The predictions of one bar, loaded: one variable per label on symbol.

Parameters:

timestampstr or datetime-like

A bar of the store.

Returns:

xr.Dataset

The row, with timestamp kept as a scalar coordinate.

Raises:

KeyError

If the bar has not been predicted.

Examples

>>> LivePredictionStore("live/live_predictions.zarr").row("2026-10-07")["ret_5"].dims
('symbol',)