polars_online.ops

The window operators: exponentially weighted means, sums and rates over a stream’s clock, looking back or ahead, a variance and a standard deviation looking back, and the increment of a running sum. Each returns a pl.Expr that composes with Polars’ own – pl.col is the current row. It is computed by polars_online.stream.with_windows() in one pass, in O(window) memory, or given to a spec as a target. A target that looks ahead is known only once its window has closed, so the bank learns it then, under an embargo (polars_online.spec) of at least the window.

Every operator is defined once, by the library that has it, and held to it in the tests. For row t at policy time tau_t with lam = 2 ** (-1 / half_life):

operator

defined as

held to

ewm_mean()

the time-weighted mean: row i’s value is held over (t_{i-1}, t_i], the first value of a stretch from before it, and the mean over (tau_t - w, tau_t] weighs each held value by integral lam**(tau_t - s) ds over its interval inside the window

Polars’ ewm_mean_by. A burst of rows does not outweigh a quiet stretch, and a row at a repeated stamp holds no interval, so it moves no mean. No mean counts rows; pandas’ ewm(times=) is that form, and is not offered

rewm_mean()

the mirror: value j is held over [t_j, t_{j+1}), until the next row, and weighed by integral lam**(s - tau_t) ds inside (tau_t, tau_t + w]

the reversed stream’s ewm_mean()

ewm_sum(), rewm_sum()

sum lam**|t_j - tau_t| * x_j over the rows of the window, each counted once at its own time

Polars’ ewm_sum_by on distinct stamps; at a repeated stamp every row holds the stamp’s whole window, so each carries the stamp’s total where ewm_sum_by’s is a running sum

ewm_rate(), rewm_rate()

the sum over the decayed time the window covers, integral_0^T lam**s ds = half_life / ln 2 * (1 - 2 ** (-T / half_life)), T the window’s span inside the stretch: a quantity per unit time

the sum and the mass, each as above

ewm_var(), ewm_std()

the variance about ewm_mean(), each value under the mean’s weight m_i: sum m_i (x_i - mu) ** 2 / V1, V1 = sum m_i; unless bias, times V1 ** 2 / (V1 ** 2 - V2), V2 = sum m_i ** 2; and its root. Looking back only

Polars’ ewm_var and ewm_std under adjust=False on a clock that steps by 1 a row, the correction for unequal weights Polars’ own

increment()

x_i - x_{i-1} within the group and session; null on a session’s first row; seconds on a temporal column

the row before it

Which rows a window holds follows Polars’ rolling windows (rolling_*_by): a window is a set of timestamps, and closed says which ends are in. Every row at one stamp gets the same backward window, later rows at that stamp included, and looking ahead the mirror holds:

closed

looking back

looking ahead

"right" (the default)

(tau_t - w, tau_t]: the stamp’s own rows are in, so the output waits for the next distinct stamp

(tau_t, tau_t + w]: none at the stamp is in; a row exactly w later is

"left"

[tau_t - w, tau_t): the stamp’s own rows are outside, and the output is known as the stamp arrives

[tau_t, tau_t + w): every row at the row’s own stamp is in, the row itself included

"both"

both ends in; waits for the next stamp

both ends in: the stamp’s rows, and a row exactly w later

"none"

neither end; known as the stamp arrives

neither end

An edge between two rows is decided from the difference of their clocks, exact in nanoseconds on a temporal clock, as Polars decides it, at any age of the stream. min_samples (Polars’ name) nulls a window holding fewer rows with a value. A value that is NaN, infinite or beyond 1e100 in magnitude is read as missing, as a null is: every model reads its inputs within that bound, where a weight of 1e100 times a square of 1e100 is 1e300, still inside a double’s range. At a row whose value is missing the operator gives the window as it stands, where Polars’ ewm_mean_by gives null. A row is in a window by its stamp, but a mean weighs it by its held interval inside the window. So a row exactly one window old under "left" or "both" counts for min_samples and weighs nothing, since its interval ends at its row, and a mean with no other value in the window is null.

A weighted mean is a ratio of two sums. A VWAP is decayed notional over decayed volume, the time mass cancelling. A side’s VWAP puts when/then inside both sums. There is no weight=:

