River developer on retainer: online StandardScaler zeros pre-normalized sensor_temperature_normalized for 10,000 rows — running std=0 cold-start fallback — rolling accuracy 0.69 vs offline baseline 0.80 — 3 weeks — River developer on monthly retainer

river 0.21.x • Python 3.11 • online binary classification • 100K-row streaming IoT sensor dataset • 50 features • compose.Pipeline([('scaler', preprocessing.StandardScaler()), ('clf', linear_model.LogisticRegression(optimizer=optim.SGD(lr=0.01)))]) • sensor_temperature_normalized pre-normalized in CSV using full-dataset z-score • river StandardScaler running std=0 at row 1 • transform = x / max(1, std) fallback • feature value corrupted for first 10,000 rows • permutation importance 0.14 (top feature) • rolling accuracy stuck at 0.69 • offline baseline 0.80 • developer checks river version, optimizer lr, feature list for 3 weeks • fix: compose.Select(normalized_features) bypasses re-scaling • rolling accuracy 0.69→0.79 • 1.5h retainer engagement

Westfield River Streaming Ltd. built an online binary classification pipeline using river 0.21.x to score 100,000 IoT sensor readings in real time. The dataset had 50 features. One of them, sensor_temperature_normalized, had been pre-normalized in the CSV using a z-score computed from all 100,000 rows before streaming began: mean and standard deviation were calculated from the full dataset and applied as a batch transformation, producing a column with values in approximately [−3.5, +3.5] with mean near 0 and standard deviation near 1. The team fed this CSV directly into river’s stream.iter_csv and placed a preprocessing.StandardScaler() at the head of the pipeline, as they would in any scikit-learn workflow. The rolling accuracy plateaued at 0.69. An offline train/test split baseline using the same features and logistic regression achieved 0.80. The developer spent three weeks checking the river version, the SGD learning rate, the feature list ordering, and the LogisticRegression L2 coefficient before engaging a River retainer developer.

The pipeline and streaming setup at the time of the engagement:

# Westfield River Streaming Ltd. — online binary classification on IoT sensor data
# river 0.21.x, Python 3.11

import river
from river import compose, preprocessing, linear_model, optim, stream, metrics

# 100K-row streaming IoT sensor dataset, 50 features.
# sensor_temperature_normalized: pre-normalized in CSV using full-dataset statistics.
#   full_dataset_mean = X['sensor_temperature'].mean()   # computed on all 100K rows
#   full_dataset_std  = X['sensor_temperature'].std()    # computed on all 100K rows
#   df['sensor_temperature_normalized'] = (
#       df['sensor_temperature'] - full_dataset_mean
#   ) / full_dataset_std
# Result: sensor_temperature_normalized in CSV has values in [-3.5, +3.5]
# with batch mean≈0 and batch std≈1.

pipeline = compose.Pipeline(
    ('scaler', preprocessing.StandardScaler()),
    ('clf',   linear_model.LogisticRegression(
        optimizer=optim.SGD(lr=0.01),
        l2=0.0,
    ))
)

# ← Problem: river's StandardScaler is an ONLINE transformer.
# It maintains running_mean and running_var using Welford's algorithm.
# At row 1: running_mean=0, running_var=0 → running_std=0.
# transform(x) = (x - running_mean) / max(1, running_std)
# At row 1: (x - 0) / max(1, 0) = x / 1 = x     ← not centered, not scaled
# At row 50: running_std may still be only ~0.08    ← denominator clamped to 1
# At row 500: running_std may be ~0.31              ← denominator clamped to 1
# At row 10,000: running_std finally converges ~0.95 ← approaching true std
#
# For sensor_temperature_normalized (already normalized, batch std≈1):
# The running_mean starts at 0 and fluctuates during cold-start.
# The running_std starts at 0 and climbs slowly.
# For the first 10,000 rows, the transform is NOT a no-op.
# It introduces running-mean noise: (x - fluctuating_running_mean) / 1
# sensor_temperature_normalized is the top-importance feature (permutation importance = 0.14)
# It carries corrupted signal for the first 10% of the stream.

metric = metrics.RollingROCAUC(window_size=1000)
rolling_acc = metrics.RollingAccuracy(window_size=1000)

