Traffic Anomaly Detection: EWMA, Prometheus, Alerts

Imagine: on Thursday evening, /api/checkout suddenly gets a spike of 500 errors. Monitoring stays silent because the static threshold isn't breached — the peak is only 30% of the daily max (just 300 requests per second instead of 1000). Three hours later you notice the problem from customer complaints. Or another scenario: a scraper starts hammering the catalog, sending 10,000 requests per minute — a standard rate limit doesn't trigger because the traffic is distributed across IPs. We deploy a traffic anomaly detection system that spots such spikes within 5 seconds and sends an alert to Slack. No false positives on seasonal peaks. We have 10+ years of experience building monitoring systems and have implemented over 50 anomaly detection projects. For example, with a client that has 50,000 unique visitors per day, we reduced detection time from 3 hours to 5 seconds. Savings from implementation can run into tens of thousands of dollars per year by cutting downtime. Get an engineer's consultation to tune thresholds for your traffic.

How Traffic Anomaly Detection Works

Our detector combines two statistical methods: EWMA (Exponentially Weighted Moving Average) and z-score on a sliding window. The first adapts to trends, the second catches sharp outliers. According to EWMA, it's an exponential smoothing method. We wrap this into a service that collects metrics in real-time via Redis, analyzes every minute, and sends alerts with automatic mitigation. Detection time is reduced by 99% compared to manual log analysis.

Why Standard Monitoring Systems Fall Short

Typical Zabbix or Nagios use static thresholds. They give false positives on traffic peaks during sales hours and miss slow leaks. Our system is adaptive: it adjusts to seasonality and trends. We tune sensitivity individually for your traffic profile, using historical data from 2–4 weeks. For example, during Black Friday, we automatically raise thresholds to avoid false alerts. This allowed one client to reduce incident response time from 2 hours to 10 minutes.

How the Anomaly Detection System Is Built

We combine EWMA and z-score on a sliding window. EWMA responds faster to trends, z-score catches sharp outliers better. For long-term baselines we use Prometheus with quantile-based rules. The detector is written in Python but can be ported to Go for high-load systems (10,000+ RPS).

What Counts as an Anomaly?

We identify three types of anomalies:

• Volume: RPS, bandwidth, number of unique IPs rise sharply (e.g., from 100 to 5000 in a minute).
• Structural: ratio of HTTP methods changes (e.g., a sharp rise in GET when the norm is 70/30 GET/POST), increase in requests to specific endpoints.
• Quality: error rate 4xx/5xx rises, p99 latency breaks historical norm, increase in 404 ratio (scanning).

Method Comparison

Which Statistical Methods Are Used?

Statistical Detection Methods

import numpy as np from collections import deque import time class TrafficAnomalyDetector: def __init__(self, window_size=60, sensitivity=3.0): """ window_size: size of the sliding window in points (seconds/minutes) sensitivity: threshold in sigmas (z-score) """ self.window_size = window_size self.sensitivity = sensitivity self.metrics = {} # {metric_name: deque of values} def _get_window(self, metric: str) -> deque: if metric not in self.metrics: self.metrics[metric] = deque(maxlen=self.window_size) return self.metrics[metric] def add_point(self, metric: str, value: float): """Add a new data point""" self.metrics.setdefault(metric, deque(maxlen=self.window_size)).append(value) def check(self, metric: str, current_value: float) -> dict: """Check if the current value is an anomaly""" window = self._get_window(metric) if len(window) < 10: # Not enough data return {'anomaly': False, 'reason': 'insufficient_data'} values = list(window) mean = np.mean(values) std = np.std(values) if std == 0: z_score = 0 if current_value == mean else float('inf') else: z_score = abs(current_value - mean) / std is_anomaly = z_score > self.sensitivity direction = 'spike' if current_value > mean else 'drop' return { 'anomaly': is_anomaly, 'z_score': round(z_score, 2), 'direction': direction if is_anomaly else None, 'current': current_value, 'baseline_mean': round(mean, 2), 'baseline_std': round(std, 2), 'threshold': round(mean + self.sensitivity * std, 2) } 

Exponential Weighted Moving Average (EWMA)

Better at responding to trends, not sensitive to single outliers:

