`polars_online`¶

Streaming / online regression models for Polars.

Three entry points share one Rust core (see docs/PLAN.md):

ModelBank, chunk-fed, memory O(state) not O(data) – and as a plan, lf.online.fit_predict(specs), a LazyFrame that streams (df.online.fit_predict(specs) for a frame in memory);
run(), or the online CLI: parquet, ipc, csv or ndjson in and out;
the expression namespace, pl.col("y").online.<model>(...), for a frame in memory only: polars calls it with the whole column in either engine, so every use warns (InMemoryExpressionWarning, polars_online._expr).

Errors follow one contract throughout, and each docstring says which of it applies: a file that cannot be read or written is the OSError subclass for what went wrong (FileNotFoundError, PermissionError, …), with the path in the message; a value that is refused – a spec parameter, a config key, what a column holds – is ValueError naming the spec and the parameter or column; a wrong type is TypeError; a spec name or position a bank has not got is KeyError or IndexError; a bank fed from two threads at once is RuntimeError. Inside a polars plan the same messages arrive as polars’ ComputeError. A refused chunk never changes a bank, and a failed run never replaces an output or a state file.

class polars_online.ModelBank(specs: Iterable[dict[str, Any]])[source]¶

Bases: object

Runs a list of specs over ordered chunks, one state per (spec, group).

Chunks must arrive in stream order within each group (the clock must not run backwards unless the spec’s reset semantics say so). Feed LazyFrame.collect_batches() for out-of-core streams.

specs are the dicts the polars_online.spec builders make. ValueError when there are none, when two share a name, or when a dict is not a spec (the message names the field).

A bank is one ordered stream, so it is not for two threads at once: a method that finds the bank in use on another thread raises RuntimeError saying so rather than interleave with it (fit_predict() releases the GIL while it works). predict() learns nothing, so any number of predict calls may overlap; only a fit_predict in flight refuses them, and they refuse it.

property specs: list[dict[str, Any]]¶

The specs this bank runs, as the dicts the builders made.

A state file carries them, so a bank loaded from one reports its specs without being told what they are – every field, including the ones left at their default. That is what makes a state file self-describing: with groups(), output_fields() and the four diagnostic tables, a file can be walked by a caller who knows nothing about it in advance.

A copy, and read-only. The bank’s behaviour comes from the Rust state built at construction, not from this list, so a mutation here could only disagree with it – and used to, silently: editing bank.specs[0]["features"] in place left coef() labelling coefficients from a spec the bank was not running. Assigning to bank.specs raises AttributeError, and mutating what it returns changes nothing.

rows_seen() → int[source]¶: Rows fed so far, over every chunk and group – skipped rows and dropped groups included, so not the sum of groups().

groups(spec: str | int | None = None) → DataFrame[source]¶

The groups the bank holds state for: one row per (spec, group).

group is the key as a string – "" for a spec without a group column, null for rows whose key was null, as in solve_failures(). rows_processed counts the group’s rows that the null policy did not skip, and last_clock is its last clock value (null before the first row, or on a row-count clock). State lives until drop_groups() removes it, so this is how a long-running bank finds the groups that have gone quiet:

stale = bank.groups().filter(pl.col("last_clock") < now - 30 * 86400)
bank.drop_groups(stale["group"])

spec, a name or a position, narrows the table to one spec: KeyError for a name the bank has not got (the message lists the names), IndexError for a position it has not got.

drop_groups(keys: Iterable[str | None], spec: str | int | None = None) → int[source]¶: Forget the state of these groups, in every spec or in one, and return how many streams were dropped. Keys are as groups() reports them; a key the bank does not hold is not an error, it just drops nothing. A dropped group starts cold if it appears again, exactly as a never-seen one would; nothing else in the bank changes, and rows_seen() still counts the rows it was fed. spec is as for groups() (KeyError / IndexError for one the bank has not got).

fit_predict(df: DataFrame) → DataFrame[source]¶

One chunk in; the chunk plus one struct column per spec out.

The struct is named after the spec (out["m"]; its fields are output_fields()), and pred in it is out-of-sample: computed from the state before the row updates it. Chunk boundaries never change the numbers, only the cadence at which coef is reported.