for x, y in stream.iter_csv(
    'westfield_sensor_100k.csv',
    target='fault_label',
    converters={f: float for f in [
        'sensor_temperature_normalized',
        'sensor_pressure_raw',
        'sensor_vibration_raw',
        # ... 47 more features
    ]},
):
    y_pred = pipeline.predict_one(x)
    metric.update(y, pipeline.predict_proba_one(x)[True])
    rolling_acc.update(y, y_pred)
    pipeline.learn_one(x, y)

print(f"Final rolling ROCAUC (last 1K rows): {metric.get():.4f}")
print(f"Final rolling accuracy (last 1K rows): {rolling_acc.get():.4f}")
# Final rolling ROCAUC (last 1K rows): 0.7231
# Final rolling accuracy (last 1K rows): 0.6903
# Offline train/test split baseline (same features, LogisticRegression): 0.8014
# Gap: 0.11 — too large for cold-start alone to explain

Why river’s online StandardScaler corrupts pre-normalized features during cold-start

River’s preprocessing.StandardScaler implements Welford’s online algorithm to maintain a running mean and running variance across the stream. It has never seen any data before the first learn_one call: running_mean = 0, running_var = 0, and therefore running_std = 0. The transform formula is z = (x − running_mean) / max(1, running_std) — the max(1, running_std) denominator is a numerical guard that prevents division by zero when the running standard deviation has not yet converged. This guard is well-designed for raw features with arbitrary scale: it ensures the pipeline can score examples before accumulating enough statistics, with the expectation that the denominator will climb past 1 as the true standard deviation of the feature becomes evident from the stream.

For sensor_temperature_normalized, the batch statistics had already been applied before streaming: the column had values in [−3.5, +3.5], batch mean ≈ 0, batch std ≈ 1. River’s StandardScaler received a stream of values that looked like a standard normal distribution. Its running mean converged to near 0 fairly quickly. Its running variance converged more slowly — Welford’s variance estimate requires many observations to stabilize for a feature with std near 1. At row 100, running_std ≈ 0.07. At row 1,000, running_std ≈ 0.22. At row 5,000, running_std ≈ 0.61. River’s max(1, running_std) clamps the denominator to 1 for all of these values — not because the guard is malfunctioning, but because running_std < 1 throughout the cold-start period. The transform is therefore z = (x − running_mean) / 1 for the first several thousand rows.

The running mean is itself non-zero and unstable during cold-start. After seeing only a few hundred rows, the running mean fluctuates by ±0.08 around the true mean of 0. This means the transform for sensor_temperature_normalized is z = x − running_mean_fluctuation — a noisy shift of an already-normalized feature. The logistic regression model receives a corrupted version of its top-importance feature for the first 10,000 rows (10% of the stream). The gradient updates during this period push the weight for sensor_temperature_normalized in a direction consistent with the corrupted values. Even after the StandardScaler converges, the online model’s weight is biased from the cold-start period and must be corrected gradually through subsequent gradient steps. The rolling accuracy never fully recovers: by the time the weight converges correctly, the model has also begun drifting toward later patterns in the stream.

# Diagnostic: plot StandardScaler running statistics for sensor_temperature_normalized
# river 0.21.x — cold-start running mean and std convergence

import river
from river import compose, preprocessing, linear_model, optim, stream, metrics
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt

# Track running statistics manually to diagnose cold-start effect
class DiagnosticScaler(preprocessing.StandardScaler):
    def __init__(self):
        super().__init__()
        self.history_running_mean = []
        self.history_running_std  = []
        self.history_transform    = []

    def transform_one(self, x):
        result = super().transform_one(x)
        feature = 'sensor_temperature_normalized'
        if feature in self.means:
            running_mean = self.means[feature]
            running_var  = self.vars[feature]
            running_std  = running_var ** 0.5
            effective_std = max(1, running_std)
            transform_val = (x[feature] - running_mean) / effective_std
            self.history_running_mean.append(running_mean)
            self.history_running_std.append(running_std)
            self.history_transform.append(transform_val)
        return result

diag_scaler = DiagnosticScaler()

pipeline_diag = compose.Pipeline(
    ('scaler', diag_scaler),
    ('clf',   linear_model.LogisticRegression(
        optimizer=optim.SGD(lr=0.01),
        l2=0.0,
    ))
)

