The online namespaces

Importing polars_online registers online on pl.LazyFrame, pl.DataFrame and pl.Expr. The frame namespaces run a model bank over the frame – as a plan that streams, or eagerly; the expression namespace runs one model over the calling column, for a frame in memory.

lf.online / df.online

The bank as a polars source: lf.online.fit_predict(specs) (ENHANCEMENTS E33).

A LazyFrame in, a LazyFrame out. Executing the plan streams the input through a fresh ModelBank in chunk_rows chunks, so a query with the bank in it is O(chunk) in memory however long the stream is – where the expression form, pl.col("y").online.<model>(...), in the same query is O(data): polars calls a user expression once with its whole column, and its streaming engine collects the column to do so (docs/PERFORMANCE.md section 11; the expression warns about it, polars_online._expr). This is polars’ IO-plugin mechanism (polars.io.plugins.register_io_source): the bank is registered as a source, the kind of node the engine pulls batches from, and what comes after it – filters, selects, joins, sink_parquet – is polars’ own.

The plan is pure: every execution starts from the same state (the specs’ initial state, or load_state, read when the plan is built), so collecting twice gives the same frame. save_state writes the state the execution ends in – after the last row the source fed the bank – atomically, and because the plan is pure that write is idempotent: polars runs a plan’s source once per execution and twice, concurrently, when one query uses the plan twice (a self-join, pl.concat, pl.collect_all of two sinks), and every run ends in the same state (docs/STATE-WORKFLOW.md).

df.online.fit_predict(specs) is the eager twin, ModelBank(specs) .fit_predict(df) in one call. online.unnest(specs) takes a bank’s output apart: each spec’s struct column becomes its fields as columns, with the coef list as one named column per coefficient (polars_online.spec.coef_fields()). Both namespaces are attached at import, which no type checker can see; fit_predict(), predict() and unnest() are the same calls with the frame as the first argument, visibly typed.

class polars_online._frame.LazyFrameOnlineNamespace(lf: LazyFrame)

Bases: object

A model bank over the plan’s rows, as a plan that streams.

fit_predict(specs: Iterable[dict[str, Any]] | None = None, *, load_state: str | PathLike[str] | None = None, save_state: str | PathLike[str] | None = None, closed_groups: str | PathLike[str] | None = None, chunk_rows: int | None = None) → LazyFrame

The plan’s rows plus one struct column per spec, learning as it goes.

Executing the returned plan – collect(), collect_batches(), sink_parquet() and the rest – streams this plan’s rows through a new ModelBank in chunk_rows chunks (default 100,000; chunking never changes the numbers, only coef’s reporting cadence), so memory is O(chunk + state) whatever the length of the stream. Rows must arrive in stream order, as for the bank. Filters, selections and head applied after are pushed into the source: a filter never changes what the bank learns from – filter before to do that – a selection is read from the input, so a wide scan reads only the columns the specs and the query need, and head(n) feeds the bank the first n rows and no more.

specs are the bank’s, or load_state names a saved bank to resume from (with specs, they are checked against the file). The file is read when the plan is built, so the plan carries that state: each execution starts from it afresh, and a plan collected twice gives the same frame. save_state writes the state the execution ends in – after the last row the source fed the bank – to that path when it ends, atomically (ModelBank.save()), so the file is the old state or the new one and never half of either; load_state and save_state may be the same path. Because the plan is pure the write is the same whenever it happens: a plan used twice in one query (a self-join, pl.concat, pl.collect_all of two sinks) runs twice and writes the same bytes twice. Nothing is written unless the source reaches the last row: a run abandoned before then, or one the bank ended with an error, leaves the file as it was; a node after the bank failing does not stop the bank, so the state is written then (docs/STATE-WORKFLOW.md) – polars_online.run() saves only after its output is committed, for the case where the two must be tied together, and a dated save_state per batch of data keeps a rerun from learning it twice.