Raises TypeError for anything but a DataFrame (a LazyFrame is told to collect, or to feed fit_predict_batches()), and ValueError – naming the spec and the column – when a column a spec reads (target, feature, clock, session, weight, group) is not in the frame; a target, feature, clock or weight column is not numeric (a datetime clock is refused rather than read as its epoch integer: cast it to the unit halflife and max_dclock are in); the clock has a null or non-finite value; a weight is negative (null skips the row); a spec is named like an input column, which the struct would replace; or a group’s clock runs backwards under on_clock_reset="error" (the other policies absorb it). A refused chunk leaves the bank exactly as it was, so the corrected chunk can be fed. RuntimeError when the bank is in use on another thread (class docstring).

predict(df: DataFrame) → DataFrame[source]¶

Score a frame against the bank as it stands, learning nothing.

Every row gets the struct fit_predict() would give it as the next row of its group’s stream, computed from the current state; the bank is left exactly as it was, so the call is safe from any number of threads at once and row order does not matter. This is the serving side of a trained bank: load once, predict per request, and fit_predict the rows later, in order, once their targets arrive.

The frame needs each spec’s feature and clock columns. Target columns are optional – present, they give resid and the standardized residual; absent, those are null. The session column is optional and feeds session_gap; a weight column is not read. A trend model (holt) extrapolates over the clock distance from the row it last learned, capped by max_dclock, and a kalman with revert_halflife shrinks its coefficients over that distance exactly as the next fit_predict row would; the other coefficient models predict from their current coefficients regardless of the clock.

Per field: n_eff, lam_selected, sigma, the residual quantiles, autocorrelation and the metrics are the values the bank holds, frozen; coef is filled on the last accepted row (the same coefficients score every row); drift never fires; rows of a group the bank has never seen, or without usable features, are null throughout, as a skipped row is in fit_predict.

Raises what fit_predict() raises for the same frame – a missing or non-numeric column, a bad clock value – except that a missing target is not an error, and RuntimeError only when a fit_predict is in flight on another thread.

fit_predict_batches(batches: Iterable[DataFrame]) → Iterable[DataFrame][source]¶: Lazily map fit_predict() over an iterator of chunks: each is fed as the generator reaches it, so lf.collect_batches() streams through the bank one chunk at a time. Whatever fit_predict raises for a chunk, this raises there; the chunks before it have been learned from.

coef(spec: str | int | None = None, group: str | None = None) → DataFrame[source]¶

The coefficients behind a fit: one row per (spec, group, instance, position), so a bank loaded from a state file answers “what are the betas?” without a row of data:

bank = po.ModelBank.load("state.bin")
betas = bank.coef()             # every spec that has coefficients
just_one = bank.coef("ols")     # or name one
wide = betas.pivot("term", index=["group", "instance"], values="coef")

Like last_row(), summary() and describe(), it takes every spec by default and leads with a spec column, so the frames of several banks stack with a plain concat. Sweeping this way skips a spec that has no coefficients – an ew_cov, which emits statistics, or a seqtest, which emits evidence – since the question was “the coefficients in this bank” and those have none. Naming one of them still raises, because then the question was about that spec. A bank with no coefficients at all gives an empty frame.

The values are what the output’s coef field reported on the last row each stream learned from: the fit after that row, which the next row’s pred is computed from. Laid out by polars_online.spec.coef_index():

group, instance: As groups() and gram() report them: the key as a string ("" for a spec without a group column) and the decay instance’s field suffix ("@h500", or "" for a single one).
n_eff: The accumulated weight behind the fit – what the next row’s n_eff field reports. The solve schedule (solve_every, default halflife/50 or every row for halflife=inf, and max_rows_between_solves) decides when a stream first solves, not min_periods: pred waits for min_periods, coef does not, so a fit with n_eff below it is over fewer rows than the spec asks for (a solve over fewer rows than terms is a jittered one, counted by solve_failures()).
position, target, ridge, feature_set, lambda, term: coef_index()’s columns: position indexes the flat coef list, term is "intercept", a feature name, or "level"/"trend" for holt.
coef: The value, in the features’ original units; null until the stream’s first solve, as coef is on those rows.

spec and group are as for gram(): KeyError / IndexError for a spec the bank has not got, and a group it has never seen gives an empty frame with the same columns. ValueError for a named ew_cov spec, which emits statistics, not coefficients, and for a named seqtest spec, which emits evidence.

