Detecting Step Changes After Sensor Maintenance #

Drift is gradual; maintenance is not. A replaced filter, a cleaned optical chamber, a new calibration coefficient or a firmware update produces a discontinuity — a step in the series where the sensor’s relationship to reality changed in an instant. Every drift-correction method assumes continuity, so a step that is not detected gets smeared across the rolling window: for half a window before and after the event, the baseline is a blend of two different sensors and the corrected values are wrong on both sides. This guide detects those steps, distinguishes them from weather, and resets the baseline cleanly. It protects the methods in sensor drift correction algorithms in Python.

Difference First, Detect Second #

The hardest part of step detection in environmental data is that the environment itself steps. A cold front drops temperature by six degrees in twenty minutes across an entire city; a change detector run on a raw series flags it, correctly identifying a change that has nothing to do with the instrument.

The fix is to detect on a differenced series: subtract a robust summary of comparable neighbours from the target before running the detector. A regional weather change moves target and neighbours together and cancels; a filter swap moves only the target and survives at full amplitude. The differenced series is also far more stationary than the raw one, which is what the detector’s statistics assume.

The neighbour set matters. Use sensors that share the target’s exposure — the same site classification rather than merely the nearest coordinates — because a street-canyon sensor differenced against a rooftop one produces a residual full of real, uncancelled variation.

Production-Ready Implementation #

CUSUM is the right detector here: it accumulates small persistent deviations, which is exactly the shape of a step, and it has two interpretable parameters.

# python 3.11 · pandas==2.2.2 · numpy==1.26.4
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass(frozen=True)
class StepDetection:
    index: int
    direction: str          # 'up' or 'down'
    magnitude: float        # in measurement units
    confidence: float       # cumulative sum at detection, in sigma


def neighbour_difference(
    target: pd.Series, neighbours: pd.DataFrame, min_neighbours: int = 3
) -> pd.Series:
    """Target minus the robust centre of its comparable neighbours.

    The median (not the mean) so that one neighbour with its own step does not
    inject a mirror-image step into every other sensor's differenced series.
    """
    usable = neighbours.notna().sum(axis=1) >= min_neighbours
    centre = neighbours.median(axis=1)
    return (target - centre).where(usable)


def cusum_steps(
    series: pd.Series, *, threshold_sigma: float = 5.0, drift_sigma: float = 0.5,
    min_separation: int = 60,
) -> list[StepDetection]:
    """Two-sided CUSUM over a differenced series.

    threshold_sigma: cumulative deviation, in sigma, that declares a step.
    drift_sigma:     the slack term; deviations below this do not accumulate, so
                     genuine slow drift is ignored and only steps trigger.
    min_separation:  intervals to wait before another detection, so one step is
                     reported once rather than at every subsequent sample.
    """
    x = series.dropna().to_numpy(dtype=float)
    idx = series.dropna().index.to_numpy()
    if len(x) < 3 * min_separation:
        return []

    sigma = float(np.median(np.abs(np.diff(x)))) * 1.4826 or 1e-9   # robust scale
    mean = float(np.median(x))

    hi = lo = 0.0
    last_hit = -min_separation
    out: list[StepDetection] = []
    for i, value in enumerate(x):
        z = (value - mean) / sigma
        hi = max(0.0, hi + z - drift_sigma)
        lo = min(0.0, lo + z + drift_sigma)

        if i - last_hit < min_separation:
            continue
        if hi > threshold_sigma or lo < -threshold_sigma:
            before = float(np.median(x[max(0, i - min_separation):i]))
            after = float(np.median(x[i:i + min_separation]))
            out.append(StepDetection(
                index=int(idx[i]) if np.issubdtype(idx.dtype, np.integer) else i,
                direction="up" if hi > threshold_sigma else "down",
                magnitude=after - before,
                confidence=abs(hi if hi > threshold_sigma else lo),
            ))
            hi = lo = 0.0
            mean = after            # re-baseline after the step
            last_hit = i
    return out

