Detecting Stuck Sensor Values with Run-Length Checks #

A stuck sensor is the most expensive failure in an environmental network because it is the one that looks healthiest. Data keeps arriving, at the right cadence, with values inside every plausible range, passing every completeness check — and the number has not changed since Tuesday. Statistical detectors are structurally blind to it, and a stuck reading can persist for weeks before someone notices a flat line on a chart. This guide implements the check that catches it: a run-length test on identical or near-identical consecutive values, tuned to each measurement’s real variability. It belongs at the cheap end of the gate chain described in anomaly detection methods for sensor networks.

Why It Needs Its Own Detector #

Consider a temperature probe frozen at 18.3 °C. Every reading is physically plausible, so the range check passes. The change between consecutive readings is zero, so the rate-of-change check passes — it is looking for values that move too fast. The reading sits near the middle of the recent distribution, so its Z-score is approximately zero.

Then it gets worse. A rolling standard deviation computed over a window that is mostly constant collapses toward zero, so the Z-score denominator shrinks, and the next genuinely varying reading — from this sensor after it recovers, or from a neighbour if the statistics are pooled — produces an enormous Z-score. A stuck sensor does not merely evade detection; it actively degrades the detector around it.

The only signal that distinguishes stuck from stable is the duration of constancy relative to what that measurement does naturally, and duration is exactly what a per-reading detector cannot see.

Production-Ready Implementation #

# 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

QC_GOOD, QC_SUSPECT, QC_STUCK = 1, 2, 3


@dataclass(frozen=True)
class StuckPolicy:
    """Run-length thresholds for one measurement type.

    tolerance is the change below which two readings count as 'the same'. It must
    be at or just above the sensor's quantisation step: a probe reporting to
    0.1 °C moves in steps of 0.1, so an exact-equality test would miss a sensor
    oscillating between two adjacent codes — the most common form of 'stuck'.
    """

    tolerance: float
    warn_intervals: int
    stuck_intervals: int
    zero_intervals: int | None = None      # tighter threshold for runs of exactly zero


def run_lengths(values: pd.Series, tolerance: float) -> np.ndarray:
    """Length of the constant run each reading belongs to, in intervals.

    A reading whose neighbours differ by more than `tolerance` has run length 1.
    """
    v = values.to_numpy(dtype=float)
    changed = np.ones(len(v), dtype=bool)
    changed[1:] = np.abs(np.diff(v)) > tolerance
    group = np.cumsum(changed)
    counts = np.bincount(group)
    return counts[group]


def flag_stuck(df: pd.DataFrame, policy: StuckPolicy, value_col: str = "value") -> pd.DataFrame:
    """Flag readings that belong to an implausibly long constant run."""
    out = df.copy()
    lengths = run_lengths(out[value_col], policy.tolerance)
    out["run_length"] = lengths

    is_zero_run = np.isclose(out[value_col].to_numpy(dtype=float), 0.0)
    zero_limit = policy.zero_intervals or policy.stuck_intervals

    stuck = ((lengths >= policy.stuck_intervals) |
             (is_zero_run & (lengths >= zero_limit)))
    warn = (~stuck) & (lengths >= policy.warn_intervals)

    out.loc[warn, "qc_flag"] = np.maximum(out.loc[warn, "qc_flag"], QC_SUSPECT)
    out.loc[stuck, "qc_flag"] = QC_STUCK
    return out

Two design choices carry the correctness. Using a tolerance rather than exact equality catches the sensor that dithers between two adjacent quantisation codes — which reads as “changing” to an equality test and is just as broken. And flagging the whole run rather than only its tail means the first reading of a stuck period is flagged too, which matters when the data is later filtered by flag rather than by time.

For a streaming path, the same test with O(1) state per sensor:

@dataclass
class StuckTracker:
    """Incremental run tracking for a live stream — one small object per sensor."""

    policy: StuckPolicy
    last_value: float | None = None
    run: int = 0

    def observe(self, value: float) -> int:
        if self.last_value is not None and abs(value - self.last_value) <= self.policy.tolerance:
            self.run += 1
        else:
            self.run = 1
        self.last_value = value
        if self.run >= self.policy.stuck_intervals:
            return QC_STUCK
        if self.run >= self.policy.warn_intervals:
            return QC_SUSPECT
        return QC_GOOD

