Implementing sliding window algorithms for excursion detection

Point-in-time threshold checks cannot tell transient sensor noise and brief door openings apart from genuine product-threatening deviations, yet cold chain regulation demands exactly that distinction. FDA 21 CFR §211.142 requires continuous monitoring with documented impact assessments, the EU GDP Guidelines (Chapter 9) require monitoring systems to capture both the magnitude and duration of out-of-range events, and USP <1079> establishes Mean Kinetic Temperature (MKT) as the accepted metric for cumulative thermal stress. A sliding window algorithm satisfies all three by evaluating temperature trajectories over a configurable, rolling timeframe — turning a noisy stream into the time-weighted evidence an inspector will scrutinize during a batch disposition investigation. This how-to builds that detector in Python and is the engine behind the duration-based excursion scoring model.

The window directly discharges these controls, each of which becomes a concrete computation in the code below.

Regulatory requirement	Window control	Regulatory anchor
Continuous monitoring with documented deviation impact	Rolling buffer evaluated on every reading; explicit state emitted per sample	21 CFR §211.142
Capture magnitude and duration of out-of-range events	Grace period + cumulative time-out-of-range tracking via the state machine	EU GDP Ch. 9
Scientifically valid cumulative thermal stress metric	Time-weighted MKT over the window	USP <1079>
Attributable, tamper-evident record of every evaluation	SHA-256 hash over each per-sample payload	21 CFR Part 11 §11.10(e)

Prerequisites

Python 3.9 or newer — the example uses deque[Tuple[...]] generic subscription in annotations and the standard-library enum, hashlib, and math modules. No third-party packages are required for the detector itself.
No external dependencies — collections.deque, datetime, and hashlib ship with CPython. For the optional plotting/validation harness you may add pip install matplotlib separately, but it is not needed in production.
Telemetry contract — readings arrive as (timestamp, temperature_c) pairs with strictly monotonic UTC timestamps. Upstream, the secure IoT gateway is assumed to discipline clocks and order packets; this detector trusts that contract and rejects violations rather than silently repairing them.
Product specification limits — upper/lower limits and the grace period must come from validated, product-specific thresholds (see establishing temperature excursion thresholds by product). Do not hard-code 2–8°C for every SKU.
Activation energy — activation_energy (kJ/mol) and gas_constant (kJ/(mol·K)) must share the same energy unit. The defaults pair 83.144 kJ/mol with 0.00831446 kJ/(mol·K).
Access control — the process consuming this detector must write to an append-only audit store; the per-sample hash is only meaningful if the records it protects cannot be silently overwritten.

Step-by-Step Implementation

Step 1 — Model the detector as a finite state machine

Before writing buffer code, fix the decision logic. The detector is a three-state machine: NORMAL (all readings in range), GRACE (a breach has started but has not yet outlasted the configured grace period — this filters door openings, defrost cycles, and calibration drift), and EXCURSION_ACTIVE (the breach persisted beyond grace, so an audit-ready alert fires). A zero-length grace period must move straight from NORMAL to EXCURSION_ACTIVE on the first breach. Modelling the transitions explicitly is what lets the downstream rule engine consume structured state changes instead of raw, noisy telemetry.

Step 2 — Maintain the rolling window with a deque

High-throughput telemetry demands deterministic latency, so the window uses a double-ended queue for O(1) append and pop, avoiding garbage-collection pauses that could stall real-time alerting. The constructor validates configuration up front (an inverted limit or negative grace period is a setup error, not a runtime condition), and _prune_window drops readings older than the configured span so the buffer always reflects the current evaluation horizon.

python

import enum
import hashlib
import math
from collections import deque
from datetime import datetime, timedelta
from typing import Any, Dict, Optional, Tuple


class ExcursionState(enum.Enum):
    NORMAL = "NORMAL"
    GRACE = "GRACE"
    EXCURSION_ACTIVE = "EXCURSION_ACTIVE"