last_row(spec: str | int | None = None, group: str | None = None) → DataFrame[source]¶

The output struct as it stood on the last row each stream learned from: one row per (spec, group), the struct’s fields unnested after spec and group.

It is the row fit_predict() reported for that row, field for field – pred, resid, sigma, the metrics, the residual quantiles, n_eff, and coef when the row carried it (a chunk’s last row does; coef() has the coefficients whichever row was last). It travels with the state, so a bank loaded from a file says how each model was doing without its output frame, and a directory of fits compares without keeping the last row of every output:

fits = sorted(Path("fits").glob("*.bin"))
table = pl.concat(
    [po.ModelBank.load(f).last_row().with_columns(file=pl.lit(f.name)) for f in fits],
    how="diagonal_relaxed",
)
table.sort("ic_y", descending=True)  # with emit_metrics=True

spec, a name or a position, narrows the table to one spec (KeyError / IndexError for one the bank has not got, as groups()); group to one group, and a group the bank has never seen gives an empty frame. Specs with different fields are stacked diagonal_relaxed, so a field one spec has not got is null on its rows. A group with no learned row yet – every row skipped so far, or a state file written before 0.2.0 – is a row of nulls. predict() does not move it, and a chunk that ends in skipped rows leaves the row before them.

summary(spec: str | int | None = None, group: str | None = None) → DataFrame[source]¶

What each stream has been fed: one row per (spec, group).

Counts and ranges over every row routed to the group since its state began – undecayed, so they say what the model was trained on rather than what it still remembers – and kept in the state file, so a bank loaded from a file says it too. The columns:

spec, group: As groups() reports them.
rows_fed: Rows routed to the group, skipped or not.
rows_processed: Rows the models saw: every feature and the weight usable.
rows_skipped: rows_fed - rows_processed: a null, NaN, infinite or out-of-bound feature or weight.
rows_learned: Processed rows with a positive weight and, for a model with targets, at least one usable target.
rows_zero_weight: Processed rows with weight 0 (the clock moved; nothing learned).
weight_sum: Sum of the processed rows’ weights (1 per row without a weight column).
clock_min, clock_max, last_clock: The clock range fed and the last value; null on a row-count clock.
session_changes: Rows whose session differed from the previous row’s.
clock_backwards: Rows whose clock fell below the previous row’s within a session (what on_clock_reset decided about).
resets: Rows at which session_gap="reset" or on_clock_reset="reset_state" restarted the stream.

spec narrows to one spec (KeyError / IndexError for one the bank has not got), group to one group; a group never seen gives an empty frame. A state file written before 0.2.0 carries no summary: its groups report spec, group, rows_processed and last_clock, and nulls elsewhere – for good, since a count that began at the load would read as the whole history. predict() moves none of it, and feeding the same rows in one chunk or a thousand gives the same numbers to the bit.

describe(spec: str | int | None = None, group: str | None = None) → DataFrame[source]¶

Per-column statistics of what each stream has been fed: one row per (spec, group, input column), in spec order – features, then targets, then the weight column.

column and role ("feature", "target", "weight") name the column; count and null_count partition the rows fed (a value counts when finite and within the input bound, as the models take it, and is a null otherwise – polars nulls, NaN, infinities and magnitudes beyond the bound alike); mean, std (sample, ddof=1; null below two values), min and max are over the counted values, undecayed and in row order, so chunking cannot move them. An unsupervised model lists no targets, an ew_class label column has its counts only, and a comparison’s target is the difference of residuals it tests, named as the spec names it.

spec and group narrow the frame as in summary(). A state file written before 0.2.0 lists its columns with every number null; see summary().

gram(spec: str | int, group: str | None = None) → list[dict[str, Any]][source]¶

The EW accumulators behind a spec’s fit, per group and instance.

Returns one dict per (group, decay instance) with:

group, instance

The group key as groups() reports it ("" for a spec without a group column, None for a null key) and the instance’s field suffix ("@h500", or "" for a single instance).

columns, targets

What the axes mean: the spec’s features, with "intercept" first when the spec has one (the term names of polars_online.spec.coef_index()), and the target names the per-target arrays are indexed by. targets is empty for ew_cov, which learns from none. They are what makes the mapping self-describing, so polars_online.gram can take a column by name.