class EWMADetector: def __init__(self, alpha=0.1, k=3.0): """ alpha: smoothing factor (0.05–0.2) k: number of standard deviations for threshold """ self.alpha = alpha self.k = k self.ewma = {} # {metric: {'mean': float, 'variance': float}} def update_and_check(self, metric: str, value: float) -> dict: if metric not in self.ewma: self.ewma[metric] = {'mean': value, 'variance': 0} return {'anomaly': False} state = self.ewma[metric] mean = state['mean'] variance = state['variance'] # Update EWMA mean and variance new_mean = self.alpha * value + (1 - self.alpha) * mean new_variance = (1 - self.alpha) * (variance + self.alpha * (value - mean) ** 2) state['mean'] = new_mean state['variance'] = new_variance std = np.sqrt(new_variance) if new_variance > 0 else 0 threshold_high = new_mean + self.k * std threshold_low = max(0, new_mean - self.k * std) is_anomaly = value > threshold_high or value < threshold_low return { 'anomaly': is_anomaly, 'direction': 'spike' if value > threshold_high else 'drop', 'current': value, 'expected': round(new_mean, 2), 'threshold_high': round(threshold_high, 2), 'deviation_pct': round(abs(value - new_mean) / max(new_mean, 1) * 100, 1) } 

Real-Time Metrics Collection

import redis from datetime import datetime import threading class MetricsCollector: def __init__(self, redis_client): self.r = redis_client self.detector = EWMADetector(alpha=0.1, k=3.5) self.alert_cooldown = {} # prevent alert spam def record_request(self, status_code: int, path: str, latency_ms: float, method: str): """Called in middleware for each request""" now = int(time.time()) minute = now - (now % 60) pipe = self.r.pipeline() # RPS counter pipe.incr(f"metrics:rps:{now}") pipe.expire(f"metrics:rps:{now}", 300) # Errors if status_code >= 400: pipe.incr(f"metrics:errors:{now}") pipe.expire(f"metrics:errors:{now}", 300) # Latency (histogram in Redis) latency_bucket = int(latency_ms / 100) * 100 pipe.hincrby(f"metrics:latency:{minute}", str(latency_bucket), 1) pipe.expire(f"metrics:latency:{minute}", 3600) # Endpoint counter endpoint = f"{method}:{path.split('?')[0][:50]}" pipe.hincrby(f"metrics:endpoints:{minute}", endpoint, 1) pipe.expire(f"metrics:endpoints:{minute}", 3600) pipe.execute() def analyze_current_window(self): """Analyze the last 60 seconds and return anomalies""" now = int(time.time()) anomalies = [] # Collect RPS for the last 60 seconds rps_values = [] for i in range(60): t = now - i val = self.r.get(f"metrics:rps:{t}") rps_values.append(int(val or 0)) current_rps = rps_values[0] # Update detector with historical data for v in reversed(rps_values[1:]): self.detector.update_and_check('rps', v) result = self.detector.update_and_check('rps', current_rps) if result['anomaly']: anomalies.append({ 'metric': 'rps', 'severity': 'high' if result['deviation_pct'] > 200 else 'medium', **result }) # Error rate total = sum(rps_values[:60]) or 1 error_keys = [self.r.get(f"metrics:errors:{now-i}") for i in range(60)] total_errors = sum(int(v or 0) for v in error_keys) error_rate = total_errors / total err_result = self.detector.update_and_check('error_rate', error_rate) if err_result['anomaly'] and error_rate > 0.1: anomalies.append({ 'metric': 'error_rate', 'severity': 'critical' if error_rate > 0.3 else 'high', **err_result }) return anomalies 

Alerting and Automatic Mitigation

