Deduplicating and Throttling SPC Alert Storms

A control chart stuck in an out-of-control state re-fires on every incoming point, and without a gate a single incident becomes hundreds of identical notifications until the channel is muted. This guide is part of real-time alert notification channels: it shows how to collapse a persistent run into one alert with a dedup key, a cooldown, and a token bucket, while guaranteeing a genuinely new condition still breaks through.

Prerequisites

  • Python 3.10+ (standard library only — the gate needs no third-party packages)
  • A stream of alert records, each carrying process id, characteristic, rule, and severity, from the out-of-control rule detection (Nelson / Western Electric) layer
  • Detection that emits at most one signal per point per rule, so the gate suppresses a real storm rather than masking a double-fire bug
  • A downstream transport (Slack, Teams, email, or SMS) via sending SPC alerts to Slack and Microsoft Teams
  • Agreement on the cooldown window and burst size with the shift owners — throttling is a policy decision, not just a parameter

Step 1 — Choose a dedup key that is neither too tight nor too loose

The routing key is the whole game. Include the fields that define one incident and exclude anything that changes point to point. Keying on process, characteristic, rule, and severity means every point of the same shifted run shares a key (so it is suppressed), while a different rule, a different characteristic, or an escalation to a higher severity yields a new key (so it breaks through). Never fold the measured value, timestamp, or point index into the key — that makes every point look new and defeats deduplication.

from __future__ import annotations

from dataclasses import dataclass
from enum import IntEnum


class Severity(IntEnum):
    INFO = 10
    WARNING = 20
    CRITICAL = 30


@dataclass(frozen=True)
class Alert:
    process_id: str
    characteristic: str
    rule: str
    severity: Severity

    def dedup_key(self) -> str:
        """One open incident per (process, characteristic, rule, severity)."""
        return f"{self.process_id}|{self.characteristic}|{self.rule}|{int(self.severity)}"

Step 2 — Suppress duplicates within a cooldown window

The first gate is a cooldown: once a key has been delivered, drop repeats of that key for cooldown_s seconds. This alone collapses a sustained run, because every point after the first shares the key and lands inside the window.

import time


class CooldownGate:
    """Drops repeats of a key seen within cooldown_s seconds."""

    def __init__(self, cooldown_s: float = 300.0) -> None:
        self.cooldown_s = cooldown_s
        self._last_seen: dict[str, float] = {}

    def allow(self, key: str, now: float | None = None) -> bool:
        now = time.time() if now is None else now
        last = self._last_seen.get(key)
        if last is not None and (now - last) < self.cooldown_s:
            return False  # duplicate inside the cooldown
        self._last_seen[key] = now
        return True

Step 3 — Rate-limit each key with a token bucket

A cooldown blocks exact repeats but a legitimately escalating condition can still generate distinct keys quickly. The second gate is a per-key token bucket: capacity b sets the allowed burst, refill rate r sets the sustained ceiling. Each delivery costs one token; when the bucket is empty, suppress.

from dataclasses import field


@dataclass
class TokenBucket:
    capacity: float
    refill_per_sec: float
    tokens: float = field(init=False)
    updated: float = field(default_factory=time.time)

    def __post_init__(self) -> None:
        self.tokens = self.capacity

    def allow(self, now: float | None = None) -> bool:
        now = time.time() if now is None else now
        elapsed = max(0.0, now - self.updated)
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_sec)
        self.updated = now
        if self.tokens >= 1.0:
            self.tokens -= 1.0
            return True
        return False

Step 4 — Combine the gates and coalesce the suppressed count

Wire the cooldown and token bucket into one throttle that returns a decision and, when it suppresses, coalesces the count so a later summary can report "40 signals collapsed to 1." A new key — new rule or escalated severity — bypasses the open incident's cooldown and is allowed immediately.

import logging

logger = logging.getLogger(__name__)


class AlertThrottle:
    """Cooldown + token-bucket gate with coalesced suppression counts."""

    def __init__(
        self,
        cooldown_s: float = 300.0,
        bucket_capacity: float = 3.0,
        refill_per_sec: float = 1.0 / 60.0,
    ) -> None:
        self._cooldown = CooldownGate(cooldown_s)
        self._bucket_capacity = bucket_capacity
        self._refill = refill_per_sec
        self._buckets: dict[str, TokenBucket] = {}
        self.suppressed: dict[str, int] = {}

    def admit(self, alert: Alert, now: float | None = None) -> bool:
        """Return True if the alert should be delivered, False if suppressed.

        A suppressed alert increments a per-key counter so the incident's
        total volume is recoverable for the audit log and any digest.
        """
        now = time.time() if now is None else now
        key = alert.dedup_key()

        if not self._cooldown.allow(key, now):
            self.suppressed[key] = self.suppressed.get(key, 0) + 1
            return False

        bucket = self._buckets.setdefault(
            key, TokenBucket(self._bucket_capacity, self._refill))
        if not bucket.allow(now):
            self.suppressed[key] = self.suppressed.get(key, 0) + 1
            logger.info("rate-limited key=%s", key)
            return False

        return True

Verification

Prove the two properties that matter: a storm collapses to one delivery, and a new condition still breaks through. Time is injected so the test is deterministic and never sleeps.

throttle = AlertThrottle(cooldown_s=300.0)

run = Alert("LINE3-STN2", "bore_dia_mm", "Nelson 1", Severity.WARNING)

# 40 points of the SAME out-of-control run, one every 5 seconds.
admitted = [throttle.admit(run, now=1000.0 + 5 * i) for i in range(40)]
assert sum(admitted) == 1                      # exactly one notification
assert admitted[0] is True                     # the first point delivers
assert throttle.suppressed[run.dedup_key()] == 39

# A genuinely NEW condition mid-run: a different rule escalates to CRITICAL.
escalation = Alert("LINE3-STN2", "bore_dia_mm", "Nelson 5", Severity.CRITICAL)
assert throttle.admit(escalation, now=1050.0) is True   # breaks through immediately

print("40 raw signals collapsed to 1; escalation broke through")

Expected output: 40 raw signals collapsed to 1; escalation broke through. The two load-bearing assertions are sum(admitted) == 1 (the storm collapses) and the escalation returning True (a real new condition is never swallowed). A throttle that passes the first but fails the second is worse than no throttle, because it hides the signal SPC exists to surface.

Root-Cause Table

Symptom Cause Fix
Every point still alerts despite the gate Dedup key includes the value, timestamp, or index, so each point is a new key Key only on process, characteristic, rule, and severity
A new excursion is never delivered Key too coarse (process only) so a new rule collapses into the open incident Add rule and severity to the key so escalations break through
Storm resumes the instant the cooldown ends Cooldown shorter than the incident duration and no open-incident tracking Lengthen the cooldown or hold suppression until the run clears and the chart returns in control
Burst of distinct keys still floods No rate limit behind the cooldown Add the token bucket; size capacity to the acceptable burst per key
Suppressed volume is invisible in the audit Dropped alerts not counted Coalesce a per-key suppressed counter and log it with the incident
Throttle test is flaky Test relies on wall-clock and sleep Inject now into admit, allow, and the bucket; advance it explicitly

Up one level: Real-time alert notification channels.