Anomaly Detection

Anomaly detection identifies rare, unusual patterns that differ from expected behavior. Unlike classification, you often have few or no labeled anomalies - the system must learn "normal" and flag deviations. This powers fraud detection, infrastructure monitoring, quality control, and security systems.

Where It Is Used

Anomaly detection protects systems and identifies opportunities:

• Finance: Fraud detection, suspicious transactions
• Infrastructure: Server monitoring, network intrusion
• Manufacturing: Equipment failure prediction, quality control
• Healthcare: Patient monitoring, unusual vital signs
• Cybersecurity: Threat detection, behavioral anomalies
• IoT: Sensor malfunction, unusual patterns

Products: Datadog, Splunk, AWS GuardDuty, manufacturing quality systems.

Typical Data You Will See

Data types: - Time series (metrics, sensor readings) - Tabular (transactions, events) - Images (visual inspection) - Text (log files, messages) - Network data (traffic, connections)

What You Do in Practice

1. Define "anomaly" - What makes something abnormal?
2. Understand normal - What does typical behavior look like?
3. Choose approach - Supervised, semi-supervised, or unsupervised
4. Engineer features - Capture patterns that distinguish anomalies
5. Select algorithm - Statistical, tree-based, or neural
6. Set thresholds - Balance false positives and false negatives
7. Evaluate carefully - Often limited labeled data
8. Deploy with monitoring - Track alert rates and feedback
9. Handle feedback - Incorporate human labels
10. Iterate continuously - Normal behavior evolves

Baseline First

Start with simple statistical methods:

import numpy as np
from scipy import stats

# Z-score baseline
def zscore_anomaly(data, threshold=3):
    """Flag points more than threshold std from mean."""
    z_scores = np.abs(stats.zscore(data))
    return z_scores > threshold

# IQR method
def iqr_anomaly(data, multiplier=1.5):
    """Flag outliers using interquartile range."""
    q1 = np.percentile(data, 25)
    q3 = np.percentile(data, 75)
    iqr = q3 - q1
    lower = q1 - multiplier * iqr
    upper = q3 + multiplier * iqr
    return (data < lower) | (data > upper)

# Moving average for time series
def moving_avg_anomaly(data, window=24, threshold=3):
    """Flag deviations from moving average."""
    rolling_mean = data.rolling(window=window).mean()
    rolling_std = data.rolling(window=window).std()
    z_score = (data - rolling_mean) / rolling_std
    return np.abs(z_score) > threshold

Why statistical baseline: - Simple and interpretable - No training required - Works well for univariate data - Establishes anomaly rate expectations

Algorithm Approaches

Isolation Forest:

from sklearn.ensemble import IsolationForest

# Train on normal data (or all data if unsupervised)
model = IsolationForest(
    n_estimators=100,
    contamination=0.01,  # Expected anomaly rate
    random_state=42
)
model.fit(X_train)

# Predict: -1 for anomaly, 1 for normal
predictions = model.predict(X_test)
anomaly_scores = model.decision_function(X_test)  # Lower = more anomalous

Autoencoder for anomaly detection:

import torch
import torch.nn as nn

class Autoencoder(nn.Module):
    def __init__(self, input_dim, hidden_dim=32):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, hidden_dim)
        )
        self.decoder = nn.Sequential(
            nn.Linear(hidden_dim, 64),
            nn.ReLU(),
            nn.Linear(64, input_dim)
        )

    def forward(self, x):
        encoded = self.encoder(x)
        decoded = self.decoder(encoded)
        return decoded

# Train on normal data
model = Autoencoder(input_dim=features)
criterion = nn.MSELoss()

for epoch in range(epochs):
    for batch in normal_data_loader:
        reconstructed = model(batch)
        loss = criterion(reconstructed, batch)
        loss.backward()
        optimizer.step()

# Anomaly score = reconstruction error
def get_anomaly_score(model, x):
    with torch.no_grad():
        reconstructed = model(x)
        error = ((x - reconstructed) ** 2).mean(dim=1)
    return error

Time Series Anomaly Detection

Time series requires temporal context:

import pandas as pd
from prophet import Prophet

# Prophet for time series decomposition
def prophet_anomaly(df, interval_width=0.99):
    """Use Prophet to detect anomalies."""
    model = Prophet(interval_width=interval_width)
    model.fit(df[['ds', 'y']])

    forecast = model.predict(df[['ds']])
    df['yhat'] = forecast['yhat']
    df['yhat_lower'] = forecast['yhat_lower']
    df['yhat_upper'] = forecast['yhat_upper']

    df['anomaly'] = (df['y'] < df['yhat_lower']) | (df['y'] > df['yhat_upper'])
    return df

