Time Series Anomaly Detection

Time series anomaly detection identifies unusual patterns in sequential data - spikes, drops, level shifts, and trend changes. Unlike static anomaly detection, time series methods must account for temporal patterns: seasonality, trends, and autocorrelation. Production systems require both accuracy and low latency for real-time alerting.

Where It Is Used

Time series anomaly detection powers monitoring across industries:

• Infrastructure: Server metrics, network traffic, application performance
• Finance: Trading anomalies, transaction monitoring, market surveillance
• IoT: Sensor readings, device health, industrial equipment
• E-commerce: Traffic spikes, conversion drops, revenue anomalies
• Healthcare: Patient vitals, medical device monitoring
• Energy: Grid monitoring, consumption patterns, outage detection

Products: DataDog, New Relic, Splunk, AWS CloudWatch Anomaly Detection.

Types of Anomalies

Typical Data You Will See

Data characteristics: - Strong temporal autocorrelation - Multiple seasonalities (daily + weekly) - Trend components - Noise and measurement error - Concept drift over time

What You Do in Practice

1. Understand the metric - What does normal look like?
2. Explore patterns - Seasonality, trends, noise level
3. Choose detection approach - Statistical, ML, or deep learning
4. Handle seasonality - Decompose or model explicitly
5. Set thresholds - Balance sensitivity vs false alarms
6. Implement alerting - Real-time detection pipeline
7. Tune for domain - Different metrics need different sensitivity
8. Handle missing data - Interpolation or robust methods
9. Evaluate carefully - Labeled anomalies rare, use proxies
10. Monitor the monitor - Track alert quality over time

Baseline First

Start with statistical methods before ML:

import numpy as np
from scipy import stats

class ZScoreDetector:
    def __init__(self, window=100, threshold=3.0):
        self.window = window
        self.threshold = threshold

    def detect(self, series):
        anomalies = []
        for i in range(self.window, len(series)):
            window_data = series[i-self.window:i]
            mean = np.mean(window_data)
            std = np.std(window_data)

            if std > 0:
                z_score = (series[i] - mean) / std
                if abs(z_score) > self.threshold:
                    anomalies.append(i)

        return anomalies

# Seasonal decomposition + residual detection
from statsmodels.tsa.seasonal import seasonal_decompose

def seasonal_anomaly_detection(series, period=24, threshold=3.0):
    # Decompose into trend, seasonal, residual
    decomposition = seasonal_decompose(series, period=period)
    residuals = decomposition.resid

    # Detect anomalies in residuals
    mean = np.nanmean(residuals)
    std = np.nanstd(residuals)
    z_scores = (residuals - mean) / std

    return np.where(np.abs(z_scores) > threshold)[0]

Why statistical baseline: - Interpretable, explainable - Fast, no training needed - Works well for simple patterns - Good benchmark for ML methods

Detection Methods

Prophet for business metrics:

from prophet import Prophet

# Fit model on historical data
model = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=True,
    daily_seasonality=True
)
model.fit(df[['ds', 'y']])  # ds=timestamp, y=value

# Predict with uncertainty intervals
forecast = model.predict(df[['ds']])

# Anomaly = outside prediction interval
df['anomaly'] = (
    (df['y'] < forecast['yhat_lower']) |
    (df['y'] > forecast['yhat_upper'])
)

LSTM Autoencoder:

import torch
import torch.nn as nn

class LSTMAutoencoder(nn.Module):
    def __init__(self, input_dim, hidden_dim, seq_len):
        super().__init__()
        self.encoder = nn.LSTM(input_dim, hidden_dim, batch_first=True)
        self.decoder = nn.LSTM(hidden_dim, input_dim, batch_first=True)
        self.seq_len = seq_len

    def forward(self, x):
        # Encode
        _, (h, c) = self.encoder(x)
        # Decode
        decoder_input = h.repeat(self.seq_len, 1, 1).permute(1, 0, 2)
        output, _ = self.decoder(decoder_input)
        return output

# Anomaly score = reconstruction error
def anomaly_score(model, x):
    reconstruction = model(x)
    error = torch.mean((x - reconstruction) ** 2, dim=[1, 2])
    return error

Evaluation and Metrics

Evaluation is challenging - anomalies are rare and often unlabeled.

Practical evaluation:

# Inject synthetic anomalies
def inject_anomalies(series, n_anomalies=10):
    indices = np.random.choice(len(series), n_anomalies, replace=False)
    anomaly_series = series.copy()

    for idx in indices:
        # Random anomaly type
        anomaly_type = np.random.choice(['spike', 'drop', 'shift'])
        if anomaly_type == 'spike':
            anomaly_series[idx] *= 3
        elif anomaly_type == 'drop':
            anomaly_series[idx] *= 0.2
        else:
            anomaly_series[idx:idx+10] += series.std() * 2

    return anomaly_series, indices

# Evaluate detection
injected, true_anomalies = inject_anomalies(series)
detected = detector.detect(injected)
recall = len(set(detected) & set(true_anomalies)) / len(true_anomalies)

Failure Modes and Debugging Checklist

• [ ] False alarms: Threshold too sensitive
• [ ] Missed anomalies: Threshold too lenient
• [ ] Seasonality ignored: Daily patterns flagged as anomalies
• [ ] Concept drift: Normal baseline shifted
• [ ] Missing data: Gaps cause false detections
• [ ] Multivariate blindness: Anomaly only visible in combination
• [ ] Detection delay: Too much smoothing
• [ ] Holiday effects: Special days not modeled

Production Checklist

Before deployment: - [ ] Historical patterns understood - [ ] Seasonality modeled correctly - [ ] Threshold tuned for acceptable false alarm rate - [ ] Missing data handling defined - [ ] Alert routing configured - [ ] Runbook for common anomalies

After deployment: - [ ] Alert volume monitored - [ ] False alarm rate tracked - [ ] Detection latency measured - [ ] Model drift monitored - [ ] Feedback loop from investigations - [ ] Periodic threshold review

Operations: - [ ] On-call rotation defined - [ ] Escalation paths clear - [ ] Root cause templates ready - [ ] Auto-remediation where possible

Tools and Stack

Example Mini-Case

Scenario: E-commerce platform needs anomaly detection on 500+ metrics (latency, errors, revenue, traffic).

Approach: 1. Categorize metrics by pattern type (stationary, seasonal, trending) 2. Statistical baseline: STL decomposition + z-score on residuals 3. ML enhancement: Prophet for business metrics with holidays 4. Multivariate: Isolation Forest for correlated metrics 5. Threshold tuning: 1-2 alerts per metric per week acceptable

Architecture:

Metrics → Kafka → Feature computation → Detection models →
Alert aggregation → PagerDuty → On-call engineer

Results: - False alarm rate: < 5% - Detection latency: < 2 minutes - MTTR reduction: 40% - On-call burden: 3 actionable alerts/day

Key learnings: - Different metrics need different methods - Seasonality modeling critical for business metrics - Alert aggregation prevents alert storms - Feedback loop essential for threshold tuning

Related Pages

• Index: ML Field Guide
• Demand Forecasting
• Anomaly Detection
• Model Monitoring

Further Reading

• Prophet Documentation
• PyOD Library
• Merlion (Salesforce)
• Anomaly Detection Survey

Try it live

Everything above runs in your browser — open Time Series Anomaly Detection Lab — Catch the Spike That Matters and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.