What the schema decides is reported while the plan is built, as polars reports its own schema errors: ValueError for neither specs nor load_state, for chunk_rows below 1, for a spec the bank refuses, and for a spec whose column the plan has not got, is not numeric, or shares the spec’s name (the checks of ModelBank and ModelBank.fit_predict(), with the same messages); FileNotFoundError for a load_state that is not there or a save_state whose directory is not, ValueError for a load_state that is not a bank this build loads or whose specs are not specs (ModelBank.load()). What only the values decide – a null clock, a negative weight, a clock running backwards – is reported when the plan runs, as polars’ ComputeError carrying the bank’s message, and so is a save_state that cannot be written when the run ends, carrying the OSError’s message and the path.

closed_groups writes the groups that finished during the run to a sidecar file, in the format its extension names (ModelBank.closed_groups(), ENHANCEMENTS E54). The queue is drained after every chunk, so the bank stays bounded, and the one file is written where save_state is written and under the same rules and caveats: only when the source reaches its last row, once per execution of the plan, and twice with the same bytes for a plan used twice in one query. It needs a spec with group_close; a run in which nothing closed writes an empty frame with the schema.

predict(bank: ModelBank | str | PathLike[str], *, chunk_rows: int | None = None) → LazyFrame

The plan’s rows scored against bank as it stands, learning nothing.

Each row gets ModelBank.predict()’s struct: what the bank would report for it as the next row of its group’s stream, from the current state, which the plan never moves. bank is a ModelBank (scored as it stands each time the plan runs; predict leaves it untouched, so sharing it with a plan is safe) or a path to a saved state, read when the plan is built – build the plan again to pick up a newer file. Target columns are optional, as for predict; chunk_rows is the read chunk.

Reported while the plan is built: FileNotFoundError for a path that is not there and ValueError for a file that is not a bank this build loads (ModelBank.load()), TypeError for a bank that is neither a bank nor a path, ValueError for chunk_rows below 1 and for a column the bank reads that the plan has not got or that is not numeric (a missing target is fine). A value the bank refuses – a null clock, a negative weight – is reported when the plan runs, as polars’ ComputeError carrying ModelBank.predict()’s message.

unnest(specs: Iterable[dict[str, Any]] | ModelBank | str | PathLike[str]) → LazyFrame

The plan with each spec’s struct column taken apart into columns.

lf.unnest(names) with the coef lists taken apart too: every scalar field becomes a column of its own name (pred_y, n_eff@h500), and each coef list becomes one column per coefficient, named coef_{target}_{term}{combo}{instance} – coef_y_intercept, coef_y_x1__r0.5@h500 – as polars_online.spec.coef_fields() lists them. The columns take the struct’s place; the rest of the frame, and any spec column not named, are left as they are. specs is the spec dicts, a ModelBank (its specs), or the path of a saved bank (which carries them). So a scored plan, or a parquet the CLI wrote, comes back flat:

betas = (
    pl.scan_parquet("out.parquet")
    .online.unnest([ols])
    .select("t", "^coef_.*$")
)

Reported while the plan is built: ValueError for a spec whose column the plan has not got, is not a struct, or lacks a field the spec produces, for a spec given twice and for a spec that is not valid; TypeError for specs that are none of the three; FileNotFoundError and ModelBank.load()’s ValueError for a path. Two specs that produce a field of the same name unnest to the same column name, which polars reports as its DuplicateError – unnest them one at a time, or rename the struct’s fields first (pl.col("m").name.prefix_fields("m_")).

class polars_online._frame.DataFrameOnlineNamespace(df: DataFrame)

Bases: object

A model bank over the frame’s rows, in one call.

fit_predict(specs: Iterable[dict[str, Any]] | None = None, *, load_state: str | PathLike[str] | None = None, save_state: str | PathLike[str] | None = None, closed_groups: str | PathLike[str] | None = None) → DataFrame

ModelBank(specs).fit_predict(df) – the frame plus one struct column per spec, from a bank that is then dropped, or saved to save_state first (ModelBank.save()); load_state starts it from a saved bank instead of the specs. Keep a bank of your own to feed it more rows.