class SlidingWindowExcursionDetector:
    """
    Production-grade sliding window detector for pharmaceutical cold chain telemetry.
    Evaluates temperature streams against configurable thresholds, grace periods,
    and rolling MKT, exposing explicit state for 21 CFR Part 11 audit logging.

    Units: activation_energy and gas_constant must share the same energy unit
    (both kJ or both J). Defaults use kJ.
    """

    def __init__(
        self,
        window_minutes: int,
        upper_limit: float,
        lower_limit: float,
        grace_period_minutes: float = 0.0,
        activation_energy: float = 83.144,  # kJ/mol (standard pharmaceutical value)
        gas_constant: float = 0.00831446,   # kJ/(mol·K) — pair with kJ Ea
    ):
        # §211.142 / GDP Ch.9: thresholds must be validated, not defaulted — reject
        # an inverted or non-positive configuration before any reading is processed.
        if window_minutes <= 0:
            raise ValueError("window_minutes must be strictly positive")
        if upper_limit <= lower_limit:
            raise ValueError("upper_limit must exceed lower_limit")
        if grace_period_minutes < 0:
            raise ValueError("grace_period_minutes cannot be negative")

        self.window_delta = timedelta(minutes=window_minutes)
        self.upper_limit = upper_limit
        self.lower_limit = lower_limit
        self.grace_delta = timedelta(minutes=grace_period_minutes)
        self.Ea = activation_energy
        self.R = gas_constant

        self._buffer: deque[Tuple[datetime, float]] = deque()
        self._state = ExcursionState.NORMAL
        self._grace_start: Optional[datetime] = None
        self._last_processed: Optional[datetime] = None

    def _prune_window(self, current_ts: datetime) -> None:
        # USP <1079> evaluates cumulative thermal stress over a bounded window, not
        # for all time; evict readings older than the configured span (O(1) each).
        cutoff = current_ts - self.window_delta
        while self._buffer and self._buffer[0][0] < cutoff:
            self._buffer.popleft()

Verify the configuration guards before going further:

python

import pytest  # pip install pytest, dev-only

# Inverted limits and non-positive windows must fail fast, never half-configure.
with pytest.raises(ValueError):
    SlidingWindowExcursionDetector(window_minutes=30, upper_limit=2.0, lower_limit=8.0)
det = SlidingWindowExcursionDetector(window_minutes=30, upper_limit=8.0, lower_limit=2.0)
assert det.upper_limit > det.lower_limit

Step 3 — Compute time-weighted MKT per USP <1079>

Sampling intervals in real telemetry are rarely uniform — packet bursts after reconnection, dropped readings, and variable polling cadence all distort an unweighted Arrhenius mean. The correct form weights each sample’s exponential term by the dwell time $Δ t_{i}$ until the next sample:

T_{K} = \frac{- E _{a} / R}{ln ⎝ ⎜ ⎜ ⎜ ⎛ \frac{i \sum Δ t _{i} e ^{- E_{a} / (R T_{i})}}{i \sum Δ t _{i}} ⎠ ⎟ ⎟ ⎟ ⎞} T_{MKT} (^{\circ} C) = T_{K} - 273.15

python

    def _calculate_mkt(self) -> float:
        """Time-weighted Mean Kinetic Temperature per USP <1079>.

        Weights each sample by the interval until the next sample so irregular
        sampling cadences (which real IoT telemetry always exhibits) do not bias
        the MKT toward whichever interval happened to be densest. # USP <1079>
        """
        if len(self._buffer) < 2:
            return float("nan")

        weighted_exp_sum = 0.0
        total_seconds = 0.0
        samples = list(self._buffer)
        for (t0, T0), (t1, _) in zip(samples, samples[1:]):
            dt = (t1 - t0).total_seconds()
            if dt <= 0:
                continue
            t_k = T0 + 273.15
            weighted_exp_sum += dt * math.exp(-self.Ea / (self.R * t_k))
            total_seconds += dt

        if total_seconds == 0:
            return float("nan")
        avg_exp = weighted_exp_sum / total_seconds
        return (-self.Ea / (self.R * math.log(avg_exp))) - 273.15

A constant temperature must return that same temperature regardless of spacing — the cheapest possible regression test for the weighting math:

python

# For a steady 5.0 °C trace, MKT must collapse to ~5.0 °C (Arrhenius is order-preserving).
det = SlidingWindowExcursionDetector(window_minutes=60, upper_limit=8.0, lower_limit=2.0)
base = datetime(2026, 3, 4, 0, 0, 0)
for i in range(6):
    det.process_reading(base + timedelta(minutes=i * 5), 5.0)
