polars-online
GitHub project: github.com/hgilde/polars-online
Version 1.0 changes the API relative to 0.x. The API is more standard and consistent now, and should remain stable from here on.
Online model fitting for Polars: linear models, streaming moments, clustering and regime detection, for data too large to hold in memory at once. When the rows have a time order, a fit can also be local, following the recent rows without refitting a window at every step. Features and targets can be windowed means over the stream, looking back or ahead. Rust core, Python API, and a standalone command line.
Introduction
Here is a first model fit and an introduction to the library.
A first fit
This first fit regresses each stock's return on two signals over a made-up
trading day with known true betas. To run it, install the library and numpy
with pip install polars-online numpy (Install has the
requirements).
A spec describes one model, and fit_predict fits a list of specs in one
pass over the rows, as a model bank.
from datetime import datetime
import numpy as np # only to make up the rows
import polars as pl
import polars_online as po
# 1. Make up the input: a trading day, 9:30 to 16:00, a row a second for stocks A, B and C in turn.
rng = np.random.default_rng(0)
n = 23_400 # six and a half hours of seconds
signal_a = rng.standard_normal(n) # new every second
signal_b = np.cumsum(rng.normal(0.0, 0.05, n)) # wanders slowly
prices = pl.DataFrame({
"ts": pl.datetime_range(datetime(2024, 1, 2, 9, 30), datetime(2024, 1, 2, 16), "1s",
closed="left", eager=True),
"stock_id": ["A", "B", "C"] * (n // 3),
"signal_a": signal_a,
"signal_b": signal_b,
# beta 0.2 rising to 0.8 on signal_a, and -0.2 on signal_b
"ret": np.linspace(0.2, 0.8, n) * signal_a - 0.2 * signal_b + rng.standard_normal(n),
})
prices.write_parquet("prices.parquet") # on disk: every fit below reads it back as a query
# 2. One fit over every row: fit the rows before 15:30, save the state, read it back, serve the rest.
whole = po.spec.ewridge(
"whole", # the spec's name: it names the output column
targets=["ret"], features=["signal_a", "signal_b"],
half_life=float("inf"), # no forgetting: every row counts the same
group="stock_id", # one regression per stock
)
flat = (
pl.scan_parquet("prices.parquet").head(21_600) # a query over the file's rows before 15:30
.online.fit_predict([whole], save_state="bank.state") # adds a column "whole", saves the state
.online.unnest([whole]) # as columns: pred_<target>, resid_<target>, ...
.collect() # runs the query
)
saved = (
po.ModelBank.load("bank.state").coef() # the fit in the file, a row per coefficient
.pivot("term", index="group", values="coef") # a row per stock, a column per term
)
served = pl.scan_parquet("prices.parquet").tail(1_800).online.predict("bank.state").collect() # scores the last half hour, learns nothing
# 3. A fit that follows the recent rows, and its betas on every row.
local = po.spec.ewridge(
"local", targets=["ret"], features=["signal_a", "signal_b"], group="stock_id",
clock="ts", half_life="10m", # a row's weight halves every ten minutes of ts
gap_cap="5m", # and a gap longer than five minutes decays as five
coef_every=0, # the coefficients on every row
)
betas = pl.scan_parquet("prices.parquet").online.fit_predict([local]).online.unnest([local]).collect() # a row of betas per input row
# 4. A target that looks ahead: each row's mean return over the next ten minutes.
fwd_ret = po.ewm_mean("ret", half_life=float("inf"), window_size="10m", reverse=True).alias("fwd_ret")
ahead = po.spec.ewridge(
"ahead", targets=[fwd_ret], features=["signal_a", "signal_b"], group="stock_id",
clock="ts", half_life="10m", gap_cap="5m",
embargo="10m", # learn each row once its ten minutes have passed
)
forecast = pl.scan_parquet("prices.parquet").online.fit_predict([ahead]).online.unnest([ahead]).collect() # each row's forecast of fwd_ret
The input, prices, begins with these rows:
| ts | stock_id | signal_a | signal_b | ret |
|---|---|---|---|---|
| 09:30:00 | A | 0.13 | 0.01 | -0.10 |
| 09:30:01 | B | -0.13 | 0.03 | 0.60 |
| 09:30:02 | C | 0.64 | 0.02 | 0.39 |
| 09:30:03 | A | 0.10 | -0.03 | 0.21 |
| 09:30:04 | B | -0.54 | -0.08 | -0.08 |
| 09:30:05 | C | 0.36 | -0.01 | 0.24 |
Step 2 weights every row equally, because half_life=float("inf") forgets
nothing. So whole, a ridge regression solved from running sums, converges
to the batch fit over every row it has seen. Its state is a complete summary
of those rows, worth saving and serving.
fit_predict adds one column per spec, holding a record of named fields
in each row (a Polars Struct column), and .online.unnest, chained after it, spreads them into columns. For
stock A, flat begins:
| ts | ret | pred_ret | resid_ret | weight_sum | withheld_reason |
|---|---|---|---|---|---|
| 09:30:00 | -0.10 | null | null | 0 | above_max_error_inflation |
| 09:30:03 | 0.21 | null | null | 1 | above_max_error_inflation |
| 09:30:06 | 0.52 | null | null | 2 | above_max_error_inflation |
| 09:30:09 | 1.22 | -0.47 | 1.69 | 3 | null |
| 09:30:12 | -1.18 | 1.64 | -2.81 | 4 | null |
pred_ret is each row's prediction from the stock's rows before it, and
resid_ret is ret less pred_ret. weight_sum is the weight behind that
prediction: here the count of the stock's rows learned so far. A stock's
first three rows have too little behind them, so pred_ret is null and
withheld_reason says why (Warm-up).
The state saved at 15:30 holds each stock's fit, near the average of the true betas until then:
| group | intercept | signal_a | signal_b |
|---|---|---|---|
| A | -0.01 | 0.48 | -0.19 |
| B | -0.01 | 0.45 | -0.19 |
| C | 0.01 | 0.48 | -0.20 |
served scores the rows from 15:30 with that state, and learns nothing from
them. To keep learning from new rows instead, call
.online.fit_predict(load_state="bank.state") on a query over them
(Saving, loading and serving).
Step 3 follows the recent rows, with a clock and the gap_cap a clock
requires. Give the same spec a clock (timestamp column ts) and
half_life="10m" and each row's weight halves every ten minutes of ts. The
fit is now local: it describes the recent past, and it moves from row to row,
weighted toward the last few tens of minutes. Because this is a local
regression, remember that saving the final state may be of limited value.
But coef_every=0 produces a time series of the betas at every row, as they
stood after learning that row. This can be used to track the betas over
time.
Each hour from 10:30, stock A's beta on signal_a climbs with the true one:
| ts | coef_ret_signal_a | coef_ret_signal_b |
|---|---|---|
| 10:30:00 | 0.30 | -0.16 |
| 11:30:00 | 0.35 | -0.16 |
| 12:30:00 | 0.47 | -0.21 |
| 13:30:00 | 0.57 | -0.18 |
| 14:30:00 | 0.66 | -0.22 |
| 15:30:00 | 0.72 | -0.18 |
Step 4 forecasts a target that looks ahead. fwd_ret, a Polars
expression written with po.ewm_mean(..., reverse=True), is the time-weighted mean of the
stock's returns over the next ten minutes.
fit_predict requires an embargo of at least the target's window_size;
otherwise it refuses the spec, so for a shorter embargo use ModelBank.fit,
which keeps no predictions. With embargo="10m", each row is scored where it
sits, before its target is known, and learned from ten minutes later. So
ahead learns nothing from 9:30 to 9:40, and resid_fwd_ret is null on
every row:
| ts | pred_fwd_ret | resid_fwd_ret | weight_sum |
|---|---|---|---|
| 09:30:00 | null | null | 0 |
| 09:35:00 | null | null | 0 |
| 09:40:00 | null | null | 0 |
| 09:45:00 | -0.17 | null | 85 |
| 09:50:00 | -0.26 | null | 145 |
The idea
Here are a few points to remember about fitting models with polars-online.
Every prediction and diagnostic is out-of-sample. The bank predicts each row and computes its diagnostics from what the models learned before it, and only then learns from the row (Checking that a model works).
The stream can be far larger than memory. A bank keeps only what its
models have learned, and the rows an embargo or a window still holds. To fit
parquet files larger than memory, start the query with
pl.scan_parquet("ticks/*.parquet") before .online.fit_predict, and end it
with .sink_parquet("fitted.parquet"), which writes the result to a file, in
place of .collect().
Row order matters when a model forgets. With a decay, older rows count
less, so sort the rows by time with Polars' sort, in the query before the
bank, unless they arrive sorted. Without a decay, the models that solve or
accumulate give the same answer in any order (Convergence without a
decay names them).
Features and targets can be windows over the stream, written as Polars
expressions. A feature can look back, such as the last minute's
time-weighted mean of the mid, added as a column in the query before the
bank. A target can look ahead, such as the next minute's VWAP less the mid,
named in the spec's targets and learned once its minute has passed.
Windows always need the rows in time order (Making features and targets
from windows).
A model bank runs three ways, with the same numbers from each. It runs
inside a Polars query with .online.fit_predict, in your own Python loop
calling ModelBank.fit_predict on each chunk, or from the standalone
polars-online command line, without Python (Running a bank).
Terminology
polars-online uses these words in a sense of its own:
| term | meaning |
|---|---|
| spec | the description of one model: which model, which columns it reads, and how it treats time. po.spec.ewridge(...) builds one |
| model bank, or the bank | a set of specs fitted together over the same rows, and the Python object that holds them, ModelBank. The bank always means a model bank |
| stream | the rows in the order the bank reads them |
| chunk | one of the pieces the bank takes the stream in, one at a time: chunk_size rows in a query, 100,000 by default, or each frame passed to ModelBank.fit_predict |
| state | everything a bank has learned |
| clock, decay | the column, named by clock=, that says how far apart two rows are, and the forgetting measured along it: each row's weight halves every half_life of the clock (Time and decay) |
weight_sum |
the weight behind the state that produced a row's prediction, which min_weight reads (Warm-up) |
What you can rely on
These hold for every spec, however a bank is run:
| honest predictions | every row is predicted before its own outcome is learned (Row order and the two guarantees), and a target that looks ahead is learned only once its window has closed (Windows as a model's inputs and target) |
| bounded memory | memory grows with the models' state and the rows a delay or a window holds, and never with the number of rows that have passed (Memory) |
| any chunking | one chunk or a thousand, with or without a save and resume in the middle, gives the same numbers, to the last bit. Only which rows carry the coefficients, coef and support_coef, can change (Row order and the two guarantees) |
| any thread count | the thread count, set with a bank's threads= (or POLARS_ONLINE_MAX_THREADS for the default pool), changes only the speed. Each spec and group is one task on the bank's thread pool: with 64 groups, 14 threads process 7.3× the rows per second of one (Parallelism) |
| unusable values | NaN, ±inf and any magnitude above 1e100 count as null, so a sentinel value never reaches a model. A null skips an update but never stops the clock (Nulls, and three ways to hold a row back) |
| zero weights | a row of weight 0 is scored and moves the clock, but teaches the model nothing (Weights) |
one weight_sum |
weight_sum means the same in every model, so one min_weight means the same for every spec in a bank (Warm-up) |
| a clock policy | a spec with a clock caps each step at gap_cap, and applies session_gap at a session boundary. A step back the model would learn from is refused, naming the row, unless restart_after_step_back makes it a restart (Sessions, gaps and steps back) |
| named mistakes | every keyword is checked against its type when the spec is built, and a missing column is reported, with the spec that wanted it and the role it had there, before the bank learns any row |
| tested | 2,207 Rust tests and 6,262 Python cases (counted on 2026-10-11), held to independent libraries such as scikit-learn, statsmodels and river and to adversarial streams, run on macOS, Windows and Linux at every push (Testing) |
Install
Install the wheel from PyPI with pip or uv:
pip install polars-online # or: uv add polars-online
| need | detail |
|---|---|
| Python | 3.12 or newer: one wheel per platform covers every CPython from 3.12 on (Testing lists the versions CI runs) |
| Polars | polars>=1.37.0,<3: the range the test suite has measured, which is not a guarantee. A weekly job and every release run the test suite on the newest Polars (Versioning and the Polars pin) |
| depends on | nothing beyond polars at run time, since the wheel carries its own copy of Polars' Rust code |
| numpy | needed only by ModelBank.gram() and the po.gram, po.corr and po.sim helpers: to use them, install it with the package, as pip install "polars-online[numpy]". Without it, those calls raise ModuleNotFoundError |
| wheels | macOS (arm64, x86_64), Windows x64, and Linux (x64 glibc and musl, aarch64 glibc), on PyPI and on each GitHub release beside the command-line binaries. On any other platform, the install builds the package from its source distribution, which needs a stable Rust toolchain (rustup) |
| size | 8 to 10 MB to download and 26 to 37 MB installed, by platform, for 0.13.0 |
To build from a checkout, install uv and a stable Rust toolchain (Development), then run from its root:
uv sync
uv run maturin develop --release -m crates/online-py/Cargo.toml
Example data
Most examples after the first fit read one of two made-up frames: df,
400 rows of numbers and labels a minute apart, and trades, a stream of
quotes with trades between them. The examples that serve or resume a bank
also read today, the rows that follow df's. Before running an example,
build all three with this code. It saves each to a parquet file, and an
example that runs a query reads the file back with pl.scan_parquet, as a
stream too large to hold would be read:
from datetime import datetime, timedelta
from pathlib import Path
import numpy as np
import polars as pl
import polars_online as po
# df: 400 rows of numbers and labels, a minute apart.
rng = np.random.default_rng(0)
x0, x1, x2, signal_a, signal_b, noise_y, noise_ret = rng.standard_normal((7, 400))
row = np.arange(400)
df = pl.DataFrame({
"t": np.arange(400.0), # a numeric clock: the row's number
"ts": pl.datetime_range(datetime(2024, 1, 2, 9, 30), datetime(2024, 1, 2, 16, 9), "1m", eager=True), # a minute apart
"x0": x0, "x1": x1, "x2": x2, "signal_a": signal_a, "signal_b": signal_b,
"y": 1.0 + 2.0 * x0 - 1.0 * x1 + 0.5 * noise_y, # the truth: intercept 1, x0 2, x1 -1; x2 plays no part
"ret": np.where(row < 200, 0.8, -0.8) * signal_a # a break: the beta on signal_a flips at row 200
+ np.where(row // 100 % 2 == 1, 2.0, 0.5) * noise_ret, # two regimes: calm (0.5), volatile (2.0) on rows 100-199 and 300-399
"stock_id": [f"b{i % 4}" for i in range(400)], # four stocks, taking turns row by row
"group": [f"g{i % 3}" for i in range(400)],
"session": ["m"] * 200 + ["a"] * 200, # a morning session, then an afternoon one
"venue": ["X", "Y"] * 200,
})
# Save df, and the rows that follow it, as the files the examples scan.
df.write_parquet("ticks.parquet") # df on disk, as one file
lf = pl.scan_parquet("ticks.parquet") # the same rows, read back from the file as a query
today = df.with_columns(pl.col("t") + 400.0) # the 400 rows that follow df's, on the clock t
today.write_parquet("today.parquet")
later = pl.scan_parquet("today.parquet") # today, read back as a query
Path("ticks").mkdir(exist_ok=True) # and df as two files, for the examples that scan ticks/*.parquet
df.head(200).write_parquet("ticks/part-0.parquet")
df.tail(200).write_parquet("ticks/part-1.parquet")
# trades: quotes, with trades between them, about two rows a second.
rng = np.random.default_rng(7)
is_trade = rng.random(3000) < 0.3 # three rows in ten are trades, the rest quotes
micros = np.cumsum(rng.exponential(5e5, 3000)).astype(int) # about two rows a second
mid = 100 + np.cumsum(rng.normal(0.0, 0.01, 3000)) # the quotes' mid, on every row
trades = pl.DataFrame({
"ts": [datetime(2024, 1, 2, 9, 30) + timedelta(microseconds=int(u)) for u in micros],
"symbol": rng.choice(["AAA", "BBB"], 3000),
"mid": mid,
"side": rng.choice(["buy", "sell"], 3000),
"quantity": rng.integers(1, 10, 3000).astype(float),
"price": mid + rng.normal(0.0, 0.02, 3000),
}).with_columns(
pl.when(pl.Series(is_trade)).then(pl.col("side", "quantity", "price")) # null on a quote
)
trades.write_parquet("trades.parquet") # on disk, for the examples that scan it
df holds a known answer, so an example's output can be read against it.
y is 1 + 2·x0 − x1 plus noise of spread 0.5, and x2, signal_a and
signal_b play no part in it. ret's beta on signal_a is 0.8 for the
morning session's 200 rows and −0.8 for the afternoon's: a break. Its noise
is calm, of spread 0.5, on rows 0 to 99 and 200 to 299, and volatile, of
spread 2, on the other two hundred: two regimes. trades holds no planted
relation: its mid is a random walk.
How a bank sees a stream
The parameters every spec shares say how its model reads the stream, one
chunk at a time. Most models take all of them, and the
polars_online.spec
reference gives their units and defaults. With a group, every parameter
applies within each group, the clock included.
| the question | the parameters | the subsection |
|---|---|---|
| which columns a model reads, and what its output column is called | the spec's name, targets, features, fit_intercept |
What a spec names |
| how far apart two rows are, and how fast the past fades | clock, half_life, gap_cap, session, session_gap, restart_after_step_back |
Time and decay |
| what a fit becomes when nothing fades | half_life=inf, lam=1.0 |
Convergence without a decay |
| when an old row stops counting at all | window_size |
A hard window |
| how to fit along a feature instead of time | clock |
A local fit along any feature |
| which row order a bank needs, and how it checks it | Row order and the two guarantees | |
| which rows belong to which model | group, group_close |
Groups |
| how much each row counts | weight |
Weights |
| when a model has seen enough to report | min_weight, min_settled_frac, max_error_inflation |
Warm-up |
| when a label may be learned | embargo |
Labels that arrive late |
| what a null does, and how to keep a row from teaching a model | a null, a weight of 0, predict |
Nulls, and three ways to hold a row back |
What a spec names
A spec describes one model: which model it is, the columns it reads, and
how it treats time. The name of the spec (the first parameter) is used to
name the output column: po.spec.ewridge("ridge", ...) adds a column named
ridge. Each row of that column is a record of the model's fields for the
row, and the input's columns pass through untouched.
This code uses df from Example data:
ridge_spec = po.spec.ewridge(
"ridge",
targets=["y"], # at least one; the targets of one spec share the features' running sums
features=["x0", "x1", "x2"], # numeric columns of any width, Decimal and Boolean included; read as 64-bit floats
fit_intercept=True, # the default: the fit has a level of its own
clock="t", half_life=600.0, gap_cap=300.0, # how it treats time: Time and decay, below
)
fitted = po.ModelBank([ridge_spec]).fit_predict(df)
print(len(df.columns), "columns in, then", fitted.columns[len(df.columns):])
print("each row of ridge holds", fitted["ridge"].struct.fields)
13 columns in, then ['ridge']
each row of ridge holds ['pred_y', 'resid_y', 'weight_sum', 'settled_frac', 'withheld_reason', 'coef', 'support_coef']
The bank refuses a spec named like an input column, since its output
column would replace that column. It refuses a text column in targets
or features too. Convert such a column with Polars' with_columns, in
the query before the bank: cast(pl.Float64) for numbers stored as text,
as the error suggests, and a comparison such as pl.col("venue") == "X"
for a category. This code prints both refusals, then fits the venue as a
comparison. Its last line finds the truth planted in y: an intercept of
1, 2 on x0, −1 on x1, and nothing on the venue.
It reads df and lf from Example data:
# Two specs the bank refuses, each refusal printed.
for refused in [
po.spec.ewridge("x0", targets=["y"], features=["x1"], half_life=600.0), # named like a column
po.spec.ewridge("by_venue", targets=["y"], features=["venue"], half_life=600.0), # a text feature
]:
try:
po.ModelBank([refused]).fit_predict(df)
except ValueError as err:
print(err)
# The venue fitted as a comparison.
by_venue = po.spec.ewridge("by_venue", targets=["y"], features=["x0", "x1", "is_x"], half_life=float("inf"))
coded = lf.with_columns(is_x=pl.col("venue") == "X").online.fit_predict([by_venue]).collect()
print("intercept, x0, x1, venue X:", [round(c, 2) for c in coded["by_venue"].struct.field("coef")[-1]])
duplicate: spec "x0" has the same name as an input column; the output struct would replace it. Rename the spec.
spec "by_venue": feature column "venue" has dtype str; it must be numeric (cast it, e.g. pl.col("venue").cast(pl.Float64))
intercept, x0, x1, venue X: [1.02, 1.99, -1.01, -0.05]
Relative and look-ahead targets
A target is often one value less another, such as the next five minutes'
mean trade price less the mid at the row. Build it in one of two ways, by
when its value is known. Only a model that regresses its targets takes a
target expression. Four kinds of model refuse one by name: a model that
reads only its features, ew_class, seqtest, and an ftrl or sgd
fitting a probability or a count.
To use a target computed from the rows after its row, write it as a
Polars expression and put it in the spec's targets. Its .alias names
the output fields, as pred_fwd_move. The bank scores each row where it
sits and learns it once the expression's window has closed, so give the
spec an embargo of at least the window's window_size. Windows as a
model's inputs and target
gives the rule in full, with what to do for a shorter one. pred,
resid, sigma and the metrics are on the target's scale, so
pred + mid predicts the mean trade price itself.
The expression is the column po.stream.with_windows makes from it, fed
back to the bank, so both forms give the same predictions. Only the
residual tells them apart: a column holds the outcome at the row, so the
bank can score it there, while the expression's outcome is not known until
its window closes. This code runs on trades.parquet, the quotes and
trades saved in Example data:
# A feature, a target that looks ahead, and a fit on the trades.
flows = pl.scan_parquet("trades.parquet").with_columns( # a query, and a feature in it: a trade's signed size
flow=pl.when(pl.col("side") == "buy").then(pl.col("quantity")).otherwise(-pl.col("quantity"))
)
fwd_move = (
po.ewm_mean("price", half_life=float("inf"), window_size="5m", reverse=True) # the mean trade price over the next five minutes
- pl.col("mid") # less the mid at the row
).alias("fwd_move") # the alias names the target
ahead_5m = po.spec.ewridge("ahead_5m", targets=[fwd_move], features=["flow"],
clock="ts", gap_cap="5m", half_life="30m", embargo="5m", group="symbol")
moves = flows.online.fit_predict([ahead_5m]).online.unnest([ahead_5m]).collect() # does buying predict the price rising?
# The same target as a column, made in the query by with_windows on the spec's clock, then named in targets.
as_column = {**ahead_5m, "targets": ["fwd_move"]} # a spec is a dict: a copy with the column as target
made = po.stream.with_windows(flows, fwd_move, like=ahead_5m).online.fit_predict([as_column]) # like= takes the spec's clock, gap_cap and group
column = made.online.unnest([as_column]).collect()
# The two forms side by side.
same = (moves["pred_fwd_move"] == column["pred_fwd_move"]).sum()
print(f"rows predicted: {moves['pred_fwd_move'].count()} by the expression, {column['pred_fwd_move'].count()} by the column, {same} equal")
# The rows of the last five minutes have no outcome yet in either form.
print(f"residuals: {moves['resid_fwd_move'].count()} from the expression, {column['resid_fwd_move'].count()} from the column")
rows predicted: 733 by the expression, 733 by the column, 733 equal
residuals: 0 from the expression, 556 from the column
Write a ratio target to sit about zero, as the quotient less 1 or as its
.log(). hit_rate asks whether prediction and outcome fall on the same
side of zero (Is another model
better?). A plain ratio sits
about 1, where two positive numbers always agree, so its hit_rate reads
1.0 whatever the fit. A difference sits about zero too. A target that looks
ahead takes the same forms: (next_5m / pl.col("mid")).log(), where
next_5m is the po.ewm_mean("price", ..., reverse=True) inside fwd_move above, and mid * exp(pred) turns its
prediction back into a price. This code regresses four forms of a trade's
price against the mid at its row on the same signed size, flow.
It reads trades.parquet from Example data:
flows = pl.scan_parquet("trades.parquet").with_columns(
flow=pl.when(pl.col("side") == "buy").then(pl.col("quantity")).otherwise(-pl.col("quantity"))
)
# Four forms of a trade's price against the mid.
returns = flows.with_columns(
ratio=pl.col("price") / pl.col("mid"), # price / mid, about 1
fwd_return=pl.col("price") / pl.col("mid") - 1, # the quotient less 1, about zero
log_ret=(pl.col("price") / pl.col("mid")).log(), # ln(price / mid), about zero
diff=pl.col("price") - pl.col("mid"), # price - mid, about zero
)
vs_mid = po.spec.ewridge("vs_mid", targets=["ratio", "fwd_return", "log_ret", "diff"], features=["flow"],
clock="ts", gap_cap="5m", half_life="30m", group="symbol", emit_metrics=True)
against_mid = returns.online.fit_predict([vs_mid]).online.unnest([vs_mid]).collect()
# How often each form's prediction had the sign right.
for target in ["ratio", "fwd_return", "log_ret", "diff"]:
print(f"hit_rate of {target}: {against_mid[f'hit_rate_{target}'].drop_nulls()[-1]:.2f}")
hit_rate of ratio: 1.00
hit_rate of fwd_return: 0.51
hit_rate of log_ret: 0.51
hit_rate of diff: 0.51
A target expression requires an operator that looks ahead
(po.ewm_mean, po.ewm_sum or po.ewm_rate with reverse=True); otherwise, make the
target a column with Polars' with_columns, chained in the query before
the bank, and name it in targets. The bank's refusal names the call
that makes the column. Here the target, how far from the mid each trade
printed, is known at the trade's own row, so the spec needs no embargo.
The quotes have no quantity, so the bank skips them, as it skips any row
with a null feature.
This code uses trades.parquet from Example data:
edges = pl.scan_parquet("trades.parquet").with_columns(edge=(pl.col("price") - pl.col("mid")).abs()) # the target as a column, null on the quotes
by_size = po.spec.ewridge("by_size", targets=["edge"], features=["quantity"],
clock="ts", gap_cap="5m", half_life="30m", group="symbol")
printed = edges.online.fit_predict([by_size]).online.unnest([by_size]).collect() # does a larger trade print farther from the mid?
is_trade = printed["quantity"].is_not_null()
print(f"trades predicted: {printed.filter(is_trade)['pred_edge'].count()} of {is_trade.sum()}")
print(f"quotes predicted: {printed.filter(~is_trade)['pred_edge'].count()} of {(~is_trade).sum()}")
trades predicted: 893 of 899
quotes predicted: 0 of 2101
If a column holds a value from after its row, such as the price five
minutes on, give the spec an embargo of that horizon (Labels that
arrive late). To make a column from a window
over the stream, looking back or ahead, use po.stream.with_windows in the
query before the bank (Windows as
columns).
The command line has no with_columns, so make such a column in the file
it reads. po.target's relative_to= and relative= are refused, and the
error shows the with_columns form.
po.target
only names a target's output fields apart from its column.
Time and decay
A model forgets old rows by its decay: each row's weight halves every
half_life along a clock, a column that says how far apart two rows
are. At each row, the model multiplies the weight of everything it has
learned by λ = 0.5 ** (Δclock / half_life), where Δclock is the
clock's step from the previous row. To fit one model per half-life, give
half_life a list.
Clock types and units
Write every clock parameter in the clock column's own kind of value, since a timestamp carries its unit and a number does not.
| the clock | a clock parameter is | for example |
|---|---|---|
a Datetime, Date or Duration column: a temporal clock |
a duration | half_life=pl.duration(minutes=10) |
| a numeric column, such as seconds or cumulative traded volume | a number of the column's own units | half_life=600.0 |
| none | a number of rows | half_life=100: a row 100 rows back counts half as much as the latest |
A duration is written three ways: pl.duration(minutes=10),
timedelta(minutes=10), or Polars' duration text "10m".
Every clock parameter acts on the clock, so a model's numbers do not
change with how densely the rows arrive. The clock parameters are
half_life, gap_cap, restart_after_step_back, session_gap, embargo,
drift_threshold and coef_every, and a model's window_size, its cadences
(snapshot_every, solve_every, pca_every, prune_every) and its own
half-lives. A few things still count rows, and each model's entry in
docs/MODELS.md names them.
Quantities measured on a temporal clock reach the output in seconds.
holt's trend, for one, is per second. The clocks themselves keep the
clock column's own type: emit_clocks (Labels that arrive
late), and the clock columns of summary(),
groups() and closed_groups(). This code fits holt, a level and its
trend, to t, a column that rises by one each minute, once on the numeric
clock t and once on the Datetime clock ts. The two clocks mark the
same rows, so the weights agree, and the trend is the same rate in two
units.
It reads df from Example data:
import textwrap
# The same model on each clock, read on the last row.
on_t = po.spec.holt("on_t", targets=["t"], clock="t", half_life=100.0, gap_cap=300.0) # numbers: t counts minutes
on_ts = po.spec.holt("on_ts", targets=["t"], clock="ts", half_life="100m", gap_cap="300m") # durations: ts is a Datetime
last = po.ModelBank([on_t, on_ts]).fit_predict(df).tail(1)
for name, unit in [("on_t", "per unit of t, a minute"), ("on_ts", "per second")]:
fields = last[name].struct
print(f"{name}: weight_sum {fields.field('weight_sum')[0]:.2f}, trend {fields.field('coef')[0][1]:.6f} {unit}")
# The same spec on ts with plain numbers: refused.
try:
po.ModelBank([po.spec.holt("wrong", targets=["t"], clock="ts", half_life=100.0, gap_cap=300.0)]).fit_predict(df)
except ValueError as err: # a plain number on a Datetime clock
print(textwrap.fill(str(err), 100))
on_t: weight_sum 135.66, trend 1.000406 per unit of t, a minute
on_ts: weight_sum 135.66, trend 0.016673 per second
spec "wrong": clock column "ts" has dtype datetime[μs], a temporal clock, but half_life is a plain
number, which it cannot read: a number has no unit, and the column's own (e.g. epoch microseconds)
would silently become it. Give half_life and the other clock parameters as durations, e.g.
pl.duration(minutes=10), timedelta(minutes=10) or "10m"; or cast the clock to the unit you mean,
e.g. pl.col("ts").dt.epoch("s").cast(pl.Float64), and use that column.
The bank refuses a clock parameter of the wrong kind before it reads a
row, as the last lines show, and its error names the column, the parameter
and the fix. Only 0 and inf, which mean the same in every unit, may
stay numbers where a parameter takes them:
| refused | on | because |
|---|---|---|
| a plain number | a Datetime clock |
it would silently take the column's storage unit: half_life=600 on a microsecond column would mean 600 microseconds |
| a duration | a numeric clock | it has nothing to measure it against |
| a spec that mixes the two kinds | any clock | |
a duration finer than the clock can act on, such as gap_cap="12h" |
a Date clock, which moves in days |
|
lam, a decay per clock unit |
a temporal clock | it has no duration form, so a temporal clock takes half_life instead |
A temporal clock is read in its own integer nanoseconds, so it must lie between the years 1677 and 2262, the range nanoseconds in a 64-bit integer cover. The same instants stored in milliseconds, microseconds or nanoseconds give the same numbers, and a time zone changes nothing. A model reads only the gap between consecutive rows, taken in integer nanoseconds before it becomes seconds, so a nanosecond timestamp keeps its nanoseconds whatever the stream's age.
An integer clock is read as its integers: any column from Int8 to
Int64 or UInt8 to UInt64. The step between two rows is taken in
integers before it becomes a double. Every decision on the clock is taken on
the integers too: a step back, a gap past gap_cap, a cadence, a window's
edge and an embargo's release. So an Int64 column of epoch nanoseconds
keeps its nanoseconds, where a double near 1.8e18 resolves 256 of them. Its
clock parameters stay numbers, and a fractional one such as gap_cap=0.5 is
compared as the number it is.
A clock column the bank cannot read exactly is refused, by row or by
name. A UInt64 value past the largest Int64 is refused by row. An
Int128 or UInt128 column is refused by name: cast it to Int64 if its
values fit, or after subtracting an origin. A UInt128 column is refused
wherever it is in the frame, since no frame carrying one can reach the
bank. A later chunk whose clock is a float, or an integer of another width,
is refused by name.
Sessions, gaps and steps back
A stream from a market has session boundaries, gaps and steps back in its clock. The spec has a parameter for each, and together they are the clock's policy:
| the stream has | the parameter | what it does |
|---|---|---|
| a session boundary, such as the start of a trading day, where the clock's jump is not elapsed time | session, a column whose value changes at a session boundary, and session_gap, required with it unless group_close="session" closes the group at the change instead (One row per finished group) |
applies a step you choose, session_gap, in place of the jump, capped at gap_cap like any step. session_gap="reset" starts the model over instead |
| a gap in the clock, such as an hour with no rows | gap_cap, required with a clock, finite and above 0 |
caps each step between two rows a model learns from, so a quiet hour ages the fit no more than gap_cap would. After a run of skipped rows, the next row's step is capped too, however long the run |
| a clock that steps back, such as a replayed day | restart_after_step_back |
starts the model over at a step back larger than the setting, and refuses one no larger as a late row. Unset, the default, every step back is refused (what each step back means) |
A stream has a break wherever gap_cap shortened a step, and wherever
the session changes. A model that keeps past rows by position, as a lag
does, empties them at a break, where the previous row no longer stands for
the moment just before.
Each policy shows in weight_sum, the weight behind each row's
prediction. To show each, this code moves the afternoon session of df
two days on, which puts a weekend between rows 199 and 200 (counting from
0), and fits one regression under four policies:
| spec | its policy | the weekend counts as |
|---|---|---|
capped |
gap_cap="5m" |
five minutes: the fit ages a little |
uncapped |
gap_cap="3d" |
two days: about a hundred half-lives empty the fit |
one_minute |
session="session", session_gap="1m" |
one minute, the step chosen for the session boundary |
reset |
session="session", session_gap="reset" |
a fresh start: the model forgets everything |
weight_sum is read before a row's own step, so a reset shows on row 200
itself, and a step's decay on row 201. A numeric clock takes the same
parameters as numbers of its units, such as half_life=600.0, gap_cap=300.0 on t, and lam, the weight kept per unit, may stand in
for half_life.
This code uses df from Example data:
# df with a weekend between its two sessions, saved.
afternoon = pl.col("session") == "a"
df.with_columns(ts=pl.when(afternoon).then(pl.col("ts") + pl.duration(days=2)).otherwise(pl.col("ts"))) \
.write_parquet("weekend.parquet")
# Four ways to treat the weekend.
common = dict(targets=["y"], features=["x0", "x1"], clock="ts", half_life="30m") # ts is a Datetime: durations
policies = [
po.spec.ewridge("capped", gap_cap="5m", **common), # the weekend counts as five minutes
po.spec.ewridge("uncapped", gap_cap="3d", **common), # the weekend counts in full
po.spec.ewridge("one_minute", gap_cap="5m", session="session", session_gap="1m", **common), # one minute at the boundary
po.spec.ewridge("reset", gap_cap="5m", session="session", session_gap="reset", **common), # a fresh start there
]
# Each policy's weight_sum around the boundary at row 200.
fitted = pl.scan_parquet("weekend.parquet").online.fit_predict(policies).collect()
print("row" + "".join(f"{p['name']:>12}" for p in policies))
for row in [199, 200, 201, 202]:
print(f"{row:<3}" + "".join(f"{fitted[p['name']].struct.field('weight_sum')[row]:>12.2f}" for p in policies))
row capped uncapped one_minute reset
199 43.34 43.34 43.34 43.34
200 43.35 43.35 43.35 0.00
201 39.62 1.00 43.36 1.00
202 39.72 1.98 43.37 1.98
Convergence without a decay
With decay off (half_life=inf or lam=1.0), a model that solves or
accumulates converges to the batch fit it defines over every row it has
seen. A model that reweights, steps, filters or tests depends on
the row order whether or not decay is on. Each model's kind is in the
model table, with the few settings that read the rows in
sequence even so.
Row order does not change the fit of a model that solves or accumulates, apart from those settings, since both keep running sums. With no clock column, rows fed forwards, backwards or shuffled give the same coefficients, to rounding. With a clock column, the bank refuses a row out of clock order (How a bank detects rows out of order).
For no forgetting, give half_life=float("inf"): a huge finite
half-life still forgets, a little. Under half_life=1e12, settled_frac
stays near zero until the stream has run for about 10¹² clock units, so a
min_settled_frac gate (Warm-up) stays closed.
This code fits ewridge with no decay to df's rows in their order,
reversed and shuffled, and solves the same least squares with numpy. The
last line reads settled_frac on the last row under half_life=1e12.
It reads df from Example data:
# Plain least squares in three orders, then numpy's.
batch = po.spec.ewridge("batch", targets=["y"], features=["x0", "x1", "x2"],
half_life=float("inf"), # no clock, no decay
ridge=0.0) # plain least squares, in place of the default ridge of 1e-6
for order, rows in [("in order", df), ("reversed", df.reverse()), ("shuffled", df.sample(fraction=1.0, shuffle=True, seed=1))]:
coef = po.ModelBank([batch]).fit_predict(rows)["batch"].struct.field("coef")[-1] # the fit after the last row
print(f"{order:<9}", " ".join(f"{c:9.6f}" for c in coef))
design = np.column_stack([np.ones(df.height), df["x0"], df["x1"], df["x2"]])
print(f"{'numpy':<9}", " ".join(f"{c:9.6f}" for c in np.linalg.lstsq(design, df["y"], rcond=None)[0]))
# A huge finite half-life, which is not inf.
huge = po.spec.ewridge("huge", targets=["y"], features=["x0"], clock="t", half_life=1e12, gap_cap=300.0)
print(f"settled_frac after 399 units of t under half_life=1e12: {po.ModelBank([huge]).fit_predict(df)['huge'].struct.field('settled_frac')[-1]:.1e}")
in order 0.992088 1.994121 -1.007945 -0.026134
reversed 0.992088 1.994121 -1.007945 -0.026134
shuffled 0.992088 1.994121 -1.007945 -0.026134
numpy 0.992088 1.994121 -1.007945 -0.026134
settled_frac after 399 units of t under half_life=1e12: 2.8e-10
A hard window
A half-life never forgets entirely: three half-lives back still carries
12.5% of the weight. To make every row w clock units old or older count
for exactly nothing, give window_size=w. Five models take it: ewridge,
lasso, ew_cov, ew_class and marginal.
weight(age) = 0.5 ** (age / half_life) if age < window_size (closed="right", the default)
or age <= window_size (closed="both")
= 0 otherwise
closed says which edge holds the row exactly window_size old, as
Polars' rolling_*_by(closed=) does. Under "right", the default, that
row has left, as it has in the window operators and in
rolling_sum_by. Under "both" it stays. A model's window always holds the
row it ends at, because the model reads its fit after it has learned that
row. So "left" and "none" are refused, and so is "both" without a
window_size.
Inside the window the weights still halve every half_life, so the
newest row counts most. For a flat rolling fit with equal weights, give
half_life=float("inf") with the window_size.
A window forgets a break that a half-life still holds. In df, the
beta of ret on signal_a flips from 0.8 to −0.8 at row 200 (counting
from 0). This code follows it with a half-life of 100 rows, and with a flat
window of 100 rows. A hundred rows after the break, the window holds only
rows after it and reads the new beta, while the half-life still gives the
old rows half their weight.
It reads df from Example data:
common = dict(targets=["ret"], features=["signal_a"], clock="t", gap_cap=300.0, coef_every=0) # the beta on every row
decayed = po.spec.ewridge("decayed", half_life=100.0, **common) # a half-life of 100 rows
windowed = po.spec.ewridge("windowed", half_life=float("inf"), window_size=100.0, **common) # the last 100 rows, equally weighted
fitted = po.ModelBank([decayed, windowed]).fit_predict(df)
print("row decayed windowed")
for row in [199, 250, 300, 399]:
beta = [fitted[name].struct.field("coef")[row][1] for name in ("decayed", "windowed")] # [intercept, signal_a]
print(f"{row:<4} {beta[0]:7.2f} {beta[1]:8.2f}")
row decayed windowed
199 0.77 0.83
250 0.29 0.01
300 0.00 -0.78
399 -0.39 -0.78
The cut is exact. At time t, the rows learned at or before a time u
contribute 0.5 ** ((t − u) / half_life) times the running sum as it stood
at u. Subtracting that leaves exactly the rest. The model keeps a ring
of snapshots for this, one every row by default. The edge is the oldest
snapshot inside the window, so a coarse spacing discards a little more
than asked, never less.
snapshot_every spaces the snapshots on the clock, as solve_every
spaces a regression's solves. It takes a number of clock units, or a
duration such as "1m" on a temporal clock. max_rows_between_snapshots
caps the rows between them, and whichever comes first takes a snapshot.
Under snapshot_every the effective window is in
[window_size − snapshot_every, window_size], whatever the row rate, with
snapshot_every the spacing in force: a thinning window_budget doubles it.
weight_sum is the weight inside the window. This code fits ew_cov
over a window of 150 units of t with a half-life of 50, and a snapshot
every 10 units. The weight a row a unit apart adds settles at
1 / (1 − 0.5 ** (1 / 50)), about 72.6, without a window. A window keeps
the share of it younger than its effective edge, 1 − 0.5 ** (edge / 50),
with the edge between 140 and 150 units back.
It reads df from Example data:
# The same model with a window and without.
cut = po.spec.ew_cov(
"cut", features=["x0", "x1"], clock="t", gap_cap=300.0, half_life=50.0,
window_size=150.0, # a row this many clock units old or older contributes exactly nothing
snapshot_every=10, # a snapshot every 10 units of t
window_budget={"refuse": 64}, # at most 64 MiB of snapshots per ring
)
whole = po.spec.ew_cov("whole", features=["x0", "x1"], clock="t", gap_cap=300.0, half_life=50.0)
# Each weight_sum on the last row, against the ceiling and the window's share of it.
last = po.ModelBank([cut, whole]).fit_predict(df).tail(1)
ceiling = 1 / (1 - 0.5 ** (1 / 50))
print(f"without a window: weight_sum {last['whole'].struct.field('weight_sum')[0]:.2f}, the ceiling {ceiling:.2f}")
print(f"inside the window: weight_sum {last['cut'].struct.field('weight_sum')[0]:.2f}, "
f"between {ceiling * (1 - 0.5 ** (140 / 50)):.2f} and {ceiling * (1 - 0.5 ** (150 / 50)):.2f}")
without a window: weight_sum 72.35, the ceiling 72.64
inside the window: weight_sum 63.43, between 62.21 and 63.56
Five things to know before reading a windowed model's numbers:
| the clock is the decayed one | a row's age adds up each step after gap_cap and any session_gap |
| the edge is a discontinuity | a row ageing out drops its whole weight at once, so the series has small steps that a plain exponential average does not |
| it is a subtraction | precision falls with the fraction discarded: negligible at a window of three half-lives, worse as the window shortens toward the half-life |
weight_sum is the weight inside the window |
so min_weight gates on a weight that settles once the window is full |
| a long step empties the window | a step longer than window_size, after gap_cap, empties the window, and the model reports nulls from the row after it, never stale numbers. A cap shorter than the window never empties it |
What the window truncates depends on the model:
| model | inside the window | refused beside window_size |
|---|---|---|
ewridge |
the sums the fit is solved from, and sigma and zscore with them. The coefficients are the window's as of the last solve |
ridge_scale="sum", session_shrink, gram_block_rows |
lasso |
the same, and the error that selects the penalty. A feature that stays constant inside the window has no evidence there, and goes to exactly zero | |
ew_cov |
every moment, and mahal, partial_corr and the principal components read from them |
lags, mahal_quantiles |
ew_class |
each class's moments, so the classifier follows class means that move | |
marginal |
every pair's weight, means and second moments, so corr, beta and t_stat describe the window. Its lag moments, under window_lags=True, are an estimate rather than exact |
bins, feature_moments="shared", and lags unless window_lags=True |
window_budget caps the snapshots' memory, per ring (one per spec and
group), in MiB. The ring's memory grows with the window: for an ew_cov over 20 columns, a
1,000-row window holds about 3 MB per group, divided by the rows between
snapshots.
window_budget |
when a chunk would take a ring past the cap |
|---|---|
| not given | the chunk is refused past 256 MiB |
{"refuse": 64} |
the chunk is refused before any of it is learned, and the bank goes on as it was. The error names the ring's size, its snapshot cadence and the ways to lift the cap |
{"thin": 64} |
the ring drops every other snapshot and doubles its spacing, on the clock or in rows, which, like snapshot_every, only ever shortens the window |
{"refuse": float("inf")} |
nothing: no cap |
A refusal comes before the bank learns any of the chunk, except for two
overruns found only as the rows go in. One is under
drift_action="reset". The other is a snapshot that grows within the
chunk, as an ewridge's or a lasso's does when its targets first go
missing on different rows. After either, the bank refuses every call that
follows, so rebuild it from its last save with po.ModelBank.load.
A local fit along any feature
To fit any model locally along a feature, sort the rows by that feature
with Polars' sort in the query before the bank, and name the feature as
the spec's clock. half_life is then a bandwidth in that feature's
units. Each row is fitted on the rows before it in that order, each
weighted by 0.5 ** (Δx / half_life), with Δx its distance from the row
in that feature: an exponential kernel. The fit is a local regression from
running sums that do not grow. It agrees to 1e-12 with a kernel-weighted
least squares recomputed from scratch at every row, which
tests/test_ewridge.py holds it to.
The kernel is one-sided, so the fit lags. Narrow the bandwidth to lag
less, and widen it to smooth the noise. This code makes 1,200 points of
sin(x) plus noise of standard deviation 0.1, for x between −3 and 3, and
fits a straight line in x locally at two bandwidths. It prints each fit's
root-mean-square distance from the curve, beside the best straight line's:
import numpy as np
import polars as pl
import polars_online as po
# 1,200 noisy points on sin(x), saved.
rng = np.random.default_rng(1)
x = rng.uniform(-3.0, 3.0, 1200)
pl.DataFrame({"x": x, "y": np.sin(x) + 0.1 * rng.standard_normal(1200)}).write_parquet("curve.parquet")
# A straight line fitted locally at a given bandwidth, over the points sorted by x.
def local(bandwidth):
kernel = po.spec.ewridge(
"kernel", targets=["y"], features=["x"],
clock="x", # the clock is a feature: the decay is a kernel in it
half_life=bandwidth, # the bandwidth, in x's own units
gap_cap=1.0, # a wider gap decays as if it were this wide
max_rows_between_solves=1, # refit at every row
coef_every=0, # write the coefficients on every row: the line through the row's neighbourhood
min_weight=20.0,
)
query = pl.scan_parquet("curve.parquet").sort("x") # sorted by the feature, in the query before the bank
return query.online.fit_predict([kernel]).online.unnest([kernel]).collect() # pred_y: the curve at each row's x
# Two bandwidths, and the best straight line through every point.
narrow, wide = local(0.25), local(1.0)
fit = narrow["pred_y"].is_not_null() # the rows past the warm-up
curve = narrow.filter(fit)
truth = np.sin(curve["x"].to_numpy())
line = np.polyval(np.polyfit(narrow["x"], narrow["y"], 1), curve["x"].to_numpy())
# Each fit's distance from the true curve.
def rms(pred):
return float(np.sqrt(np.mean((np.asarray(pred) - truth) ** 2)))
print(f"bandwidth 0.25: {rms(curve['pred_y']):.2f}")
print(f"bandwidth 1.0: {rms(wide.filter(fit)['pred_y']):.2f}")
print(f"straight line: {rms(line):.2f}")
bandwidth 0.25: 0.08
bandwidth 1.0: 0.29
straight line: 0.39
The narrow kernel follows the curve the straight line cannot. The wide one is most of the way back to the line.
The clock column can stay out of features. With other columns as
features, the fit is a regression whose coefficients move along the clock.
Row order and the two guarantees
Two guarantees hold for every run, and both rest on a fixed row order, which the caller supplies.
Every prediction and diagnostic is out-of-sample. Each row is scored from the state as it stood before the row's own target was learned, so no row's outcome reaches its own prediction or the diagnostics measured on it.
The numbers do not depend on the chunking. One chunk or a thousand,
with or without a save and resume in the middle, gives bit-identical
numbers. Only which rows carry coef, and support_coef and the standard
errors of emit_se_coef beside it, can differ, and only by default. The bank then writes them on each group's last
accepted row in every chunk, so smaller chunks report them more often.
Under coef_every or max_rows_between_coefs they do not move
(Coefficients).
This code checks both on df. It predicts row 300 with a bank that has
learned only rows 0 to 299, and compares that with the same row of one
run over every row. It then runs the stream again one row per chunk,
through fit_predict_batches (In a loop).
It reads df from Example data:
# One run over every row.
honest = po.spec.ewridge("honest", targets=["y"], features=["x0", "x1"], clock="t", half_life=100.0, gap_cap=300.0)
one_run = po.ModelBank([honest]).fit_predict(df)["honest"].struct.field("pred_y")
# Row 300 from a bank that learned only the rows before it.
first_300 = po.ModelBank([honest])
first_300.fit_predict(df.head(300)) # learns rows 0 to 299
alone = first_300.predict(df.slice(300, 1))["honest"].struct.field("pred_y")[0]
print(f"row 300: {one_run[300]:.6f} in one run, {alone:.6f} from rows 0 to 299 alone")
# The same stream, one row per chunk.
by_row = pl.concat(po.ModelBank([honest]).fit_predict_batches(df, chunk_size=1)) # 400 chunks of one row
print(f"one chunk against 400: {(one_run - by_row['honest'].struct.field('pred_y')).abs().max()} largest difference")
row 300: 2.142699 in one run, 2.142699 from rows 0 to 299 alone
one chunk against 400: 0.0 largest difference
A model learns in row order, so a query whose row order Polars does not
guarantee can give a different model each time it runs. Run a chunk at a
time, as a bank runs it, a join, a group_by or a unique can deliver
rows in another order than lf.collect() gives. Give each such step its
guarantee in the query:
| a query step that may reorder rows | give it |
|---|---|
join |
maintain_order="left" |
group_by |
maintain_order=True |
unique |
a sort after it |
| anything else | a sort before the bank |
How a bank detects rows out of order
A bank checks the order it is given in three places, warning about the query and refusing a chunk that runs backwards before learning any of it.
| check | what it reads | when it finds disorder |
|---|---|---|
| the query | the query handed to the bank: a join, group_by or unique whose order Polars does not guarantee, unless a sort follows it, and a sort by several keys without maintain_order=True |
raises OrderNotGuaranteedWarning, naming the step and its fix. The run goes on |
| the clock | each group's clock, row by row, for every spec with a clock |
refuses the chunk on a backwards step the settings below do not allow |
| the group keys | with group_close="monotone", the keys in the column's own order, an integer column as numbers and any other as text |
refuses the chunk on a key below the one before it, since a closed group cannot reopen |
The query check names the step and its fix. This code joins each
row's venue to a fee, once as Polars' join does by default and once with
maintain_order="left", and prints what the bank warns of each. The
warning calls the query a plan, and its steps nodes, as Polars does.
It reads lf from Example data:
import textwrap
import warnings
# A fee per venue, and a spec that reads it.
pl.DataFrame({"venue": ["X", "Y"], "fee": [0.1, 0.2]}).write_parquet("fees.parquet")
priced = po.spec.ewridge("priced", targets=["y"], features=["x0", "fee"], half_life=100.0)
# Join both ways, and print the bank's warning, if any.
for order in [None, "left"]:
joined = lf.join(pl.scan_parquet("fees.parquet"), on="venue", **({"maintain_order": order} if order else {}))
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
joined.online.fit_predict([priced]).collect()
said = [str(w.message) for w in caught if issubclass(w.category, po.OrderNotGuaranteedWarning)]
print(f"maintain_order={order!r}:", textwrap.fill(said[0], 100) if said else "no warning")
maintain_order=None: lf.online.fit_predict: the plan's row order is not guaranteed, and an online model learns in row
order. 1 node(s) leave it unspecified: a join without maintain_order (pass maintain_order="left" to
keep the order of the left input). The streaming engine that runs the plan may deliver rows in an
order lf.collect() would not, and it may differ between runs. Give the plan a fixed order --
maintain_order="left" on a join, maintain_order=True on a group_by, a sort after a unique() or
before the bank -- or, for a plan whose order is fixed by other means,
warnings.simplefilter("ignore", polars_online.OrderNotGuaranteedWarning).
maintain_order='left': no warning
The query check is best-effort. It reads the query's explain() text,
and walks the query's steps only when that names a join, an aggregation, a
unique or a sort. A step followed by a sort counts as ordered, unless
that sort is by several keys without maintain_order=True. Such a sort
leaves rows with equal keys in no particular order: 2,109 of 10,000 rows
moved when measured. A join with maintain_order="left" is followed on its
left side only. A query Polars cannot serialize passes without a warning,
and the check never fails a run. For a query whose order you know, silence
the warning with
warnings.simplefilter("ignore", po.OrderNotGuaranteedWarning).
ModelBank.fit does not warn when its state cannot depend on the row
order. That is when every spec is an ewridge or rls with decay off
and none of the settings that read the order. Those include a window, a
session, a weight, an embargo, a drift reset, a coef cadence, a window
expression as a target, and any diagnostic but emit_drift, emit_clocks
and emit_error_inflation. Such sums reach the same state in any order, to
rounding, and fit keeps only the state.
The clock is checked group by group, so groups may interleave freely:
only the rows of one group need to be in clock order. Equal clock values
are zero apart. A row with a null feature is skipped, but its clock is
still checked. A row that predict scores is never refused (Serving
without learning).
restart_after_step_back and the size of a step back decide what it
means. It has no default, since only the caller knows how late a row can
be, and 0 restarts the model at every step back:
| a step back | is read as | and the bank |
|---|---|---|
any size, with restart_after_step_back unset, the default |
rows out of order | refuses the chunk |
no larger than restart_after_step_back |
a late row, such as a transposed pair or a row a minute late | refuses the chunk |
larger than restart_after_step_back |
a new start, such as a replayed day or a restarted feed | restarts the model |
any size, on a row whose session value changes |
a new session, where the step is no measure of time | applies session_gap, or restarts the model under session_gap="reset" |
A refused chunk leaves the bank as it was. The bank runs the clock of
every group of every spec over the chunk before it learns any row. So one
late row in one group changes nothing: correct the chunk and feed it again.
To resume a saved state on input that overlaps it, pass the input through
bank.skip_learned(...), which drops the rows the state has learned (Save
and load).
The error locates the step and names the way out, and the summary counts
what the clock rules met. The row the error names is counted from the
start of the input however it arrives (one frame, batches, a query or the
command line's file). On a temporal clock the step is a duration.
bank.summary() reports clock_backwards, the rows whose clock fell below
the previous row's within a session, and resets, the rows where a stream
restarted. This code feeds ten rows of df with the row at t = 6 two
rows late, first with restart_after_step_back unset, then set below the
step:
It reads df from Example data:
import textwrap
late = df.head(10)[[0, 1, 2, 3, 4, 5, 7, 8, 6, 9]] # t runs 0 ... 5, 7, 8, 6, 9
# Left unset, any step back is refused.
strict = po.spec.ewridge("strict", targets=["y"], features=["x0"], clock="t", half_life=100.0, gap_cap=300.0)
try:
po.ModelBank([strict]).fit_predict(late)
except ValueError as err:
print(textwrap.fill(str(err), 100))
# Set below the step: the model starts over, and summary() counts it.
lenient = po.spec.ewridge("lenient", targets=["y"], features=["x0"], clock="t", half_life=100.0, gap_cap=300.0,
restart_after_step_back=1.0) # a step back larger than 1 starts the model over
bank = po.ModelBank([lenient])
bank.fit_predict(late)
counts = bank.summary().row(0, named=True)
print(f"{counts['spec']}: clock_backwards {counts['clock_backwards']}, resets {counts['resets']}")
spec "strict": clock column "t" goes backwards by 2 at row 8 (restart_after_step_back is unset, so
every step back is refused); the bank was not updated. Sort each group by the clock; to resume a
saved state on input that overlaps it, drop the rows it has learned, with
ModelBank.skip_learned(frame) in Python or --skip-learned on the command line; or, if a step back
this large starts the stream over, set restart_after_step_back below it (a step back no larger than
the setting is a late row).
lenient: clock_backwards 1, resets 1
Groups
To fit one model per value of a column, name the column with group: the
bank fits a separate model for each distinct value, in the same pass. Each
group's model sees only its own rows, on its own clock, so a group's
numbers are those of the same spec fitted to that group's rows alone.
This code uses df from Example data:
# One model per stock, in one pass.
per_stock = po.spec.ewridge(
"per_stock", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0,
group="stock_id", # one model per distinct value of this column
)
bank = po.ModelBank([per_stock])
together = bank.fit_predict(df)
for row in bank.groups().iter_rows(named=True):
print(f"group {row['group']}: {row['rows_fed']} rows, the last at t = {row['clock_last']}")
# Stock b0's rows alone give the same predictions.
alone = {**per_stock, "group": None} # the same spec without a group, fed stock b0's rows alone
b0 = po.ModelBank([alone]).fit_predict(df.filter(pl.col("stock_id") == "b0"))
in_bank = together.filter(pl.col("stock_id") == "b0")["per_stock"].struct.field("pred_y")
print(f"b0 in the bank against b0 alone: {(in_bank - b0['per_stock'].struct.field('pred_y')).abs().max()} largest difference")
group b0: 100 rows, the last at t = 396.0
group b1: 100 rows, the last at t = 397.0
group b2: 100 rows, the last at t = 398.0
group b3: 100 rows, the last at t = 399.0
b0 in the bank against b0 alone: 0.0 largest difference
A group or session column keeps the type its first chunk had. A key is
its value's text, so an Int64 1 and a Float64 1.0 would be two
groups. The bank refuses a column of another kind by name, and keeps the
kind in its state: integer, float, text, boolean, or a temporal type with
its unit and zone. An Int32 after an Int64, or a Categorical after a
String, is the same kind and goes on. bank.groups() and the other
readers list an integer column's keys as numbers, the null group first.
Without group_close, the bank keeps every group it has seen. To free
the groups that have gone quiet, find them with bank.groups() and drop
them with bank.drop_groups(...) (In a loop).
With group_close, the bank writes out a finished group's running sums
and frees its memory, which keeps its memory bounded when new group
values never stop appearing. group_close="monotone" finishes a group once
a higher key arrives, so it needs the keys in order, as a block number or a
date is. group_close="session" finishes a group once its session value
changes. One row per finished group shows
what is written.
Weights
To make some rows count more than others, name a column of row weights
with weight. A row of weight 2 counts as two rows of weight 1.
A weight of 0 is legal, and a negative weight is refused. A row of weight 0 is scored and moves the clock, but teaches the model nothing (Nulls, and three ways to hold a row back).
In a model that keeps means, scaling every weight by the same factor
changes nothing but weight_sum, so scale min_weight by the same
factor. This code gives row 100 a weight of 0 and every other row a
weight of 1, then scales every weight by 10. With no decay, weight_sum
counts the weight learned before each row: row 100 is scored, but the row
after it finds the weight unchanged.
It reads df from Example data:
weighted = po.spec.ewridge(
"weighted", targets=["y"], features=["x0", "x1"], half_life=float("inf"),
weight="w", # a column of row weights
)
# Row 100 weighs nothing; then every weight times 10.
ones = df.with_columns(w=pl.when(pl.int_range(pl.len()) == 100).then(0.0).otherwise(1.0))
tens = ones.with_columns(pl.col("w") * 10)
fields = [po.ModelBank([weighted]).fit_predict(frame)["weighted"].struct for frame in (ones, tens)]
# The rows around row 100, and what the scaling moved.
for row in [99, 100, 101, 102]:
print(f"row {row}: pred_y {fields[0].field('pred_y')[row]:.4f}, weight_sum {fields[0].field('weight_sum')[row]:g}")
print(f"weights times 10: pred_y moves by at most {(fields[0].field('pred_y') - fields[1].field('pred_y')).abs().max():g}, "
f"weight_sum is {fields[1].field('weight_sum')[-1] / fields[0].field('weight_sum')[-1]:g} times as large")
row 99: pred_y -2.2004, weight_sum 99
row 100: pred_y 0.6661, weight_sum 100
row 101: pred_y 2.5369, weight_sum 100
row 102: pred_y 2.4487, weight_sum 101
weights times 10: pred_y moves by at most 0, weight_sum is 10 times as large
Eight models read a weight differently, and
tests/test_weight_scale.py holds every other model to that rule:
| model | a row's weight is |
|---|---|
rls |
on the sum scale, against a ridge of fixed size: a heavier stream outweighs it sooner |
kalman |
a scale on the observation's precision: the update divides the observation noise R_j by w, so with a fixed obs_var= the observation's variance is obs_var / w |
sgd |
a step size: the gradient carries it |
ftrl |
an importance weight, as Vowpal Wabbit's, against penalties of fixed size |
pa |
a step size below 1; a weight above 1 counts as 1 |
hmm |
on the sum scale, against the transitions' prior count |
rcov |
0 or 1, since a realised covariance is a sum over returns; any other weight is refused |
seqtest |
refused, since every learned row is one trial: to skip a row, null its target |
Warm-up
A model should not report a number it is not yet informed enough to give.
Every model with a decay shares two gates that say what "informed enough"
means, and the linear fits add a third, on by default on ewridge alone.
The gates hold back output only: the model learns from every row either
way.
| setting | default | withholds a prediction while | read from |
|---|---|---|---|
min_weight |
by model, below | weight_sum is below it |
the weight behind the state |
min_settled_frac |
0, off |
the decay window is less full than this fraction of its steady state | settled_frac = 1 − 2^(−T / half_life), with T the decay time the model has seen: 0.5 at one half-life, 0.75 at two, whatever the row rate |
max_error_inflation |
sqrt(2) on ewridge; off on rls, kalman and lasso |
estimation error would inflate the prediction's error over the noise floor by more than this factor | error_inflation = sqrt(1 + edf / n_kish): the effective degrees of freedom the fit used, over Kish's effective sample size behind it (below) |
weight_sum means the same in every model, so one min_weight means
the same for every spec in a bank. It is the weight behind the state
that produced a row's prediction: the weight as the previous row left it,
before this row's decay and its update, and 0 on a stream's first row. It
runs one behind the row count while nothing is forgotten. For rows of
weight 1 a step Δclock apart, it settles at 1 / (1 − λ), with λ from
Time and decay. It is a weight and not a sample size: at
a half-life of 600 with rows 0.1 apart it settles near 8,657. For the
sample size, read Kish's n_kish from gram().
Every output is null until the model's weight reaches min_weight. A
regression model checks each target against that target's own weight, the
weight of the rows it was present on, so an often-null target reports
later than the others. For rls, which learns a row only when every target
is present, that is the weight of the rows it learned from. The
weight_sum field is the shared weight either way. To set one threshold
per target, give min_weight a list. The default depends on the model:
| default | models |
|---|---|
| one per unknown: the features, and the intercept when there is one | lasso, kalman, huber, quantile, rls, sgd, pa, ftrl |
the feature count plus one (1 for holt) |
ew_cov, ew_class, kmeans, micro, holt |
| 3 | marginal, deco |
| 1 | bocpd |
| 0, each having a gate of its own or nothing to gate | ewridge, seqtest, rcov, hmm, corrchange, audit |
min_settled_frac is off by default, because a fit kept as weighted
means is unbiased from its first row when the process is stationary. Set it
when the half-life was chosen to average over regimes or seasons that a
shorter history would not represent.
max_error_inflation follows the size of the fit: add a feature, and
the gate holds predictions back longer. It reads Kish's count, so one row
carrying a hundred times the weight of the others counts as barely one. Each
model that has it reads its own estimation variance. The others refuse the
setting and say why:
| model | error_inflation |
|
|---|---|---|
ewridge |
sqrt(1 + edf / n_kish) |
on by default at sqrt(2) |
rls |
sqrt(1 + k / n_kish), k the coefficients: its edf is at most k under the fading prior |
off unless set |
lasso |
sqrt(1 + df / n_kish) per penalty, df the active coefficients and the intercept |
off unless set |
kalman |
sqrt(1 + z' P z / R) for each row: the prediction's own variance, its coefficients' covariance carried through the row's clock gap, over the noise |
off unless set; read per row, so a row far from the design is held back where others pass |
Each row says how ready its model was, and why a prediction is
missing. withheld_reason names the gate that held a row back, and the
gates are checked in the order of its values, so the first that applies is
the one named. This code sets all three gates on one spec and counts the
rows each withholds:
It reads df from Example data:
warm = po.spec.ewridge(
"warm", targets=["y"], features=["x0", "x1"], clock="t", half_life=100.0, gap_cap=300.0,
min_settled_frac=0.02, # withhold until the decay window is 2% full
min_weight=8.0, # and until the weight behind the fit reaches 8
max_error_inflation=1.05, # and while estimation error would add more than 5% over the noise floor
)
# How many rows each gate withheld, and the first row predicted.
rows = po.ModelBank([warm]).fit_predict(df).unnest("warm")
counts = rows.group_by("withheld_reason").len().sort("withheld_reason", nulls_last=True)
for reason, n in counts.iter_rows():
print(f"{reason or 'predicted'}: {n} rows")
first = rows.filter(pl.col("pred_y").is_not_null()).row(0, named=True)
print(f"first prediction at t = {first['t']:g}: weight_sum {first['weight_sum']:.2f}, settled_frac {first['settled_frac']:.3f}")
below_min_settled_frac: 4 rows
below_min_weight: 5 rows
above_max_error_inflation: 21 rows
predicted: 370 rows
first prediction at t = 30: weight_sum 27.18, settled_frac 0.182
The readiness fields:
| field | on | what it holds |
|---|---|---|
settled_frac |
every model that writes a row | how full the decay window is |
withheld_reason |
every model that writes a row | why pred_y is null, or null when it is not: below_min_settled_frac, below_min_weight, above_max_error_inflation or below_warm_rows (the rows kmeans, hmm and bocpd hold to seed themselves from), in that order of precedence |
error_inflation_y |
ewridge, rls, kalman, under emit_error_inflation |
the row's own estimation variance over the noise: high on a row leaning on a direction the data never showed, at one triangular solve a row |
support_coef |
ewridge, huber, quantile, beside coef |
each coefficient's data share, 1 - ridge * (S^-1)_jj on the system the solve inverts. A duplicated pair of features reads 0.5 each |
se_coef |
ewridge, rls, kalman, under emit_se_coef, beside coef |
each coefficient's standard error in its own units: sigma times the fit's own covariance, kalman's posterior exactly. A report, not a gate; it errs large in warm-up and under a ridge |
summary() carries the same readings per group, and warns of a gate
that will not open. It adds weight_sum_settled, the weight the stream
settles at. Once the stream is 95% settled, a ReadinessWarning names,
once, a coefficient more ridge than data. It also names a noise gate or a
min_weight that has withheld every row for a further half-life, and that
the rows, at their rate so far, would not meet. It gives the half-life or
the ceiling that would change that, and promises nothing of rows that come
faster. The command line closes a run with one line per spec whose groups
ended withheld or with a coefficient more ridge than data. Is it
ready? reads these fields to decide
when a model can be trusted.
Labels that arrive late
Some labels arrive after their row: the next five minutes' return is known
five minutes after its row. A model that learns it at its own row sees
that much of the future before it predicts the rows in between. Every
out-of-sample number after that is contaminated, and with a feature
correlated with its own past, even a pure-noise column starts to look
predictive. To learn each label only once it would have arrived, give the
spec an embargo. To write the same delay out as rows,
use po.stream.embargo in the query before the bank.
The built-in delay
Give a spec embargo="5m" (any finite delay above 0), and the bank scores
each row where it sits and learns from it five minutes later on the clock.
Everything computed from the labels sees only those that had arrived: the
prediction, sigma, zscore, the metrics, break detection, the conformal
interval, weight_sum and the min_weight gate.
This code fits df's minute-apart rows under embargo="5m", with
emit_clocks=True, and prints the clocks of five rows, counting from 0.
Each row is scored from the rows at least five minutes older, the one
exactly five minutes older included, so the first five rows have learned
nothing.
It reads df from Example data:
fwd = po.spec.ewridge("fwd", targets=["y"], features=["x0", "x1"],
clock="ts", gap_cap="1h", half_life="30m",
embargo="5m", # a label takes five minutes to arrive: learned five minutes later
emit_clocks=True) # scored_clock and learned_clock on every row
rows = po.ModelBank([fwd]).fit_predict(df).unnest("fwd")
for row in [0, 4, 5, 6, 300]:
r = rows.row(row, named=True)
learned = f"the rows up to {r['learned_clock']:%H:%M}" if r["learned_clock"] else "no row yet"
print(f"row {row}: scored at {r['scored_clock']:%H:%M}, having learned {learned}, weight_sum {r['weight_sum']:.2f}")
row 0: scored at 09:30, having learned no row yet, weight_sum 0.00
row 4: scored at 09:34, having learned no row yet, weight_sum 0.00
row 5: scored at 09:35, having learned the rows up to 09:30, weight_sum 1.00
row 6: scored at 09:36, having learned the rows up to 09:31, weight_sum 1.98
row 300: scored at 14:30, having learned the rows up to 14:25, weight_sum 43.74
| case | what embargo does |
|---|---|
| what the delay counts | the time that passed on the clock column, skipped rows included, and session_gap where a session restarts the clock. It is not the capped step the model decays by, so the release depends on the rows alone and survives any chunking |
| exactly at the delay | the row is learned. The time that passed is held exactly: in integer nanoseconds on a temporal clock, and by one subtraction of the two rows' values on a number clock, in integers on an integer clock. Rows 1 ms apart under embargo="2s" are each learned 2,000 rows on |
| with no clock column | one unit is one row of the group, a skipped row included: embargo=20 is twenty rows |
| at a break | nothing is released early. A break's own effect, such as a lag ring's clearing or session_shrink's blend, waits with the row after it, and runs when that row is learned. A reset drops the rows still waiting |
| at the end of the stream | the rows still waiting are not learned, unless the state is saved and another run resumes from it |
| in the state | the waiting rows are saved with it: one row's values for each row inside the delay, in each group |
| row by row | emit_clocks=True, on any model, writes scored_clock, the row's own clock, and learned_clock, the clock of the newest row the model had learned from when it was scored, both in the clock column's own type. With no clock they are the row's index in its group |
The delay as data
Use
po.stream.embargo
when the delay has to be visible in the frame, or when another engine fits
the model. It returns every row twice: a copy to score at t with weight
0, and a copy to learn from at t + delay.
Sort its input by the clock across all rows, as
lf.sort("t", maintain_order=True), before the call. Order within each
group is not enough, since the call merges the two copies by the clock
alone. It adds two columns: _online_role, which says which copy a row is,
and _online_role_weight, which you pass to the spec's weight.
The built-in delay agrees with the doubled stream to the bit on pred,
resid and weight_sum when each row's t + delay is another row's
clock. An evenly spaced clock with a delay of whole steps is such a case.
On an irregular clock they differ, since the doubled stream decays a label
from t + delay and the built-in delay learns it at the row that releases
it.
Every residual diagnostic differs in any case, from sigma and
zscore to the conformal band and drift. embargo takes in the residual
of the prediction the row was scored with. The doubled stream's learning
copy forms its residual at t + delay, from a model that has since learned
every row before it.
This code doubles lf, whose clock t moves a row at a time, and holds
the scored copies to the built-in delay of the same five units.
It reads lf from Example data:
doubled = po.stream.embargo(lf, clock="t", delay=5.0) # every row twice, in clock order
scored = doubled.online.fit_predict(
[po.spec.ewridge("m", targets=["y"], features=["x0"], clock="t", gap_cap=10.0,
half_life=50.0, weight="_online_role_weight")] # 0 on the copies to score, 1 on the rest
).filter(pl.col("_online_role") == "predict").collect() # keep the scored copies
# The built-in delay on the same rows: t is a row apart, so each t + 5 is another row's clock.
built_in = lf.online.fit_predict(
[po.spec.ewridge("m", targets=["y"], features=["x0"], clock="t", gap_cap=10.0, half_life=50.0, embargo=5.0)]
).collect()
for field in ["pred_y", "weight_sum"]:
a, b = scored["m"].struct.field(field), built_in["m"].struct.field(field)
print(f"{field}: {a.count()} rows, {(a == b).sum()} equal to the bit")
pred_y: 393 rows, 393 equal to the bit
weight_sum: 400 rows, 400 equal to the bit
Nulls, and three ways to hold a row back
A null skips an update but never stops the clock. NaN, ±inf and any
magnitude above 1e100 count as null, so sentinel values never reach a
model. A null target is one of three ways to keep a row from teaching a
model, beside a weight of 0 and predict:
| the row is scored | the clock advances | the fit moves | weight_sum |
use it to | |
|---|---|---|---|---|---|
| a null target | yes | yes | not that target's, except under target_gaps="pairwise" |
counts the row. In a regression model the target's own weight only decays, so a long stretch of nulls can take that target below its min_weight |
leave a label out |
weight 0 |
yes | yes | no | keeps decaying, so a long stretch can fall below min_weight |
keep a row's place in the stream |
predict |
yes | no | no | frozen | serve (Serving without learning) |
This code holds row 100 of df back from a model in each of the three
ways, beside a run that learns it. For each it prints whether row 100 was
scored, the clock the model stands at after it, whether the fit moved, and
weight_sum on the row after it, on df from Example data:
# One spec, and four copies of the stream that differ only at row 100.
held = po.spec.ewridge("held", targets=["y"], features=["x0", "x1"], clock="t", half_life=100.0, gap_cap=300.0,
weight="w", max_rows_between_solves=1) # a solve on every row, so each fit read is current
rows = df.with_columns(w=pl.lit(1.0))
row_100 = pl.int_range(pl.len()) == 100
ways = {
"learned": rows,
"a null target": rows.with_columns(y=pl.when(row_100).then(None).otherwise(pl.col("y"))),
"weight 0": rows.with_columns(w=pl.when(row_100).then(0.0).otherwise(pl.col("w"))),
"predict": rows,
}
# A bank that has seen rows 0 to 99, to tell whether the fit moved.
before = po.ModelBank([held])
before.fit_predict(rows.head(100)) # rows 0 to 99
print(f"{'':<14} {'scored':>6} {'clock':>6} {'fit moved':>9} {'weight_sum':>10}")
# Each way: rows 0 to 99, then row 100, then row 101.
for way, frame in ways.items():
bank = po.ModelBank([held])
bank.fit_predict(frame.head(100))
if way == "predict":
scored = bank.predict(frame.slice(100, 1)) # scores row 100 and learns nothing
else:
scored = bank.fit_predict(frame.slice(100, 1))
# What row 100 did: the clock, the fit, and the next row's weight_sum.
clock = bank.groups()["clock_last"][0]
moved = bank.coef()["coef"].to_list() != before.coef()["coef"].to_list()
after = bank.fit_predict(frame.slice(101, 1))["held"].struct.field("weight_sum")[0]
is_scored = scored["held"].struct.field("pred_y")[0] is not None
print(f"{way:<14} {'yes' if is_scored else 'no':>6} {clock:>6g} {'yes' if moved else 'no':>9} {after:>10.2f}")
scored clock fit moved weight_sum
learned yes 100 yes 72.89
a null target yes 100 no 72.89
weight 0 yes 100 no 71.89
predict yes 99 no 72.39
A null in any feature, or in the weight, skips the row: its outputs are null, and no update happens.
A null in one target skips only that target's update. The row still
writes that target's pred, and leaves its resid null. In rls, whose
targets share one factor, the row updates none of them. In ewridge and
lasso, whose targets share running sums, the row still reaches the other
targets' sums. Under the default target_gaps="own_rows", the missing
target keeps its own copy of the sums from that row on, and over the rows
it lacks the copy only ages, so its fit holds still. Under "pairwise" one
set of sums learns every row, so a null target's
coefficients drift with the feature noise.
Preparing a stream
Two tools turn a stream's raw rows into the columns a model reads. Each
reads the rows in one pass, returns a DataFrame for a DataFrame and a
query for a query, and can save its state with save_state= for the next
run:
| tool | what it makes | where it runs | subsection |
|---|---|---|---|
the window operators of po.ops |
exponentially weighted means, variances, sums and rates over a window of the clock, looking back or ahead | in po.stream.with_windows, before the bank, as columns; or in a spec's targets, as a target the bank computes |
Making features and targets from windows |
po.stream.refresh_time |
one grid for series that tick at their own times | before the bank | Series that tick at their own times |
Making features and targets from windows
A window operator of po.ops returns a Polars expression whose value on
each row is computed over the rows near it on the clock. A window that
looks back from the row makes a feature, and one that looks ahead, under
reverse=True, makes a target. po.ewm_mean in either direction is a
time-weighted mean, as Polars' ewm_mean_by computes one, and the sums and rates divide
into weighted means such as a VWAP (volume-weighted average price). Each
operator keeps running sums, so it does the same work on a row however long
its window.
docs/WINDOWS.md shows each part of this with a recipe. Its recipes hold the nine operators to Polars' own, and show which ends of a window are in and what a gap does to a window. Others fit a window target beside its column form, make features in units of their spread, and resume a window run.
To add windows as columns, call po.stream.with_windows before the
bank. It takes formulas as with_columns does, computes them over the
stream in one pass, and returns each row once every window over it has
closed. Here the feature mid_trend is the mid against its time-weighted
mean over the last two minutes. The target fwd_edge is the next minute's
VWAP of the trades, less the mid now.
This code uses trades.parquet from Example data:
# The next minute's VWAP, from two sums over the same window.
notional = pl.col("price") * pl.col("quantity")
next_minute = dict(half_life="10s", window_size="1m", reverse=True) # weights halve every 10 s from the row; the window ends a minute on
windowed = pl.scan_parquet("trades.parquet").online.with_windows(
mid_trend=pl.col("mid") - po.ewm_mean("mid", half_life="30s", window_size="2m"), # a feature: looks back
fwd_edge=po.ewm_sum(notional, **next_minute) / po.ewm_sum("quantity", **next_minute) - pl.col("mid"), # a target: looks ahead
clock="ts", gap_cap="5m", group="symbol", # the clock, the longest step it ages by, and one set of windows per symbol
).collect()
# Five rows, with both windows beside the mid.
print("time symbol mid mid_trend fwd_edge")
for r in windowed.slice(200, 5).iter_rows(named=True):
print(f"{r['ts']:%H:%M:%S} {r['symbol']} {r['mid']:.3f} {r['mid_trend']:+.4f} {r['fwd_edge']:+.4f}")
time symbol mid mid_trend fwd_edge
09:31:48 AAA 100.107 +0.0536 -0.1181
09:31:49 BBB 100.105 +0.0504 -0.0299
09:31:50 BBB 100.094 +0.0395 -0.0196
09:31:50 BBB 100.096 +0.0403 -0.0253
09:31:50 BBB 100.083 +0.0275 -0.0127
To have a model learn a target that looks ahead, put its expression in
the spec's targets, with an embargo of at least its window. The bank
then computes the window itself, scores each row where it sits, and learns
from the row once its window has closed. A target's value is a minute in
the future when its row arrives, so a model that learned it at once would
score that minute's rows in sample.
Series that tick at their own times
Series observed at different instants cannot be compared row by row: a fine
common grid pushes their correlation toward zero (the Epps effect), and
filling values forward invents observations. So put them on one grid with
po.stream.refresh_time
before the bank. Its grid, which Barndorff-Nielsen, Hansen, Lunde and
Shephard defined, has a point at each refresh time: the first instant by
which every series has ticked at least once since the previous point.
Nothing is interpolated.
Give it the ticks in long form, one row per tick. Three series on a clock
t:
ticks = pl.DataFrame({
"symbol": ["AAA", "BBB", "AAA", "CCC", "BBB", "AAA", "AAA", "CCC"], # which series ticked
"t": [0.4, 0.9, 1.3, 1.6, 2.2, 2.5, 2.8, 3.1], # when, in clock order
"px": [100.0, 20.0, 100.2, 50.0, 20.1, 100.1, 100.4, 49.9], # what it observed
})
grid = po.stream.refresh_time(ticks, series="symbol", names=["AAA", "BBB", "CCC"],
clock="t", value="px")
| time_refresh | AAA_value | BBB_value | CCC_value | n_obs_AAA | n_obs_BBB | n_obs_CCC | retained_fraction |
|---|---|---|---|---|---|---|---|
| 1.6 | 100.2 | 20.0 | 50.0 | 2 | 1 | 1 | 0.75 |
| 3.1 | 100.4 | 20.1 | 49.9 | 2 | 1 | 1 | 0.75 |
The first refresh time is 1.6, when CCC ticks for the first time. AAA's value there is 100.2, from its tick at 1.3, and its tick at 0.4 is dropped. After 1.6, BBB ticks at 2.2 and AAA twice, so CCC's tick at 3.1 completes the second point.
Each column of the grid holds:
| column | what it holds |
|---|---|
time_refresh |
the refresh time: the clock of the tick that completed the point |
AAA_value, ... |
each series' last observed value at that time. The output looks synchronous and is not: each value is up to one of its own inter-tick intervals old |
n_obs_AAA, ... |
that series' ticks since the previous point, of which the grid kept the last |
retained_fraction |
the share of those ticks the grid kept. Read it before trusting a correlation |
The grid runs at the pace of the slowest series, so a fast series loses
most of its ticks. The series holding the grid up, whose tick completes
each point, keeps an n_obs near 1. When one series is much slower than
the others, pass pairs=True for an independent two-series grid per pair,
which keeps far more ticks.
Sort the ticks by the clock within each group before the call: a
step back stops the run with an error naming the row.
To build the grid in two runs, give the first save_state= and the
second load_state=, with the same names and pairs. The second run
goes on from the first one's last tick, even part-way through an interval.
On the eight ticks of the first grid:
names = ["AAA", "BBB", "CCC"]
first = po.stream.refresh_time(ticks.head(5), series="symbol", names=names, clock="t", value="px",
save_state="refresh.state") # the ticks to 2.2: part-way through the second interval
rest = po.stream.refresh_time(ticks.tail(3), series="symbol", names=names, clock="t", value="px",
load_state="refresh.state") # the other three, from where the first run stopped
assert pl.concat([first, rest]).equals( # the two grids together are the grid of one run
po.stream.refresh_time(ticks, series="symbol", names=names, clock="t", value="px")
)
To fit models on the grid, compute each series' change between points
with Polars' diff(), and give the specs the grid's clock, time_refresh.
Given a query, refresh_time returns one, so
lf.online.fit_predict(specs) chained
after it runs the grid and the models in one pass. This code plants a
correlation: three prices share one move and add one of their own, as
large, so their changes correlate at 0.5. Each is observed at times of its
own, about once per clock unit. ew_cov reads the correlations on the
grid, and ewridge regresses one series' change on the other two's, whose
true coefficients are 1/3 each. It then reads the same ticks on a
grid every 0.1 clock units, each series' last price carried forward:
# Three prices that share one move, each observed at times of its own.
rng = np.random.default_rng(1)
steps = 200_000 # a hidden price path, 0.01 clock units a step
shared = np.cumsum(rng.normal(0.0, 0.1, steps)) # the move the three series share
# One frame per series, then one stream in clock order.
names = ["AAA", "BBB", "CCC"]
frames = []
for s in names:
price = shared + np.cumsum(rng.normal(0.0, 0.1, steps)) # plus a move of its own, as large
t = np.cumsum(rng.exponential(1.0, 2_100))
t = t[t < 2_000] # about one tick per clock unit, at times of its own
frames.append(pl.DataFrame({"symbol": s, "t": t, "px": price[(t * 100).astype(int)]}))
pl.concat(frames).sort("t").write_parquet("own_times.parquet") # one stream, in clock order
# The refresh grid and the models on it, in one query.
grid = po.stream.refresh_time(pl.scan_parquet("own_times.parquet"), series="symbol", names=names,
clock="t", value="px") # a query, since it was given one
returns = grid.select("time_refresh", *(pl.col(f"{s}_value").diff().alias(s) for s in names)) # each series' change since the last point
specs = [
po.spec.ew_cov("comove", features=names, stats=["corr"],
clock="time_refresh", gap_cap=10.0, half_life=float("inf")),
po.spec.ewridge("beta", targets=["AAA"], features=["BBB", "CCC"],
clock="time_refresh", gap_cap=10.0, half_life=float("inf")),
]
out = returns.online.fit_predict(specs).collect() # the first point has no change, so both models skip it
# The correlations and the betas on the last point.
last = out.row(-1, named=True)
print(f"{out.height} refresh times from {pl.read_parquet('own_times.parquet').height} ticks")
print("on the refresh grid: corr " + ", ".join(f"{last['comove'][f'corr_{a}_{b}']:.2f}" for a, b in [("AAA", "BBB"), ("AAA", "CCC"), ("BBB", "CCC")])
+ f"; AAA on BBB {last['beta']['coef'][1]:.2f}, on CCC {last['beta']['coef'][2]:.2f}")
# The same ticks on a grid every 0.1 clock units, for comparison.
ticks_by_series = pl.read_parquet("own_times.parquet").partition_by("symbol", as_dict=True)
fine = pl.DataFrame({"t": np.arange(0.0, 2_000.0, 0.1)})
for s in names: # each series' last price at each point of the fine grid
fine = fine.join_asof(ticks_by_series[(s,)].select("t", pl.col("px").alias(s)), on="t")
moves = fine.select(pl.col(names).diff()).drop_nulls()
print(f"on a grid every 0.1: corr {moves.corr()['AAA'][1]:.2f}, {moves.corr()['AAA'][2]:.2f}, {moves.corr()['BBB'][2]:.2f}")
1064 refresh times from 5835 ticks
on the refresh grid: corr 0.34, 0.32, 0.36; AAA on BBB 0.27, on CCC 0.23
on a grid every 0.1: corr 0.03, 0.02, 0.03
The fine grid finds almost no correlation, since in most of its steps one series moved and the others did not: the Epps effect. The refresh grid finds about two thirds of the planted 0.5. What it misses comes from each value's age at a refresh time, which is up to one of its own series' intervals between ticks.
Running a bank
Run a bank in one of three ways. Each gives the same numbers, whatever the chunking of the rows. From a bank object, the output can leave as Arrow, for a consumer that is not Polars:
| way | the call | use it for | where |
|---|---|---|---|
| inside a Polars query | lf.online.fit_predict(specs) |
ordinary Polars steps chained after the bank | As a query |
| in your own Python loop | bank.fit_predict(chunk) |
reading the bank between chunks | In a loop |
| from a standalone command line, with no live Python at all | polars-online |
a scheduled job, or a deployment with no Python | Running a bank as a job, without Python |
| output as Arrow, from a bank object | fit_predict_arrow, predict_arrow |
a consumer that is not Polars | Output as Arrow |
As a query: lf.online.fit_predict
lf.online.fit_predict(specs)
adds a model bank to a Polars query, a LazyFrame, which runs only when
you ask for its result. df.online.fit_predict(specs) does the same for a
DataFrame in memory. When the query runs, it feeds its rows to a bank
that starts with nothing learned, chunk_size rows at a time (100,000 by
default). Run it with collect() for one frame, sink_parquet() to write
a file without holding the result in memory, or collect_batches() for one
chunk at a time.
The chunk size trades memory for overhead, and never changes the
numbers (Row order and the two guarantees).
Memory is the bank's state plus one chunk, however long the files. Here the
query runs the same spec over ticks/*.parquet in one chunk, and over
ticks.parquet one row at a time, and the two agree to the bit.
df's y is 1 + 2·x0 − x1 plus noise, so the fit of each stock finds
those coefficients, and 0 for x2.
This code uses the files ticks.parquet and ticks/*.parquet from Example data:
spec = po.spec.ewridge( # a ridge regression of y for each stock, at two ridge values
"ridge",
targets=["y"], features=["x0", "x1", "x2"],
clock="t", half_life=600.0, gap_cap=300.0,
group="stock_id", ridge=[1e-6, 0.1], standardize=True,
)
# Run the query to a file, then read what it wrote.
(
pl.scan_parquet("ticks/*.parquet") # a query over the files; nothing is read yet
.online.fit_predict([spec], chunk_size=100_000) # a bank with nothing learned yet; every run starts from the same place
.filter(pl.col("ridge").struct.field("weight_sum") > 50) # after the bank: filters what comes out, never what the bank learns from
.select("t", "stock_id", "ridge") # Polars reads only these columns, and those the specs read, from the files
.sink_parquet("fitted.parquet") # runs the query; memory is state + one chunk, however long the files
)
fitted = pl.read_parquet("fitted.parquet")
print(f"{fitted.height} rows past the filter; the record's fields: {fitted['ridge'].struct.fields}")
# The same spec, one row per chunk.
one_row_chunks = pl.scan_parquet("ticks.parquet").online.fit_predict([spec], chunk_size=1).collect()
same = one_row_chunks.filter(pl.col("ridge").struct.field("weight_sum") > 50)["ridge"].struct.field("pred_y__r0.1")
print(f"largest difference, chunks of 100,000 against chunks of 1: {(fitted['ridge'].struct.field('pred_y__r0.1') - same).abs().max()}")
# One stock's fit on its last row.
last = fitted.filter(pl.col("stock_id") == "b3").row(-1, named=True)
intercept, b0, b1, b2 = last["ridge"]["coef"][:4] # the fit at the first ridge value; the second's follow
print(f"b3's coefficients on its last row: intercept {intercept:.2f}, x0 {b0:.2f}, x1 {b1:.2f}, x2 {b2:.2f}")
172 rows past the filter; the record's fields: ['pred_y__r0.000001', 'resid_y__r0.000001', 'pred_y__r0.1', 'resid_y__r0.1', 'weight_sum', 'settled_frac', 'withheld_reason', 'coef', 'support_coef']
largest difference, chunks of 100,000 against chunks of 1: 0.0
b3's coefficients on its last row: intercept 1.04, x0 1.99, x1 -0.97, x2 -0.13
"ridge" is one column whose value in each row is a record with named
fields (Output field names lists them). weight_sum
is the weight a fit rests on, so the filter keeps each stock's rows once
its fit has seen about 50 rows' worth. lf.online.fit_predict([spec]).head(5)
feeds the bank the first 5 rows and no more.
Filter after the bank, not before it. A filter after the bank keeps
the query running a chunk at a time. One before the bank makes Polars hold
several blocks of each parquet file per thread: 3.1 GB at 12M rows,
against 1.8 GB for the same filter after, which is what the query takes
with no filter at all (Memory says how this was measured).
To keep rows from teaching the model, give them weight 0 before the
bank, in place of a filter: make the column with with_columns, and name
it in the spec's weight. The rows still come out scored, and the clock
advances through them, so no gap opens where they were. Here only venue
X's 200 rows teach the model. Under half_life=float("inf") no row's
weight decays, so weight_sum counts the rows learned before each row.
This code uses ticks.parquet from Example data:
learn_from_x = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"],
clock="t", half_life=float("inf"), gap_cap=300.0, weight="w")
fitted = (
pl.scan_parquet("ticks.parquet")
.with_columns(pl.when(pl.col("venue") == "X").then(1.0).otherwise(0.0).alias("w")) # learn from venue X only
.online.fit_predict([learn_from_x])
.collect()
)
# How many rows were scored, and how much weight was learned.
scored = fitted["ridge"].struct.field("pred_y").is_not_null().sum()
print(f"rows scored: {scored} of {fitted.height}; weight_sum on the last row: {fitted['ridge'].struct.field('weight_sum')[-1]:.0f}")
rows scored: 395 of 400; weight_sum on the last row: 200
lf.online.fit_predict checks each spec against the query's columns when
it is called, so a missing or non-numeric column raises ValueError before
any row is read. The bank refuses what only the values reveal, such as a
null clock or a step back, while the query runs. That error arrives as
polars.exceptions.ComputeError under py-polars 1.x, and as the bank's own
ValueError under 2.0, since how a query wraps an error is Polars' to
change. Either way its message names the row and the way out.
This code uses ticks.parquet from Example data:
misspelt = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x9"], half_life=100.0)
try:
pl.scan_parquet("ticks.parquet").online.fit_predict([misspelt]) # refused here: the query never runs
except ValueError as e:
print(e)
spec "ridge": feature column "x9" not found; the input has columns ["t", "ts", "x0", "x1", "x2", "signal_a", "signal_b", "y", "ret", "stock_id", "group", "session", "venue"]
With a type checker, call po.fit_predict,
po.predict
and po.unnest
in place of the .online methods. The package adds those methods to
Polars' frames when it is imported, so a type checker cannot see them. Each
function takes the frame first.
This code uses lf and later from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
fitted = po.fit_predict(lf, [spec], save_state="ridge.state") # lf.online.fit_predict([spec], save_state="ridge.state")
columns = po.unnest(fitted, [spec]).collect() # fitted.online.unnest([spec]); running it saves the state
scored = po.predict(later, "ridge.state").collect() # later.online.predict("ridge.state")
In a loop: ModelBank
To feed a bank from your own Python loop, build a
ModelBank
from the specs, pass it each chunk with bank.fit_predict(chunk), and ask
it what it holds between chunks (What a bank holds).
bank.fit_predict_batches(query, chunk_size=...) runs that loop for you,
reading a query or a DataFrame in chunks of the size given, and yields
each chunk's output. Here the bank reads 100 rows at a time, and stock
b0's coefficient on x0 settles near the 2 that df plants.
This code uses ticks.parquet from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"],
clock="t", half_life=600.0, gap_cap=300.0, group="stock_id")
bank = po.ModelBank([spec]) # nothing learned yet
for out in bank.fit_predict_batches(pl.scan_parquet("ticks.parquet"), chunk_size=100):
# out: the chunk's columns, plus one column per spec. Between chunks, the bank answers questions:
x0 = bank.coef().filter(pl.col("group") == "b0", pl.col("term") == "x0")["coef"][0]
print(f"after {bank.rows_fed} rows: b0's coefficient on x0 is {x0:.3f}")
bank.save("bank.state") # written whole or not at all: a temporary file, then a rename
# With chunks of your own: for chunk in pl.scan_parquet("ticks.parquet").collect_batches(): bank.fit_predict(chunk)
# A run whose product is the state: po.ModelBank([spec]).fit(query) learns from every row and keeps no output.
after 100 rows: b0's coefficient on x0 is 2.105
after 200 rows: b0's coefficient on x0 is 2.055
after 300 rows: b0's coefficient on x0 is 2.087
after 400 rows: b0's coefficient on x0 is 2.023
Feed a bank from one thread at a time, though predict, which learns
nothing, may run from any number of threads at once. A bank reads one
ordered stream, so a call that finds it busy on another thread raises
RuntimeError: wait for that call to return, or give each thread its own
bank.
A bank keeps every group until you drop it, so in a long-running bank,
drop the quiet ones with bank.drop_groups. bank.groups() gives each
group's clock_last in the clock column's own dtype: a Datetime in its
unit and zone, an integer clock in its own width, a float clock as a
Float64. Compare it with the current time of the same kind. A dropped
group starts over, with nothing learned, if it appears again. In df the
four stocks take turns, so each last ticked at a different t.
This code uses df and ticks.parquet from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"],
clock="t", half_life=600.0, gap_cap=300.0, group="stock_id")
bank = po.ModelBank([spec])
bank.fit(pl.scan_parquet("ticks.parquet"))
for group, rows, clock_last in bank.groups().select("group", "rows_fed", "clock_last").iter_rows():
print(f"{group}: {rows} rows, the last at t = {clock_last:.0f}")
now = df["t"].max() # the current time on the clock t; on a temporal clock, a datetime
stale = bank.groups().filter(pl.col("clock_last") < now - 2) # quiet for more than 2 units of t
print(f"dropped {bank.drop_groups(stale['group'])}; left: {bank.groups()['group'].to_list()}")
b0: 100 rows, the last at t = 396
b1: 100 rows, the last at t = 397
b2: 100 rows, the last at t = 398
b3: 100 rows, the last at t = 399
dropped 1; left: ['b1', 'b2', 'b3']
Running a bank as a job, without Python
For a scheduled job, or a deployment with no Python, run the same bank from
a file to a file with the standalone polars-online command line, a binary on
each GitHub release. docs/RUNNER.md has every key and
flag, with a recipe for each kind of run: a check before the run,
resuming, scoring, a run that writes only its state, and the groups that
close.
polars-online reads the input, the output and the specs from a TOML file. Each [[specs]] table takes the keywords every spec shares, and
its [specs.model] table names the model and takes the model's own. This
bank.toml is the first example's spec of As a
query:
input = "ticks.parquet" # read a chunk at a time; the extension names the format
output = "fitted.parquet" # the input's columns, plus one column per spec
save_state = "bank.state" # the state after the last row; load_state resumes from one
[[specs]] # one table per spec
name = "ridge"
targets = ["y"]
features = ["x0", "x1", "x2"]
clock = "t"
half_life = 600.0
gap_cap = 300.0
group = "stock_id"
[specs.model] # the model, and its own keywords
type = "ewridge"
ridge = [1e-6, 0.1]
standardize = true
Save it beside the example data's ticks.parquet: a relative path in the
file is read from the file's own directory. The predictions, residuals and
weights the run writes are the bank's in Python, bit for bit:
polars-online --config bank.toml
wrote 400 rows (1 chunk) to fitted.parquet
saved state to bank.state
On a terminal, a progress line above the summary counts the rows as the
run goes, and --quiet turns it off. After wrote, a run prints a line
for each spec that ended with groups not ready, and nothing for the
others. Such a line reads
spec "ridge": 2 groups whose last row was withheld (below_min_weight 2).
The same file builds the same bank in Python, as
po.ModelBank(tomllib.load(f)["specs"]) with the file opened in binary
mode. tomllib reads each [[specs]] table as the dict a builder
returns, and Specs in TOML in the runner
guide checks both the spec and the predictions against the command line's.
Output as Arrow
To hand a bank's output to a consumer that is not Polars, call
bank.fit_predict_arrow(df) or bank.predict_arrow(df) in place of
fit_predict or predict. Each returns one ArrowStruct per spec, which
holds that spec's output through the Arrow PyCapsule interface, an Arrow
specification, so any reader of __arrow_c_array__ takes it directly. Its
values are fit_predict's, field for field and null for null. The input
still goes in as a Polars DataFrame. fit_predict itself hands its
output over as a Polars Series, on py-polars' private methods, so this
package measures a Polars range and promises none (Which interfaces carry
a promise).
Read each struct once. Exporting hands its buffers to the consumer, so
a second read raises ValueError.
This code uses df and today from Example data:
# One spec's output as Arrow, read once as a Series.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
bank = po.ModelBank([spec])
structs = bank.fit_predict_arrow(df) # one per spec, as Arrow
ridge = pl.Series(structs[0]) # Polars as the reader here: any reader of __arrow_c_array__ takes it
print(f"{len(structs)} {type(structs[0]).__name__}; read as a Series: {ridge.name}, {ridge.len()} rows")
print(f"the same as fit_predict's column: {ridge.equals(po.ModelBank([spec]).fit_predict(df)['ridge'])}")
# A second read of the same struct is refused.
try:
pl.Series(structs[0]) # the second read
except ValueError as e:
print(e)
scored_structs = bank.predict_arrow(today) # the same for predict, on today: the rows after df's
1 ArrowStruct; read as a Series: ridge, 400 rows
the same as fit_predict's column: True
this ArrowStruct has already been exported: the Arrow PyCapsule interface hands its buffers to the consumer, so it can be read once
Here is how two consumers take a struct, as measured:
| consumer | measured on | takes |
|---|---|---|
| pyarrow | pyarrow 25.0.1 | a struct as it is: pa.array(s) and pa.table(s) each take one, values and types unchanged |
| DuckDB | duckdb 1.5.5 | pl.Series(s). It refuses an ArrowStruct, because __arrow_c_stream__ is the method it looks for. A spec's output arrives table-shaped, one column per field |
Saving, loading and serving
Save a bank's state (everything it has learned, for every spec and group) to one file, and load it back to keep learning or to score new rows without learning. Or export it as JSON. docs/STATE-WORKFLOW.md says what each step of the workflow guarantees and what a saved bank carries, each with a recipe. The three uses:
| use | the call | what holds |
|---|---|---|
| resume | po.ModelBank.load(path), or load_state= on a query |
the next row is learned as one run would learn it |
| serve | bank.predict(frame), or lf.online.predict(path) |
each row is scored as fit_predict would score it as the next row, and the state never moves |
| read elsewhere | bank.write_json() |
everything save writes, as JSON |
This code fits df, saves the bank, and loads it back. Served, the first
row of today gets the prediction one run over df and today gives it.
Resumed on input that overlaps the state, through skip_learned, the bank
learns each row once and gives one run's predictions on every row.
This code uses df and today from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
one_run = po.ModelBank([spec]).fit_predict(pl.concat([df, today])) # df and today as one stream, for comparison
# Fit df, save the bank, and load it back.
bank = po.ModelBank([spec])
bank.fit_predict(df) # the stream so far
bank.save("bank.state")
loaded = po.ModelBank.load("bank.state", specs=[spec]) # specs= checks the file holds this model, not another
# Serve today from the loaded bank.
served = loaded.predict(today) # score today's rows, learning nothing
pred = lambda out: out["ridge"].struct.field("pred_y")
print(f"served, today's first row: {pred(served)[0]:.6f}; one run: {pred(one_run)[400]:.6f}")
# Resume on input that overlaps what the state has learned.
rerun = pl.concat([df.tail(50), today]) # input that overlaps the state by 50 rows
kept = loaded.skip_learned(rerun) # the rows the state has not learned
resumed = loaded.fit_predict(kept)
print(f"kept {kept.height} of {rerun.height} rows; largest difference from one run: {(pred(resumed) - pred(one_run.tail(400))).abs().max()}")
served, today's first row: 1.603543; one run: 1.603543
kept 400 of 450 rows; largest difference from one run: 0.0
A state saved by 0.13.0 or an earlier release does not load in the
releases after it: refit the bank from its input. From 1.0, a state a 1.x
build saved loads in every later 1.x (This package's own
versioning). with_windows keeps a state
of its own (Saving and resuming a window
run).
Save and load
Each way of running a bank saves and loads the same file, written whole or not at all, and the same to the byte whichever way wrote it. Each writes the file at its own moment:
| run by | saves with | loads with | the file is written |
|---|---|---|---|
| a bank object | bank.save(path) |
po.ModelBank.load |
when bank.save is called |
| a query | save_state= |
load_state= |
when the run reaches its last row, with the same bytes a ModelBank would write |
| the command line | --save-state |
--load-state |
only after its output is committed |
This code uses df and ticks.parquet from Example data:
from pathlib import Path
# The same state, saved from a bank and from a query.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
bank = po.ModelBank([spec])
bank.fit_predict(df)
bank.save("from_bank.state") # from a bank object
(pl.scan_parquet("ticks.parquet") # the same rows, from a query
.online.fit_predict([spec], save_state="from_query.state")
.sink_parquet("fitted.parquet"))
print(f"the same bytes: {Path('from_bank.state').read_bytes() == Path('from_query.state').read_bytes()}")
blob = bank.save_bytes() # the same file as bytes, for a checkpoint that is not a file
print(f"as bytes too: {blob == Path('from_bank.state').read_bytes()}")
the same bytes: True
as bytes too: True
A query loads and saves with load_state= and save_state=, and the
two may name one file. This code uses lf and later from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
lf.online.fit_predict([spec], save_state="bank.state").sink_parquet("fitted.parquet") # fit, then save at the last row
served = later.online.predict("bank.state").collect() # serve from the file
later.online.fit_predict(load_state="bank.state", save_state="bank.state").sink_parquet("more.parquet") # the next rows: continue, save again
bank = po.ModelBank.load_bytes(po.ModelBank.load("bank.state").save_bytes(), specs=[spec]) # bytes load as a file does
To resume on input that overlaps the state, pass it through
bank.skip_learned(frame) before the bank. Without it, the overlap steps
every group's clock back, which the bank refuses unless
restart_after_step_back reads it as a new start. skip_learned keeps the
rows after each group's last clock (a row at that clock counts as learned)
and every row of a group the bank has not seen. It keeps a LazyFrame
lazy. For a bank whose specs all count rows, it raises ValueError; drop
the learned rows with frame.slice(bank.rows_fed) instead, when the
input starts where the saved one did.
A query abandoned before its last row, or ended by an error inside the
bank, leaves the save_state file as it was. An error in a step after
the bank, such as a full disk under sink_parquet, does not stop the bank
on Polars 1.x. The state is then written although the output is missing
(docs/STATE-WORKFLOW.md
has the measurements). To write the state only with the output, use the
command line, which saves it after committing the output. In a query, give
each batch of data its own dated save_state file, so a rerun loads the
previous batch's state.
This code uses lf and later from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
(lf.online.fit_predict([spec], save_state="bank-2024-01-02.state") # the first batch, and a state file of its own
.sink_parquet("fitted-2024-01-02.parquet"))
(later.online.fit_predict(load_state="bank-2024-01-02.state", # the next batch loads the one before ...
save_state="bank-2024-01-03.state") # ... and writes a file of its own,
.sink_parquet("fitted-2024-01-03.parquet")) # so a rerun of it loads bank-2024-01-02 again
po.ModelBank.load raises an error that names what is wrong with the
file:
| the file | raises | what to do |
|---|---|---|
| does not exist yet | FileNotFoundError |
on a first run, build the bank from its specs with po.ModelBank(specs) |
is not a bank, or holds different specs from the specs= given |
ValueError |
check the path, and pass the specs the state was saved with |
| holds a state that contradicts its own spec | ValueError |
refit the bank from its input |
| was written under a newer state schema or file format than this build reads | ValueError |
load it with the version of polars-online that wrote it |
| was saved by 0.13.0 or an earlier release | ValueError, naming the range |
refit the bank from its input. po.schema_version() gives the installed version's schema |
This code uses df and ticks.parquet from Example data:
# A saved bank, and a spec that does not match it.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
bank = po.ModelBank([spec])
bank.fit_predict(df)
bank.save("bank.state")
other = po.spec.ewridge("ridge", targets=["y"], features=["x0"], clock="t", half_life=600.0, gap_cap=300.0)
# Three loads that fail: no file, the wrong spec, not a state file.
for path, specs in [("nowhere.state", [spec]), ("bank.state", [other]), ("ticks.parquet", None)]:
try:
po.ModelBank.load(path, specs=specs)
except (FileNotFoundError, ValueError) as e:
print(f"{type(e).__name__}: {e}")
FileNotFoundError: [Errno 2] No such file or directory: 'nowhere.state'
ValueError: saved specs do not match the bank's specs: spec "ridge" (1 of 1) differs at features: ["x0","x1"] in the file, ["x0"] in the bank; refusing to load
ValueError: not a polars-online bank state file (invalid type: integer `80`, expected struct BankHeader)
Serving without learning
To score rows without learning from them, call bank.predict(frame) on a
bank object, or lf.online.predict(...) in a query. Every row is scored
from the same state, which predict never moves. Since it skips each
row's update, ewridge scores 1.8 times as fast as it learns at 5
features, and 2.9 times at 20 (docs/PERFORMANCE.md
§9). A query takes the bank object, read as it stands when the query runs,
or a path, read when the query is built.
This code uses df, today and later from Example data:
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", half_life=600.0, gap_cap=300.0)
bank = po.ModelBank([spec])
bank.fit_predict(df)
bank.save("bank.state")
# Three ways to score the rows after df's.
scored = bank.predict(today) # every row scored from the same state: nothing moves
served = later.online.predict(bank).collect() # a bank, as it stands when the query runs
from_file = later.online.predict("bank.state").collect() # a path, read when the query is built
# Every row saw the same state, and the three agree.
weights = scored["ridge"].struct.field("weight_sum")
print(f"weight_sum on every row: {weights.min():.4f} to {weights.max():.4f}; the three agree: "
f"{scored['ridge'].equals(served['ridge']) and served['ridge'].equals(from_file['ridge'])}")
weight_sum on every row: 320.4975 to 320.4975; the three agree: True
Row i of scored carries what fit_predict would have reported had it
been the next row of its group's stream: pred, weight_sum, sigma,
zscore, the selection and the metrics, field for field
(docs/DIAGNOSTICS.md). Two fields differ: drift never
fires, and coef is filled on each group's last accepted row only, since
the same coefficients score every row. How predict reads a row:
| the row | predict |
|---|---|
| its target column | may be absent; resid is then null |
| its weight column | is not read |
| of a group the bank has never seen | scores null |
| before the last clock its group learned | is scored against the state as it stands, with a clock step of 0: never refused, never a restart |
where session_gap="reset" or group_close="session" would restart its group |
scores null, with weight_sum 0 |
after rows an embargo still holds |
sees the state as it stands, since predict releases nothing: a held row whose delay has passed by the scored row's clock stays held until the next fit_predict learns it, so weight_sum and learned_clock are the state's on every row |
Reading a state without this library
To read a state from a program without this library, export it with
bank.write_json(), which returns the text, or write it to a file with
bank.write_json(path). Keep
the binary file for resuming, since load reads only that form. Every
export is read back and checked against the state before you get it, and
one that does not match raises ValueError.
A program reading the JSON must turn the strings "nan", "inf" and
"-inf" back into numbers. JSON has no literal for NaN or ±inf, and
an ordinary state can hold them, as a spec with half_life=inf (no
forgetting) does. The strings are the spelling a spec's half_life takes.
This code uses df from Example data:
import json
bank = po.ModelBank([po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], half_life=float("inf"))])
bank.fit_predict(df)
text = bank.write_json() # everything save() writes, as JSON
bank.write_json("bank.json") # the same, to a file
state = json.loads(text) # any JSON reader
half_life = state["specs"][0]["half_life"]
print(f"{state['magic']}: half_life {half_life!r}, as a number {float(half_life)}")
polars-online-bank: half_life 'inf', as a number inf
Reading the fit
Read a fit from the bank, with its methods, between chunks or after the run, or from the output, row by row. The bank reports each group as it stood after the last row it learned from.
| to read | call | subsection |
|---|---|---|
| what a bank holds and what it was fed | repr(bank), bank.groups(), bank.coef(), bank.summary(), bank.describe(), bank.last_row() |
What a bank holds |
| any field of the output, by name | po.spec.output_index, po.spec.coef_fields |
Output field names |
| the coefficients, at the end or row by row | bank.coef(), or coef_every=0 and .online.unnest |
Coefficients |
| fits over any subset of the rows and columns, from the running sums one pass leaves | bank.gram(spec), po.gram |
Fitting many features from one pass |
| a finished group's running sums | bank.closed_groups() |
One row per finished group |
what a correlation matrix says, from an array, a gram() dict or a closed row |
po.corr |
Reading a correlation matrix |
| what is wrong with the data | bank.check(), bank.audit() |
Checking your data's quality |
What a bank holds
A bank describes itself without its data, live or loaded from its file.
Here a spec with a group per stock learns from lf, and the bank it saves
is loaded back with nothing but the file.
This code uses lf from Example data:
# A bank fitted per stock, and saved.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1", "x2"], clock="t", gap_cap=300.0,
half_life=600.0, group="stock_id")
fitted = po.ModelBank([spec])
fitted.fit(lf) # learn from every row of the file, keep only the state
fitted.save("bank.state")
# Loaded back from the file alone, it describes itself.
bank = po.ModelBank.load("bank.state") # no specs=: the file is enough
print(repr(bank)) # its specs, its groups, the rows it was fed
print(bank.rows_fed, bank.solve_failures()["solve_failures"].to_list()) # rows fed; solves that needed help, per group
print(bank.output_fields["ridge"]) # the fields of the spec's column
for spec_name, group, rows, last in bank.groups().iter_rows(): # one row per (spec, group)
print(spec_name, group, rows, last)
ModelBank(['ridge'], groups=4, rows_fed=400)
400 [0, 0, 0, 0]
['pred_y', 'resid_y', 'weight_sum', 'settled_frac', 'withheld_reason', 'coef', 'support_coef']
ridge b0 100 396.0
ridge b1 100 397.0
ridge b2 100 398.0
ridge b3 100 399.0
bank.groups() gives each group's rows and its last clock, and
bank.solve_failures() counts the solves that needed jitter or kept the
previous fit: none here. bank.specs gives every spec back, as the dict
its builder made.
bank.specs returns a read-only copy. The bank runs the state it was
built from, so an editable list could drift from it and mislabel what
coef() reports. To change a spec, build and feed a new bank.
Four tables say how each fit is doing and what it was fed. Each has
a row per group, or per group and coefficient or column. Here each stock's
coefficients come back near df's truth: intercept 1, x0 2, x1 −1
and x2 0.
This code uses lf from Example data:
# A ridge per stock, learned from every row of lf.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1", "x2"], clock="t", gap_cap=300.0,
half_life=600.0, group="stock_id")
bank = po.ModelBank([spec])
bank.fit(lf)
# The four tables, each a frame.
betas = bank.coef() # one row per coefficient: spec, group, instance, ..., term, coef
fed = bank.summary("ridge") # one row per group: what it was fed, and its warm-up readings
cols = bank.describe("ridge") # one row per input column per group: role, count, nulls, moments
last = bank.last_row("ridge") # one row per group: the record fit_predict wrote on its last row
# Read three of them: the coefficients, what was fed, and b0's columns.
wide = betas.pivot("term", index="group", values="coef") # a row per group, a column per term
for row in wide.iter_rows(named=True):
print(row["group"], " ".join(f"{t} {row[t]:.2f}" for t in ["intercept", "x0", "x1", "x2"]))
for g, n_fed, n_learned, w in fed.select("group", "rows_fed", "rows_learned", "weight_sum").iter_rows():
print(f"{g}: fed {n_fed}, learned {n_learned}, weight_sum {w:.1f}")
for c, role, n, mean, std in cols.filter(group="b0").select("column", "role", "count", "mean", "std").iter_rows():
print(f"b0 {c} ({role}): {n} values, mean {mean:.2f}, std {std:.2f}")
b0 intercept 1.00 x0 2.02 x1 -1.02 x2 -0.01
b1 intercept 1.01 x0 2.02 x1 -1.06 x2 0.05
b2 intercept 0.93 x0 1.94 x1 -0.99 x2 -0.04
b3 intercept 1.04 x0 1.99 x1 -0.97 x2 -0.13
b0: fed 100, learned 100, weight_sum 100.0
b1: fed 100, learned 100, weight_sum 100.0
b2: fed 100, learned 100, weight_sum 100.0
b3: fed 100, learned 100, weight_sum 100.0
b0 x0 (feature): 100 values, mean -0.04, std 0.95
b0 x1 (feature): 100 values, mean 0.03, std 1.01
b0 x2 (feature): 100 values, mean -0.09, std 0.99
b0 y (target): 100 values, mean 0.88, std 2.12
last_row() matches fit_predict field for field: pred, sigma,
the metrics and the interval when the spec asks for them
(docs/DIAGNOSTICS.md), weight_sum, and coef when that row
carried it. Until a group learns from a row, its record is all nulls.
summary()'s counts and describe() cover every row fed to the group,
without decay. They say what the model was trained on, and predict
moves neither. So summary()'s weight_sum is 100 above, each stock's
100 rows of weight 1, where the fit's own weight_sum has decayed below
it. describe() counts a value as the models do: a null, a NaN, an
infinity or a magnitude beyond 1e100 counts in unusable_count, not in
count. The columns of summary(), each count an Int64:
| columns | what they hold |
|---|---|
rows_fed |
rows routed to the group |
rows_processed |
rows the model accepted |
rows_skipped |
rows the model skipped, because a feature or the weight was missing |
rows_learned |
rows that moved the fit: a weight above zero and a target present. Under an embargo a row is counted as it arrives; a window target's row, when it is released |
rows_zero_weight |
rows that advanced the clock and nothing else |
weight_sum |
the sum of the accepted rows' weights, undecayed: 1 a row without a weight column |
clock_min, clock_max, clock_last |
the clock's range and last value, in the clock column's own dtype |
session_changes, clock_backwards, resets |
what the clock rules met |
solve_failures |
solves that needed jitter or kept the previous fit, which bank.solve_failures() lists alone |
settled_frac, weight_sum_settled, error_inflation, support_coef_min and the feature it belongs to, n_coef |
the warm-up readings after the last row. weight_sum_settled is the weight the stream settles at, which says whether a min_weight can be met |
All four cover every spec by default, with spec as the first column,
so the tables of banks from different runs stack. coef(),
summary() and describe() have the same columns for every spec, so a
plain concat stacks them. The fields of last_row() differ between
specs, so stack it with "diagonal_relaxed", which fills a missing field
with nulls. Here two fits at different half-lives are saved to files of
their own and compared in one table.
This code uses lf from Example data:
from pathlib import Path
Path("fits").mkdir(exist_ok=True)
for name, half_life in [("fast", 50.0), ("slow", 1000.0)]: # two half-lives, a bank and a file each
one = po.ModelBank([po.spec.ewridge(name, targets=["y"], features=["x0", "x1"], clock="t",
gap_cap=300.0, half_life=half_life)])
one.fit(lf)
one.save(f"fits/{name}.state")
# Load the files back, and stack their tables in one.
banks = [po.ModelBank.load(f) for f in sorted(Path("fits").glob("*.state"))]
betas = pl.concat([b.coef() for b in banks]) # the same columns for every spec
last = pl.concat([b.last_row() for b in banks], how="diagonal_relaxed") # fields that differ by spec
for row in betas.pivot("term", index="spec", values="coef").iter_rows(named=True):
print(row["spec"], " ".join(f"{t} {row[t]:.3f}" for t in ["intercept", "x0", "x1"]))
print(last.columns)
fast intercept 0.994 x0 1.936 x1 -1.004
slow intercept 0.993 x0 1.992 x1 -1.007
['spec', 'group', 'pred_y', 'resid_y', 'weight_sum', 'settled_frac', 'withheld_reason', 'coef', 'support_coef']
Output field names
Your code reads the output by field name, and
docs/OUTPUTS.md lists every field of every model. To
find a name without building it, filter one of two tables:
po.spec.output_index for every field, or po.spec.coef_fields for every
coefficient. The code that renders the names builds both from the spec, so
they always agree with the output. Since a test holds every name, default
and signature to a checked-in snapshot, no release renames your columns
silently (Versioning and the Polars pin).
Give a spec several values of ridge or half_life, several
feature_sets or penalties, and it fits a grid: one fit per
combination of values, each a point of the grid. A field's name says
which point it holds. Here grid has two half-lives and two ridge values,
and its coefficient on x1 at the smaller ridge comes back as df's −1.
This code uses df from Example data:
grid = po.spec.ewridge("m", targets=["y"], features=["x0", "x1"], clock="t",
gap_cap=300.0, half_life=[100.0, 500.0], ridge=[1e-6, 0.5])
out = df.online.fit_predict([grid]) # df's columns, plus the spec's column "m"
# A field's name, found from the values it holds.
idx = po.spec.output_index(grid) # every field, with the values its name encodes, and its dtype
name = idx.filter((pl.col("kind") == "pred") & (pl.col("target") == "y")
& (pl.col("ridge") == 1e-6) & (pl.col("half_life") == 500.0))["field"].item()
print(name, out["m"].struct.field(name).dtype)
# A coefficient's place in its list, found the same way.
row = po.spec.coef_fields(grid).filter( # one row per coefficient: its list, position, and unnest column
(pl.col("term") == "x1") & (pl.col("ridge") == 1e-6) & (pl.col("half_life") == 500.0)
).row(0, named=True)
x1 = out["m"].struct.field(row["field"]).drop_nulls().list.get(row["position"])
print(row["field"], row["position"], f"{x1[-1]:.2f}") # x1's coefficient on the last row
pred_y__r0.000001@h500 Float64
coef@h500 2 -1.00
If your own code parses the field names after the bank, keep __ and
@ out of target names and feature-set labels. A target named y__r0.5
renders like a ridge grid on y. Filtering po.spec.output_index needs no
such care.
Each name follows a grammar, so you can read it by eye. A number in a
name renders as a plain decimal in [1e-6, 1e7), as 0.000001 above, and
in compact scientific notation outside it.
pred_{target}{combo}{instance}
resid_{target}{combo}{instance}
sigma_{target}{combo}{instance}
abs_resid_q{level}_{target}...
weight_sum{instance}
settled_frac{instance}
withheld_reason{instance}
coef{instance}
support_coef{instance}
scored_clock, learned_clock under emit_clocks, once per spec
{combo} |
when |
|---|---|
"" |
a single ridge, no feature sets |
__r{ridge} |
a ridge grid |
__{set} |
feature sets, a single ridge |
__{set}_r{ridge} |
feature sets and a ridge grid |
__p{penalty} |
a lasso path |
{instance} |
when |
|---|---|
"" |
a single half_life |
@h{half_life} |
a half_life grid; a duration as its text, @h10m |
Coefficients
Read the coefficients from the bank with bank.coef(), or from the
output, as columns that show the fit as it moved. Both are the fit after
a row's update: bank.coef() after the last row each group learned from,
and the output's coef after its own row. A row's own pred came from the
fit before it. Here the bank, the output and a saved output each give
the same fit, near df's truth.
This code uses lf from Example data:
ols = po.spec.ewridge("ols", targets=["y"], features=["x0", "x1"], clock="t",
half_life=600.0, gap_cap=300.0, group="stock_id",
coef_every=0) # write coef on every row but a skipped one
# 1. From a bank -- live, or loaded from a state file with no data at hand.
bank = po.ModelBank([ols])
bank.fit(lf)
betas = bank.coef() # one row per coefficient: spec, group, instance, weight_sum, ..., term, coef
wide = betas.pivot("term", index=["group", "instance"], values="coef") # a row per (group, instance), a column per term
print(wide.select("group", pl.col("intercept", "x0", "x1").round(2)).rows())
# 2. From the output, as columns: the fit as it moved, one row per row.
path = (
lf.online.fit_predict([ols])
.online.unnest([ols]) # pred_y, resid_y, weight_sum, coef_y_intercept, coef_y_x0, coef_y_x1, ...
.select("t", "stock_id", "^coef_.*$")
.collect()
)
for t, x0 in path.filter(pl.col("t").is_in([0.0, 4.0, 40.0, 200.0, 396.0])).select("t", "coef_y_x0").iter_rows():
print(f"b0 at t {t:>3.0f}: coef on x0 {x0:.2f}")
# 3. From a saved output: unnest takes the specs, a bank, or the path of the state the run saved.
lf.online.fit_predict([ols], save_state="ols.state").sink_parquet("fitted.parquet")
saved = pl.scan_parquet("fitted.parquet").online.unnest("ols.state").collect()
print(saved.select(path.columns).equals(path))
# 4. As a matrix: the rows that carry coef, fixed to the layout's width.
k = po.spec.coef_index(ols).height # 3: intercept, x0, x1
matrix = (
pl.read_parquet("fitted.parquet")
.select(pl.col("ols").struct.field("coef"))
.filter(pl.col("coef").is_not_null())
.select(pl.col("coef").list.to_array(k))
.to_series()
.to_numpy()
)
print(matrix.shape)
[('b0', 1.0, 2.02, -1.01), ('b1', 1.01, 2.03, -1.06), ('b2', 0.93, 1.94, -0.99), ('b3', 1.03, 1.99, -0.97)]
b0 at t 0: coef on x0 0.00
b0 at t 4: coef on x0 0.66
b0 at t 40: coef on x0 2.24
b0 at t 200: coef on x0 2.06
b0 at t 396: coef on x0 2.02
True
(400, 3)
Stock b0's rows are every fourth, from t 0. Its path starts at 0,
since one row fixes no slope, is near 2 by its eleventh row, and ends on
the 2.02 bank.coef() reports.
coef_every=0 writes coef on every row but a skipped one, as a list
of one float per term. Without a cadence the bank writes it on each
group's last accepted row in a chunk, and on none for a group whose rows
there were all skipped. A cadence can be a number of clock units, or a
duration such as "5m" on a temporal clock. It writes coef once the
clock has moved that far since the group's last coef row, as
solve_every schedules a solve.
max_rows_between_coefs writes it after that many rows. With both, it
is written by whichever comes first. A cadence reads only the clock and the rows, so the
rows that carry coef do not depend on the chunking.
To spread that list into columns, chain .online.unnest(specs) after
fit_predict. It takes the specs, a bank, or the path of a saved state,
and works on a saved output. With separator="." it names each column after
its spec too, ols.coef_y_x0, as Polars' own unnest(separator=) does, so
two specs that fit one target unnest side by side.
For a matrix, fix the list's width on the rows that carry it, as item 4
does with .list.to_array(k), k being the height of
po.spec.coef_index(spec). coef is a list because a list stores nothing
for a null row, and without a cadence coef is null on most rows.
Several targets, or a grid, put several blocks in one coef list: one
per target and point of a grid of ridge values, feature_sets or a
penalties. Each half-life of a half_life grid has a coef field of its
own. Here grid, as Output field names builds it,
has two half-lives and two ridge values, and the ridge of 0.5 shrinks each
slope by more than a third.
This code uses df from Example data:
# A grid of two half-lives by two ridge values, fitted on df.
grid = po.spec.ewridge("m", targets=["y"], features=["x0", "x1"], clock="t",
gap_cap=300.0, half_life=[100.0, 500.0], ridge=[1e-6, 0.5])
grid_bank = po.ModelBank([grid])
out = grid_bank.fit_predict(df) # df's columns, plus "m"
# The blocks spread into columns.
cols = out.online.unnest([grid]) # coef@h100 and coef@h500, spread into columns
print(cols.select("pred_y__r0.5@h500", "coef_y_x0__r0.5@h500").columns) # each block's columns named the way its pred field is
# The same blocks from the bank, a row each.
betas = grid_bank.coef() # target, ridge, feature_set and penalty tell the blocks apart
blocks_wide = betas.pivot("term", index=["instance", "ridge"], values="coef") # a row per block
for instance, ridge, b0, b1, b2 in blocks_wide.sort("instance", "ridge").iter_rows():
print(f"{instance} ridge {ridge}: intercept {b0:.2f}, x0 {b1:.2f}, x1 {b2:.2f}")
['pred_y__r0.5@h500', 'coef_y_x0__r0.5@h500']
@h100 ridge 1e-06: intercept 0.99, x0 1.97, x1 -1.00
@h100 ridge 0.5: intercept 0.97, x0 1.25, x1 -0.60
@h500 ridge 1e-06: intercept 0.99, x0 1.99, x1 -1.00
@h500 ridge 0.5: intercept 0.97, x0 1.30, x1 -0.63
Fitting many features from one pass
Three models, ewridge, lasso and ew_cov, keep a matrix of running
sums, their Gram in least-squares terms: the weight of the rows, the
columns' means and co-moments, and each target's moments. Every
least-squares fit on those columns reads its rows only through the Gram,
whose size depends on the number of columns alone. So one pass over a
stream too large to hold can leave a Gram per week, or per hour, and every
fit after the pass reads those Grams and never the rows. bank.gram(spec)
returns them as numpy arrays, one dict per group, and po.gram fits
them, as docs/GRAM.md shows for each task. Both need
numpy, an optional extra, which pip install "polars-online[numpy]" adds.
Give the spec a block column as its group, and the bank keeps a Gram
for each block. Here a column made in the query names each row's hour,
and one pass leaves eight hourly Grams of df's five features. Pooled,
they give least squares on all 400 rows, with df's truth and the t of
each coefficient. Any run of hours, on any of the columns, is one more
call.
This code uses lf from Example data:
# One Gram per hour, from one pass.
features = ["x0", "x1", "x2", "signal_a", "signal_b"]
hours = po.spec.ewridge(
"hours", targets=["y"], features=features, group="hour", # a Gram per hour
half_life=float("inf"), solve_every=1e18, # least squares' sums; the pass solves nothing
)
bank = po.ModelBank([hours])
bank.fit(lf.with_columns(hour=pl.col("ts").dt.truncate("1h"))) # the hour, made in the query
hourly = bank.gram("hours") # one dict of numpy arrays per hour, oldest first
# Pool every hour into one fit.
pooled = po.gram.merge(hourly) # the Gram of all the hours' rows
coef = po.gram.solve(pooled) # least squares: the intercept, then the features
t = po.gram.coef_stats(pooled, coef)["t"] # and each coefficient's t
print(f"{len(hourly)} hourly Grams, pooled over {pooled['weight_sum']:.0f} rows")
for name, b, tb in zip(pooled["columns"], coef, t):
print(f"{name:>9} {b:+.2f} t {tb:+6.1f}")
morning = po.gram.solve(po.gram.merge(hourly[:3]), features=["x0", "x1"]) # any hours, any columns, from the same pass
print("x0 and x1 before 12:00:", " ".join(f"{b:+.2f}" for b in morning[1:3]))
8 hourly Grams, pooled over 400 rows
intercept +0.99 t +39.1
x0 +1.99 t +78.3
x1 -1.01 t -39.9
x2 -0.03 t -1.0
signal_a +0.02 t +0.6
signal_b +0.03 t +1.3
x0 and x1 before 12:00: +2.00 -1.05
The same code serves thousands of columns. A Gram of 200 features and an intercept is 323 KB, and one of 10,000 columns 800 MB, whatever its rows.
docs/GRAM.md is the guide to fitting from Grams. It gives the theory, the keys of a Gram and what a Gram cannot answer, with a recipe that runs for each task:
| task | its section of GRAM.md |
|---|---|
| a Gram per week from one pass | Weekly blocks from one pass |
| pool any weeks, and score a fit on others | Pooling blocks |
| pool decayed blocks, each aged first | Aging a block before it is pooled |
| pool weeks and leave their levels out | Pooling within blocks |
| a ridge and its t | A ridge and its t |
| the lasso path | Choosing columns on the lasso path |
| columns chosen on most halves of the weeks, and a null for them | Stability selection over halves of the weeks |
| a quarter of the bytes, and one wide Gram on several threads | A compact Gram, gram_threads |
One row per finished group
When new group values never stop appearing, such as a day id, a session
id or a block number, set group_close to keep the bank's memory bounded.
The bank then writes each finished group's running sums out, a row per
half-life and per Gram, and frees its state. Without it, every group's
state stays until bank.drop_groups removes it. group_close is refused
beside an embargo or a window target, since both hold a group's rows past
its end.
Each closed row holds, bit for bit, what bank.gram() would have
returned for that group just before it closed, its matrix packed as the
upper triangle. It adds the span's own rows_fed, rows_learned,
clock_min and clock_max. A model with coefficients adds its coef, an
ew_cov with pca its eigendecomposition, and a marginal its pairs.
Here block 0's closed row is set beside the Gram of a bank that has seen
only block 0's rows, and the two agree to the bit. Blocks 0 to 2 close,
and block 3, the last, stays open.
This code uses df from Example data:
blocks = po.spec.ew_cov("cov", features=["x0", "x1"], lam=1.0, # lam=1.0: no decay
group="block", group_close="monotone") # a block is finished once a higher one arrives
by_block = df.with_columns(block=pl.int_range(pl.len()) // 100) # a block id that rises every 100 rows
bank = po.ModelBank([blocks])
bank.fit_predict(by_block)
closed = bank.closed_groups() # blocks 0, 1 and 2, oldest first: one half-life and one Gram, so one row each
print(closed.select("group", "rows_fed", "rows_learned").rows())
# Block 0 in a bank that has seen only its rows, so it is still open: its Gram, packed as a closed row's is.
open_bank = po.ModelBank([blocks])
open_bank.fit_predict(by_block.head(100))
still_open = open_bank.gram("cov", layout="packed")[0]
row = closed.row(0, named=True)
print(all(row[k] == still_open[k].tolist() for k in ["means", "comoments"]), row["weight_sum"] == still_open["weight_sum"])
first = po.gram.from_row(closed.head(1)) # block 0's row, as the dict gram() returns
print(f"{po.gram.correlation(first)[0, 1]:.3f}") # everything in po.gram works on it: x0 and x1's correlation in block 0
[('0', 100, 100), ('1', 100, 100), ('2', 100, 100)]
True True
0.021
Two modes decide when a group is finished. In either, the last group
never closes, since no row proves it finished, and bank.gram() still
reads it.
group_close |
a group is finished when | refused |
|---|---|---|
"monotone" |
its key is below the largest seen, since the key never goes backwards | a null key, or a chunk whose keys are out of order, naming the row; a float or temporal key column, which Polars' cast turns into an integer in the query before the bank |
"session" |
its session value changes |
a spec without a session column, or with session_gap |
Under "monotone", the bank orders an integer key as a number and a
String or Categorical key as text, so "9" comes after "10". Polars'
own sort agrees, so sort the input by the key before the bank.
Closed rows wait in the bank until bank.closed_groups() takes them
out. Called with drop=False, it reads them and leaves them queued. Rows
not yet read are saved with the state, so a loop that saves between chunks
loses none.
To write the closed rows to a file as the run goes, pass a path as
closed_groups= to lf.online.fit_predict, fit_predict_batches or
fit, or --closed-groups to the command line
(docs/RUNNER.md). With fit, which keeps no per-row
output, the file is the run's whole product: a stream too large for memory
goes in, and one row per block comes out. Here the blocked stream is saved
to disk, learned from as it is read back, and each closed block's row
goes to blocks.parquet.
This code uses df from Example data:
blocks = po.spec.ew_cov("cov", features=["x0", "x1"], lam=1.0, group="block", group_close="monotone")
df.with_columns(block=pl.int_range(pl.len()) // 100).write_parquet("by_block.parquet") # the stream, on disk
po.ModelBank([blocks]).fit(pl.scan_parquet("by_block.parquet"), closed_groups="blocks.parquet") # read back as a query
print(pl.read_parquet("blocks.parquet").select("group", "rows_fed").rows())
[('0', 100), ('1', 100), ('2', 100)]
Reading a correlation matrix
po.corr repairs, shrinks, summarizes and scores a correlation matrix,
given as an array, a gram() dict or a closed row (po.gram.correlation
turns a Gram into one). Each function needs only numpy, and
follows the paper it names. A matrix pieced together pair by pair, each
pair on its own rows, need not be a correlation matrix at all, as here:
its smallest eigenvalue is negative. po.corr.nearest repairs it.
r = np.array([[1.0, 0.9, 0.7], # three correlations estimated pair by pair, each on its own rows:
[0.9, 1.0, 0.1], # x0 follows both x1 and x2 closely, yet x1 and x2 barely move together,
[0.7, 0.1, 1.0]]) # which no single set of rows can give
print(f"smallest eigenvalue {np.linalg.eigvalsh(r).min():+.3f}")
# The repair, and how far it moved the matrix.
fixed, dist, iters = po.corr.nearest(r) # Higham (2002): the nearest correlation matrix, by alternating projections
print(f"repaired: smallest eigenvalue within 1e-8 of 0: {abs(np.linalg.eigvalsh(fixed).min()) < 1e-8}, moved by {dist:.3f}")
print(np.round(fixed, 3).tolist())
# Shrinkage, the noise edges, and a correlation's standard error.
shrunk, alpha = po.corr.shrink(fixed, alpha=0.2) # Ledoit-Wolf shrinkage toward a constant-correlation target
lo, hi = po.corr.mp_edge(n=2000, m=50) # the Marchenko-Pastur edges: eigenvalues inside them are what pure noise gives
print(f"noise edges for 50 series over 2,000 rows: {lo:.3f} to {hi:.3f}")
print(f"standard error of a correlation of 0.3 over 2,000 rows: {po.corr.fisher_se(n=2000, rho=0.3):.4f}")
smallest eigenvalue -0.093
repaired: smallest eigenvalue within 1e-8 of 0: True, moved by 0.120
[[1.0, 0.837, 0.655], [0.837, 1.0, 0.135], [0.655, 0.135, 1.0]]
noise edges for 50 series over 2,000 rows: 0.709 to 1.341
standard error of a correlation of 0.3 over 2,000 rows: 0.0204
The repair lands on the edge of the valid matrices, so its smallest eigenvalue is 0 to the iteration's tolerance. It moves each correlation a little, 0.120 in all as a Frobenius distance.
The rest, by kind, each with its arguments in the API reference:
| kind | function | what it gives |
|---|---|---|
| input | matrix |
a correlation matrix from an array, a gram() dict, or a closed row |
| shrinkage | shrink |
Ledoit-Wolf shrinkage toward a constant-correlation (or identity) target, and its weight |
| transform | to_z, from_z |
Fisher's transform and its inverse, clipped so a degenerate ±1 is finite |
| spectrum | spectral, from_spectral |
the top eigenpairs, and the completion back to a correlation matrix |
| spectrum | absorption |
the absorption ratio: how much of the variance the top k eigenpairs carry |
| spectrum | mp_edge, mp_density |
the Marchenko–Pastur edges and density: what pure noise gives |
| spectrum | shift |
the standardised absorption shift, (fast − slow) / scale |
| one number | equicorr |
deco's one number for the whole matrix, offline |
| one number | equicorr_row |
the equicorrelation estimate of one standardised row, the u deco computes per row |
| one number | equicorr_loglik |
the Gaussian log-density of a standardised row under an equicorrelation matrix |
| sampling error | fisher_se |
the standard error of a correlation; phi_a= and phi_b= add the AR(1) inflation, with its caveats |
| sampling error | signal_share |
how much of a correlation's movement between blocks is not sampling noise, at Kish's sample size |
| blocks | block_means, from_blocks |
the mean correlation within and between labelled blocks, and the block matrix back from those means |
| scale | epps_invert |
the correlation at a coarser scale, from ew_cov's lagged co-moments |
| forecast | loss |
how wrong a forecast correlation matrix was: qlike, z_mse or Engle–Colacito minvar |
Checking the data and the model
A fit can go wrong for two kinds of reason, and the library checks each.
The data can be wrong, such as a feed that writes -999 where it measured
nothing, so that no model can learn from it well. Or the model can be
wrong for data that is fine, such as a relationship that broke halfway.
Each kind has its call or switch, and a guide of its own.
| to check | call or switch | guide |
|---|---|---|
| whether the data can be learned from at all | bank.check(), beside an audit spec |
Checking your data's quality, docs/DATA-QUALITY.md |
| whether the model learned from it well | a spec's emit_* switches, and po.eval after the run |
Checking that a model works, docs/DIAGNOSTICS.md |
Checking your data's quality
bank.check() reads what a bank kept during a pass and lists what is
wrong with the data, with what to do about each finding.
docs/DATA-QUALITY.md has a section on each
finding. An audit
spec in the same bank counts what no model keeps, so run it beside the
models in the same pass. It keeps about 10 KiB of state a column,
whatever the rows. Here three faults are planted in df: a sentinel, a
missing feature and a constant one. An audit and a ridge regression learn
from the faulty file, and check() names each fault.
This code uses df from Example data:
# df with three faults planted, saved.
faulty = df.with_columns(
x1=pl.when(pl.col("t") % 20 == 0).then(-999.0).otherwise(pl.col("x1")), # -999 where x1 was not measured: 5% of rows
x2=pl.when(pl.col("t") % 3 == 0).then(None).otherwise(pl.col("x2")), # x2 missing on a third of the rows
x3=pl.lit(1.0), # a feature that never moves
)
faulty.write_parquet("faulty.parquet")
# An audit beside a ridge, learned from the faulty file.
features = ["x0", "x1", "x2", "x3"]
audit = po.spec.audit("audit", columns=[*features, "y"], clock="t", gap_cap=300.0) # counts what no model keeps
model = po.spec.ewridge("ridge", targets=["y"], features=features, clock="t", gap_cap=300.0, half_life=100.0)
bank = po.ModelBank([audit, model])
bank.fit(pl.scan_parquet("faulty.parquet"))
# Each finding, and the advice on the worst.
findings = bank.check() # one row per problem, the worst first
for row in findings.select("severity", "code", "spec", "column", pl.col("value").round_sig_figs(3)).iter_rows():
print(*row)
print(findings.row(0, named=True)["message"]) # what to do, and the section of docs/DATA-QUALITY.md on it
warning constant audit x3 0.0
warning sentinel audit x1 0.0475
warning constant ridge x3 0.0
warning missing ridge x2 0.335
info heavy_tails audit x1 969.0
info missing audit x2 0.335
column 'x3' took one value (1) on every usable row of the stream: as a feature it is collinear with the intercept, and as a target there is nothing to learn. See https://github.com/hgilde/polars-online/blob/main/docs/DATA-QUALITY.md#constant
The audit finds all three faults: x3 constant, -999 on 4.75% of
x1's values, and x2 missing on a third of the rows. It also finds the
heavy tail the sentinel gives x1, a value 969 robust standard
deviations from the median. The ridge regression finds the two faults it
can measure. To a model -999 is a number like any other, so only the
audit sees a sentinel.
Each finding is a row, the worst first. Its severity is error (the
model cannot learn), warning (the fit is unreliable) or info. Its
code names the check, such as missing, constant, collinear or
leakage. The spec, group and column say where it was found, the
measured value sits beside its threshold, and the message says what
to do.
check() adds no work to the run, because it reads only what the bank
keeps anyway: the tables of What a bank holds, and
the Gram where a spec keeps one. So call it after a first pass, before
trusting a fit, on a live bank or one loaded from its file. Without an
audit in the bank, it says nothing about sentinels, frozen feeds,
duplicated columns or the clock's gaps.
Every threshold was measured for how often it finds a planted problem
and how often it fires on clean data. On five clean shapes over ten seeds
no check raised an error or a warning, and each planted problem was found
on every seed. Each message ends with a link to the section of
docs/DATA-QUALITY.md on its code. That section
shows what the problem does to each model and how to fix it with Polars
before the bank, in a recipe that runs, with the theory behind it. The
docstring of ModelBank.check lists every check with its threshold.
Checking that a model works
Judge a model as the stream runs, by switching diagnostics on in its spec,
or after the run, by passing its output to po.eval.
docs/DIAGNOSTICS.md takes each question about a
model in turn. A spec's diagnostics measure each row against what the
models learned before it, never against the row's own outcome, and their
memory does not grow with the stream.
emit_breaks asks whether the relationship has broken. Its
break_wald compares two fits of the same regression, one at the
diagnostic's memory and one at four times it, and grows when a
coefficient moves. Here it finds df's planted break: the beta of ret
on signal_a flips from 0.8 to −0.8 at row 200.
This code uses lf from Example data:
beta = po.spec.ewridge(
"beta", targets=["ret"], features=["signal_a"], clock="t", gap_cap=300.0, half_life=100.0,
coef_every=0, # the coefficients on every row
emit_breaks=True, # studentized_ret, cusum_ret, cusum_sq_ret, break_wald_ret
)
out = lf.online.fit_predict([beta]).online.unnest([beta]).collect()
# break_wald is chi2 with a degree of freedom per coefficient, here 2, whose 0.01% value is 18.4.
wald = out.select("t", "coef_ret_signal_a", "break_wald_ret")
print(f"largest break_wald before row 200: {wald.filter(pl.col('t') < 200)['break_wald_ret'].max():.1f}")
first = wald.filter(pl.col("break_wald_ret") > 18.4)["t"].min()
for row, b, w in wald.filter(pl.col("t").is_in([199.0, first, 299.0, 399.0])).iter_rows():
print(f"row {row:.0f}: beta on signal_a {b:+.2f}, break_wald {w:.1f}")
largest break_wald before row 200: 17.5
row 199: beta on signal_a +0.77, break_wald 3.4
row 235: beta on signal_a +0.41, break_wald 19.5
row 299: beta on signal_a -0.01, break_wald 60.9
row 399: beta on signal_a -0.39, break_wald 48.8
break_wald stays below 18.4 for the 200 rows before the flip, and
passes it at row 235, 35 rows after. By then the fit's beta has moved only
from 0.77 to 0.41, and at a half-life of 100 rows it is still on its way
to −0.8 at the end. The CUSUMs of the same switch stay null on this short
stream: they wait until the fit has settled, 4.3 half-lives in.
Every switch but emit_clocks and emit_feature_health reads the
residuals, and feature health reads the features a fit's coefficients
lean on. So only the ten linear models,
which predict a target, take them, and any other model refuses them by
name. A switch adds its fields to the spec's output, one per slot: one
prediction of one target, at one point of a grid.
docs/DIAGNOSTICS.md answers "is this model working?" one question at a time. For each diagnostic it gives the definition and its update, and the threshold that means something, with its false-alarm rate as measured. It gives the time the diagnostic adds a row, the theory it rests on, and a recipe that runs with its output shown:
| question | switch or call | fields, per slot <t> |
|---|---|---|
| is it scaled right? | emit_calibration |
calibration_slope_<t>, calibration_intercept_<t>, calibration_wald_<t> |
| has it broken? | emit_breaks, emit_drift |
studentized_<t>, cusum_<t>, cusum_sq_<t>, break_wald_<t>, drift_<t> |
| can its t be trusted? | emit_se_coef, emit_robust_se |
se_coef, se_coef_hc0, se_coef_hac, on coef's rows (Coefficients) |
| what is it missing? | emit_specification, emit_autocorr |
ljung_box_<t>, breusch_pagan_<t>, ramsey_reset_<t>, autocorr_<t> |
| are its tails heavy? | emit_tails |
skew_<t>, kurtosis_<t>, jarque_bera_<t>; drift_threshold is set against the kurtosis |
| which row moved it? | emit_influence |
influence_<t>: ewridge, rls and kalman |
| are its features healthy? | emit_feature_health |
spread_ratio_<feature>, mean_shift_<feature>, once per instance |
| is another model better? | emit_metrics, emit_selected, emit_averaged; po.eval.metrics, window_metrics, compare_specs, diebold_mariano, clark_west, seqtest, and po.eval.sums for an output too large to hold |
ic_<t>, r2_<t>, hit_rate_<t>; selected_<t>, pred_<t>__selected, pred_<t>__averaged, one per target |
| is it ready? | min_weight, min_settled_frac, max_error_inflation, emit_error_inflation (Warm-up) |
weight_sum, settled_frac, withheld_reason, error_inflation_<t> |
| how wide is its error? | emit_sigma, emit_zscore, resid_quantiles, conformal |
sigma_<t>, zscore_<t>, abs_resid_q<p>_<t>, lo_<t>, hi_<t>, coverage_<t> |
| when was a row scored, and from what? | emit_clocks, on every model |
scored_clock, learned_clock, once per spec (Labels that arrive late) |
| does a model find a truth it was given? | po.sim.regimes |
a stream whose regimes are known, and the truth beside it |
Models
polars-online has twenty-two models in four families, and each family
answers one kind of question about a stream. A spec's clock, grouping and
warm-up mean the same whichever model it names. Where a model takes a
half-life, it is a half-life on the clock (How a bank sees a
stream). To look up a model's keywords and
their defaults, open its builder's reference page, such as the one for
po.spec.ewridge. docs/OUTPUTS.md lists every field a
model writes, and docs/TESTING.md what each is checked
against.
| model | learns by | what it is |
|---|---|---|
| Linear models | predict a numeric target, from features or from its own past | |
ewridge · math |
solve | exponentially weighted ridge regression, the workhorse; several ridge values and feature sets are solved from the same running sums at almost no extra work |
rls · math |
solve | recursive least squares, in the numerically safe square-root form |
lasso · math |
solve | lasso and elastic-net path, with the penalty chosen as the stream runs |
kalman · math |
filter | a Kalman filter whose coefficients drift as random walks |
huber · quantile · math |
reweight | robust and quantile regression |
sgd · math |
step | stochastic gradient descent with squared, Huber, quantile, ε-insensitive, Poisson and logistic losses |
pa · math |
step | passive-aggressive regression, with no learning rate to tune |
ftrl · math |
step | FTRL-proximal regression, logistic by default, with an L1 penalty that zeroes coefficients |
holt · math |
step | Holt's linear trend, the baseline that uses no features |
| Moments and correlation | track the running moments and correlations of the columns you name | |
ew_cov · math |
accumulate | running mean, variance, covariance, correlation, partial correlation, Mahalanobis distance and principal components |
marginal · math |
accumulate | every (feature, target) pair's running mean, variance, covariance, correlation, slope and t, for a wide set of columns, kept in the state and read back as a table; optionally at a set of lags, and with binned target moments for the relations a correlation cannot see |
deco · math |
accumulate | one correlation for the whole matrix: Engle & Kelly's equicorrelation, or one per block and per pair of blocks |
rcov · math |
accumulate | a block's realised covariance, robust to microstructure noise: the Barndorff-Nielsen–Hansen–Lunde–Shephard kernel or Christensen–Kinnebrock–Podolskij pre-averaging, reported when a group closes |
audit · math |
accumulate | what each column holds, counted in one pass: nulls, NaNs and infinities apart, repeated values, runs, persistence, tails and duplicated columns, and the clock's duplicate stamps and gaps, read back as a table and by bank.check() |
| Clustering and classification | put labels on rows | |
kmeans · math |
step | exponentially weighted k-means: cluster labels assigned before the row is learned from, with a split–merge move that finds a cluster born after seeding |
micro · math |
step | density-based clustering: DenStream micro-clusters linked into clusters of any shape and number, flagging the rows that belong to none |
ew_class · math |
accumulate | Gaussian classification, QDA, LDA or naive Bayes, with one set of running moments per class: a label column in, class probabilities out |
| Sequential tests and regimes | give evidence that something holds or has changed, and which regime the stream is in | |
seqtest · math |
test | a sequential test of a sign by betting, giving evidence you can read at any row: on its own, of a column's sign; with a and b, of whether one spec of the bank predicts closer than another |
corrchange · math |
test | has the correlation structure changed: the Wied–Krämer–Dehling constancy test span by span, the Wied–Galeano detector row by row against a stable history, or the size of a change between two windows against a permutation null |
bocpd · math |
filter | how long this regime has lasted: Adams & MacKay's run-length posterior, so the answer is the regime's age and not a flag |
hmm · math |
filter | a Gaussian hidden Markov model, filtered as the stream runs: ew_class without the labels, with a transition matrix that can be learned |
How a model learns from a row decides whether, with decay off, its answer depends on the row order:
| learns by | what the model does with a row | with decay off | order-dependent even so |
|---|---|---|---|
| solve | keeps running sums and computes its coefficients from them | converges to the batch answer, in any row order | lasso's penalty_selected, ranked by out-of-sample error, though its path converges in any order |
| accumulate | keeps running sums and reports them | converges to the batch answer, in any row order | a lag, in ew_cov or marginal, which counts learned rows; rcov's block and deco's per-row estimate; audit's runs, smallest change, lag pairs and clock steps, and past distinct_cap the digest's median, mad and robust_z and the counters' top_value, top_count, second_count and equal_by_chance, each within its bound |
| reweight | solves from running sums, but lets the fit before each row decide how that row enters them | depends on the row order | |
| step | moves its coefficients a little on each row | depends on the row order | |
| filter | carries a belief forward from row to row | depends on the row order | |
| test | counts evidence as it goes | depends on the row order |
Choosing a model
Choose the family by the question you are asking, then the model by what
your data need. Each model has an entry in
docs/MODELS.md: when to reach for it, its update rule,
its parameters, a recipe on data whose answer is known, and what it is
checked against. That page opens with the notation the update rules share.
The four models that watch for change, seqtest, corrchange, bocpd
and hmm, have a section of their own, Detecting changes and
regimes.
| your question | the family | start with | change to another when |
|---|---|---|---|
| what will the target be? | Linear models | ewridge |
the coefficients drift (kalman); some targets are wild (huber, quantile); few features matter (lasso, ftrl); each row must take O(k) time (sgd, pa); the target is a count (sgd) or 0/1 (ftrl); you want a baseline with no features (holt) |
| how do these columns vary and move together? | Moments and correlation | ew_cov |
there are thousands of columns (marginal); one correlation must stand for a whole matrix (deco); the prices are noisy ticks (rcov); you want to know what the columns hold (audit) |
| which group does this row belong to? | Clustering and classification | kmeans |
the clusters have any shape (micro); you have the labels to learn from (ew_class) |
| has something held, or changed? | Sequential tests and regimes | seqtest |
the correlations may have changed (corrchange); you want the regime's age (bocpd) or which regime it is (hmm) |
A linear model predicts a number, and a model whose coefficients drift
follows a relationship that breaks. In df, ret's slope on
signal_a is 0.8 on rows 0 to 199 and −0.8 from row 200. ewridge, with
every row counted the same, averages the two slopes. kalman follows the
new one, as far as the afternoon's noisier rows let it.
This code uses df from Example data:
common = dict(targets=["ret"], features=["signal_a"], clock="t", gap_cap=10.0)
specs = [
po.spec.ewridge("ridge", half_life=float("inf"), **common), # every row counts the same
po.spec.kalman("kalman", half_life=50.0, coef_half_life=20.0, **common), # a coefficient that may drift
]
bank = po.ModelBank(specs)
bank.fit(df) # learn every row, keep no output
for r in bank.coef().filter(pl.col("term") == "signal_a").iter_rows(named=True):
print(f"{r['spec']}: slope at the last row {r['coef']:.2f}")
ridge: slope at the last row 0.10
kalman: slope at the last row -0.66
A moments model tracks how columns vary and move together, and computes
each row's statistics from the rows before it. So a row's correlation
can serve as a feature for that row without leaking the row's own values. Here the correlation of
signal_a and ret changes sign after the break at row 200.
This code uses df from Example data:
moves = po.spec.ew_cov("moves", features=["signal_a", "ret"], clock="t", gap_cap=10.0,
half_life=50.0, stats=["corr"])
corr = df.online.fit_predict([moves])["moves"].struct.field("corr_signal_a_ret")
print(f"corr(signal_a, ret) at row 199: {corr[199]:.2f}, at row 399: {corr[399]:.2f}")
corr(signal_a, ret) at row 199: 0.41, at row 399: -0.33
A clustering model labels each row with its nearest cluster before
learning from the row, and moves the clusters as the stream runs. Here
kmeans finds three clusters planted at (0, 0), (6, 0) and (0, 6). The
code makes its own points, since df has no clusters:
rng = np.random.default_rng(3)
centres = np.array([[0.0, 0.0], [6.0, 0.0], [0.0, 6.0]]) # three clusters, planted
points = centres[rng.integers(0, 3, 3000)] + rng.standard_normal((3000, 2))
pl.DataFrame(points, schema=["a", "b"]).write_parquet("points.parquet")
clusters = po.spec.kmeans("clusters", features=["a", "b"], n_clusters=3, half_life=float("inf"))
bank = po.ModelBank([clusters])
bank.fit(pl.scan_parquet("points.parquet"))
found = bank.coef()["coef"].to_numpy().reshape(3, 2).round(1) # one row of (a, b) per centre
print("centres found:", sorted(found.tolist()))
centres found: [[0.0, -0.1], [0.0, 6.0], [6.0, 0.0]]
A sequential test says when the evidence is enough, and you may read it
at any row. Here seqtest asks whether a regression on x0 and x1
predicts y closer than holt, which reads only y's own past. Its
evidence passes the 5% level a few rows into df.
This code uses df from Example data:
import math
# Two forecasts of y, and a test between them.
common = dict(targets=["y"], clock="t", gap_cap=10.0, half_life=100.0)
specs = [
po.spec.ewridge("ridge", features=["x0", "x1"], **common),
po.spec.holt("holt", **common), # y's own level and trend, no feature
po.spec.seqtest("closer", targets=["y"], a="ridge", b="holt"), # does ridge predict closer?
]
out = df.online.fit_predict(specs)
passed = out["closer"].struct.field("log_e_a_y") >= math.log(20) # the 5% level, however often you look
print(f"ridge predicts closer than holt at the 5% level from row {passed.arg_true()[0]}")
ridge predicts closer than holt at the 5% level from row 18
Detecting changes and regimes
Four models watch a stream for change, and each answers its own question. What each finds on streams whose answer is known, and what it misses, is measured in docs/REGIMES.md. Their update rules and parameters are under Sequential tests and regimes. Each answer is read before its row is learned from, as every prediction is, so it can drive a decision on that row:
| the question | the model | what it writes on each row |
|---|---|---|
| does a column tend to be positive, or does one spec predict closer than another? | seqtest |
the log wealth of a bet on each side; past ln 20 is the 5% level, however often you look |
| have the correlations among these columns changed? | corrchange |
a test statistic, its critical value, and a flag where one passes the other |
| how long has the current regime lasted? | bocpd |
run_mode, the most likely number of rows since the regime began |
| which regime is the stream in? | hmm |
the probability of each hidden state, and the most probable one |
This code makes two series whose returns take turns, 500 rows at a time,
between a calm regime and one 2.7 times as wide. It uses po.sim.regimes,
which returns the regime each row was drawn in beside the rows. bocpd
dates each switch, and hmm says which regime each row is in:
import polars as pl
import polars_online as po
# Make the data, and save it with its truth.
sim = po.sim.regimes( # two series, and the regime each row was drawn in
2, states=[0.3, 0.3], transition=[[0.0, 1.0], [1.0, 0.0]], # the two regimes take turns, a block each
n_blocks=6, rows_per_block=500,
scale_state=[0.0, 1.0], # regime 1's returns are e = 2.7 times as wide
seed=3,
)
truth = sim["truth_rows"].select("t", regime="state")
sim["rows"].join(truth, on="t").write_parquet("regimes.parquet")
returns = pl.scan_parquet("regimes.parquet").with_columns(pl.col("x_1", "x_2").diff()) # from the levels
# One model dates each switch, the other names the regime.
age = po.spec.bocpd("age", features=["x_1", "x_2"], hazard=500.0) # how long has this regime lasted?
which = po.spec.hmm( # which regime is this row in?
"which", features=["x_1", "x_2"], n_components=2, half_life=float("inf"),
precision_prior=1e-2, warm_rows=500,
transition=[0.998, 0.002, 0.002, 0.998], transition_prior=500.0, # a regime lasts about 500 rows
)
out = returns.online.fit_predict([age, which]).collect().with_columns(
began=pl.col("t") - pl.col("age").struct.field("run_mode"), # the row the current regime began on
guess=pl.col("which").struct.field("state"),
)
# Score both against the planted switches and regimes.
for switch in (500, 1000, 1500, 2000, 2500):
began = out.filter(pl.col("t") == switch + 50)["began"][0]
print(f"a switch at row {switch}: 50 rows later, bocpd dates the regime to row {began}")
scored = out.filter(pl.col("t") >= 500) # hmm seeds its states from the first 500 rows
agree = (scored["guess"] == scored["regime"]).mean() # its states are numbered as seeded, so read either way
print(f"hmm puts {max(agree, 1 - agree):.0%} of the rows after seeding in their true regime")
a switch at row 500: 50 rows later, bocpd dates the regime to row 497
a switch at row 1000: 50 rows later, bocpd dates the regime to row 1000
a switch at row 1500: 50 rows later, bocpd dates the regime to row 1501
a switch at row 2000: 50 rows later, bocpd dates the regime to row 1998
a switch at row 2500: 50 rows later, bocpd dates the regime to row 2501
hmm puts 94% of the rows after seeding in their true regime
Read a regime's age from bocpd's run_mode. Fifty rows after each
switch, bocpd dates the new regime to within three rows of the truth. Its
alarm, p_change, scores each row on its own, so it barely moves when the
spread grows 2.7 times: no single row of the wider regime is surprising
enough on its own.
Tell hmm that regimes last. One return says little about which regime
it came from. A sticky transition matrix, counted by transition_prior
as 500 rows, lets the filter weigh each row against what the rows before
it said. Without those two arguments, the filter puts 59% of the rows in
their true regime.
Performance
A bank spends its time on the models' own arithmetic, and its memory on
their state, a few chunks in flight, and what Polars has read ahead of the
bank. The thread counts and chunk_size change only the speed and the
memory, never the numbers a bank returns.
docs/PERFORMANCE.md has the measurement behind
every figure here, and a table of what to change when a number is not what
you expected. Every figure was measured on one Apple M4 Pro with 14 cores
(10 performance, 4 efficiency), so your own times will differ.
A model's speed is set by the work it does on each row. ewridge
solves a system of k features, so it does O(k²) work a row, and ten
times the features take several times as long. This code writes 200,000
rows of 50 features to a file, then times one ewridge over it at 5
features and at 50, best of three runs. ModelBank.fit learns from every
row and keeps no output, so the time is the reading and the model:
import time
import numpy as np
import polars as pl
import polars_online as po
# 200,000 rows of 50 features, on disk.
rng = np.random.default_rng(1)
n = 200_000
X = rng.standard_normal((n, 50))
pl.DataFrame(
{"y": X[:, 0] + rng.standard_normal(n)} | {f"x{j}": X[:, j] for j in range(50)}
).write_parquet("wide.parquet")
# Time one fit at 5 features and at 50, best of three.
for k in (5, 50):
spec = po.spec.ewridge("ridge", targets=["y"], features=[f"x{j}" for j in range(k)],
half_life=1000.0)
seconds = []
for _ in range(3):
start = time.perf_counter()
po.ModelBank([spec]).fit(pl.scan_parquet("wide.parquet")) # learn every row, keep no output
seconds.append(time.perf_counter() - start)
print(f"time at {k} features: {n / min(seconds) / 1e6:.1f} million rows a second")
time at 5 features: 2.2 million rows a second
time at 50 features: 0.8 million rows a second
These rates sit below the table's, most of all at five features, because
they include reading the file. The table times fit_predict on a frame
already in memory, which is the model alone.
Throughput
Here are rows a second for every model at its usual settings, all under these conditions:
| machine | an Apple M4 Pro, one process |
| runs | best of 3, 200k rows per run, measured on 2026-09-29 at commit 5e96018 with uv run python scripts/benchmark.py --markdown, beside one other process on about one core, with no load average recorded; run 2 of PERFORMANCE §28 |
ewridge and rls |
re-measured the same day at commit d0f8616, once ewridge's solve had been made cheaper, beside a load average of 3.2 to 3.5 (§30) |
| noise | a run moves by up to about 11% from the last on this machine, so read a gap smaller than that as noise |
| half-life | 1,000, in every row that gives none of its own |
k, K |
the number of features; the number of clusters or hidden states |
The linear models:
| model | settings | rows/sec | what sets its speed |
|---|---|---|---|
ewridge |
k=5, 1 target, 1 half-life | 6,199,532 | |
ewridge |
k=20, 1 target, 1 half-life | 3,068,424 | |
ewridge |
k=50, 1 target, 1 half-life | 946,975 | |
ewridge |
k=20, 10 targets | 1,653,010 | one set of feature sums for every target (below) |
ewridge |
k=20, 5 half-lives, 500 to 2,500 | 1,714,699 | one set of sums per half-life, run in parallel (below) |
ewridge + conformal |
k=20, 90% interval | 3,070,074 | nothing measurable: the interval reads the residual the model already has |
rls |
k=20, 1 target | 1,662,687 | its square-root form, at 54% of ewridge's speed: the form is what keeps one extreme row from destroying it by cancellation |
kalman |
k=20, 1 target | 1,876,826 | |
kalman |
k=20, revert_half_life |
1,677,506 | |
lasso |
k=20, 1 target (3-point path) | 1,629,739 | |
huber |
k=20, 1 target | 3,267,509 | |
sgd |
k=20, squared loss, standardize=False |
8,423,373 | the raw step; the default, standardize=True, scales each feature on the row first |
sgd |
k=20, coef_min=0, coef_sum=1, standardize=False |
2,288,779 | a simplex constraint sorts 2k breakpoints per row, which makes sgd 3.7 times slower |
pa |
k=20, standardize=False |
8,284,518 | the raw step, as for sgd |
ftrl |
k=20, 1 target | 3,873,282 |
List several targets on the same features in one spec's targets, where
they share one set of feature sums: 10 targets take 1.86 times as long as
one.
A grid of half-lives keeps one set of sums per half-life, and runs them in parallel. So the grid of five takes 1.43 times as long as its shortest half-life, 500, which runs alone at 2,443,773 rows a second. That is slower than at the half-life of 1,000 that every other row here uses, because a shorter half-life solves more often.
The models of moments and correlation:
| model | settings | rows/sec | what sets its speed |
|---|---|---|---|
ew_cov |
k=20: mean, std, corr (230 statistics) | 1,655,993 | |
ew_cov |
k=20: mean, mahal, mahal_q0.99 |
692,690 | the Mahalanobis distance factors a k × k matrix by Cholesky once per row |
ew_cov |
k=20: mean, cov, lags 1–5 | 1,123,565 | |
deco |
k=20, one equicorrelation | 2,093,753 | one number for the whole matrix of its m series, computed in O(m) a row with no m × m matrix to factor |
deco |
k=20 in 4 blocks | 1,563,248 | |
rcov |
4 features, kernel, blocks of 1000 | 3,385,427 | it accumulates per row and computes its kernel only when the block closes |
The models of clustering and classification:
| model | settings | rows/sec | what sets its speed |
|---|---|---|---|
ew_class |
k=20, 3 classes, full covariance | 466,466 | a k × k matrix factored by Cholesky once per learned row |
ew_class |
k=20, 3 classes, tied covariance | 531,384 | |
ew_class |
k=20, 3 classes, diagonal | 2,448,700 | |
kmeans |
4 features, K=8 | 4,590,376 | a distance to each centre |
kmeans |
k=20, K=8 | 2,383,197 | a distance to each centre |
micro |
4 features, eps=1 |
8,799,815 | a distance to each centre |
The sequential tests and regimes:
| model | settings | rows/sec | what sets its speed |
|---|---|---|---|
seqtest |
sign of one column | 14,919,853 | a handful of operations |
corrchange |
4 features, monitor, window_rows=500 |
449,526 | |
corrchange |
4 features, window 100, permute every 500 | 173,711 | its permutation null (below) |
bocpd |
4 features, diagonal, max_run=200 |
884,387 | the number of runs it keeps (below) |
bocpd |
4 features, full covariance, max_run=200 |
568,724 | the number of runs it keeps (below) |
hmm |
4 features, K=2 | 1,182,841 | a k × k covariance factorized per state per row, which is ew_class's work with the classes hidden |
hmm |
k=20, K=2 | 325,937 |
corrchange's window kind is the slowest model here, because its
permutation null redraws n_perm statistics every permute_every_rows rows.
To skip the redraws, give crit a number, which fixes the threshold.
bocpd slows with the number of run lengths it keeps, so a larger
prune_below makes it faster. It does O(runs · d²) work a row for d
features. prune_below drops every run holding less than that share of the
probability mass, and max_run, 10,000 by default, caps how many runs it
keeps. With neither bound the runs grow by one every row, and a stream
takes O(rows²). A changepoint collapses them to a few dozen, so bocpd
runs faster on data that breaks. A stream that does not break spreads them
over thousands of run lengths, and there max_run is the bound. On 20,000
i.i.d. Gaussian rows with no max_run
(PERFORMANCE §15):
prune_below |
rows/sec |
|---|---|
1e-8 |
204,000 |
1e-6, the default |
324,000 |
1e-4 |
687,000 |
Parallelism
To use more cores, give the bank more tasks: set group= on each spec,
pass more specs, or both. On every chunk the bank splits its work into
tasks, one per spec and group, or one per spec without a group. It runs
them on its own thread pool, separate from Polars', largest first, so a
few big groups do not leave cores idle at the end. Within a task the rows
go one at a time, because each row's update depends on the last. So a bank
with one spec, one group and one half-life is one thread's work per chunk,
while Polars reads and writes in parallel around it. All of it runs in one
process, by design (What this is not).
| what to add | how it runs in parallel | measured |
|---|---|---|
groups, with a spec's group= |
one task per group | k=20 over 64 groups: 7.3× from 1 to 14 threads on a 14-core machine (below) |
| specs, in the list the bank takes | one task per spec | eight single-group specs, k=20 over 300k rows, run in 155 ms in one bank, against 641 ms one at a time |
half-lives, as a list given to half_life= |
each half-life in a grid is its own set of running sums, and a task's half-lives run in parallel on the bank's pool. Ridge and feature-set grids share one set of sums and are expanded at solve time, so they need no thread | |
shards, on a wide marginal |
a task that runs on several threads: shards splits its pairs across the pool within a group (marginal) |
|
gram_threads, on a wide ewridge or ew_cov |
a task that runs on several threads: each row's k × k update, or a block's merge, is split across the pool within a group, from 725 columns. A count caps the share of the pool. The default, "auto", splits only when the groups running at once leave half the pool idle, since a split pays only on idle threads |
one group at 2,000 features: 3.0× under "auto" on 14 threads; four groups at 800 features: 1.6×; 28 groups: even, where taking the whole pool was 9-11% slower (gram_threads) |
| a Python thread that reads ahead | Python's global lock is released while a chunk is in the bank, so a Python reader thread can run ahead of ModelBank.fit_predict |
Specs run side by side, and each gives the numbers it gives alone.
This code runs the comparison of the table's second row on a generated
parquet file: eight ewridge specs of 20 features, one per half-life, in
one bank and then one at a time. Its times include reading the file, and
are your machine's, and the outputs are equal to the bit:
import time
import numpy as np
import polars as pl
import polars_online as po
# A file of 20 features, and eight specs that differ only in half-life.
rng = np.random.default_rng(2)
n = 300_000
features = [f"x{j}" for j in range(20)]
pl.DataFrame(
{"y": rng.standard_normal(n)} | {f: rng.standard_normal(n) for f in features}
).write_parquet("specs.parquet")
specs = [po.spec.ewridge(f"h{h}", targets=["y"], features=features, half_life=float(h))
for h in (100, 200, 400, 800, 1600, 3200, 6400, 12800)] # eight specs, one task each
# The eight in one bank, then one at a time.
start = time.perf_counter()
together = pl.scan_parquet("specs.parquet").online.fit_predict(specs).collect()
in_one_bank = time.perf_counter() - start
start = time.perf_counter()
alone = [pl.scan_parquet("specs.parquet").online.fit_predict([s]).collect() for s in specs]
one_at_a_time = time.perf_counter() - start
# The times, and whether the outputs agree.
print(f"time: {in_one_bank:.2f} s in one bank, {one_at_a_time:.2f} s one at a time")
same = all(together[s["name"]].equals(a[s["name"]]) for s, a in zip(specs, alone))
print("the same output either way:", same)
time: 0.42 s in one bank, 1.56 s one at a time
the same output either way: True
Rows a second for k=20 over 64 groups, at each thread count (PERFORMANCE §31):
| threads | 1 | 2 | 4 | 8 | 14 |
|---|---|---|---|---|---|
| rows/s | 0.98M | 1.89M | 3.44M | 5.98M | 7.08M |
Set the bank's thread count with threads=, on the bank or the call
that makes one. ModelBank(specs, threads=8), ModelBank.load(path, threads=8), po.fit_predict(..., threads=8), po.predict, the frame
namespaces' fit_predict and predict, and po.gram's lars_path,
lars_paths and solve_subsets all take it, and so does the command line,
as --threads 8 or a threads key in the TOML. An integer is a pool of
that many threads. Every bank given the same count shares one pool, which
goes away with the last of them, and asking for the size the process pool
already has uses that pool. None, the default, is the process pool. The
count is not saved with a state, and the output and the saved state are the
same at any count. A count below 1, a float or a bool is refused, naming
threads: 0 is not "one per core" here, None is.
This code fits one spec on one thread and on four, and checks the two agree.
It uses the file ticks.parquet from Example data:
import polars as pl
import polars_online as po
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t",
half_life=600.0, gap_cap=300.0, group="stock_id")
one = pl.scan_parquet("ticks.parquet").online.fit_predict([spec], threads=1).collect()
four = pl.scan_parquet("ticks.parquet").online.fit_predict([spec], threads=4).collect()
print("the same output on 1 and 4 threads:", one.equals(four))
print("threads=4 is a pool of", po.thread_pool_size(threads=4)) # reports, builds nothing
the same output on 1 and 4 threads: True
threads=4 is a pool of 4
Two environment variables size the default pools, one for each:
| variable | sizes | unset or 0 |
read at | what took effect |
|---|---|---|---|---|
POLARS_ONLINE_MAX_THREADS |
the bank's default pool, for a bank or call given no threads |
one thread per core | the first chunk a bank without threads fits or scores, or the first po.thread_pool_size() |
po.thread_pool_size() |
POLARS_MAX_THREADS |
Polars' readers and writers, and their read-ahead (Tuning memory with Polars' own settings) | one thread per core | import polars, which import polars_online runs |
pl.thread_pool_size() |
Polars has no per-call thread count, so its pool is set by its variable
alone, before import polars: in the shell, as
POLARS_MAX_THREADS=8 python fit.py, or with os.environ at the top of the
script. POLARS_ONLINE_MAX_THREADS is read once, when the default pool is
first used, and ignored after; threads= never reads it. A
POLARS_ONLINE_MAX_THREADS that is not a whole number makes the first call
that uses the default pool raise ValueError, naming the variable.
Giving both pools every core slows neither, because a bank task never calls
back into Polars' pool.
To search over factor sets, build one spec per set and pass the whole
list to one lf.online.fit_predict call. Each spec runs as its own task
on each group, and the query reads the file once. Its output has one
column per spec, which is what po.eval.compare_specs reads, and one state
file, grid.state, holds them all. df's target is built from x0 and
x1, so the set of those two scores best, and adding x2, which plays no
part, scores a little lower.
This code uses the file ticks.parquet from Example data:
import os
os.environ["POLARS_MAX_THREADS"] = "8" # Polars' readers and writers: read at import polars
import polars as pl
import polars_online as po
from itertools import product
# Three factor sets, each with and without standardizing: six specs.
factors = {"mkt": ["x0"], "mkt-sz": ["x0", "x1"], "mkt-sz-val": ["x0", "x1", "x2"]}
def factor_spec(name, features, standardize):
return po.spec.ewridge(f"{name}-std{standardize:d}", # "mkt-std0", "mkt-std1", ...
targets=["y"], features=features, clock="t", gap_cap=300.0,
group="stock_id", session="session", session_gap=60.0,
half_life=[100.0, 1000.0], ridge=[1e-3, 0.1], # gridded inside the spec
standardize=standardize)
specs = [factor_spec(n, f, s) for (n, f), s in product(factors.items(), [False, True])]
# One pass over the file fits all six.
(pl.scan_parquet("ticks.parquet")
.online.fit_predict(specs, chunk_size=200_000, save_state="grid.state", threads=8)
.sink_parquet("grid.parquet"))
# one row per spec and grid point; keep each spec's best
scores = po.eval.compare_specs(pl.read_parquet("grid.parquet"), [s["name"] for s in specs])
best = scores.group_by("spec").agg(pl.col("r2").max()).sort("r2", descending=True)
for spec_name, r2 in best.iter_rows():
print(f"{spec_name:<15} best R² {r2:.3f}")
mkt-sz-std0 best R² 0.938
mkt-sz-std1 best R² 0.938
mkt-sz-val-std0 best R² 0.927
mkt-sz-val-std1 best R² 0.926
mkt-std0 best R² 0.707
mkt-std1 best R² 0.707
On 2.56M rows over 64 groups, the six specs make 6 × 64 tasks every chunk, and the same query takes 29.0 s at one thread and 4.3 s at fourteen (PERFORMANCE §31).
Chunk size
Set how many rows the bank takes at a time with the keyword chunk_size,
100,000 by default. lf.online.fit_predict, lf.online.predict,
ModelBank.fit_predict_batches, ModelBank.fit, with_windows and
refresh_time take it. A frame passed to ModelBank.fit_predict is one
chunk, whatever its size.
The chunk size changes the speed and the memory, never a prediction.
It changes only which rows carry coef and support_coef: each group's
last accepted row of every chunk. This code reads the same file 10 rows
at a time and all at once, with four groups, so the small chunks write the
coefficients on 4 rows of each of 40 chunks.
This code uses the file ticks.parquet and df from Example data:
# The same spec over chunks of 10 rows, the whole file, and a frame.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", gap_cap=300.0,
half_life=50.0, group="stock_id")
query = pl.scan_parquet("ticks.parquet")
small = query.online.fit_predict([spec], chunk_size=10).collect() # read 10 rows at a time
large = query.online.fit_predict([spec], chunk_size=100_000).collect() # the whole file at once
whole = po.ModelBank([spec]).fit_predict(df) # a frame: one chunk, whatever its size
# Compare the predictions, and count the rows that carry coef.
def field(out, name):
return out["ridge"].struct.field(name)
print("largest difference in pred_y:", (field(small, "pred_y") - field(large, "pred_y")).abs().max())
print("the same as the frame's:", field(large, "pred_y").equals(field(whole, "pred_y")))
print("rows carrying coef:", field(small, "coef").is_not_null().sum(), "at 10 rows a chunk,",
field(large, "coef").is_not_null().sum(), "at 100,000")
largest difference in pred_y: 0.0
the same as the frame's: True
rows carrying coef: 160 at 10 rows a chunk, 4 at 100,000
For a wide frame, use larger chunks, because each chunk carries a fixed overhead. At 10,000 columns, handing a chunk across from Polars alone takes about 8 ms a call: 4 µs of every row at 2,000 rows a call, and 0.4 µs at 20,000 (PERFORMANCE §20).
For less memory, use smaller chunks, because a few are in flight at once: three on the command line, and a few more in Python (Memory).
Memory
Memory grows with the models' state and the rows a delay or a window
holds, and never with the number of rows that have passed. Every way of
running a bank works a chunk at a time. Measured as the most memory the
process ever held (macOS's peak footprint), with Polars 2.0.0 on 14
cores. The file has 20 features over 32 groups, parquet in and parquet out,
and ewridge reads it with a clock, a weight and a group per row. The 12M
rows are the 3M rows four times over, so the state is the same at both
sizes. scripts/memory_bench.py makes the file and runs each case:
| what you write | 3M rows | 12M rows | what it is |
|---|---|---|---|
lf.online.fit_predict([spec]) |
1.2 GB | 1.8 GB | the bank inside a query |
for chunk in lf.collect_batches(): bank.fit_predict(chunk) |
1.1 GB | 1.5 GB | your own loop |
The bank does not grow from 3M to 12M rows: the growth is the memory
allocator keeping pages it has freed. Told to return them at once, with
_RJEM_MALLOC_CONF=dirty_decay_ms:0,muzzy_decay_ms:0, the query holds
0.86 GB at 3M rows and 0.95 GB at 12M. Of the rest, nearly all is Polars
reading ahead in the parquet file. The command line, whose allocator
returns freed pages at once, holds 0.77 GB at 12M rows against 0.74 GB at
3M (docs/RUNNER.md).
Memory has four parts, each with its own bound:
| part | grows with | what bounds it |
|---|---|---|
| the state | the models and their settings, and the number of groups: a bank's group= keeps one set of running sums per group, and grows with the number of groups, not the number of rows |
it does not grow with the stream, but a window's snapshots and marginal's bins grow with their settings, and each group's is capped, at 256 MiB by default |
| the chunks in flight | chunk_size |
a few chunks at once, three on the command line (Chunk size) |
| the rows a delay or a window holds | the delay or the window, times the rows' rate | never the stream's length |
| whatever Polars' reader has read ahead | Polars' thread count | POLARS_MAX_THREADS shrinks it, and Polars' prefetch settings tune it directly (Tuning memory with Polars' own settings) |
Measure the first part by saving the state, whose size is what the models
keep. This code fits one ewridge of 20 features on the first
10,000 rows of a generated file and on all 1,000,000, once with no groups
and once over 64. The state stays the same size as the rows grow, and it
grows with the groups:
import numpy as np
import polars as pl
import polars_online as po
# A million rows of 20 features and 64 groups, on disk.
rng = np.random.default_rng(3)
n = 1_000_000
features = [f"x{j}" for j in range(20)]
pl.DataFrame(
{"g": rng.integers(0, 64, n), "y": rng.standard_normal(n)}
| {f: rng.standard_normal(n) for f in features}
).write_parquet("long.parquet")
# The state's size after 10,000 rows and after all of them, with and without groups.
for group in (None, "g"):
spec = po.spec.ewridge("ridge", targets=["y"], features=features, half_life=1000.0, group=group)
for rows in (10_000, 1_000_000):
bank = po.ModelBank([spec])
bank.fit(pl.scan_parquet("long.parquet").head(rows)) # learn, keep no output
kb = len(bank.save_bytes()) / 1024 # the state, as saved
print(f"{'64 groups' if group else 'one group'}, after {rows:>9,} rows: state {kb:,.0f} KB")
one group, after 10,000 rows: state 10 KB
one group, after 1,000,000 rows: state 10 KB
64 groups, after 10,000 rows: state 581 KB
64 groups, after 1,000,000 rows: state 583 KB
For a statistic per group over a long stream, use a spec's group=,
which keeps one set of running sums per group. Polars' rolling over
groups, with .over("group") or group_by=, holds every row of its input.
Peak memory on the same 12M rows:
| a rolling window | peak memory |
|---|---|
with .over("group") |
6.5 GB |
with group_by= |
1.7 GB |
| without groups | 0.25–0.28 GB |
Tuning memory with Polars' own settings
Polars' streaming engine reads blocks of a parquet file ahead of whatever
consumes them, sized from its thread count. While a bank is the slowest
step, a local disk needs none of that read-ahead, so it is the part of
memory you can cut. Three of Polars' environment variables set it, in the
shell or with os.environ in Python:
import os
os.environ["POLARS_ROW_GROUP_PREFETCH_SIZE"] = "1" # blocks (row groups) read ahead: read when a query starts to run
os.environ["POLARS_ROW_GROUP_PREFETCH_KBYTES_BUDGET"] = "65536" # a byte cap on the same read-ahead: read when a query starts to run
os.environ["POLARS_MAX_THREADS"] = "4" # the read-ahead is sized from this too: read at import polars
POLARS_ROW_GROUP_PREFETCH_SIZE=1 saves the most. Polars reads it each
time a query that scans the file starts to run, so set it at any point
before the call that runs the query, such as sink_parquet or
bank.fit_predict_batches(lf).
Measured as peak resident memory on 8M rows of 12 columns in 80 blocks,
with the allocator's page retention off so that the figure is live data:
| what runs | default prefetch | POLARS_ROW_GROUP_PREFETCH_SIZE=1 |
|---|---|---|
lf.online.fit_predict(...).sink_parquet(...) |
1.63 GB | 1.12 GB |
bank.fit_predict_batches(lf) |
1.41 GB | 1.07 GB |
The saving depends on how much data a block holds, so compare a run at
the default with one at 1 on your own files. On a file whose blocks
hold 262,000 rows, the setting takes a query from 1.86 GB to 0.51 GB
(PERFORMANCE §11).
The byte cap, POLARS_ROW_GROUP_PREFETCH_KBYTES_BUDGET, rarely binds,
because it counts compressed bytes, which cost almost nothing for a
memory-mapped local file.
To fit a run in less memory, give Polars fewer threads and the bank every
core. Polars sizes its read-ahead from POLARS_MAX_THREADS, while the
bank's threads changes only the bank's speed.
This code uses the file ticks.parquet from Example data:
import os
os.environ["POLARS_MAX_THREADS"] = "4" # before import polars: sizes the reader's read-ahead
import polars as pl
import polars_online as po
# The run itself does not change.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1", "x2"], clock="t",
half_life=600.0, gap_cap=300.0, group="stock_id")
(pl.scan_parquet("ticks.parquet") # written in Example data
.online.fit_predict([spec], chunk_size=200_000, threads=14) # the bank keeps every core
.sink_parquet("fit.parquet"))
On 12M rows over 64 groups, with one spec of four features and two half-lives (PERFORMANCE §31):
| Polars threads | bank threads | time | peak memory |
|---|---|---|---|
| 14 | 14 | 1.6 s | 0.86 GB |
| 4 | 14 | 1.9 s | 0.58 GB |
Keeping Polars at four threads holds 33% less memory, for 15% more time.
Window operators
A window operator, such as po.ewm_mean, does the same work on a row
however long its window, and holds about one window of rows. So a window
adds little memory: within 0.04 GB of reading and writing the file alone,
and a 4-minute window takes what a 1-minute one does. Measured on 16M rows
arriving about two a second on the clock, parquet in and parquet out, one
group (PERFORMANCE §33,
and §34
for the window targets):
| what runs | time | peak memory |
|---|---|---|
| reading and writing the file, and nothing else | 0.40 s | 1.03 GB |
a forward VWAP through with_windows |
2.43 s | 0.99 GB |
the same VWAP through Polars' rolling recipe |
19.52 s | 5.24 GB |
| a model whose target is a window expression, which the bank resolves itself: the native form | 9.37 s | 1.09 GB |
the window written by with_windows(..., like=spec) and fed back as a plain target: the column form |
5.87 s | 1.23 GB |
An operator whose input is null on most rows runs slower, because each read scans the rows between its values. On 300k rows, one value in 5,000 takes 0.24 s, against 0.04 s with a value on every row (PERFORMANCE §35).
To have operators share one queue, give them the same direction, back or
ahead, and the same half_life, window_size and closed. Sixteen
operators on one queue run in 57% of the time of sixteen with queues of
their own.
Write a look-ahead target as a column for speed, or give its expression
in targets for less memory. For the column form, call
with_windows(..., like=spec) in the query before the bank, and name the
column it writes in the spec's targets, under the same embargo. Polars
then runs the window and the bank at once, as two stages of one query. In
the native form the bank resolves the windows itself, one after the other
in the same source as the model. With many groups, that adds a Polars
query per group per chunk on the calling thread.
Windows as a model's inputs and
target runs both
forms.
Against scikit-learn
This library's main regressions solve exactly, where scikit-learn's
streaming regressor takes gradient steps.
SGDRegressor.partial_fit,
the scikit-learn estimator most people compare with, is a first-order
stochastic optimiser: its answer depends on the learning rate, the
schedule, the feature scaling and the row order. ewridge, rls,
lasso, huber and quantile accumulate sufficient statistics and
solve, so they have no learning rate. With decay off
(half_life=float("inf")) and ridge=0, ewridge is ordinary least
squares to 2e-13 of numpy.linalg.lstsq, in any row order. The like for
like is sgd, the cheap
O(k) baseline: at the same constant step it gives SGDRegressor's R² to
four places, because the two run the same recursion.
The comparison ran on one generated stream of 100,000 rows with k = 20,
on 2026-09-08. Each contender ran at its best over a sweep of its own
settings, by scripts/sklearn_comparison.py on scikit-learn 1.9.0.
ewridge solves on every row here, and its rate was re-measured on
2026-09-29. The noise ceiling is the R² of the generating signal itself:
| contender | R² stationary | R² drifting | rows/sec | what a prediction saw |
|---|---|---|---|---|
| noise ceiling | 0.9831 | 0.9923 | ||
SGDRegressor, row by row |
0.9829 | 0.9899 | 3,400 | every row before it |
SGDRegressor, batches of 1,000 |
0.9830 | 0.9820 | 2,200,000 | every row before its batch |
po.spec.sgd |
0.9826 | 0.9906 | 6,000,000 | every row before it |
po.spec.ewridge |
0.9831 | 0.9907 | 480,000 | every row before it |
Every contender reaches the noise ceiling on a stationary stream, and
the gaps come from what a prediction saw. sklearn's fast form updates
once per 1,000-row batch, so a prediction can come up to 999 rows before
its update, and on the drifting stream it scores 0.9820. Called on every
row, partial_fit runs at 3,400 rows a second, which is Python's overhead
on each row. sklearn has the ecosystem: pipelines, GridSearchCV,
calibration, and far more use.
The two differ in what one fit can hold:
| sklearn | here | |
|---|---|---|
| a grid of six penalties | 3.8× the work: six estimators | 1.6× the work: one accumulator, six solves |
| several targets | one estimator each | one shared X'X |
| one fit per key | a dict of estimators | group=, one state per key |
| standardisation | a StandardScaler fitted on the whole frame can leak |
streaming, so it cannot leak |
| chunking | chunk invariance is a test | |
| saved model | a pickle | a versioned file that loads on every OS |
The exact solve wins in the first hundred rows of every group, and
sklearn wins on a very wide row by batching. An exact solve is right as
soon as its Gram, X'X, is full rank, about k rows in, where a
first-order method needs about 1/eta rows per direction. On 500 groups of
200 rows, ewridge scores an R² of 0.9693 on rows 25 to 50 against
SGDRegressor's 0.7277. At k = 10,000, ewridge's (k+1)² matrix
holds 860 MB and moves all of it on every row, so it runs 53 rows a
second, or 379 with gram_block_rows=1024. There, sgd at a learning
rate of 0.2 / k runs 33,271 rows a second from 0.89 MB, faster than
SGDRegressor's 20,007 in batches.
PERFORMANCE §19
has the full tables, the sweeps, and why a half-life gives no edge on
evenly spaced rows.
Scope and integrations
polars-online fits models to rows that are already read and in clock
order, so reading, joining and ordering the rows is the job of Polars or of
another tool before the bank. A bank can take its rows from a database
through Polars, or run as an operator inside a stream processor. Chunk
invariance means another engine's
batching cannot change the numbers. For a model this library does not fit,
such as gradient-boosted trees, po.gbdt builds XGBoost's and LightGBM's
training objects from files too large to hold.
What this is not
When a spec names a clock, sort the rows by it before the bank, because
polars-online neither buffers nor reorders rows that arrive late. A row that
arrives out of clock order is refused, unless its step back is larger than
restart_after_step_back, which starts the model over. A label that
arrives late can be held back with embargo.
clock, gap_cap, restart_after_step_back and session describe time
within a stream, and none of them waits for a late row.
polars-online does two things to a stream itself
(Preparing a stream). po.stream.refresh_time puts
series that tick at their own times on one grid, and
po.stream.with_windows computes exponentially weighted means, variances,
sums and rates over a window of the clock. It deliberately does not
provide the rest:
| not provided | use instead |
|---|---|
| connectors or ingestion | a Polars reader in the query before the bank, such as pl.scan_parquet, or pl.scan_arrow_c_stream for a database (Databases) |
| windows beyond those, such as a tumbling or sliding aggregation of any function, and asof or interval joins | Polars' group_by_dynamic, rolling, join_asof or join_where in the query before the bank, or a streaming framework such as Pathway |
| watermarks or a late-arrival policy | rows put in clock order before the bank, by Polars' sort or by a stream processor |
| distributed execution | one process, with a thread pool across (spec × group) (Parallelism) |
Databases: DuckDB and ADBC
To feed a bank from a database, sort the query by the clock with
ORDER BY in its SQL. Read its result with Polars'
pl.scan_arrow_c_stream (py-polars 1.43.0 or later), and chain
.online.fit_predict(specs) on that query. The rows cross through the
Arrow PyCapsule interface, so no pyarrow is needed. Install DuckDB or an
ADBC driver yourself: the package depends on neither.
This code uses the file ticks.parquet from Example data:
import duckdb
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", gap_cap=300.0,
half_life=100.0)
con = duckdb.connect() # an in-memory database
query = "SELECT * FROM 'ticks.parquet' ORDER BY t" # clock order, sorted by the database
fitted = (
pl.scan_arrow_c_stream(con.cursor().sql(query)) # a DuckDB relation, on a cursor of its own
.online.fit_predict([spec])
.collect()
)
# With ADBC, on a cursor of its own: cur.execute(query), then pl.scan_arrow_c_stream(cur.fetch_arrow())
examples/duckdb_cursors.py and examples/adbc_cursors.py run these steps against real databases.
Give every lazy query its own cursor, from con.cursor(), because a
cursor holds one open result. Both examples show the rule and the trap
of breaking it, and exit non-zero if either stops holding:
| database | what springs the trap | what comes back | what says so |
|---|---|---|---|
| DuckDB | a second query built on the same connection | the first query yields no rows | ConsumedSourceWarning |
| ADBC | the cursor executed again before its first query is read | the newer query one batch short, in clock order, and the older one with that batch added out of order: on SQLite, 198,976 and 201,024 of 200,000 rows | nothing, for the newer query; for the older one, a spec with a clock refuses the step back |
Do not close an ADBC cursor just after a read of its stream stopped
early. A read stops early on an error, such as a spec refusing a clock
that steps back, or when the consumer stops. Closing the cursor at once
then crashed the process with a segmentation fault, measured on SQLite,
adbc_driver_manager 1.12.0 and polars 1.44.2 at 200,000 rows. A wait of
3 s before the close, or no close before the process exits, avoided it.
Pathway
To run a bank inside a Pathway pipeline, wrap a
ModelBank in a stateful operator. The operator calls fit_predict on
each batch Pathway hands it, and Pathway checkpoints it with save_bytes
and load_bytes. Pathway does the ingestion, the event-time alignment and
the windowing before the operator. This code feeds the operator two
plain batches with a checkpoint and a restore between them, and its
predictions equal one run's over the same rows.
This code uses df from Example data:
# The operator: a bank that Pathway feeds and checkpoints.
class BankOperator: # the operator's state is the bank
def __init__(self, specs):
self.specs, self.bank = specs, po.ModelBank(specs)
def __call__(self, batch: pl.DataFrame) -> pl.DataFrame:
return self.bank.fit_predict(batch) # each ordered batch Pathway hands it
# Pathway's checkpoint, and its restore.
def snapshot(self) -> bytes:
return self.bank.save_bytes() # what Pathway's persistence stores
def restore(self, blob: bytes) -> None:
self.bank = po.ModelBank.load_bytes(blob, specs=self.specs)
# Feed it two batches, restarting between them.
spec = po.spec.ewridge("ridge", targets=["y"], features=["x0", "x1"], clock="t", gap_cap=300.0,
half_life=100.0)
op = BankOperator([spec])
first = op(df.head(200)) # plain batches stand in for Pathway's
op.restore(op.snapshot()) # a checkpoint and a restart between two batches ...
rest = op(df.tail(200)) # ... and the bank goes on where it was
# Compare with one run over the same rows.
one_run = po.ModelBank([spec]).fit_predict(df) # the same rows, with no restart
print("rows scored:", first.height + rest.height)
restarted = pl.concat([first, rest])
print("the same predictions as one run:",
restarted["ridge"].struct.field("pred_y").equals(one_run["ridge"].struct.field("pred_y")))
rows scored: 400
the same predictions as one run: True
examples/pathway_integration.py runs this operator over plain batches, so it runs without Pathway, which is not a dependency. Its sketch of the pipeline around the operator needs Pathway, installed under its own licence, and an input connector.
Training XGBoost and LightGBM on files larger than memory
po.gbdt builds XGBoost's and LightGBM's own training objects from
files too large to hold, so xgb.train and lgb.train run on them
unchanged. Both libraries bin each feature before they boost, and
binning reads every row. po.gbdt reads the rows in batches, in Rust, and
hands each library one batch at a time.
docs/GBDT.md says how it bins, what each library is
handed, and what Polars' read-ahead does to its memory. po.gbdt is
unstable: its keywords may change in any release.
This code writes a training file and a validation file whose target
depends on x0 and x1 but not x2, with noise of spread 0.3. The
validation rows are binned on the training rows' bins (ref=train), as
XGBoost requires of a set it evaluates on:
import numpy as np
import polars as pl
import polars_online as po
import xgboost as xgb
# A training file and a validation file, where y ignores x2.
rng = np.random.default_rng(4)
for name, n in (("train", 20_000), ("valid", 5_000)):
x0, x1, x2 = rng.standard_normal((3, n))
pl.DataFrame({
"x0": x0, "x1": x1, "x2": x2, # x2 plays no part
"y": np.sin(2 * x0) + (x1 > 0.5) + 0.3 * rng.standard_normal(n), # noise of spread 0.3
}).write_parquet(f"{name}.parquet")
# Bin both files, and train for 50 rounds.
features = ["x0", "x1", "x2"]
train = po.gbdt.xgboost_matrix("train.parquet", label="y", features=features) # an xgb.QuantileDMatrix
valid = po.gbdt.xgboost_matrix("valid.parquet", label="y", features=features, ref=train) # on train's bins
history = {}
booster = xgb.train({"tree_method": "hist", "max_depth": 4, "eta": 0.3}, train, 50,
evals=[(valid, "valid")], evals_result=history, verbose_eval=False)
# The validation error, and each feature's share of the gain.
rmse = history["valid"]["rmse"]
print(f"validation RMSE: {rmse[0]:.2f} after 1 round, {rmse[-1]:.2f} after 50 (the noise is 0.30)")
gain = booster.get_score(importance_type="total_gain")
print("share of the gain:", ", ".join(f"{f} {g / sum(gain.values()):.2f}" for f, g in gain.items()))
validation RMSE: 0.68 after 1 round, 0.31 after 50 (the noise is 0.30)
share of the gain: x0 0.70, x1 0.30, x2 0.00
Versions, testing and development
Version 1.0 starts this package's promise of a stable API: a minor release adds to it, and only a major release removes or renames a stable part. Each guarantee in What you can rely on is held by tests that run on Linux, macOS and Windows at every push. A checkout builds with uv and a stable Rust toolchain.
Versioning and the Polars pin
Two versions matter when you install this package. Its own version says
when to expect a breaking change. The range of Polars it accepts,
polars>=1.37.0,<3, is measured by this project, and Polars does not
promise it.
This package's own versioning
To keep your code working across the 1.x series, pin the major version
in your requirements, as polars-online~=1.0 or polars-online>=1,<2.
A minor release adds models, parameters, outputs and functions, and breaks
nothing stable. It can also correct numbers that were wrong, so pin the
minor as well if your numbers must not move. polars-online~=1.0.0 pins
the current minor: it takes that minor's patch releases and stops before
the next minor. The package follows semantic versioning
(CHANGELOG.md).
Code written for 0.x needs the changes the CHANGELOG lists under Changed (breaking) before it runs on 1.0. Version 1.0 renamed and removed parts of the 0.x API, to make it more standard and consistent. A spec keyword that 1.0 renamed is refused with an error that names the new one.
Stable means part of this package's API, and a breaking change to a
stable part needs a new major version. tests/test_api_surface.py
compares the stable parts with a snapshot of them, tests/api_surface.txt,
so a change to one shows as a diff in review. The command line's exit
status is held by a test of its own. An unstable part can change its form
in any minor release:
| what it covers | |
|---|---|
| stable | everything in polars_online.__all__; the po.spec constructors, their keywords and their defaults; the functions of the helper modules po.eval, po.gram, po.ops and po.stream; the .online namespace; the output field names; the columns, and their types, of the frames a bank and po.spec return, and of the closed-groups frame; the words each string-valued parameter takes; the TOML config keys; the command line's flags, its exit status and the files it writes; and the environment variables the package reads |
| unstable | the file format of a window run's state (with_windows(save_state=...)); the written form of a formula target, in a TOML config or inside a bank's state; the Arrow export: fit_predict_arrow, predict_arrow and ArrowStruct; and the modules po.sim, po.corr and po.gbdt. Each says so in its docstring, and warns with UnstableWarning when used with POLARS_ONLINE_WARN_UNSTABLE=1 set |
| not stable | any name that starts with an underscore, apart from the two columns po.stream.embargo adds, _online_role and _online_role_weight, which are stable; the Rust crates (online-core, online-polars, online-cli and online-py), which are not published to crates.io, so nothing outside this repository builds against them; and a model's numbers within the 1e-12 tolerance the golden tests pin, which depend on the platform and the Polars version; and the text the command line prints (its summary, notices and errors) and the timing lines POLARS_ONLINE_TIMING=1 asks for, which are written for a person and may be reworded in a minor release |
A changed default or meaning needs a major release, and a fix to numbers that were wrong a minor one. A number is wrong when it disagrees with the model's stated definition, the update equations its docstring gives:
| a change | the release it needs |
|---|---|
| a fix that moves no number | patch |
| a fix to numbers that were wrong | minor, declared in the CHANGELOG with the difference scripts/compare_release.py measures from the last release |
| a new model, parameter, output or function | minor |
| a stable name renamed | minor, and the old name keeps working until the next major. Using it warns with a PolarsOnlineDeprecationWarning that names the new one, as the next paragraph says |
| a default or a meaning changed, or a stable name removed | major |
| a change to an unstable part | minor, declared in the CHANGELOG |
A spec parameter renamed after 1.0 is forwarded by a table, and any
other renamed name by a forwarding of its own. The table reads a spec
builder's old keyword, a spec dict's old key and a TOML file's old key as
the new one, at any depth of a spec. It does not reach another function's
keyword, a function, a word a parameter takes, a command-line flag, an
environment variable, an output field or a frame column. Each of those
ships its own forwarding in the release that renames it: a check at the
function's top, a stub, an alias or a second column. Both warn with
PolarsOnlineDeprecationWarning, and both last until the next major.
A state file a 1.x build saved loads in every 1.x build after it, and a state file a 0.x release saved does not load in 1.0. The oldest layout a build loads stays at 1.0's for the whole 1.x series, and each change to a state's layout ships a loader for the layout before it. A saved state of each kind, frozen in the test suite, checks it three ways. The current layout's fixtures load, continue to the bit, and save back byte for byte. A previous layout's fixtures load through its loader, continue to the bit, and save back as the current layout's fixture bytes of the same case. So the loader turns a state into exactly what this build writes from the same rows. The two unstable formats, a window run's state and a formula target inside a bank's, are outside that promise. A 0.x state is refused by its schema version, with an error that says to refit it from its input (Saving, loading and serving).
Moving the Polars floor up needs a minor release, and dropping a Polars major version a major one. A resolver that installs the new release upgrades Polars with it, inside the same Polars major, so no code has to change for a higher floor:
| a change to the Polars range | the release it needs |
|---|---|
| widening it to admit a newer Polars | minor |
| moving the floor up | minor |
| capping it below a Polars that broke this library, as How the pin moves describes | patch |
| dropping a Polars major version | major |
What is pinned
The wheel carries its own Rust polars, pinned exactly, and shares one
process with the py-polars you install. Data crosses between the two only
through the interfaces listed under
Which interfaces carry a promise, so
py-polars can be any version in the range. The lowest py-polars depends on
what you use:
| py-polars | rust polars | pyo3-polars | pyo3 | Python |
|---|---|---|---|---|
| >= 1.37.0, < 3 (built and tested against 2.0.0) | 0.55.2 | 0.28 | 0.29 | ≥ 3.12 (abi3-py312) |
| what you use | the lowest py-polars | why |
|---|---|---|
ModelBank alone |
1.28.1 | |
lf.online.fit_predict, ModelBank.fit_predict_batches, ModelBank.fit, with_windows and refresh_time |
1.37.0 | they read with LazyFrame.collect_batches. Before 1.37.0 it reads ahead of a slower reader without limit: draining a 1 GB parquet scan held 1.6 GB on 1.34.0 and 1.36.1, and 0.33 GB on 1.37.0 |
| the examples that stream a DuckDB, ADBC or pyarrow source into a bank | 1.43.0 | for pl.scan_arrow_c_stream |
The suite passes on 2.0.0, the pin, at every change, and on the floor,
1.37.0, before every release and once a month
(How the pin moves). It
passed on the floor on 2026-10-12. There, 25 tests skip, each naming the
newer Polars it needs: most check against a Polars function that has
changed since, and the rest stream a database or a pyarrow reader through
pl.scan_arrow_c_stream. The suite has also passed on 1.38.1, 1.44.1,
1.44.2 (the pin until 2026-10-08) and 2.0.0-rc.1 with identical numbers.
tests/test_scaffold.py asserts the pin and the range, and
docs/RELEASE-READINESS.md has each run with
its date and what differed.
Which interfaces carry a promise
Treat a break on a new Polars as expected maintenance, because three of
the four ways this library crosses into Polars carry no promise from
Polars. The two that stream data, ModelBank and the IO plugin behind
lf.online.fit_predict, are among the three, so check those paths first:
| interface | used by | its promise |
|---|---|---|
| pyo3-polars' extension types | ModelBank |
none beyond the latest definitions working with the latest Polars: provided "for convenience" |
| the IO plugin | lf.online.fit_predict, lf.online.predict, and with_windows and refresh_time on a query |
none: documented, but @unstable in py-polars |
an expression's serialized form, expr.meta.serialize(format="json") |
the window formulas, read into this library's own tree | none: Polars calls it unstable across versions; its shapes were measured the same on 1.34.0 and 1.44.2 |
| the Arrow PyCapsule interface | fit_predict_arrow |
an Arrow specification, which py-polars and pyarrow consume. Only the output side uses it, and the frame still goes in as a Polars frame |
A mismatch raises an AttributeError before any data moves, and does
not crash the process. ModelBank moves data across the boundary
through the Arrow C Data Interface, with two private methods of py-polars,
_export and _import, so a Polars without them fails at once.
Which interfaces carry a
promise in
the release guide says how a break shows on each interface, and which
tests watch for it.
Testing
Each guarantee in What you can rely on is held
by a test, against an oracle or an invariant. An oracle is a reference
the model cannot share a bug with, such as scikit-learn's Ridge for
ewridge. An invariant is a property every run must keep, checked over
streams Hypothesis generates with mixed nulls, clocks that repeat or jump,
values at ±1e8, zero weights and tiny groups:
| guarantee | held by |
|---|---|
| honest predictions | changing a row's own target never changes that row's prediction, for every model (tests/test_properties.py) |
| bounded memory | a 10-million-row run whose memory and state stay bounded (tests/test_soak.py, opt-in with pytest -m soak, weekly) |
| any chunking | a stream fed in chunks of 1 to 13 rows gives one chunk's output, on generated streams for every regression model (tests/test_properties.py) and on a fixed stream for every model, which is also saved and loaded halfway (tests/test_semantics_all_models.py) |
| any thread count | a bank's pool at 1 thread against 8, its output compared bit for bit (tests/test_portability.py) |
zero weights and one weight_sum |
every model held to the same weight_sum recursion (crates/online-core/tests/model_contract.rs) |
| each model's numbers | a reference written from its update equations, and another library wherever one computes the same quantity, such as statsmodels, scipy and river |
The suite has 2,207 Rust tests and 6,262 pytest cases, counted on
2026-10-11. It runs on Linux at Python 3.12, 3.13 and 3.14, and on macOS
and Windows at 3.12 and 3.14, at every push to main and every pull
request. docs/TESTING.md is the ledger of what each part
proves: every reference and its agreement, every invariant, and where the
suite runs.
Development
CONTRIBUTING.md has the commands to build and test from a checkout, and how to make a change.
| to find | read |
|---|---|
| every document, and which to read for what | docs/README.md |
| a map for coding agents, at the repo root and on the docs site | llms.txt (llmstxt.org) |
the API reference, built from the docstrings and published from every green push to main |
https://hgilde.github.io/polars-online/ |
| what each release changed | CHANGELOG.md |
| how to cut a release, and which Polars versions are promised | docs/RELEASE-READINESS.md |
| how to add a model: every layer it touches, and the test that catches each one skipped | docs/EXTENDING.md |
| the rules this documentation is written to | docs/WRITING.md |
License
Apache-2.0. See SECURITY.md to report a vulnerability.
1 maintainer
Metadata
1 maintainer
Release files for polars-online 1.0.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| polars_online-1.0.0.tar.gz | 2.6 MB | Details |
Built distributions (wheels)
| File | Reset | |||
|---|---|---|---|---|
| polars_online-1.0.0-cp312-abi3-win_amd64.whl | CPython 3.12 | abi3 | Windows x86-64 | Details |
| polars_online-1.0.0-cp312-abi3-musllinux_1_2_x86_64.whl | CPython 3.12 | abi3 | Linux musl 1.2+ x86-64 | Details |
| polars_online-1.0.0-cp312-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl | CPython 3.12 | abi3 | Linux glibc 2.17+ x86-64 | Details |
| polars_online-1.0.0-cp312-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl | CPython 3.12 | abi3 | Linux glibc 2.17+ ARM64 | Details |
| polars_online-1.0.0-cp312-abi3-macosx_11_0_arm64.whl | CPython 3.12 | abi3 | macOS 11.0+ ARM64 | Details |
| polars_online-1.0.0-cp312-abi3-macosx_10_12_x86_64.whl | CPython 3.12 | abi3 | macOS 10.12+ x86-64 | Details |
Total release size: 163.2 MB
Release files / polars_online-1.0.0.tar.gz
| Download URL | polars_online-1.0.0.tar.gz |
|---|---|
| Size | 2.6 MB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
2aedd251c013a53c802507eb3345492862411fd8cf9a5992724713d3f5ba598f
|
|
BLAKE2b-256 checksum How to use checksums |
8bbdabe9e917e8db99ab1c7ceb12495e96289c431bf44fe28b0dd7f8d8bb8ec5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-win_amd64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-win_amd64.whl |
|---|---|
| Size | 29.4 MB |
| Tags | CPython 3.12 Windows x86-64 abi3 |
|
SHA-256 checksum How to use checksums |
68da3a5412da3cd858ca4c0551e19cdbbfbb8a442447dcda83e276c04415a065
|
|
BLAKE2b-256 checksum How to use checksums |
bb6011a5b1ca3175481273541f0d838675f5deb74f6f1003b423adc4467c47b1
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-musllinux_1_2_x86_64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-musllinux_1_2_x86_64.whl |
|---|---|
| Size | 27.9 MB |
| Tags | CPython 3.12 Linux musl 1.2+ x86-64 abi3 |
|
SHA-256 checksum How to use checksums |
5a3106ea79bb1e9181e4e460c9f1671425ab6800403a707521ed6ddd99dd0c30
|
|
BLAKE2b-256 checksum How to use checksums |
2bde4233a8b0ff918f1fa94409bf46d59c572006401fb24eb83b5217df24fcc5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl |
|---|---|
| Size | 27.5 MB |
| Tags | CPython 3.12 Linux glibc 2.17+ x86-64 abi3 |
|
SHA-256 checksum How to use checksums |
62b885e115cd2d0a813dc4346883b40b0fb554bc8f92bfd798917d78189f69e0
|
|
BLAKE2b-256 checksum How to use checksums |
fa945c7cf837b169b4cfec6578ffae037ce7e875078162dd3a04c403c65bf93f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl |
|---|---|
| Size | 25.0 MB |
| Tags | CPython 3.12 Linux glibc 2.17+ ARM64 abi3 |
|
SHA-256 checksum How to use checksums |
81946d350b610f39274b40664e0f42e26e13095c291f8be54b82e9e9f29697b4
|
|
BLAKE2b-256 checksum How to use checksums |
a215bfc111a89c128e243c0d9358ce32658df9600b5496056c263fd0ded692b3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-macosx_11_0_arm64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-macosx_11_0_arm64.whl |
|---|---|
| Size | 24.1 MB |
| Tags | CPython 3.12 abi3 macOS 11.0+ ARM64 |
|
SHA-256 checksum How to use checksums |
7b9cd46bcc3c16a2b4369d3ded101e6856fce0848a45e6e196139ba6a256b34e
|
|
BLAKE2b-256 checksum How to use checksums |
c0e5f48b8f3e4725847410e7a727da069ad99bfd3d34991b2b8622f29766a541
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency logRelease files / polars_online-1.0.0-cp312-abi3-macosx_10_12_x86_64.whl
| Download URL | polars_online-1.0.0-cp312-abi3-macosx_10_12_x86_64.whl |
|---|---|
| Size | 26.8 MB |
| Tags | CPython 3.12 abi3 macOS 10.12+ x86-64 |
|
SHA-256 checksum How to use checksums |
d328fbb42bc2f567b6ec979be6b518a484834c544ca7d8be49e14e64ae99855e
|
|
BLAKE2b-256 checksum How to use checksums |
e5230524ef3432fb177effcab5db7b8d770a630f804bb7d5cfe6da846366f294
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 13, 2026.
Transparency log
