Table of Contents

Introduction

In intensive care units (ICUs), every heartbeat matters. Patient monitors stream vital signs—heart rate, blood pressure, oxygen saturation—24/7. But sensors glitch, leads disconnect, and artifacts appear. A single false alarm can trigger unnecessary interventions; a missed anomaly can be fatal.

Traditional outlier detection using the Interquartile Range (IQR) requires a full dataset. But in live healthcare telemetry, you can’t wait for hours of data. You need real-time IQR-based outlier detection that works on the fly.

This article reveals how to adapt IQR for streaming data using reservoir sampling, with a life-critical ICU scenario and production-ready Python code.

What Is the Interquartile Range (IQR)?

The IQR measures statistical dispersion as the difference between the 75th percentile (Q3) and 25th percentile (Q1):

IQR=Q3−Q1

Outliers are typically defined as values outside the range:

[Q1−1.5×IQR, Q3+1.5×IQR]

This non-parametric method is robust to noise and doesn’t assume normal distribution—ideal for physiological signals.

Why Outlier Detection Saves Lives in Critical Care

Imagine an ICU patient with sepsis. Their heart rate normally hovers around 90 bpm. Suddenly, a loose ECG lead causes a 5-second spike to 220 bpm.

Without filtering:

With real-time IQR outlier detection:

In high-stakes environments, intelligent filtering isn’t convenience—it’s patient safety.

The Challenge of Streaming IQR

Classic IQR needs all data to compute percentiles. But streaming systems:

Solution: Maintain a representative window of recent data using reservoir sampling or a fixed-size buffer, then compute IQR on that window in real time.

Efficient Reservoir Sampling for Live Data

We use a fixed-size rolling window (e.g., last 100 readings) to approximate the distribution. When the window is full, new values replace the oldest. This balances:

For small windows (<1000 points), sorting is fast enough. For larger systems, consider t-digest or streaming quantile algorithms—but for most real-world cases, a simple buffer suffices.

Complete Implementation with ICU Simulation

PlantUML Diagram
import bisect
import random
import time
from collections import deque
from typing import Tuple, Optional
import math

# Define the new return type for clarity: 
# (is_outlier, lower_bound, upper_bound, q1_idx, q3_idx)
AddReturnType = Tuple[bool, float, float, int, int]

class StreamingIQRDetector:
    """
    Real-time outlier detection using Interquartile Range (IQR) on a 
    TRUE sliding window. Maintains a fixed-size window of recent values.
    """
    def __init__(self, window_size: int = 100, multiplier: float = 1.5):
        if window_size <= 0:
            raise ValueError("Window size must be positive")
        self.window_size = window_size
        self.multiplier = multiplier
        
        # Sorted list for efficient percentile calculation (O(log N) insertion)
        self._sorted_window: list[float] = [] 
        
        # Deque to track insertion order for true sliding window behavior
        self._history = deque(maxlen=window_size)
        
        # Store last calculated quartile values/indices for external access (for display)
        self.q1_val: Optional[float] = None
        self.q3_val: Optional[float] = None
        self.q1_idx: int = -1
        self.q3_idx: int = -1

    def add(self, value: float) -> AddReturnType:
        """
        Add a new value and return (is_outlier, lower_bound, upper_bound, q1_idx, q3_idx).
        """
        # --- 1. Window Maintenance (O(N) removal complexity) ---
        
        # If the window is full, remove the OLDEST element first.
        if len(self._history) == self.window_size:
            oldest_value = self._history.popleft()
            
            # Find and remove the oldest value from the sorted list.
            removal_idx = bisect.bisect_left(self._sorted_window, oldest_value)
            
            # Check for value existence before popping (essential if duplicates are possible)
            if removal_idx < len(self._sorted_window) and self._sorted_window[removal_idx] == oldest_value:
                self._sorted_window.pop(removal_idx)

        # Add the new value to both structures
        self._history.append(value)
        bisect.insort(self._sorted_window, value)
        
        # --- 2. Outlier Calculation ---
        
        n = len(self._sorted_window)
        # Use placeholders if not enough data
        lower_bound = 0.0
        upper_bound = 0.0
        is_outlier = False
        self.q1_idx, self.q3_idx = -1, -1 # Reset indices
        self.q1_val, self.q3_val = 0.0, 0.0
        
        if n >= 20: # Minimum size for better stability
            # Compute Q1 and Q3 indices (0-based)
            q1_idx = math.floor(n * 0.25)
            q3_idx = math.ceil(n * 0.75) - 1 
            
            # Ensure indices are within bounds
            q1_idx = max(0, min(q1_idx, n - 1))
            q3_idx = max(0, min(q3_idx, n - 1))

            q1 = self._sorted_window[q1_idx]
            q3 = self._sorted_window[q3_idx]
            
            iqr = q3 - q1
            
            # The IQR outlier limits
            lower_bound = q1 - self.multiplier * iqr
            upper_bound = q3 + self.multiplier * iqr
            
            is_outlier = value < lower_bound or value > upper_bound
            
            # Update instance attributes for last calculated quartiles (optional, but useful)
            self.q1_idx = q1_idx
            self.q3_idx = q3_idx
            self.q1_val = q1
            self.q3_val = q3
        
        return is_outlier, lower_bound, upper_bound, self.q1_idx, self.q3_idx

    def reset(self) -> None:
        """Clear the window (e.g., on patient transfer)."""
        self._sorted_window.clear()
        self._history.clear()
        self.__init__(self.window_size, self.multiplier) # Re-initialize state