assert abs(det._calculate_mkt() - 5.0) < 1e-6

Step 4 — Drive state transitions and emit an audit-ready payload

process_reading is the public entry point. It rejects out-of-order timestamps (a non-monotonic stream corrupts the audit trail), prunes and appends, recomputes rolling statistics, applies the state-machine logic from Step 1, and returns a compliance-ready payload whose SHA-256 hash makes each evaluation tamper-evident.

python

    def process_reading(self, timestamp: datetime, temperature: float) -> Dict[str, Any]:
        """Ingest one reading, update the window, transition state, return an
        audit-ready payload."""
        # §11.10(e): a chronological audit trail requires monotonic time; a
        # backdated or duplicate reading is rejected, never silently reordered.
        if self._last_processed and timestamp <= self._last_processed:
            raise ValueError("Timestamps must be strictly monotonic for deterministic audit trails")
        self._last_processed = timestamp

        self._prune_window(timestamp)
        self._buffer.append((timestamp, temperature))

        temps_in_window = [t for _, t in self._buffer]
        rolling_mean = sum(temps_in_window) / len(temps_in_window)
        rolling_mkt = self._calculate_mkt()

        is_breach = temperature > self.upper_limit or temperature < self.lower_limit

        # GDP Ch.9 duration logic: a zero grace period moves NORMAL -> EXCURSION_ACTIVE
        # on the first breach; otherwise the breach must outlast the grace window.
        if is_breach:
            if self._state == ExcursionState.NORMAL:
                self._grace_start = timestamp
                if self.grace_delta == timedelta(0):
                    self._state = ExcursionState.EXCURSION_ACTIVE
                else:
                    self._state = ExcursionState.GRACE
            elif self._state == ExcursionState.GRACE:
                if timestamp - self._grace_start >= self.grace_delta:
                    self._state = ExcursionState.EXCURSION_ACTIVE
        else:
            if self._state in (ExcursionState.GRACE, ExcursionState.EXCURSION_ACTIVE):
                self._state = ExcursionState.NORMAL
                self._grace_start = None

        grace_remaining = 0.0
        if self._state == ExcursionState.GRACE and self._grace_start:
            grace_remaining = max(0.0, (self.grace_delta - (timestamp - self._grace_start)).total_seconds())

        # §11.10(e) tamper evidence: hash the attributable fields of this evaluation.
        audit_payload = f"{timestamp.isoformat()}|{temperature}|{self._state.value}"
        audit_hash = hashlib.sha256(audit_payload.encode("utf-8")).hexdigest()

        return {
            "timestamp": timestamp.isoformat(),
            "temperature_c": temperature,
            "rolling_mean_c": round(rolling_mean, 2),
            "rolling_mkt_c": None if math.isnan(rolling_mkt) else round(rolling_mkt, 2),
            "window_sample_count": len(self._buffer),
            "current_state": self._state.value,
            "excursion_active": self._state == ExcursionState.EXCURSION_ACTIVE,
            "grace_remaining_seconds": round(grace_remaining, 1),
            "audit_hash": audit_hash,
        }

Prove the grace period actually suppresses a transient and escalates a sustained breach:

python

det = SlidingWindowExcursionDetector(
    window_minutes=60, upper_limit=8.0, lower_limit=2.0, grace_period_minutes=10,
)
t = datetime(2026, 3, 4, 0, 0, 0)
det.process_reading(t, 5.0)                               # NORMAL
r1 = det.process_reading(t + timedelta(minutes=2), 9.5)  # breach starts -> GRACE
assert r1["current_state"] == "GRACE" and not r1["excursion_active"]
r2 = det.process_reading(t + timedelta(minutes=4), 5.0)  # recovers inside grace
assert r2["current_state"] == "NORMAL"                   # transient suppressed
det.process_reading(t + timedelta(minutes=6), 9.8)       # breach restarts
r3 = det.process_reading(t + timedelta(minutes=20), 9.9) # persists past grace
assert r3["excursion_active"] is True                     # genuine excursion fired