# Sliding window features
def create_window_features(series, window_sizes=[6, 12, 24]):
    """Create features from sliding windows."""
    features = pd.DataFrame()

    for w in window_sizes:
        features[f'mean_{w}'] = series.rolling(w).mean()
        features[f'std_{w}'] = series.rolling(w).std()
        features[f'min_{w}'] = series.rolling(w).min()
        features[f'max_{w}'] = series.rolling(w).max()
        features[f'diff_{w}'] = series - series.shift(w)

    return features.dropna()

Threshold Selection

Choosing the right threshold balances precision and recall:

def optimize_threshold(scores, labels, optimize_for='f1'):
    """Find threshold that optimizes given metric."""
    thresholds = np.percentile(scores, np.arange(90, 100, 0.5))
    best_score = 0
    best_threshold = thresholds[0]

    for thresh in thresholds:
        predictions = scores > thresh

        if optimize_for == 'f1':
            score = f1_score(labels, predictions)
        elif optimize_for == 'precision':
            score = precision_score(labels, predictions)
        elif optimize_for == 'recall':
            score = recall_score(labels, predictions)

        if score > best_score:
            best_score = score
            best_threshold = thresh

    return best_threshold, best_score

# Business-driven threshold
def cost_based_threshold(scores, labels, cost_fp=1, cost_fn=10):
    """Find threshold that minimizes business cost."""
    thresholds = np.percentile(scores, np.arange(80, 100, 0.5))
    costs = []

    for thresh in thresholds:
        predictions = scores > thresh
        fp = ((predictions == 1) & (labels == 0)).sum()
        fn = ((predictions == 0) & (labels == 1)).sum()
        total_cost = fp * cost_fp + fn * cost_fn
        costs.append((thresh, total_cost))

    return min(costs, key=lambda x: x[1])

Evaluation and Metrics

Evaluation without labels: - Domain expert review of flagged anomalies - Comparison with known incidents - A/B testing (did alerts correlate with issues?) - Manual labeling of sample

Failure Modes and Debugging Checklist

• [ ] Concept drift: Normal behavior changes over time
• [ ] Seasonal patterns: Regular variations flagged as anomalies
• [ ] Feature leakage: Anomaly-related features in training
• [ ] Contaminated training: Anomalies in "normal" training data
• [ ] Threshold sensitivity: Small changes cause large alert swings
• [ ] Alert fatigue: Too many false positives
• [ ] Missing context: Anomalies without business context

Production Checklist

Before deployment: - [ ] Baseline anomaly rate established - [ ] Threshold tuned on validation data with labels - [ ] Alert volume acceptable (not too many false positives) - [ ] Escalation path defined - [ ] Feedback mechanism for label collection

After deployment: - [ ] Alert rate monitored - [ ] Precision tracked via feedback - [ ] Model retrained on new labeled data - [ ] Seasonal patterns accounted for - [ ] Drift detection on input features

Operationalizing alerts: - [ ] Severity levels defined (low/medium/high) - [ ] Routing rules to appropriate teams - [ ] Playbooks for common anomaly types - [ ] Feedback loop to improve model

Handling Drift

Normal behavior changes - models must adapt:

def detect_drift(current_data, reference_data, threshold=0.05):
    """Detect distribution drift using KS test."""
    from scipy.stats import ks_2samp

    drift_detected = {}
    for column in current_data.columns:
        stat, p_value = ks_2samp(
            reference_data[column], 
            current_data[column]
        )
        drift_detected[column] = p_value < threshold

    return drift_detected

# Adaptive threshold
class AdaptiveThreshold:
    def __init__(self, window_size=1000, percentile=99):
        self.window_size = window_size
        self.percentile = percentile
        self.scores = []

    def update(self, score):
        self.scores.append(score)
        if len(self.scores) > self.window_size:
            self.scores.pop(0)

    def get_threshold(self):
        return np.percentile(self.scores, self.percentile)

Tools and Stack

Example Mini-Case

Scenario: E-commerce platform needs to detect fraudulent transactions in real-time.

Approach: 1. Features: Transaction amount, frequency, location, device, time patterns 2. Baseline: Z-score on amount by user - Catches 40% of fraud 3. Model: Isolation Forest trained on normal transactions 4. Threshold: Tuned for 1% false positive rate 5. Real-time: Score computed on transaction, alert if above threshold

Results: - Fraud detection rate: 78% - False positive rate: 0.8% - Alert volume: 500/day (manageable for review team) - Latency: 15ms per transaction

Key learnings: - User-level features critical (deviation from personal baseline) - Time-of-day patterns important for threshold adjustment - Feedback loop from fraud team improved precision 20%

Related Pages

• Index: ML Field Guide
• Fraud Detection
• Model Monitoring
• Time Series Forecasting

Further Reading

• PyOD: Python Outlier Detection
• Isolation Forest Paper
• Anomaly Detection Survey
• Time Series Anomaly Detection with MERLION

Try it live

Everything above runs in your browser — open AI Anomaly Detection Lab — A Manufacturing Sensor Case Study and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.