row_count = 0
for x, y in stream.iter_csv(
    'westfield_sensor_100k.csv',
    target='fault_label',
    converters={f: float for f in ['sensor_temperature_normalized',
        'sensor_pressure_raw', 'sensor_vibration_raw']},
):
    pipeline_diag.predict_one(x)
    pipeline_diag.learn_one(x, y)
    row_count += 1
    if row_count >= 20000:
        break

rows = list(range(len(diag_scaler.history_running_std)))
fig, axes = plt.subplots(3, 1, figsize=(10, 9))
axes[0].plot(rows, diag_scaler.history_running_mean, linewidth=0.8)
axes[0].axhline(0, color='red', linestyle='--', label='true batch mean')
axes[0].set_title('Running mean (sensor_temperature_normalized)')
axes[0].set_xlabel('Row number')
axes[0].legend()
axes[1].plot(rows, diag_scaler.history_running_std, linewidth=0.8)
axes[1].axhline(1, color='red', linestyle='--', label='true batch std')
axes[1].axhline(1, color='orange', linestyle=':', label='max(1, std) clamp threshold')
axes[1].set_title('Running std (sensor_temperature_normalized)')
axes[1].set_xlabel('Row number')
axes[1].legend()
axes[2].plot(rows, diag_scaler.history_transform, linewidth=0.5, alpha=0.6)
axes[2].set_title('Actual transform output (should equal raw input if truly no-op)')
axes[2].set_xlabel('Row number')
plt.tight_layout()
plt.savefig('cold_start_diagnostic.png', dpi=120)
print('Cold-start diagnostic plot saved.')

# Running mean stabilises near 0 by row 1,000
# Running std does NOT reach 1.0 until row 8,500
# max(1, running_std) clamps denominator to 1 for first 8,500 rows
# → transform is NOT a no-op for first 8,500 rows despite pre-normalization

# Permutation importance (computed offline using river.utils.rolling)
# to confirm sensor_temperature_normalized is top feature:
from river import evaluate, utils
import pandas as pd, numpy as np

data = pd.read_csv('westfield_sensor_100k.csv')
X_all = data.drop('fault_label', axis=1).to_dict(orient='records')
y_all = data['fault_label'].tolist()

# After training the converged pipeline, permute each feature and measure accuracy drop
from sklearn.metrics import accuracy_score
from copy import deepcopy

trained_pipeline = deepcopy(pipeline_diag)  # already trained on 100K rows above

baseline_preds = [trained_pipeline.predict_one(x) for x in X_all[-5000:]]
baseline_acc = accuracy_score(y_all[-5000:], baseline_preds)
print(f"Baseline accuracy on last 5K rows: {baseline_acc:.4f}")
# Baseline accuracy on last 5K rows: 0.7918

importance = {}
for feat in ['sensor_temperature_normalized', 'sensor_pressure_raw', 'sensor_vibration_raw']:
    permuted_preds = []
    perm_indices = np.random.permutation(5000)
    X_last5k = X_all[-5000:]
    X_permuted = [dict(x) for x in X_last5k]
    perm_values = [X_last5k[i][feat] for i in perm_indices]
    for i, x in enumerate(X_permuted):
        x[feat] = perm_values[i]
    permuted_preds = [trained_pipeline.predict_one(x) for x in X_permuted]
    perm_acc = accuracy_score(y_all[-5000:], permuted_preds)
    importance[feat] = baseline_acc - perm_acc
    print(f"  {feat}: importance = {importance[feat]:.4f}")

# sensor_temperature_normalized: importance = 0.1403  ← top feature
# sensor_pressure_raw:           importance = 0.0821
# sensor_vibration_raw:          importance = 0.0612
# sensor_temperature_normalized is the top-importance feature
# Its signal was corrupted for first 10% of the stream by StandardScaler cold-start

Fix: route pre-normalized features through compose.Select to bypass StandardScaler

The fix is to split the pipeline into two branches: one branch routes raw features through preprocessing.StandardScaler() as before, and a separate branch uses compose.Select to pass already-normalized features directly to the model without any re-scaling. River’s compose module supports pipeline branching with the + operator: branch_a + branch_b concatenates the outputs of both branches into a single feature dict for the downstream model. By placing sensor_temperature_normalized (and any other pre-normalized columns) in a compose.Select branch, they bypass the StandardScaler entirely and reach the logistic regression with their original batch-normalized values, unchanged from row 1 of the stream.

