Online normalization and feature extraction

"You trained me to expect a mean of zero. The sensor warmed up, the mean walked off, and nobody told me. I have been confidently wrong for three hours."

A Drifting AI Agent

This section assumes you understand the offline feature pipeline from Chapter 8 and, above all, the leakage discipline from Chapter 5: a statistic must never be computed from data the model has not yet seen at that moment. It builds directly on the windowing machinery of Section 60.1; here we ask what to compute inside each window, and how to carry state across windows without ever holding the full history.

Why offline z-scoring quietly breaks on a stream

The default preprocessing move is the z-score: subtract the training-set mean, divide by the training-set standard deviation, so each channel is centered and unit-scaled. Offline this is one line. On a live sensor it fails in two distinct ways. First, causality: if you standardize a live window using statistics that include future samples, or refit the scaler on the incoming batch, you have leaked information backward in time exactly as Chapter 5 warned, and your online accuracy will be optimistically wrong. Second, non-stationarity: the mean and variance a sensor produces are not constants. An IMU's bias drifts with temperature (Chapter 2), a PPG baseline wanders with skin contact, a vibration sensor's amplitude climbs as a bearing degrades. A scaler frozen at training time slowly centers the wrong point. The fix is to compute the statistics online, from the past only, and to let them adapt at a rate you choose deliberately.

Running statistics in constant memory

Two algorithms cover almost every case. When you want an unbiased estimate over the whole stream so far and the distribution is roughly stationary, use Welford's method, which updates a running mean and a running sum of squared deviations without ever storing the samples and without the catastrophic cancellation of the naive "sum of squares minus square of sum" formula. For the count \(n\), mean \(\mu_n\), and aggregate \(M_{2,n}=\sum_i (x_i-\mu_n)^2\):

$$ \mu_n = \mu_{n-1} + \frac{x_n - \mu_{n-1}}{n}, \qquad M_{2,n} = M_{2,n-1} + (x_n-\mu_{n-1})(x_n-\mu_n), \qquad \sigma_n^2 = \frac{M_{2,n}}{n-1}. $$

When the stream is non-stationary, an all-history mean is exactly wrong, because it weights a reading from an hour ago the same as the one just in. There you want an exponentially weighted mean and variance, which decay old evidence geometrically with a smoothing factor \(\alpha \in (0,1)\):

$$ \delta = x_t - \mu_{t-1}, \qquad \mu_t = \mu_{t-1} + \alpha\,\delta, \qquad \sigma_t^2 = (1-\alpha)\left(\sigma_{t-1}^2 + \alpha\,\delta^2\right). $$

The choice of \(\alpha\) is the whole design. Its effective memory is roughly \(1/\alpha\) samples, so for a 100 Hz stream, \(\alpha = 0.01\) remembers about one second and tracks fast drift, while \(\alpha = 0.0001\) remembers minutes and ignores it. Pick \(\alpha\) from the physical timescale of the drift you want to follow, not by trial and error. The code below implements both, then uses the exponentially weighted version as a live z-score normalizer.

import numpy as np

class Welford:
    "Exact running mean/variance over all history, constant memory."
    def __init__(self):
        self.n, self.mean, self.M2 = 0, 0.0, 0.0
    def update(self, x):
        self.n += 1
        d = x - self.mean
        self.mean += d / self.n
        self.M2 += d * (x - self.mean)
    @property
    def var(self):
        return self.M2 / (self.n - 1) if self.n > 1 else 0.0

class EWZScore:
    "Causal exponentially weighted normalizer for a non-stationary stream."
    def __init__(self, alpha=0.01, eps=1e-6, warmup=50):
        self.alpha, self.eps, self.warmup = alpha, eps, warmup
        self.mean, self.var, self.n = 0.0, 1.0, 0
    def transform(self, x):
        self.n += 1
        d = x - self.mean
        self.mean += self.alpha * d
        self.var = (1 - self.alpha) * (self.var + self.alpha * d * d)
        if self.n < self.warmup:          # do not trust stats before warmup
            return 0.0
        return d / np.sqrt(self.var + self.eps)   # note: uses PRE-update mean

# 100 Hz signal whose mean ramps (sensor warm-up) then holds
sig = np.concatenate([np.linspace(0, 5, 500), np.full(500, 5)]) + np.random.randn(1000)*0.3
scaler = EWZScore(alpha=0.02)
z = np.array([scaler.transform(x) for x in sig])
print(f"raw mean drift: {sig[:500].mean():.2f} -> {sig[500:].mean():.2f}; "
      f"normalized mean ~ {z[100:].mean():.2f}")

Two details in EWZScore are load-bearing, not stylistic. The normalizer divides by the pre-update mean and variance, so the current sample never normalizes itself, preserving causality. And it returns zeros during a warmup period, because a z-score computed from three samples is noise; a model fed that noise during the first second of a session will produce garbage exactly when a wearable is being put on.

Incremental features without a full window scan

Normalization is the simplest running statistic; the same constant-time discipline extends to the features themselves. Rolling mean and variance over a fixed window come free from the recursions above. Rolling min and max look like they need a scan of the window every step, but a monotonic deque delivers them in amortized \(O(1)\): you keep a deque of candidate extrema, evicting any that a newer, larger (or smaller) value dominates, so the front is always the current extremum. Rolling quantiles and counts yield to reservoir or histogram sketches. The most valuable streaming features on sensor data are usually spectral (Chapter 7), and here two tricks matter. If you need the energy in one or a few known frequency bins, say the mains hum or a motor's rotation rate, the Goertzel algorithm tracks each bin incrementally at a fraction of the cost of a full FFT. If you need the whole spectrum on every hop, the sliding DFT updates the previous window's transform with the one sample that entered and the one that left, rather than recomputing from scratch. Both turn a per-window batch operation into a per-sample update, which is what the tight latency and memory budgets of Chapter 59 demand.

from river import preprocessing, feature_extraction, stats

scaler = preprocessing.StandardScaler()          # running z-score, causal
roll   = feature_extraction.RollingMean(window_size=100)
for x in stream:                                 # x is a dict of channels
    xs = scaler.learn_one(x).transform_one(x)    # update, then normalize

Keeping train and serve in agreement

The subtle failure in streaming preprocessing is a mismatch between how features were normalized during training and how they are normalized in production. Two policies are defensible, and mixing them silently is the trap. In the frozen policy you fit the scaler once on the training set and ship those constants; it is simple and reproducible but blind to drift, and it needs a separate drift monitor (Section 60.5) to tell you when the constants have gone stale. In the adaptive policy the normalizer keeps updating on the live stream, so it tracks drift for free, but you must then train the model on features that were normalized the same causal, adaptive way, by replaying the training stream through the identical online scaler. If you train on globally z-scored features and serve with an adaptive one, the model meets a distribution it never saw. The rule is blunt: whatever normalizer runs at serve time must have run, sample for sample, over the training data too. This is the online face of the leakage-safe evaluation thread, and it is why test-time adaptation (Chapter 66) treats the normalization layer as part of the model, not as inert plumbing.

What's Next

In Section 60.4, we let the model itself learn from the stream. The same River library that gave us causal normalizers provides incremental classifiers and regressors that update one sample at a time, so the model can adapt alongside the features it consumes, turning the frozen predictor of the offline world into a living one.