How to Group Alerts by Time Window in Python

Group alert occurrences that fall within a sliding time window per alert key, reducing noise and summarizing bursts into single events.

Medium Python 3.9+ Aug 9, 2026 Observability & SRE 12 views 0 copies

alerts grouping monitoring time-window observability defaultdict

Python code

54 lines
Python 3.9+
Run it 
from collections import defaultdict
from datetime import datetime, timedelta

def group_alerts(alerts, window_minutes=10):
    """Group alerts that occur within the same time window."""
    alerts_by_key = defaultdict(list)
    
    for alert in alerts:
        key = alert["key"]
        timestamp = alert["timestamp"]
        
        # Check if this alert fits in an existing group
        placed = False
        for group in alerts_by_key[key]:
            if abs(group["start"] - timestamp) <= timedelta(minutes=window_minutes):
                group["alerts"].append(alert)
                group["latest"] = max(group["latest"], timestamp)
                placed = True
                break
        
        if not placed:
            alerts_by_key[key].append({
                "start": timestamp,
                "latest": timestamp,
                "alerts": [alert]
            })
    
    # Flatten grouped alerts
    result = []
    for key, groups in alerts_by_key.items():
        for group in groups:
            result.append({
                "key": key,
                "first_occurrence": group["start"].isoformat(),
                "last_occurrence": group["latest"].isoformat(),
                "count": len(group["alerts"]),
                "alerts": group["alerts"]
            })
    return result

if __name__ == "__main__":
    base_time = datetime(2024, 1, 1, 10, 0, 0)
    alerts = [
        {"key": "CPU_HIGH", "timestamp": base_time},
        {"key": "CPU_HIGH", "timestamp": base_time + timedelta(minutes=5)},
        {"key": "CPU_HIGH", "timestamp": base_time + timedelta(minutes=15)},
        {"key": "MEM_LOW", "timestamp": base_time + timedelta(minutes=2)},
        {"key": "CPU_HIGH", "timestamp": base_time + timedelta(minutes=20)},
    ]
    
    grouped = group_alerts(alerts, window_minutes=10)
    for group in grouped:
        print(f"{group['key']}: {group['count']} alerts "
              f"({group['first_occurrence']} to {group['last_occurrence']})")

Output

stdout 
CPU_HIGH: 2 alerts (2024-01-01T10:00:00 to 2024-01-01T10:05:00)
CPU_HIGH: 2 alerts (2024-01-01T10:15:00 to 2024-01-01T10:20:00)
MEM_LOW: 1 alerts (2024-01-01T10:02:00 to 2024-01-01T10:02:00)

How it works

The function uses defaultdict(list) to maintain a list of groups per alert key. For each incoming alert, it scans existing groups for that key and checks if the timestamp falls within the configured window of the group's start time using abs() and timedelta. If matched, the alert is appended and the latest timestamp is updated with max(). Otherwise a new group is started. Finally, it flattens all groups into a summary list with first/last occurrence and count. This greedy approach is simple and works well for streaming or batch alert feeds, though it only compares against group start—not the previous alert—so large bursts can still be missed.

Common mistakes

  • Comparing timestamps as strings instead of datetime objects, causing TypeError or wrong comparisons
  • Forgetting to convert alert timestamps to datetime if they arrive as ISO strings or Unix epochs
  • Using `=` instead of `<=` in the window check, skipping alerts exactly at the window boundary
  • Not sorting alerts by timestamp before grouping, which can produce overlapping groups

Variations

  1. Sort `alerts` by `timestamp` before processing for deterministic sequential grouping
  2. Use a priority queue to process alerts and close groups as soon as the window expires

Real-world use cases

  • Aggregating repeated CPU/memory alerts from monitoring agents to reduce on-call pager noise.
  • Bundling consecutive failed login attempts within a few minutes into one security incident.
  • Collapsing repeated API 500 error alerts during a deployment into a single incident ticket.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Observability & SRE

Keep learning

Related tutorials and quizzes for this topic.

Tutorials

Quizzes