n_eff

Accumulated weight behind these moments.

n_kish

Kish’s effective sample size, n_eff**2 / sum(w**2) – the number of equally weighted rows these moments are worth, and what a standard error computed from them divides by. n_eff counts weight, not rows, so it is not a sample size: (1 + lam) / (1 - lam) is the Kish size of an exponentially weighted window, whatever the halflife’s units. None before the first row.

It is scale-free: decay divides n_eff and sum(w**2) by the same factor, so n_kish does not shrink when a stream goes quiet. It says how many rows these moments average, not how old they are – n_eff and target_weights are what say that.

means

EW column means, shape (k,).

comoments

Centered co-moments, shape (k, k) – the EW analogue of a centered X'X / n. Centered is what makes it accurate at large offsets (E11b).

cross_moments

Per-target uncentered cross-moments E[z*y], shape (n_targets, k). Empty for ew_cov.

target_weights

Per-target accumulated weight, shape (n_targets,). Differs from n_eff when targets have different null patterns.

target_means, target_vars

Per-target EW mean and centered variance of the target itself, shape (n_targets,), in the same arithmetic as comoments – a target’s variance here is the variance an ew_cov over that column would report, to the bit. Empty for ew_cov.

target_n_kish

Per-target Kish effective sample size, target_weights**2 / sum(w**2) over that target’s rows; nan for a target that has not seen a weighted row. Empty for ew_cov.

lags, lag_comoments

The lags an ew_cov(lags=[...]) accumulates at, in the order given, and their cross-moments as an (L, k, k) array: lag_comoments[l][a][b] is E_w[d_a(t) * d_b(t - lags[l])], both deviations against the mean before the row (ENHANCEMENTS E56). Both None for a spec without lags. The matrix is not symmetric for a lag above zero – a leading b is not b leading a – and lag 0 would be comoments exactly.

The target moments are what makes the export a complete sufficient statistic (ENHANCEMENTS E45). With the cross-moments alone there is no residual variance, no R^2, no information criterion and no standard error to be had from a saved Gram, because every one of them needs Var[y]:

beta = solve(raw, cross_moments[t])           # the model's own fit
resid_var = target_vars[t] - beta[1:] @ comoments[1:, 1:] @ beta[1:]
r2 = 1 - resid_var / target_vars[t]

Under a window everything here is the window’s – the accumulators the fit coef reports was solved from, so po.gram.solve on it is that fit – except the target moments, which the window’s snapshots do not carry: target_means, target_vars and target_n_kish are None there rather than the whole history’s (review 2026-09-12, S19).

A state saved before task 38 has none of them: n_kish, target_means, target_vars and target_n_kish are None there, for that state’s whole remaining life. The weight sums behind them cannot be replayed, and a sum(w**2) accumulated from the resume point against an n_eff from the whole stream would report an effective size too large by the length of the history – a wrong number where None is the true answer.

The two moment forms differ, and mixing them gives a silently wrong answer rather than an error, so the bridging identity is worth stating plainly:

raw = comoments + np.outer(means, means)
raw @ beta[t] == cross_moments[t]     # up to the ridge term

Values are in the features’ original units. The intercept, when the spec has one, is column 0: a constant 1, so it has zero variance in comoments and raw[0] == means.

Only models that keep a co-moment matrix report – ewridge, lasso and ew_cov. The others yield nothing: rls and kalman track an inverse, and the gradient models keep no second moment at all.

Why this exists (ENHANCEMENTS E30): the accumulators are the expensive part, and they are already exact, centered, decayed on the model’s own clock with session and max_dclock handling, and resumable. Anyone wanting to do something other than our solve with them – a custom penalty, an information criterion, cond(G), a scree plot, forward stepwise, orthogonal matching pursuit, or simply to check a fit by hand – previously had to recompute X'X from raw data in a second pass. These come from one pass over data that is never materialized, at every point in the stream rather than one, and they are the same matrices the deployed model solves against.

This is not a speed claim: for a single batch Gram over materialized data, BLAS dgemm is blocked, vectorized, and comfortably faster.