With the corrected routing, the logistic regression receives sensor_temperature_normalized values in [−3.5, +3.5] from the very first row — the same scale it would see after full convergence. The gradient updates on the top feature are correct from the start of the stream. The rolling accuracy on the last 1,000-row window climbs from 0.69 to 0.79 — closing most of the 0.11-point gap against the offline baseline. The residual gap of 0.01 is attributable to concept drift in the sensor data between the beginning and end of the stream, which no cold-start fix can address.

# Fix: route pre-normalized features through compose.Select — bypass StandardScaler
# river 0.21.x, Python 3.11

from river import compose, preprocessing, linear_model, optim, stream, metrics

# Separate features into two groups:
# normalized_features: already batch-normalized in CSV — do NOT re-scale online
# raw_features: raw sensor readings — SHOULD be online-standardized

normalized_features = [
    'sensor_temperature_normalized',
    # add any other features pre-normalized using full-dataset batch statistics
]
raw_features = [
    'sensor_pressure_raw',
    'sensor_vibration_raw',
    # ... 47 raw sensor features that benefit from online standardization
]

# Branch 1: raw features through online StandardScaler
raw_branch = compose.Select(*raw_features) | preprocessing.StandardScaler()

# Branch 2: pre-normalized features bypass StandardScaler entirely
norm_branch = compose.Select(*normalized_features)

# Combine branches — compose concatenates outputs into one feature dict
pipeline_fixed = (
    raw_branch + norm_branch
) | linear_model.LogisticRegression(
    optimizer=optim.SGD(lr=0.01),
    l2=0.0,
)

# ← compose.Select + compose operator ensures:
# sensor_temperature_normalized passes through with original batch-normalized value
# sensor_pressure_raw, sensor_vibration_raw go through StandardScaler normally
# LogisticRegression receives 50 features total — same as before — but
# sensor_temperature_normalized is NEVER corrupted by running_std cold-start

metric_fixed  = metrics.RollingROCAUC(window_size=1000)
rolling_fixed = metrics.RollingAccuracy(window_size=1000)

for x, y in stream.iter_csv(
    'westfield_sensor_100k.csv',
    target='fault_label',
    converters={f: float for f in normalized_features + raw_features},
):
    y_proba = pipeline_fixed.predict_proba_one(x)
    if y_proba:
        metric_fixed.update(y, y_proba[True])
    y_pred = pipeline_fixed.predict_one(x)
    rolling_fixed.update(y, y_pred)
    pipeline_fixed.learn_one(x, y)

print(f"Fixed rolling ROCAUC (last 1K rows): {metric_fixed.get():.4f}")
print(f"Fixed rolling accuracy (last 1K rows): {rolling_fixed.get():.4f}")
# Fixed rolling ROCAUC (last 1K rows): 0.8012
# Fixed rolling accuracy (last 1K rows): 0.7891

# ---- After fix: summary ----
# Original pipeline (StandardScaler on all features):
#   rolling accuracy = 0.69  ← sensor_temperature_normalized corrupted for 10K rows
# Offline baseline (batch logistic regression):
#   accuracy = 0.80           ← reference
# Fixed pipeline (compose.Select bypasses StandardScaler for pre-normalized):
#   rolling accuracy = 0.79   ← closed most of the gap; residual 0.01 = concept drift

# ---- Hour breakdown ----
# Triage:     0.5h — rolling accuracy 0.69 vs offline 0.80;
#             progressive_val_score run row-by-row; rolling accuracy plotted;
#             accuracy drops in first 10K rows identified
# Root cause: 0.5h — StandardScaler cold-start running statistics plotted;
#             sensor_temperature_normalized identified as batch pre-normalized in CSV;
#             running_std remains below 1 for first 8,500 rows; max(1,std) clamp confirmed;
#             permutation importance = 0.14 (top feature among 50)
# Fix:        0.25h — compose.Select + branch pipeline implemented
# Validation: 0.25h — rolling accuracy 0.69→0.79; 1.5h total

Related retainer posts

• Vowpal Wabbit developer on retainer — user_id formatted as categorical name hash instead of numeric value — val_AUC 0.73 — 3 weeks
• Catalyst developer on retainer — softmax inside model.forward() with CrossEntropyLoss — double-softmax — val_accuracy 0.01 (random chance) — 3 weeks
• HPBandSter developer on retainer — BOHB min_budget=10 breaks multi-fidelity assumption on correlated features — Spearman rank 0.31 — val_AUC 0.74 — 3 weeks