Implementing sliding window algorithms for excursion detection

Pharmaceutical cold chain telemetry operates at high sampling frequencies, generating continuous streams of temperature, humidity, and shock data. Point-in-time threshold checks are insufficient for modern GMP environments because they cannot distinguish between transient sensor noise, brief door openings, and genuine product-threatening deviations. Implementing sliding window algorithms for excursion detection resolves this ambiguity by evaluating temperature trajectories over configurable, rolling timeframes. This approach aligns automated rule evaluation with stability science, reduces false-positive alert fatigue, and produces auditable decision trails required by regulatory bodies.

Compliance Mapping and Regulatory Alignment

Cold chain compliance frameworks explicitly require time-weighted evaluation of temperature deviations rather than instantaneous breach logging. FDA 21 CFR § 211.142 mandates that drug product storage conditions be continuously monitored and that deviations be investigated with documented impact assessments. The EU GDP Guidelines (Chapter 9) require that monitoring systems capture both the magnitude and duration of out-of-range events, while USP <1079> establishes Mean Kinetic Temperature (MKT) as the scientifically accepted metric for evaluating cumulative thermal stress.

A sliding window architecture directly satisfies these requirements by maintaining a fixed-length buffer of recent sensor readings. As each new data point arrives, the algorithm evaluates the window against upper/lower specification limits, calculates rolling statistics, and determines whether the cumulative time-out-of-range (TTR) or rolling MKT exceeds product-specific tolerance bands. This methodology transforms raw telemetry into compliance-ready evidence, ensuring that every excursion flag is accompanied by a deterministic, reproducible calculation path suitable for 21 CFR Part 11 validation. When paired with downstream Duration-Based Scoring for Temperature Excursions, the sliding window provides the foundational temporal context required for risk-based impact assessments and batch disposition decisions.

Algorithm Architecture and State Management

High-throughput telemetry pipelines demand deterministic latency and memory efficiency. The sliding window implementation relies on a double-ended queue (deque) to maintain O(1) append and pop operations, preventing garbage collection pauses that could disrupt real-time alerting. The architecture operates as a finite state machine with three primary states:

1. NORMAL: All readings within the window remain within specification limits.
2. GRACE: A threshold breach occurs, but the duration has not yet exceeded the configured grace period. This filters transient spikes caused by door openings, defrost cycles, or sensor calibration drift.
3. EXCURSION_ACTIVE: The breach persists beyond the grace period, or cumulative TTR/MKT crosses the regulatory threshold. An audit-ready alert is generated.

This stateful design integrates seamlessly into broader Temperature Excursion Detection & Automated Rule Engines, allowing rule engines to consume structured state transitions rather than raw, noisy telemetry. The algorithm strictly enforces monotonic timestamp progression to prevent out-of-order data corruption, a critical requirement for validated pharmaceutical data pipelines.

Time-weighted MKT per USP <1079>

Sampling intervals in real cold-chain telemetry are rarely uniform — packet bursts after reconnection, dropped readings, and variable polling cadences all distort an unweighted Arrhenius mean. The implementation below weights each sample’s exponential term by the dwell time $\Delta t_i$ until the next sample:

$$T_K=\frac{-E_a/R}{\ln \left(\frac{\sum _i\Delta t_i{e}^{-E_a/(R{T}_i)}}{\sum _i\Delta t_i}\right)}{T}_{\text{MKT}}{(}^{\circ }\mathrm{C})=T_K-273.15$$

Production-Grade Python Implementation

The following implementation uses collections.deque for memory-efficient window management, supports configurable grace periods, computes rolling MKT per USP <1079>, and exposes explicit state tracking for audit logging.

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 calculations. Designed for deterministic latency and 21 CFR Part 11 compliance.

    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
    ):
        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 _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 the prose explicitly anticipates) do not bias
        the MKT toward whichever interval happened to be densest.
        """
        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

    def _prune_window(self, current_ts: datetime) -> None:
        """Removes readings older than the configured window size."""
        cutoff = current_ts - self.window_delta
        while self._buffer and self._buffer[0][0] < cutoff:
            self._buffer.popleft()

    def process_reading(self, timestamp: datetime, temperature: float) -> Dict[str, Any]:
        """
        Ingests a single telemetry reading, updates the sliding window,
        evaluates state transitions, and returns a compliance-ready payload.
        """
        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

        # Finite state machine transition logic. A zero grace period must move
        # NORMAL -> EXCURSION_ACTIVE on the first breach, never linger in GRACE.
        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())

        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,
        }

Validation, Deployment, and Troubleshooting

Deploying this algorithm in a regulated environment requires rigorous validation and proactive troubleshooting. The following steps address common operational challenges.

1. Deterministic Testing and 21 CFR Part 11 Validation

Before production deployment, validate the algorithm against known synthetic datasets. Inject controlled temperature profiles that include:

• Transient spikes (should trigger GRACE, then revert to NORMAL)
• Sustained breaches exceeding grace period (should transition to EXCURSION_ACTIVE)
• Out-of-order timestamps (must raise ValueError to prevent audit trail corruption)

Maintain version-controlled unit tests that assert exact floating-point outputs for MKT and rolling mean. Regulatory auditors expect reproducible calculations; document the test matrix and expected state transitions in your validation protocol.

2. Timestamp Synchronization Drift

Cold chain sensors often operate on independent internal clocks. If telemetry arrives with microsecond drift, the sliding window may incorrectly prune valid readings or retain stale data. Fix: Implement NTP synchronization at the edge gateway level. Add a configurable max_clock_skew_ms parameter to your ingestion pipeline. Reject or buffer readings that deviate beyond the threshold before passing them to the detector.

3. MKT Calculation Instability at Extreme Temperatures

At very low temperatures (e.g., ultra-cold storage at -70°C), the exponential term in the MKT formula can approach machine epsilon, causing floating-point underflow. Fix: Clamp input temperatures to a validated operational range (e.g., -100°C to +100°C) before MKT computation. Use math.fsum() instead of sum() for higher precision when aggregating exponential terms across large windows. Reference the official Python collections documentation for deque performance characteristics and memory guarantees under sustained load.

4. Alert Fatigue and Grace Period Tuning

Overly aggressive grace periods generate false positives during routine loading/unloading operations. Conversely, excessively long grace periods delay critical interventions. Fix: Calibrate grace periods using historical telemetry. Analyze the distribution of transient breaches during normal operations. Set the grace period to the 95th percentile of observed transient durations. Log all state transitions to a structured audit table for continuous improvement. Consult FDA guidance on continuous monitoring to ensure grace configurations do not violate product-specific stability commitments.

5. Memory Management in High-Frequency Streams

While deque provides O(1) operations, unbounded growth from misconfigured window sizes can exhaust container memory. Fix: Enforce strict maxlen constraints on the deque during initialization. Implement circuit breakers that halt processing if ingestion rates exceed expected throughput, preventing cascading failures in downstream alerting systems.

Conclusion

Implementing sliding window algorithms for excursion detection transforms raw cold chain telemetry into a scientifically rigorous, compliance-ready decision engine. By replacing brittle point-in-time checks with rolling temporal evaluation, organizations achieve accurate excursion classification, reduced operational noise, and defensible audit trails. When integrated with validated state machines and calibrated grace periods, this architecture forms the computational backbone of modern pharmaceutical temperature monitoring systems.