spec is a spec name or position (KeyError / IndexError for one the bank has not got, as for groups()); group narrows the list to one group. A group the bank has never seen gives an empty list, as does a model that keeps no co-moments – neither is an error.

Requires numpy, which is not a dependency of this package – polars does not require it either, and one optional accessor is no reason to put it on every install. pip install polars-online[numpy] adds it; without it the call raises ModuleNotFoundError saying so.

marginal(spec: str | int, group: str | None = None) → DataFrame[source]¶

The pairs a marginal spec keeps: one row per (group, instance, feature, target), in spec order – groups sorted, targets in spec order, features in spec order within each.

group and instance: As gram() reports them ("" for a spec without a group column; "" for a single decay instance, else the field suffix such as "@h500").
feature, target: The pair’s columns, by name.
n_eff: The target’s accumulated weight W_t: rows where the target was present, weighted and decayed. Differs from the struct’s n_eff when targets have different null patterns.
n_kish: W_t^2 / Q_t with Q_t the accumulated squared weight: the Kish effective sample size, the number of equally weighted rows that carry the same information. Null before the target’s first row.
mean_x, var_x, mean_y, var_y, cov: The pair’s EW moments – population form, over the decayed weights, so var is never negative.
corr, beta, t: cov / sqrt(var_x var_y), cov / var_x and corr * sqrt((n_kish - 2) / (1 - corr^2)); null until n_eff reaches the target’s min_periods, and null where undefined (a constant column, n_kish <= 2).

With lags (E66), four more list columns and four numbers:

lagcorr_xx, lagcorr_yy: Each series’ own autocorrelation at the configured lags.
lagcorr_xy, lagcorr_yx: The feature now against the target l rows back, and the reverse. A feature whose lagcorr_yx[0] exceeds its corr leads the target; one whose lagcorr_xy[0] does follows it.
n_serial, t_serial: n_kish divided by Bartlett’s serial-dependence factor, and the statistic against it. Null without serial_rule.
phi_x, phi_y: The per-row decays fitted under serial_rule="geometric"; null otherwise, and null when fewer than two kept lags are positive.

With bins or bin_edges (E67), the nonlinear view. These columns are present whenever the spec asked for bins, holding empty lists and nulls until the edges are fixed:

bin_edges: The feature’s interior edges, fixed once and never moved. Ragged: a feature keeps only the bins it can support, so a binary feature has two bins and a constant one has a single bin.
bin_n, bin_mean_y, bin_var_y: The target’s weight, mean and variance inside each bin – the feature’s response curve. One more entry than bin_edges, since the outer two bins are open. A bin no row has landed in has bin_n = 0 and null for the two moments.
split_gain, split_at: The fraction of the target’s variance removed by the best single cut of the feature, and where that cut falls. Directly comparable with corr ** 2, so split_gain - corr ** 2 is the nonlinear surplus.
split_gain_t: The t a corr would need to match that gain, against n_serial where there is one. Optimistic, because the cut was chosen by maximising over the candidates – a ranking, not a p-value.

The pairs are read from the state, so a bank loaded from a file reports them as the bank that saved it would, and feeding the rows in one chunk or a thousand gives the same numbers to the bit. A group the bank has never seen gives an empty frame; a spec that is not a marginal is refused (ValueError) – an ew_cov’s moments are read with gram(). spec is a name or position (KeyError / IndexError for one the bank has not got).

closed_groups(spec: str | int | None = None, *, drop: bool = True) → DataFrame[source]¶

The groups that have finished and not yet been read (docs/ENHANCEMENTS.md E54), oldest first, as one long frame.

A spec with group_close emits a group’s accumulators at the moment the bank can prove no further row will join it – a key smaller than the largest one fed so far under "monotone", or a session that has ended under "session" – and then drops the stream. That is what keeps a bank over an unbounded key space bounded: without it, every key ever seen stays in memory.

One row per (group, decay instance), with the common columns

spec, group, instance, session: Which stream closed. session is the value of the span that ended under group_close = "session", and null under "monotone".
n_eff, n_kish: As gram() reports them, at the moment of the close.
rows_fed, rows_learned, clock_min, clock_max: The span’s own summary() counts and clock range.