# --- Live Simulation: ICU Heart Rate Monitoring (Fixed) ---

def simulate_icu_monitoring():
    """Simulate real-time heart rate with artifacts and IQR-based filtering."""
    detector = StreamingIQRDetector(window_size=40, multiplier=1.5)
    
    print(" ICU Heart Rate Monitor (Normal: 60–100 bpm) - TRUE Sliding Window")
    print("Window Size: 40 | Multiplier: 1.5 (Standard Outlier Rule)")
    print("-" * 110)
    
    base_hr = 85.0
    
    print(f"{'Min':>3} | {'HR':>5} | {'Status':>10} | {'Window N':>8} | {'Q1':>6} | {'Q3':>6} | {'Lower Bound':>13} | {'Upper Bound':>13} | {'Event':<15}")
    print("-" * 110)
    
    for minute in range(1, 41):
        # Simulate normal variation
        base_hr += random.uniform(-0.1, 0.1) # Simulate minor drift
        hr = base_hr + random.gauss(0, 3.0)
        
        event = ""
        
        # Inject real event: fever increases HR at minute 15
        if minute == 15:
            base_hr = 110.0
            hr = 112.0
            event = " FEVER START"
            
        # Inject artifact: lead disconnect at minute 30
        if minute == 30:
            hr = 210.0 # False high reading
            event = " ARTIFACT"

        # Inject real event: HR stabilizes at new baseline
        if minute == 35:
            base_hr = 105.0
            event = " HR STABILIZED"

        # FIX: Unpack all return values from the add method
        is_outlier, lower_b, upper_b, q1_idx, q3_idx = detector.add(hr)
        
        status = " OUTLIER" if is_outlier else " NORMAL"
        
        # Determine Q1 and Q3 values for printing, handling the initial state
        q1_val_str = f"{detector._sorted_window[detector.q1_idx]:6.1f}" if len(detector._sorted_window) >= 20 else "   N/A"
        q3_val_str = f"{detector._sorted_window[detector.q3_idx]:6.1f}" if len(detector._sorted_window) >= 20 else "   N/A"
        
        print(f"{minute:3d} | {hr:5.1f} | {status:>10} | {len(detector._history):8d} | {q1_val_str} | {q3_val_str} | {lower_b:13.1f} | {upper_b:13.1f} | {event}")
        time.sleep(0.1)
    
    print("-" * 110)
    print("\n  Results Analysis:")
    print("The TRUE sliding window successfully adapts to the new baseline HR (Fever) after a short delay but flags the non-physiological reading (Artifact) as an OUTLIER based on the learned distribution (Q1/Q3).")


if __name__ == "__main__":
    simulate_icu_monitoring()
s

Best Practices for Clinical and IoT Systems

Conclusion

In live monitoring systems—from ICUs to industrial plants—outlier detection must be both statistically sound and temporally aware. By combining IQR with a streaming window, you get a simple, interpretable, and effective filter that runs in real time with minimal resources. This approach turns raw sensor streams into trustworthy signals, reducing alarm fatigue and highlighting true anomalies. Whether you’re saving lives or safeguarding machinery, streaming IQR gives you the clarity to act—only when it matters.