closed_groups writes the groups that finished to a sidecar file in the format its extension names, before save_state (ModelBank.closed_groups()).

Raises what ModelBank, ModelBank.fit_predict(), ModelBank.load() and ModelBank.save() raise, and ValueError for neither specs nor load_state. A save_state whose directory is not there is FileNotFoundError before the fit, not after it.

predict(bank: ModelBank | str | PathLike[str]) → DataFrame

ModelBank.predict() over the frame: scored against bank – a ModelBank, or the path of a saved one – as it stands, which does not move. Raises what ModelBank.load() (for a path) and ModelBank.predict() raise, and TypeError for a bank that is neither.

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

The frame with each spec’s struct column taken apart into columns, as LazyFrameOnlineNamespace.unnest() does for a plan: scalar fields under their own names, each coef list as one named column per coefficient. Raises what the plan form does, on the call.

pl.col("y").online

The online expression namespace (docs/PLAN.md section 6) – in-memory only.

pl.col("y").online.ewridge(features=[...], halflife=...) runs one spec over the column the expression receives; use .over(group) for per-group streams. Features are column names or named expressions (pl.col("x").shift(1) .alias("x_lag")), evaluated per group under .over. The implementation is the model bank itself, so expression == bank by construction.

Every call warns (InMemoryExpressionWarning). Polars hands a stateful user expression its whole column in either engine – its streaming engine collects the input to do so – so this form is O(data) where lf.online.fit_predict(specs) and po.run are O(chunk): 7.3 GB against 1.35 GB at 12M rows for the same model (docs/PERFORMANCE.md section 11). That is polars’ contract for a user expression, not something a plugin can change, and a reader who takes the expression for the natural streaming form gets the collecting one. The namespace stays for a frame already in memory, where it is the shortest way to write the model and features can be expressions; the warning exists so that nobody learns the difference from a memory profile. See the README’s closing section.

class polars_online._expr.OnlineNamespace(expr: Expr)

Bases: object

Online models over the expression’s column as a target.

For a frame in memory. Polars calls the plugin once with the whole column, in either engine – its streaming engine collects a user expression’s input to do so – so in a plan the column is O(data). For a stream, lf.online.fit_predict(specs) is the same bank as a plan that stays O(chunk) (polars_online._frame). Every method warns with InMemoryExpressionWarning (module docstring).

Each method takes the model’s parameters as polars_online.spec’s builder of the same name does, minus name, targets and group: the calling column is the target (extra_targets adds more, sharing one fit), features are column names or named expressions, and a group is .over(group). Building the expression raises what the builder raises (polars_online.spec): TypeError for a keyword the model has not got or a value of the wrong shape, ValueError for a value the model refuses; and its own TypeError for group= (written .over instead) or a feature that is neither a name nor an expression, ValueError for a calling or feature expression whose output name polars cannot determine (give it an .alias), and for extra_targets naming the calling column or a column twice. When the expression runs, a column it reads that the frame has not got is polars’ ColumnNotFoundError as the plan is resolved, and what the bank refuses on the data – a column that is not numeric, a null clock, a negative weight, a clock running backwards under on_clock_reset="error" – is polars’ ComputeError (the plugin failed with message: ...) carrying the message polars_online.ModelBank.fit_predict() gives for the same frame.

ewridge(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[EwridgeKwargs]) → Expr

EW-ridge over this column as the target. Same parameters as polars_online.spec.ewridge minus name/targets/group.

rls(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[RlsKwargs]) → Expr

Recursive least squares over this column as the target.

lasso(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[LassoKwargs]) → Expr

Lasso path with online lambda selection over this column as target.

kalman(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[KalmanKwargs]) → Expr

Kalman / random-walk-beta filter over this column as the target.

huber(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[HuberKwargs]) → Expr

Huber regression over this column as the target.

quantile(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[QuantileKwargs]) → Expr

Quantile regression over this column as the target.

ftrl(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[FtrlKwargs]) → Expr

Online logistic regression (FTRL-proximal) over this column as the binary target. pred is a probability.