and then a block per kind, present when any spec of the bank closes groups and is of that kind, null on the rows of other kinds: columns, means, comoments, targets, target_means, target_vars, target_weights, target_n_kish and cross_moments for a kind that keeps accumulators; coef for every kind that reports one; eig_vals and eig_vecs for an ew_cov with pca; pair_* for a marginal.

The pair_* block is marginal()’s frame turned on its side: pair_feature and pair_target name the pairs, and every other marginal column becomes pair_<column> holding one entry per pair in the same order. A column that is a list per pair there – the lagcorr_* family with lags, bin_edges, bin_n, bin_mean_y and bin_var_y with bins – is a list of lists here, and those blocks follow the same rule as the rest: present when any closing marginal asked for them, null on the rows of one that did not. A nested list has no CSV form, so a closed frame with them is for parquet or the frame itself.

comoments is the upper triangle with the diagonal, row by row (k(k+1)/2 numbers), and cross_moments is row-major (n_targets, k). polars_online.gram.from_row() expands both and hands back exactly what gram() would have returned for that group – field for field and bit for bit, since one builder makes them both. The exact solve on a closed group is then one line:

po.gram.solve(po.gram.from_row(row))

eig_vecs are signed for continuity with the previous closed row of the same (spec, instance) – under "monotone" the previous group, and under "session" the same group’s previous close, never the previous chunk – so a component’s sign is stable along the sequence of closes and a sign flip between two rows is a real rotation rather than an eigensolver’s arbitrary choice.

drop (the default) removes what it returns from the queue, which is what a driver draining per chunk wants; drop=False peeks. The streams are dropped when they close, never when this is called: a bank’s memory must not depend on the caller polling. What is undrained is saved with the state, so a driver that saves between chunks does not lose rows silently.

spec narrows the frame to one spec’s rows (a name or a position; KeyError / IndexError for one the bank has not got). predict() never closes anything.

solve_failures() → dict[str, dict[str | None, int]][source]¶

Jittered or failed matrix factorizations so far, per spec and group.

A solve never returns NaN silently (docs/PLAN.md section 7): a near-singular system is retried with escalating diagonal jitter, and total failure keeps the previous coefficients. Both cases are counted here, so a nonzero value means the inputs are degenerate (constant or collinear features, or far too few observations for the feature count), not that anything crashed. Models that do not factorize – rls, kalman, ftrl – always report 0.

bocpd counts rows here too: a row whose predictive could not be evaluated reports nulls and leaves the run-length posterior where it stands, and the count is what makes a run of them visible.

output_fields() → dict[str, list[str]][source]¶: Spec name -> the field names of its output struct, in order – polars_online.spec.output_fields() for every spec in the bank. The fields are fixed by the spec, so this is the output schema before any row is fed.

save(path: str | Path) → None[source]¶

Versioned msgpack state; loads on any supported OS.

Written to a temporary sibling and renamed into place, so an interrupted save leaves the previous state where it was rather than truncating it (docs/IMPROVEMENTS.md C6). The rename is preceded by a filesystem sync, which is what a resumable file costs: ~4 ms on macOS, against ~0.5 ms for serializing 500 groups. Save every chunk and the sync dominates; save every hundredth and it disappears.

Raises the OSError for what went wrong – FileNotFoundError for a directory that is not there, PermissionError for one that cannot be written – with the path in the message; the file, if it existed, is untouched. RuntimeError while a fit_predict is in flight on another thread.

save_bytes() → bytes[source]¶: What save() writes, as bytes – for a store that is not a file (load_bytes() reads them back). This is also what pickle and copy.deepcopy carry. RuntimeError while a fit_predict is in flight on another thread.

to_json(*, pretty: bool = True) → str[source]¶

The state as JSON: everything save() writes, in a form that can be read without this library.

An export, not a second state format – load() reads msgpack and only msgpack. Use it to look at a state, diff two of them, or hand one to something that is not Python.

It refuses rather than lying. serde_json writes NaN and ±inf as null and says nothing about it, where msgpack round-trips all three, so every export is read back and re-encoded and the msgpack must match the real state byte for byte. A state holding a value JSON cannot carry raises ValueError naming the problem instead of returning a file that is quietly wrong.

RuntimeError while a fit_predict is in flight on another thread, as save() does.

save_json(path: str | Path, *, pretty: bool = True) → None[source]¶