Parameter Tuning Guide #

Intervals, not minutes: the thresholds scale with the reporting cadence, so the same policy works for a one-minute and a fifteen-minute feed. Rainfall is disabled outright — a tipping-bucket gauge reporting zero for three dry weeks is a working gauge, and no run-length rule can distinguish that from a jammed bucket. Use a separate check that compares against neighbouring gauges during recorded rainfall instead.

Derive your own thresholds from the observed distribution rather than copying the table:

def suggest_thresholds(healthy: pd.Series, tolerance: float,
                       warn_pct: float = 99.0, stuck_pct: float = 99.9) -> tuple[int, int]:
    """Percentiles of the run-length distribution across known-good sensors."""
    lengths = run_lengths(healthy, tolerance)
    return int(np.percentile(lengths, warn_pct)), int(np.percentile(lengths, stuck_pct))

Verification and Testing #

# python 3.11 · pandas==2.2.2 · pytest==8.2.0
import numpy as np
import pandas as pd

POLICY = StuckPolicy(tolerance=0.5, warn_intervals=20, stuck_intervals=60)


def _frame(values):
    return pd.DataFrame({"value": values, "qc_flag": [QC_GOOD] * len(values)})


def test_a_frozen_sensor_is_flagged_and_a_varying_one_is_not():
    rng = np.random.default_rng(3)
    healthy = _frame(list(rng.normal(18, 4, 200)))
    frozen = _frame(list(rng.normal(18, 4, 100)) + [18.3] * 100)

    assert (flag_stuck(healthy, POLICY)["qc_flag"] == QC_STUCK).sum() == 0
    assert (flag_stuck(frozen, POLICY)["qc_flag"] == QC_STUCK).sum() >= 60


def test_dithering_between_two_codes_still_counts_as_stuck():
    """Alternating 18.3/18.4 is a frozen sensor, not a varying one."""
    values = [18.3 if i % 2 else 18.4 for i in range(120)]
    assert (flag_stuck(_frame(values), POLICY)["qc_flag"] == QC_STUCK).sum() >= 60


def test_the_whole_run_is_flagged_not_just_the_tail():
    values = list(np.linspace(10, 20, 40)) + [20.0] * 80
    out = flag_stuck(_frame(values), POLICY)
    stuck_idx = out.index[out["qc_flag"] == QC_STUCK]
    assert stuck_idx.min() == 40           # flagged from the first repeated reading


def test_streaming_tracker_matches_the_batch_result():
    values = list(np.linspace(10, 20, 40)) + [20.0] * 80
    tracker = StuckTracker(POLICY)
    streamed = [tracker.observe(v) for v in values]
    batch = flag_stuck(_frame(values), POLICY)["qc_flag"].tolist()
    assert streamed[-1] == batch[-1] == QC_STUCK

The last test is worth keeping permanently. A batch implementation and a streaming one that disagree produce a dataset whose flags depend on how it was processed, which is the hardest class of inconsistency to debug months later.

Gotchas #

Exact equality instead of a tolerance. A sensor whose last bit flickers reports 18.3, 18.4, 18.3 forever. Equality sees constant change; a tolerance sees the truth.

Thresholds in minutes rather than intervals. A policy written for one-minute data silently becomes fifteen times stricter when applied to a fifteen-minute feed, and every sensor on that feed gets flagged.

Running the check before resampling. On an irregular series, “consecutive readings” is not a fixed duration, so a run of twenty readings might be twenty minutes or two hours. Run this after resampling to a fixed grid.

Counting filled values in a run. An interpolated stretch is constant-ish by construction and will trigger the check. Exclude imputed rows, or the detector will flag your own gap filling as a hardware fault.

FAQ #

Related #

• Anomaly Detection Methods for Sensor Networks — the detection stage this check belongs at the front of
• Isolation Forest vs Z-Score for Sensor Anomaly Detection — the statistical detectors this check exists to complement
• Automating QC Flags for Missing Environmental Readings — the flag vocabulary a stuck-value detection writes into