Step 5 — Validate against synthetic profiles before production

Before deployment, run controlled synthetic traces through the detector and assert exact outputs. A version-controlled test matrix covering transient spikes (GRACE then NORMAL), sustained breaches (EXCURSION_ACTIVE), and out-of-order timestamps (must raise ValueError) is the artefact a GAMP 5 computerized-system-validation protocol references. Capturing the full payload stream also feeds the cache warming strategy that primes rule-engine state on restart.

python

def replay_profile(profile, **config):
    """Deterministically replay a (timestamp, temp) profile and return payloads.
    Used in CSV/IQ-OQ-PQ evidence: identical input must yield identical output."""
    det = SlidingWindowExcursionDetector(**config)  # 21 CFR Part 11 reproducibility
    return [det.process_reading(ts, temp) for ts, temp in profile]

t = datetime(2026, 3, 4, 0, 0, 0)
profile = [(t + timedelta(minutes=i), 5.0) for i in range(3)]
profile += [(t + timedelta(minutes=3), 9.9)]  # single breach, grace=0
states = [p["current_state"] for p in replay_profile(
    profile, window_minutes=30, upper_limit=8.0, lower_limit=2.0,
)]
assert states == ["NORMAL", "NORMAL", "NORMAL", "EXCURSION_ACTIVE"]

Compliance validation checklist

Run this as part of computerized-system validation; every item is something an auditor can independently confirm for the sliding window control.

Thresholds traced to product data — upper/lower limits and grace period derive from validated, SKU-specific stability data, not a global default (§211.142).
Duration captured, not just magnitude — the test matrix proves a sub-grace transient stays NORMAL while a sustained breach reaches EXCURSION_ACTIVE (EU GDP Ch. 9).
MKT is time-weighted — a unit test feeds an irregularly spaced trace and confirms the result matches the $Δ t_{i}$ -weighted form, not the simple mean (USP <1079>).
Monotonic time enforced — a backdated or duplicate timestamp raises ValueError and is never silently reordered (§11.10(e)).
Per-sample tamper evidence — every payload carries a SHA-256 audit_hash and is written to an append-only store.
Deterministic replay — identical input profiles yield byte-identical state sequences across runs and environments, evidenced in the CSV protocol.
Grace period justified — the configured grace value is documented against the observed distribution of transient breaches during normal loading/unloading.

Troubleshooting

Symptom	Root cause	Fix
`ValueError: Timestamps must be strictly monotonic` floods the log	Edge gateways on independent clocks delivering microsecond drift or replayed packets	Discipline edge clocks via NTP to a NIST-traceable source; add a `max_clock_skew_ms` gate that buffers or rejects skewed readings before the detector
`rolling_mkt_c` returns `None` or `nan` at ultra-cold storage (-70 °C)	The exponential term underflows toward machine epsilon	Clamp inputs to a validated range (e.g. -100 °C to +100 °C); switch the aggregation to `math.fsum()` for higher precision over large windows
Excessive false-positive alerts during routine door operations	Grace period shorter than real transient durations	Calibrate the grace period to the 95th percentile of observed transient breaches from historical telemetry; log all transitions for continuous tuning
Real interventions arrive late	Grace period set too long to suppress noise	Shorten grace toward the product’s time-out-of-range budget; for tight biologic budgets pair with multi-sensor correlation instead of a longer grace
Container memory grows unbounded under high-frequency streams	Window misconfigured, or a time-prune and a `maxlen` cap mixed together	Keep a single eviction strategy — this design time-prunes to handle irregular sampling; if you cap with `maxlen` instead, do not also time-prune

Conclusion

The single most operationally important correctness detail is the MKT calculation: it uses the time-weighted form, with each Arrhenius term weighted by $Δ t_{i}$ to the next sample, not the simple average over $n$ samples. With irregular IoT telemetry the unweighted form systematically overweights dense bursts (post-reconnection replay) and underweights sparse pre-gap readings, producing an MKT that misrepresents the actual thermal history — precisely the figure regulators examine during a batch disposition investigation.

For broader context, see Duration-Based Scoring for Temperature Excursions, part of the Temperature Excursion Detection & Automated Rule Engines section.