ew_cov(others: list[str | Expr], **kwargs: Unpack[EwCovKwargs]) → Expr

EW moments of this column together with others.

Unlike the model namespaces this one has no target: the column the expression is called on becomes the first feature.

deco(others: list[str | Expr], **kwargs: Unpack[DecoKwargs]) → Expr

Dynamic equicorrelation of this column together with others.

No target, as for ew_cov: the column the expression is called on becomes the first feature. blocks names subsets of the whole feature list, this column included.

bocpd(others: list[str | Expr], **kwargs: Unpack[BocpdKwargs]) → Expr

Bayesian online changepoint detection over this column together with others.

No target, as for ew_cov: the calling column becomes the first feature.

corrchange(others: list[str | Expr], **kwargs: Unpack[CorrChangeKwargs]) → Expr

A correlation-constancy test over this column together with others.

No target, as for ew_cov: the calling column becomes the first feature.

hmm(others: list[str | Expr], **kwargs: Unpack[HmmKwargs]) → Expr

A hidden Markov model over this column together with others.

No target, as for ew_cov: the calling column becomes the first feature.

rcov(others: list[str | Expr], **kwargs: Unpack[RcovKwargs]) → Expr

Refused: rcov has no expression form.

Its value is the block it emits when a group closes, and an expression has neither a group (.over() does the grouping, and the plugin never sees it) nor a close (it returns one column of the frame’s own height, with nowhere to put a block). Run it in a ModelBank with group= and group_close= and read the blocks with ModelBank.closed_groups().

sgd(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[SgdKwargs]) → Expr

SGD with pluggable losses over this column as the target.

pa(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[PaKwargs]) → Expr

Passive-aggressive regression over this column as the target.

holt(extra_targets: list[str] | None = None, **kwargs: Unpack[HoltKwargs]) → Expr

Holt’s linear trend over this column – level plus slope, no features.

The only namespace method without a features argument, because the model has none: it extrapolates this column’s own level and trend.

kmeans(others: list[str | Expr], **kwargs: Unpack[KMeansKwargs]) → Expr

EW k-means over this column together with others.

Like ew_cov this has no target: the column the expression is called on becomes the first feature. k is required. The struct holds cluster, dist, dist2, n_eff and coef (the centres).

micro(others: list[str | Expr], **kwargs: Unpack[MicroKwargs]) → Expr

Density-based (micro-cluster) clustering over this column together with others.

Like kmeans this has no target: the column the expression is called on becomes the first feature. eps is required. The struct holds cluster, dist, micro, outlier, n_clusters, n_micro, n_eff and coef (the established summaries).

ew_class(features: list[str | Expr], **kwargs: Unpack[EwClassKwargs]) → Expr

Class-conditional Gaussian classifier (QDA / LDA / naive Bayes) with this column as the label.

The column the expression is called on holds the class of each row (null: score, do not learn); classes and precision_prior are required. The struct holds class, one p_<class> per class, n_eff and coef (the class means).

seqtest(extra_targets: list[str] | None = None, **kwargs: Unpack[SeqTestKwargs]) → Expr

Sequential test of this column’s sign – two e-processes, one per direction, read at any row.

No features: the column is the test. The struct holds log_e_pos, log_e_neg, n_pos, n_neg and n_eff, all read before the row is counted. The builder’s a/b compare two specs of a bank and an expression is one spec, so they are not taken here: to compare two models over a frame in memory, test the sign of |resid_b| - |resid_a| as a column, or run the bank (lf.online.fit_predict, ModelBank), where a/b read the other specs’ residuals.

marginal(features: list[str | Expr], extra_targets: list[str] | None = None, **kwargs: Unpack[MarginalKwargs]) → Expr

Per-(feature, target) EW moments with this column as the target, as polars_online.spec.marginal keeps them.

The pairs live in the state, and an expression has no state to read after the fact: the struct holds n_eff alone, so over an expression this only walks the stream. To read the pairs, run the spec in a bank (ModelBank, lf.online.fit_predict) and call ModelBank.marginal().