Re-baselining mean after each detection is what allows several steps in one series to be found. A detector that keeps the original mean accumulates forever after the first step and reports the rest of the record as one continuous anomaly.

Resetting the drift baseline then becomes a matter of segmenting the series:

def segments_between_steps(n: int, steps: list[StepDetection]) -> list[slice]:
    """Index ranges over which a drift baseline may be computed continuously."""
    edges = [0, *(s.index for s in steps), n]
    return [slice(a, b) for a, b in zip(edges, edges[1:]) if b - a > 0]


def baseline_per_segment(values: pd.Series, steps: list[StepDetection],
                         window: int = 720) -> pd.Series:
    """Rolling baseline that restarts at every confirmed step.

    Without the reset, the window spanning a step averages two different sensors
    and the correction is wrong on both sides for half a window in each
    direction.
    """
    out = pd.Series(index=values.index, dtype=float)
    for seg in segments_between_steps(len(values), steps):
        chunk = values.iloc[seg]
        out.iloc[seg] = chunk.rolling(window, min_periods=window // 2).mean()
    return out

Parameter Tuning Guide #

The last row is the one detection earns its keep on: a blockage is never logged, presents as a step with a slightly soft edge, and is otherwise indistinguishable from a sensor that has simply become less sensitive.

Verification and Testing #

# python 3.11 · numpy==1.26.4 · pandas==2.2.2 · pytest==8.2.0
def _series_with_step(n=3000, step_at=1500, magnitude=1.6, seed=11):
    rng = np.random.default_rng(seed)
    x = rng.normal(0, 0.4, n)
    x[step_at:] += magnitude
    return pd.Series(x)


def test_detects_a_step_close_to_where_it_was_injected():
    steps = cusum_steps(_series_with_step(), threshold_sigma=5.0, min_separation=200)
    assert len(steps) == 1
    assert abs(steps[0].index - 1500) < 250
    assert abs(steps[0].magnitude - 1.6) < 0.4


def test_a_shared_weather_change_is_cancelled_by_differencing():
    rng = np.random.default_rng(5)
    weather = np.concatenate([np.zeros(1500), np.full(1500, -6.0)])   # a front
    target = pd.Series(weather + rng.normal(0, 0.4, 3000))
    neighbours = pd.DataFrame({
        f"n{i}": weather + rng.normal(0, 0.4, 3000) for i in range(4)
    })

    assert len(cusum_steps(target)) >= 1                       # raw series: false positive
    assert cusum_steps(neighbour_difference(target, neighbours)) == []


def test_baseline_resets_rather_than_smearing_across_the_step():
    values = pd.Series(np.concatenate([np.full(1500, 20.0), np.full(1500, 21.6)]))
    steps = [StepDetection(index=1500, direction="up", magnitude=1.6, confidence=8.0)]
    baseline = baseline_per_segment(values, steps, window=360)
    assert abs(baseline.iloc[1560] - 21.6) < 0.05              # no blend from before the step

The middle test is the one to run against real data as well as synthetic: replay a week containing a known frontal passage and confirm the differenced series produces no detections. If it does, the neighbour set is not comparable enough.

Gotchas #

Detecting on the raw series. Every weather transition becomes a step. Difference first.

A min_separation shorter than the drift window. Two detections inside one baseline window make the segments too short to compute a baseline at all, and the correction silently stops emitting.

Trusting the maintenance log alone. Logged dates are often the date of the paperwork, not of the visit. Use the detected step as the authoritative boundary and the log as corroboration.

Retroactively correcting data before a step. Readings before a recalibration were produced under the old coefficients and were correct under them. Close the old validity interval and open a new one instead of rewriting history.

FAQ #

Related #

• Sensor Drift Correction Algorithms in Python — the drift correction stage this protects
• Correcting Temperature Sensor Drift Using Rolling Averages — the baseline that a step change corrupts if it is not reset
• Modelling Sensor Deployment History with Validity Intervals — where a confirmed step becomes a new validity interval