Anomaly Detection for Application Monitoring
Anomaly detection for application metrics: statistical methods, z-scores, moving averages, seasonal decomposition and alerting without alert fatigue.
5 min read
Static thresholds are the default approach to monitoring alerts: "alert when error rate exceeds 5%" or "alert when p99 latency exceeds 500ms." They work until your traffic patterns change — holiday spikes, weekend dips, gradual growth. An error rate of 3% might be normal during a traffic surge but alarming at 3 AM when only internal services are hitting your API.
Anomaly detection replaces static thresholds with dynamic baselines. Instead of "is this number above X?", it asks "is this number unusual for this time and context?" This reduces false alerts during expected traffic changes while catching real incidents that static thresholds miss.
Statistical Baselines with Z-Scores
A z-score measures how many standard deviations a value is from the mean. A z-score above 3 means the value is more than 3 standard deviations from normal — roughly a 0.3% chance of occurring naturally. This is the simplest anomaly detector.
import numpy as np
from dataclasses import dataclass
@dataclass
class AnomalyResult:
value: float
mean: float
std: float
z_score: float
is_anomaly: bool
def detect_zscore_anomaly(
values: list[float],
current: float,
threshold: float = 3.0
) -> AnomalyResult:
"""Detect anomalies using z-score on recent history."""
arr = np.array(values)
mean = float(np.mean(arr))
std = float(np.std(arr))
# Avoid division by zero for constant metrics
if std == 0:
z_score = 0.0 if current == mean else float('inf')
else:
z_score = abs(current - mean) / std
return AnomalyResult(
value=current,
mean=mean,
std=std,
z_score=z_score,
is_anomaly=z_score > threshold
)
# Example: error rate over the last 24 hours (1-minute intervals)
error_rates = [0.02, 0.018, 0.022, 0.019, ...] # 1440 data points
current_rate = 0.085
result = detect_zscore_anomaly(error_rates, current_rate)
# z_score: 8.2, is_anomaly: True
# Mean 2%, current 8.5% → clearly anomalousExponential Moving Average (EMA)
Z-scores treat all historical data equally. EMA gives more weight to recent values, adapting faster to trends. The smoothing factor alpha controls how quickly the baseline adjusts — higher alpha means faster adaptation.
class EMADetector {
private ema: number | null = null;
private emaVariance: number = 0;
constructor(
private alpha: number = 0.1,
private threshold: number = 3.0
) {}
update(
value: number
): { isAnomaly: boolean; expected: number; deviation: number } {
if (this.ema === null) {
this.ema = value;
return { isAnomaly: false, expected: value, deviation: 0 };
}
// Update EMA (exponentially weighted moving average)
const prevEma = this.ema;
this.ema = this.alpha * value + (1 - this.alpha) * this.ema;
// Update variance estimate
const diff = value - prevEma;
this.emaVariance =
this.alpha * diff * diff +
(1 - this.alpha) * this.emaVariance;
const stdDev = Math.sqrt(this.emaVariance);
const deviation =
stdDev > 0 ? Math.abs(value - prevEma) / stdDev : 0;
return {
isAnomaly: deviation > this.threshold,
expected: prevEma,
deviation,
};
}
}
// Usage: detect latency anomalies
const latencyDetector = new EMADetector(0.05, 3.0);
// Feed in normal values to establish baseline
for (const latency of normalLatencies) {
latencyDetector.update(latency);
}
// Then check new values
const result = latencyDetector.update(850);
// { isAnomaly: true, expected: 120, deviation: 4.2 }Seasonal Decomposition
Many metrics have daily and weekly patterns. Request volume peaks at noon and drops at midnight. A 50% traffic increase at noon is normal; at 3 AM it is suspicious. Seasonal decomposition separates the trend, seasonal pattern, and residual components.
from dataclasses import dataclass
import numpy as np
@dataclass
class SeasonalBaseline:
hourly_means: np.ndarray # 24 values (one per hour)
hourly_stds: np.ndarray # 24 values
day_of_week_factors: np.ndarray # 7 values
def build_seasonal_baseline(
timestamps: list[float],
values: list[float],
min_weeks: int = 2
) -> SeasonalBaseline:
"""Build hourly baselines from historical data."""
from datetime import datetime
# Group values by hour-of-day
hourly_buckets: dict[int, list[float]] = {h: [] for h in range(24)}
dow_buckets: dict[int, list[float]] = {d: [] for d in range(7)}
for ts, val in zip(timestamps, values):
dt = datetime.fromtimestamp(ts)
hourly_buckets[dt.hour].append(val)
dow_buckets[dt.weekday()].append(val)
hourly_means = np.array([
np.mean(hourly_buckets[h]) if hourly_buckets[h] else 0
for h in range(24)
])
hourly_stds = np.array([
np.std(hourly_buckets[h]) if len(hourly_buckets[h]) > 1 else 1
for h in range(24)
])
global_mean = np.mean(values) if values else 1
day_factors = np.array([
np.mean(dow_buckets[d]) / global_mean if dow_buckets[d] else 1
for d in range(7)
])
return SeasonalBaseline(hourly_means, hourly_stds, day_factors)
def is_seasonal_anomaly(
baseline: SeasonalBaseline,
current_value: float,
hour: int,
day_of_week: int,
threshold: float = 3.0
) -> bool:
"""Check if value is anomalous for this time and day."""
expected = baseline.hourly_means[hour] * baseline.day_of_week_factors[day_of_week]
std = baseline.hourly_stds[hour]
if std == 0:
return current_value != expected
z = abs(current_value - expected) / std
return z > thresholdBuilding an Alerting Pipeline
Anomaly detection is only useful if it produces actionable alerts without drowning the on-call engineer in noise.
// ❌ Alerting on raw anomaly detection output
// Every z-score > 3 triggers a page
// Result: 15 alerts per day, most are benign blips
// On-call engineer ignores all alerts after 2 days
// ✅ Multi-stage alerting pipeline
interface AlertPipeline {
stage1_detection: {
description: "Raw anomaly detection on 1-minute data";
output: "Boolean: is this data point anomalous?";
};
stage2_confirmation: {
description: "Require N consecutive anomalous points";
config: { consecutiveRequired: 3; windowMinutes: 5 };
output: "Boolean: sustained anomaly confirmed?";
};
stage3_severity: {
description: "Classify severity based on deviation magnitude";
config: {
warning: "z-score 3-5 OR error rate increase 2-5x";
critical: "z-score > 5 OR error rate increase > 5x";
};
};
stage4_dedup: {
description: "Suppress duplicate alerts for same incident";
config: { silenceMinutes: 30; groupBy: "metric_name + service" };
};
}class AlertManager {
private activeAlerts = new Map<string, { since: Date; count: number }>();
shouldAlert(
metricKey: string,
isAnomaly: boolean,
severity: "warning" | "critical"
): { alert: boolean; action: "page" | "slack" | "none" } {
const existing = this.activeAlerts.get(metricKey);
if (!isAnomaly) {
// Anomaly resolved
if (existing) {
this.activeAlerts.delete(metricKey);
return { alert: true, action: "slack" }; // Resolution notification
}
return { alert: false, action: "none" };
}
if (existing) {
existing.count++;
// Already alerted, suppress until silence window expires
return { alert: false, action: "none" };
}
// New anomaly
this.activeAlerts.set(metricKey, { since: new Date(), count: 1 });
return {
alert: true,
action: severity === "critical" ? "page" : "slack",
};
}
}Combining Multiple Signals
Single-metric anomaly detection produces false positives. Combining correlated metrics dramatically improves accuracy.
interface CorrelatedCheck {
primary: { metric: string; isAnomaly: boolean };
supporting: Array<{ metric: string; isAnomaly: boolean }>;
confidence: number;
}
function correlatedAnomalyCheck(
checks: CorrelatedCheck
): { isReal: boolean; confidence: number } {
if (!checks.primary.isAnomaly) {
return { isReal: false, confidence: 0 };
}
const supportingAnomaly = checks.supporting.filter(
(s) => s.isAnomaly
).length;
const totalSupporting = checks.supporting.length;
// Higher confidence when multiple related metrics are anomalous
const correlationScore =
totalSupporting > 0 ? supportingAnomaly / totalSupporting : 0;
const confidence = 0.5 + 0.5 * correlationScore;
return {
isReal: confidence > 0.6,
confidence,
};
}
// Example: high error rate is more likely real if:
// - Latency is also elevated (correlated)
// - Traffic volume is normal (not a traffic spike causing errors)
// - CPU usage is elevated (resource pressure)
const check: CorrelatedCheck = {
primary: { metric: "error_rate", isAnomaly: true },
supporting: [
{ metric: "p99_latency", isAnomaly: true },
{ metric: "request_rate", isAnomaly: false },
{ metric: "cpu_usage", isAnomaly: true },
],
confidence: 0,
};
// 2/3 supporting anomalies → confidence: 0.83 → alertKey Takeaways
- Static thresholds break when traffic patterns change — replace "alert above X" with dynamic baselines that account for time-of-day and day-of-week patterns
- Z-scores are the simplest anomaly detector — if a value is more than 3 standard deviations from the mean, it is unusual; this catches 99.7% of genuine anomalies with minimal false positives
- Require sustained anomalies before alerting — a single anomalous data point might be noise; require 3+ consecutive anomalous points to confirm a real incident
- Combine correlated metrics to reduce false positives — high error rate plus high latency plus normal traffic is more likely a real incident than high error rate alone
- Seasonal baselines prevent time-of-day false alerts — build separate baselines for each hour and day-of-week to avoid alerting on normal daily traffic patterns