notional = pl.col("price") * pl.col("quantity")
buys = pl.when(pl.col("side") == "buy")
out = po.stream.with_windows(
    trades,
    mid_trend=po.ewm_mean("mid", half_life="5s", window_size="1m") - pl.col("mid"),
    fwd_vwap=po.ewm_sum(notional, half_life="10s", window_size="1m", reverse=True)
             / po.ewm_sum("quantity", half_life="10s", window_size="1m", reverse=True)
             - pl.col("mid"),
    buy_vwap=po.ewm_sum(buys.then(notional), half_life="10s")
             / po.ewm_sum(buys.then("quantity"), half_life="10s"),
    clock="ts", gap_cap="5m", group="symbol",
)

A feature in units of its spread is a ratio of a deviation and an operator. (x - ewm_mean(x)) / ewm_std(x) is the feature’s z-score: a model fitted on it has a coefficient per standard deviation of the feature as its spread stands on each row, where a model on x has one per unit. Where a target responds to a feature relative to its volatility, fit on the z-score. Every model reads it as a column (docs/WINDOWS.md, Features in units of their spread, with the measured gain):

spread = {"half_life": "30m"}
z_mid = (pl.col("mid") - po.ewm_mean("mid", **spread)) / po.ewm_std("mid", **spread)
zscored = pl.scan_parquet("trades.parquet").online.with_windows(
    z_mid=z_mid, clock="ts", gap_cap="5m", group="symbol"
).collect()

The clock and its policy – clock, gap_cap, restart_after_step_back, session, session_gap, group – are the call’s, shared by every operator in it, in a spec’s own words (polars_online.spec). A gap past gap_cap or a session change ends every window open across it, and partial says what such a window gives. A backward window whose span reaches before its stretch’s first row is partial too, the stream’s own start included, so partial also says what every stream’s first window_size gives:

partial

the window gives

the default

"keep"

the value over what it saw, the window ending at the last row seen

looking back

"null"

null

looking ahead

"drop"

the row leaves the output

A reset discards it: null, never dropped. Under session_gap="reset" a session change is a reset, so it discards rather than cuts. A forward window under closed="right" or "both" whose far edge is the last row before a cut or a reset is whole: every row it covers has arrived, so it gives its value. A forward window still open when the input ends is null. Under group a forward window needs a clock column, since only a clock ends the windows of a group that falls silent. partial needs a window_size: without one no window is cut short, so partial alone is refused.

An operator’s input is a column name or an element-wise expression of the row, increment() included – not another window operator: a formula over an operator’s output is a second call. half_life and window_size are in the clock’s units: numbers on a numeric clock, and durations ("10s", a timedelta, pl.duration) on a temporal one. half_life=float("inf") weighs the window evenly, and needs a window_size for a mean, a variance or a standard deviation.

An operator checks its own arguments when it is called, and names the one it refuses:

  • TypeError for an input that is neither a column name nor a pl.Expr, a half_life or window_size that is neither a number nor a duration, and a bias that is not a bool;

  • ValueError for a half_life or window_size that is NaN or not above 0, an infinite window_size, a closed or partial it does not know, partial without window_size, and a min_samples below 1 or above 2**32 - 1;

  • ValueError for an input expression that holds another window operator, or a node that is not element-wise, or that reads a column whose name starts with @po:, the prefix the operators’ own columns carry.

polars_online.stream.with_windows() refuses the rest while its plan is built: half_life=float("inf") with no window_size for a mean, a variance or a standard deviation, and a window looking ahead under group with no clock column.

polars_online.ops.ewm_mean(input: str | pl.Expr, *, half_life: float | Duration, window_size: float | Duration | None = None, closed: Literal['right', 'left', 'both', 'none'] = 'right', min_samples: int = 1, partial: Literal['keep', 'null', 'drop'] | None = None, reverse: bool = False) → pl.Expr[source]

The time-weighted mean of input over the rows at or before each row, less than window_size older (none: the running mean).

Polars’ ewm_mean_by recursion, y_i = a_i x_i + (1 - a_i) y_{i-1} with a_i = 1 - lam ** (t_i - t_{i-1}) and y_1 = x_1: each value weighed by the decayed time of the interval ending at its row. Under a window_size only the part of each interval inside the window counts.

reverse=True looks ahead instead, as Polars’ cum_sum(reverse=True) runs from the end: the mean over the rows after each row, at most window_size later, which it then requires. It is the mirror, each value held until the next row and weighed by the decayed time from the row. As a spec’s target it is known only once the window has closed, so the bank learns it under an embargo of at least window_size.