Write to_json() to path.

Plain write_text, not save()’s atomic rename: this is an export of a state that lives elsewhere, so a half-written one costs nothing but a re-run.

classmethod load(path: str | Path, specs: Iterable[dict[str, Any]] | None = None) → ModelBank[source]¶

A bank from a file save() wrote, on this or any other OS.

The file carries the specs, so none need be given; passing specs asserts they are the file’s, which is how a resuming job checks that the state it found is the state of the bank it is about to run.

Raises FileNotFoundError (or the OSError for what went wrong) when the file cannot be read, and ValueError when it can but is not a bank this build loads: not a bank state file at all, written by a newer build (the file’s format or state schema version is above this build’s), or specs differ from the file’s.

classmethod load_bytes(data: bytes, specs: Iterable[dict[str, Any]] | None = None) → ModelBank[source]¶: load() from the bytes save_bytes() gave, with the same specs check and the same ValueError for bytes that are not a bank this build loads.

Stream rows through a model bank and write them out with its columns.

input is a path (parquet, ipc, csv or ndjson, told from the extension or named by input_format; globs and cloud URLs as pl.scan_* takes them), a LazyFrame (any query: the scan is polars’, with whatever options it needs), a DataFrame, or any iterable of DataFrames in stream order – chunks from a database cursor, a socket, a generator. output is a path in any of the four formats, told the same way. Leave it out – and out of the config – for a run whose product is its state: the per-row output of an accumulator-only spec is n_eff and nothing else, which over a billion rows is 8 GB of file written so it can be deleted (ENHANCEMENTS E50). save_state is then required, since a run that writes nothing and saves nothing has done nothing. no_output=True says the same thing over a config that names an output, and is what the CLI’s --no-output sets. An output that is written is written through a temporary and renamed into place, so a run that fails leaves the previous file where it was. CSV cannot hold the bank’s struct columns, so there each spec’s struct is flattened to <spec>.<field> columns and a list field (coef) becomes a JSON string – pl.col("ridge.coef").str.json_decode(pl.List(pl.Float64)) reads it back.

config is a dict, a path to a TOML file, or None to build the config from the keyword arguments. Keywords override whatever the config supplies, so a checked-in TOML can be reused with a different input:

po.run("bank.toml", input="today.csv", output="today-out.parquet")

Returns {"rows": ..., "chunks": ...}. Chunking never changes the numbers – it only trades memory for overhead – so chunk_rows (the reader’s chunk; frames passed in directly are taken as they come) is purely a resource knob. On data sorted by group, a chunk should span several groups: the bank fits groups in parallel within a chunk.

keep_columns selects input columns before the bank sees them (and before the scan reads them). progress(rows, chunks) is called after each chunk; raising in it stops the run without publishing the output.

closed_groups writes the groups that finished during the run to a sidecar file beside the output, in the format its extension names (ENHANCEMENTS E54; see ModelBank.closed_groups() for the schema). It needs a spec with group_close and refuses predict, which closes nothing. The file is written once, at the end, through a temporary renamed into place – before save_state, so a state file always has the closed rows that go with it. output may be left out at the same time: that is the accumulate-only pass whose product is the closed groups. A run in which nothing closed writes an empty frame with the schema.

predict=True scores instead of learning: every row gets what the bank loaded from load_state predicts for it as it stands (ModelBank.predict()), and the bank is not updated – so it needs load_state and refuses save_state. One TOML can serve both runs: the keyword drops the config’s save_state, which belongs to the learning run, unless save_state= is passed alongside it.

What is wrong with the call or the config is ValueError, before a row is read: no input; no specs; a spec the bank refuses (ModelBank); a key the config, a spec or its model has not got, named with the keys there are (a misspelt key is never kept at its default in silence); chunk_rows below 1; a format that cannot be told from a path’s extension, or that is not one of the four; predict=True without load_state, or with save_state; an iterable that produced no frames; a load_state that is not a bank this build loads or whose specs are not specs; and a TOML that does not parse (tomllib.TOMLDecodeError). TypeError for a config that is none of the three, a progress that is not callable, or an item of input that is not a DataFrame. A file fails as the OSError for what went wrong, with the path in the message: a config or input that is not there (the scan is polars’, so its FileNotFoundError), a load_state that cannot be read, an output whose directory is not there, and a save_state whose directory is not – found out before the run, since after it the output would be written and the state lost. A column the specs read that the input has not got, or that keep_columns dropped, is the bank’s ValueError (a keep_columns name the input has not got is polars’ ColumnNotFoundError); a value the bank refuses – a null clock, a negative weight, a clock running backwards – is its ValueError mid-run. Whatever stops the run – the bank, the writer, progress or the iterable raising (both come through as themselves) – leaves the previous output where it was and save_state unwritten: the state is saved last, after the output is in place, so a state file always has an output to go with it.