class AnomalyAlertManager: def __init__(self, slack_webhook, pagerduty_key): self.slack = slack_webhook self.pd = pagerduty_key self.active_incidents = {} def handle_anomalies(self, anomalies: list): for anomaly in anomalies: key = f"{anomaly['metric']}_{anomaly['direction']}" # Cooldown: don't spam the same alert if self.active_incidents.get(key, 0) > time.time() - 300: continue self.active_incidents[key] = time.time() if anomaly['severity'] == 'critical': self._page_oncall(anomaly) self._auto_mitigate(anomaly) elif anomaly['severity'] == 'high': self._notify_slack(anomaly) def _notify_slack(self, anomaly: dict): import requests icon = ':rotating_light:' if anomaly['direction'] == 'spike' else ':arrow_down:' requests.post(self.slack, json={ 'text': f"{icon} *Traffic anomaly detected*\n" f"Metric: `{anomaly['metric']}`\n" f"Current: `{anomaly['current']}` (expected: `{anomaly['expected']}`)\n" f"Deviation: `+{anomaly['deviation_pct']}%`\n" f"Z-score: `{anomaly.get('z_score', 'N/A')}`" }) def _auto_mitigate(self, anomaly: dict): """Automatic protective actions for critical anomalies""" if anomaly['metric'] == 'rps' and anomaly['direction'] == 'spike': # Enable emergency rate limit redis.setex('emergency_rate_limit', 300, '50') # 50 req/s globally # Notify Cloudflare to enable Under Attack Mode via API self._enable_cloudflare_attack_mode() def _enable_cloudflare_attack_mode(self): import requests requests.patch( f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/settings/security_level", headers={'Authorization': f'Bearer {CF_API_TOKEN}'}, json={'value': 'under_attack'} ) 

Prometheus + Grafana Alerting

# prometheus/alerts.yml groups: - name: traffic_anomalies rules: - alert: RequestRateSpike expr: | rate(http_requests_total[1m]) > (avg_over_time(rate(http_requests_total[1m])[1h:1m]) * 3) for: 2m labels: severity: critical annotations: summary: "Request rate spike: {{ $value }} req/s" - alert: ErrorRateCritical expr: | rate(http_requests_total{status=~"5.."}[5m]) / rate(http_requests_total[5m]) > 0.1 for: 1m labels: severity: critical annotations: summary: "Error rate {{ $value | humanizePercentage }}" - alert: LatencyP99Spike expr: | histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) > 2 for: 3m labels: severity: high annotations: summary: "P99 latency {{ $value }}s" 

How to Integrate the Detector into Your Existing Infrastructure

We connect to any metric sources: Prometheus, StatsD, Cloudflare Analytics, nginx logs. We support export to OpenMetrics and OpenTelemetry formats. For quick integration, we provide ready Docker Compose and Terraform configs. The entire process takes 1 to 3 days depending on stack complexity. Get an engineer's consultation to adapt to your stack.

How to Implement the Anomaly Detector: Step-by-Step Guide

1. Audit current infrastructure. We collect historical metric and log data for 2–4 weeks, analyze traffic profile.
2. Select algorithm and tune thresholds. We set EWMA parameters (alpha 0.05–0.2) and z-score (k 3–5) for your stack.
3. Integrate with metric sources. Deploy the collector on Redis/Prometheus, connect log aggregator.
4. Configure alerts and auto-mitigation. Set up channels (Slack, PagerDuty, Telegram) and automatic protection scripts (Cloudflare Under Attack Mode, rate limiting).
5. Test and launch. Prototype on 1 week of data, formal testing, and activation in production.

Common Implementation Mistakes

A frequent issue is incorrect window size. A too-small window causes many false positives, a too-large window misses fast anomalies. We select the window size individually, analyzing daily and weekly cycles. Another typical mistake is ignoring seasonality (e.g., Black Friday). Our detector uses historical data from similar periods.

What's Included in the Work

• Diagnostics: analysis of current metrics and logs, threshold selection.
• Implementation: detector code (Python, Go, or Lua), Prometheus rules configuration.
• Integration: Slack, PagerDuty, Telegram, Cloudflare API.
• Documentation: algorithm description, deployment instructions, playbook scenarios.
• Training: how to configure and retrain the detector for traffic changes.
• Support: 2 weeks post-production guarantee (SLA on response time).

Timeline and Cost

Turnkey implementation takes 3 to 5 business days depending on infrastructure complexity. Pricing is determined individually after an audit. Implementation savings can amount to tens of thousands of dollars per year by cutting detection time by 99% and reducing downtime by up to 80%. Contact us for a consultation and receive a free audit of your monitoring system.