31. Rolling Error-Rate Alert Threshold

Problem

An on-call alerting job watches a stream of per-request outcomes (success or error) for a real-time voice pipeline, each with a timestamp. Rather than alerting on any single failure, it should only fire when the error rate within a trailing time window gets too high — a noisy but mostly healthy stream shouldn't page anyone.

For every event, decide whether the error rate within the trailing window_seconds (the event itself included) exceeds threshold.

Source: src/31_rolling_error_rate_alert.py

def rolling_error_alerts(events: list[tuple[float, bool]], window_seconds: float, threshold: float) -> list[bool]: ...

>>> rolling_error_alerts([(0, True), (1, True), (2, False), (3, False)], window_seconds=2, threshold=0.5)
[True, True, True, False]
>>> rolling_error_alerts([(0, False), (1, False), (2, False)], window_seconds=5, threshold=0.5)
[False, False, False]

Step-by-Step Approach

1. Maintain a deque of every event currently inside the trailing window, in arrival order, plus a running error_count of how many of those buffered events are errors.
2. For each incoming event, append it to the deque and, if it's an error, increment error_count — this is the window's right edge advancing.
3. Evict from the deque's left edge (oldest first) while the oldest event's timestamp is earlier than current_time - window_seconds, decrementing error_count whenever an evicted event was itself an error.
4. Because timestamps arrive in sorted order, the window's left edge only ever moves forward — each event is pushed once and popped at most once across the whole run, giving amortized O(1) work per event.
5. Once the window is current, compute error_count / len(window) and compare it against threshold; record whether it's strictly greater.
6. Repeat for every event in order; the recorded booleans, in order, are the answer.

The key insight is that the window is defined by elapsed time, not a fixed element count, so a two-pointer/deque sweep has to check timestamps rather than just capping the window at k elements — but the amortized cost is still O(1) per event because each event crosses the left edge exactly once, the same argument that makes the classic fixed-size sliding window O(n).

Reference solution

from collections import deque


def rolling_error_alerts(events: list[tuple[float, bool]], window_seconds: float, threshold: float) -> list[bool]:
    # deque holds (timestamp, is_error) for every event currently inside
    # the trailing window; error_count is a running total kept in sync as
    # events enter/leave, so no window is ever rescanned from scratch.
    window: deque[tuple[float, bool]] = deque()
    error_count = 0
    alerts: list[bool] = []

    for ts, is_error in events:
        window.append((ts, is_error))
        if is_error:
            error_count += 1
        # evict everything that fell out of [ts - window_seconds, ts]
        while window[0][0] < ts - window_seconds:
            old_ts, old_is_error = window.popleft()
            if old_is_error:
                error_count -= 1
        rate = error_count / len(window)
        alerts.append(rate > threshold)

    return alerts

Key Functions & Tricks

• collections.deque — O(1) append on the right and popleft on the left, the two operations this sweep needs.
• Running error_count — avoids recomputing the error total by scanning the window on every event.
• while window[0][0] < ts - window_seconds: window.popleft() — the time-based eviction condition, checked against a duration rather than an index offset.
• Amortized O(1) per event — each event enters the deque once and leaves at most once across the entire sweep.
• error_count / len(window) — the window's current size is always len(window), not a fixed k, since events can be sparse or dense in time.

How to Recognize This Pattern