frame.online.fit_predict(...) as a plain function, so that a type checker can see it.

A LazyFrame gives a plan that streams the rows through a bank when it runs (LazyFrameOnlineNamespace.fit_predict()); a DataFrame gives the frame with the bank’s columns (DataFrameOnlineNamespace.fit_predict()). load_state starts the bank from a saved one and save_state writes where it ends up; chunk_rows is the plan’s read chunk, and a frame already in memory is fitted in one call. TypeError for a frame that is neither; otherwise raises what the namespace method does.

polars_online.predict(frame: LazyFrame, bank: ModelBank | str | PathLike[str], *, chunk_rows: int | None = None) → LazyFrame[source]¶

polars_online.predict(frame: DataFrame, bank: ModelBank | str | PathLike[str], *, chunk_rows: int | None = None) → DataFrame

frame.online.predict(bank) as a plain function, so that a type checker can see it.

Scores the rows against bank as it stands and learns nothing: a plan from a LazyFrame (LazyFrameOnlineNamespace.predict()), a frame from a DataFrame (DataFrameOnlineNamespace.predict()). TypeError for a frame that is neither; otherwise raises what the namespace method does.

polars_online.unnest(frame: LazyFrame, specs: Iterable[dict[str, Any]] | ModelBank | str | PathLike[str]) → LazyFrame[source]¶

polars_online.unnest(frame: DataFrame, specs: Iterable[dict[str, Any]] | ModelBank | str | PathLike[str]) → DataFrame

frame.online.unnest(specs) as a plain function, so that a type checker can see it.

Takes each spec’s struct column apart into columns, the coef lists as one named column per coefficient: a plan from a LazyFrame (LazyFrameOnlineNamespace.unnest()), a frame from a DataFrame (DataFrameOnlineNamespace.unnest()). TypeError for a frame that is neither; otherwise raises what the namespace method does.

polars_online.online(expr: Expr) → OnlineNamespace[source]¶

expr.online as a plain function, so that a type checker can see it.

A registered namespace is attached to pl.Expr at runtime, so to a type checker pl.col("y").online is an attribute that does not exist. This returns the same namespace, with its methods and their typed keywords (docs/IMPROVEMENTS.md U4) visible:

df.with_columns(po.online(pl.col("y")).ewridge(features=["x0"], halflife=10.0))

class polars_online.InMemoryExpressionWarning[source]¶

Bases: UserWarning

Issued by every pl.col(...).online.<model>(...) call: the expression form runs on the whole column at once.

Polars calls a stateful user expression once with its whole column, in either engine, so in a plan over a file this form is O(data) where lf.online.fit_predict(specs) is O(chunk) – the same model, the same numbers, and only one of them streams (module docstring). The warning is a UserWarning, shown by default wherever the call is made; a DeprecationWarning would be hidden outside __main__, which is the one place – a pipeline module – where it matters. Using the expression on a frame that is in memory anyway is fine; say so once:

warnings.filterwarnings("ignore", category=po.InMemoryExpressionWarning)

polars_online.native_version()¶: Version of the compiled extension, checked against the Python package version.

polars_online.schema_version()¶: State-file schema version (see online_core::SCHEMA_VERSION).

polars_online.thread_pool_size()¶

Size of the bank’s thread pool.

The pool is built at the first bank call from POLARS_ONLINE_MAX_THREADS (unset: one thread per core) and never resized; this builds it if nothing has yet. Polars’ own pool, POLARS_MAX_THREADS, is separate – pl.thread_pool_size() reports that one.

Raises ValueError when the variable is set to anything but a non-negative integer.

polars_online¶

`polars_online`¶