polars_online.ops.ewm_rate(input: str | pl.Expr, *, half_life: float | Duration, window_size: float | Duration | None = None, closed: Literal['right', 'left', 'both', 'none'] = 'right', min_samples: int = 1, partial: Literal['keep', 'null', 'drop'] | None = None, reverse: bool = False) → pl.Expr[source]

ewm_sum() over the decayed time the window covers, half_life / ln 2 * (1 - 2 ** (-T / half_life)), with T the window’s span inside the stretch. It is a quantity per unit of clock, such as a volume rate from increment() of a running volume.

reverse=True looks ahead instead: the sum ahead over the decayed time the window ahead covers, window_size then required.

polars_online.ops.ewm_std(input: str | pl.Expr, *, half_life: float | Duration, window_size: float | Duration | None = None, closed: Literal['right', 'left', 'both', 'none'] = 'right', min_samples: int = 1, partial: Literal['keep', 'null', 'drop'] | None = None, bias: bool = False) → pl.Expr[source]

The square root of ewm_var(), as Polars’ ewm_std is the root of its ewm_var. (x - ewm_mean(x)) / ewm_std(x) is a feature in units of its current spread: a z-score.

polars_online.ops.ewm_sum(input: str | pl.Expr, *, half_life: float | Duration, window_size: float | Duration | None = None, closed: Literal['right', 'left', 'both', 'none'] = 'right', min_samples: int = 1, partial: Literal['keep', 'null', 'drop'] | None = None, reverse: bool = False) → pl.Expr[source]

sum lam ** (tau_t - t_j) * x_j over the rows at or before each row, less than window_size older. On distinct stamps it is Polars’ ewm_sum_by, each value counted once at its own time. At a repeated stamp every row carries the stamp’s total, where ewm_sum_by’s is a running sum. Two sums make a weighted mean.

reverse=True looks ahead instead: sum lam ** (t_j - tau_t) * x_j over the rows after each row, at most window_size later, which it then requires.

polars_online.ops.ewm_var(input: str | pl.Expr, *, half_life: float | Duration, window_size: float | Duration | None = None, closed: Literal['right', 'left', 'both', 'none'] = 'right', min_samples: int = 1, partial: Literal['keep', 'null', 'drop'] | None = None, bias: bool = False) → pl.Expr[source]

The variance of input about ewm_mean(), under the mean’s own weights, over the rows at or before each row, less than window_size older (none: the running variance). It looks back only.

Each value weighs m_i, the decayed mass of its held interval inside the window, as in ewm_mean(). The variance is sum m_i (x_i - mu) ** 2 / V1, with mu = sum m_i x_i / V1 and V1 = sum m_i. Unless bias, it is multiplied by V1 ** 2 / (V1 ** 2 - V2), with V2 = sum m_i ** 2: the correction for unequal weights that Polars’ ewm_var applies to its own. It is null while one row carries all the weight, as Polars’ first row is.

On a clock that steps by 1 a row, with half_life in rows, this is Polars’ ewm_var(half_life=..., adjust=False, bias=...). adjust=False is the form ewm_mean() already mirrors: it is ewm_mean_by, which holds a stretch’s first value from before it. So the first row weighs (1 - alpha) ** (n - 1), with alpha = 1 - lam, as adjust=False weighs it. A window that holds one value gives 0 exactly, at any level.

polars_online.ops.increment(input: str | Expr) → Expr[source]

x_i - x_{i-1} of input within the group and session, x_{i-1} the input’s last value with a value. A null input makes the row’s increment null and is skipped, as the operators hold a value from the last valued row ([1, null, 3] gives [null, null, 2], where diff() gives three nulls). Null on a session’s first row, and after a step back restart_after_step_back reads as a new start, in the row’s group or on the stream across groups. Seconds on a temporal column; on a time of day, the step in its seconds since midnight, so a step across midnight is negative, as Polars’ diff on a Time gives. On an integer column, Int8 to Int128 or UInt8 to UInt64, the step is taken in integers, then made a float, rounded once, so a running count past 2**53 keeps its steps of 1. Give it a running total before a sum or a rate: its increments are the trades themselves, where a window’s notional of 1e6 read as the difference of two running totals near 1e10 loses about four of a double’s sixteen digits.