LLM-powered anomaly detection in time-series data

Traditional anomaly detection in time-series data relies on statistical methods: z-scores, rolling averages, IQR bounds. These methods work well for simple deviations but fail in two common situations: seasonality that shifts the baseline (a traffic spike on Black Friday is not anomalous) and multi-signal correlations (CPU at 90% is fine when you just deployed; it is alarming at 3am with no deployments).

LLMs do not replace statistical detection — they are slower and more expensive per data point than a threshold check. But they excel at the interpretation layer: given that a statistical method flagged something, is it actually worth alerting on, and why?

The two-layer architecture

The most practical architecture combines fast statistical detection with LLM reasoning:

Time-series data → Statistical detector → Candidate anomalies →
LLM reasoning layer → Verified alerts with explanation

The statistical layer runs continuously and cheaply. The LLM layer sees only the candidates the statistical layer surfaces, which is typically a small fraction of total data points. This keeps costs manageable and latency acceptable.

Statistical pre-filtering

A Z-score filter is the simplest effective pre-filter:

import numpy as np
from dataclasses import dataclass
from datetime import datetime

@dataclass
class TimeSeriesPoint:
    timestamp: datetime
    metric_name: str
    value: float
    labels: dict[str, str]  # e.g. {"host": "web-01", "env": "prod"}

@dataclass
class AnomalyCandidate:
    point: TimeSeriesPoint
    z_score: float
    baseline_mean: float
    baseline_std: float
    direction: str  # "high" or "low"
    context_window: list[TimeSeriesPoint]  # surrounding data points

def detect_candidates(
    series: list[TimeSeriesPoint],
    window_size: int = 60,     # data points for baseline
    z_threshold: float = 3.0,
) -> list[AnomalyCandidate]:
    """
    Z-score based anomaly candidate detection.
    Returns points that deviate significantly from their local baseline.
    """
    candidates = []
    values = [p.value for p in series]

    for i in range(window_size, len(series)):
        baseline = values[i - window_size:i]
        mean = np.mean(baseline)
        std = np.std(baseline)

        if std == 0:
            continue  # Constant series, nothing to detect

        z = (values[i] - mean) / std
        if abs(z) >= z_threshold:
            candidates.append(AnomalyCandidate(
                point=series[i],
                z_score=z,
                baseline_mean=mean,
                baseline_std=std,
                direction="high" if z > 0 else "low",
                context_window=series[max(0, i-10):i+5],
            ))

    return candidates

LLM reasoning over candidates

Once you have candidates, pass them to an LLM with relevant context to determine severity and generate human-readable explanations:

import json
from anthropic import Anthropic
from enum import Enum

client = Anthropic()

class AlertSeverity(str, Enum):
    IGNORE = "ignore"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class AnomalyAssessment:
    severity: AlertSeverity
    is_anomalous: bool
    explanation: str
    probable_cause: str
    recommended_action: str
    confidence: float

def assess_anomaly(
    candidate: AnomalyCandidate,
    system_context: dict,  # Additional context: recent deployments, incidents, etc.
) -> AnomalyAssessment:
    """
    Use an LLM to assess whether a statistical anomaly candidate is
    actually worth alerting on and why.
    """
    # Format the context window for readability
    context_values = [
        f"{p.timestamp.strftime('%H:%M:%S')}: {p.value:.2f}"
        for p in candidate.context_window
    ]

    prompt = f"""You are an SRE analysing a metric anomaly. Assess whether this
is worth alerting on.

METRIC: {candidate.point.metric_name}
LABELS: {json.dumps(candidate.point.labels)}
ANOMALOUS VALUE: {candidate.point.value:.2f} at {candidate.point.timestamp}
DIRECTION: {candidate.direction} (z-score: {candidate.z_score:.2f})
BASELINE: mean={candidate.baseline_mean:.2f}, std={candidate.baseline_std:.2f}

RECENT VALUES (last 15 data points):
{chr(10).join(context_values)}

SYSTEM CONTEXT:
{json.dumps(system_context, indent=2, default=str)}

Assess this anomaly and respond with JSON matching this schema:
{{
  "is_anomalous": boolean,
  "severity": "ignore" | "low" | "medium" | "high" | "critical",
  "explanation": "1-2 sentence explanation of what happened",
  "probable_cause": "most likely reason for this anomaly",
  "recommended_action": "what the on-call engineer should do first",
  "confidence": 0.0 to 1.0
}}"""

    response = client.messages.create(
        model="claude-haiku-4-5",  # Haiku for cost-effective high-volume use
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )

    # Parse the JSON response
    text = response.content[0].text
    # Extract JSON from potential markdown code blocks
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    data = json.loads(text)
    return AnomalyAssessment(**data)

Adding system context

The LLM's assessment quality improves dramatically with context. What recent deployments happened? Are there ongoing incidents? What time of day is it? Is this a weekend?

from datetime import datetime, timezone
import pytz

def build_system_context(
    metric_labels: dict[str, str],
    recent_deployments: list[dict],
    active_incidents: list[dict],
) -> dict:
    """
    Build a context dictionary that helps the LLM reason about
    whether an anomaly is expected given recent system changes.
    """
    now = datetime.now(timezone.utc)
    local_tz = pytz.timezone("Asia/Kolkata")  # or from config
    local_time = now.astimezone(local_tz)

    return {
        "current_time_utc": now.isoformat(),
        "local_time": local_time.strftime("%Y-%m-%d %H:%M %Z"),
        "is_weekend": local_time.weekday() >= 5,
        "is_business_hours": 9 <= local_time.hour < 18,
        "environment": metric_labels.get("env", "unknown"),
        "recent_deployments": [
            {
                "service": d["service"],
                "time_ago_minutes": int(
                    (now - datetime.fromisoformat(d["deployed_at"])).total_seconds() / 60
                ),
                "change_type": d.get("type", "unknown"),
            }
            for d in recent_deployments
            if (now - datetime.fromisoformat(d["deployed_at"])).total_seconds() < 3600
        ],
        "active_incidents": [
            {"title": i["title"], "severity": i["severity"]}
            for i in active_incidents
        ],
    }

Deduplication and alert grouping

LLM assessment per data point gets expensive if many metrics spike together (as they do during incidents). Deduplicate before sending to the LLM:

from collections import defaultdict
from typing import Callable

def group_correlated_candidates(
    candidates: list[AnomalyCandidate],
    time_window_seconds: int = 60,
) -> list[list[AnomalyCandidate]]:
    """
    Group anomaly candidates that occur within the same time window.
    Correlated anomalies likely have the same root cause and should
    be assessed together rather than independently.
    """
    if not candidates:
        return []

    sorted_candidates = sorted(candidates, key=lambda c: c.point.timestamp)
    groups = [[sorted_candidates[0]]]

    for candidate in sorted_candidates[1:]:
        latest_in_group = groups[-1][-1].point.timestamp
        delta = (candidate.point.timestamp - latest_in_group).total_seconds()

        if delta <= time_window_seconds:
            groups[-1].append(candidate)
        else:
            groups.append([candidate])

    return groups

def assess_anomaly_group(
    group: list[AnomalyCandidate],
    system_context: dict,
) -> AnomalyAssessment:
    """
    Assess a group of correlated anomalies together for cost efficiency.
    """
    if len(group) == 1:
        return assess_anomaly(group[0], system_context)

    metrics_summary = "\n".join([
        f"- {c.point.metric_name} ({c.point.labels}): "
        f"{c.point.value:.2f} (z={c.z_score:.2f}, {c.direction})"
        for c in group
    ])

    prompt = f"""Multiple metrics anomalous simultaneously — likely correlated.

ANOMALOUS METRICS:
{metrics_summary}

SYSTEM CONTEXT:
{json.dumps(system_context, indent=2, default=str)}

Assess this group of correlated anomalies as a single incident.
Return JSON: {{"is_anomalous": bool, "severity": str, "explanation": str,
"probable_cause": str, "recommended_action": str, "confidence": float}}"""

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}],
    )

    text = response.content[0].text
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    data = json.loads(text)
    return AnomalyAssessment(**data)

Seasonal baseline adjustment

A flat rolling window fails for metrics with predictable patterns. API request rates are higher on weekdays than weekends. Error rates spike every morning at 9am when batch jobs run. A Friday at 3pm looks anomalous compared to a Monday at 3pm even though both are completely expected.

The fix is a seasonality-aware baseline that compares the current value against historical values at the same time of week, not just the preceding N data points.

from collections import defaultdict

def build_seasonal_baseline(
    history: list[TimeSeriesPoint],
    bucket_minutes: int = 30,
) -> dict[tuple[int, int], tuple[float, float]]:
    """
    Build a baseline keyed by (day_of_week, time_bucket).
    Returns {(weekday, bucket): (mean, std)} for use in seasonal detection.
    """
    buckets: dict[tuple[int, int], list[float]] = defaultdict(list)

    for point in history:
        weekday = point.timestamp.weekday()  # 0=Monday, 6=Sunday
        bucket = (point.timestamp.hour * 60 + point.timestamp.minute) // bucket_minutes
        buckets[(weekday, bucket)].append(point.value)

    return {
        key: (np.mean(vals), np.std(vals))
        for key, vals in buckets.items()
        if len(vals) >= 3  # Require at least 3 data points per bucket
    }


def detect_seasonal_candidates(
    series: list[TimeSeriesPoint],
    baseline: dict[tuple[int, int], tuple[float, float]],
    z_threshold: float = 3.0,
    bucket_minutes: int = 30,
) -> list[AnomalyCandidate]:
    """
    Detect anomalies relative to the seasonal (same time of week) baseline
    instead of a flat rolling window. Falls back to rolling baseline if
    no seasonal data is available for this bucket.
    """
    candidates = []

    for i, point in enumerate(series):
        weekday = point.timestamp.weekday()
        bucket = (point.timestamp.hour * 60 + point.timestamp.minute) // bucket_minutes
        key = (weekday, bucket)

        if key not in baseline:
            continue  # Not enough history for this time slot

        mean, std = baseline[key]
        if std == 0:
            continue

        z = (point.value - mean) / std
        if abs(z) >= z_threshold:
            candidates.append(AnomalyCandidate(
                point=point,
                z_score=z,
                baseline_mean=mean,
                baseline_std=std,
                direction="high" if z > 0 else "low",
                context_window=series[max(0, i-10):i+5],
            ))

    return candidates

Build the seasonal baseline offline from several weeks of history and update it nightly. Pass the baseline type as additional context to the LLM assessment so it can reason appropriately: "This spike is 3.5 standard deviations above the typical Monday 2pm baseline, not just the rolling mean."

Feedback loop: improving assessments over time

LLM assessments drift toward false positives without a feedback mechanism. Engineers who dismiss alerts without acting on them are giving you a signal: the model assessed that alert as important, but the human disagreed. Capture that signal.

from datetime import datetime
from enum import Enum

class AlertOutcome(str, Enum):
    ACTIONED = "actioned"       # Engineer investigated and took action
    DISMISSED = "dismissed"     # Engineer confirmed no action needed
    FALSE_POSITIVE = "fp"       # Flagged as incorrect assessment

@dataclass
class AlertFeedback:
    assessment_id: str
    outcome: AlertOutcome
    engineer_note: str
    created_at: datetime

def store_feedback(feedback: AlertFeedback, db) -> None:
    """Persist engineer feedback for offline analysis."""
    db.execute(
        """INSERT INTO alert_feedback
           (assessment_id, outcome, engineer_note, created_at)
           VALUES (?, ?, ?, ?)""",
        (feedback.assessment_id, feedback.outcome,
         feedback.engineer_note, feedback.created_at.isoformat())
    )

def compute_false_positive_rate(
    metric_name: str,
    db,
    lookback_days: int = 30,
) -> float:
    """
    Compute the false positive rate for a given metric over the last N days.
    Use this to dynamically adjust z-score thresholds per metric.
    """
    rows = db.execute(
        """SELECT outcome FROM alert_feedback af
           JOIN assessments a ON af.assessment_id = a.id
           WHERE a.metric_name = ?
             AND af.created_at > date('now', ?)
        """,
        (metric_name, f"-{lookback_days} days")
    ).fetchall()

    if not rows:
        return 0.0

    fp_count = sum(1 for r in rows if r[0] == AlertOutcome.FALSE_POSITIVE)
    return fp_count / len(rows)

Use the false positive rate per metric to automatically tighten the z-score threshold. A metric that generates false positives 40% of the time should require a higher z-score before triggering an LLM assessment, not just a higher severity for the output. Feeding the false positive rate into the context prompt also helps: "Note: this metric has a 35% historical false positive rate. Apply extra scrutiny before recommending alert."

Cost and latency considerations

At scale, even Haiku adds up. A few controls keep costs manageable:

Rate-gate LLM calls: Only send candidates to the LLM if the z-score exceeds a higher threshold (e.g. 4.0) or the metric is in a high-priority list. Let lower-priority anomalies accumulate and batch-assess them hourly.

Cache similar contexts: If the same metric on the same host flagged twice within an hour and system context has not changed, reuse the prior assessment.

Throttle by metric importance: Tag metrics as tier-1 (customer-facing, assess immediately), tier-2 (internal, assess within 5 minutes), or tier-3 (infrastructure, assess hourly).

A rough cost estimate for production scale: at 10,000 statistical candidate anomalies per day with 80% filtered by grouping and caching, you send roughly 2,000 LLM assessments per day. At Haiku pricing and 500 tokens per assessment, that is well under $1 per day. The cost is negligible against the value of eliminating alert fatigue.

The right balance is: statistical detection for speed, LLM reasoning for quality, cost controls for scale. That combination produces alert quality that neither approach achieves alone.

Writing alert notifications with LLM-generated summaries

Once you have a verified anomaly assessment, the alert notification itself becomes a communication problem. A PagerDuty alert that says "cpu_usage_percent z=4.2 high" wakes someone up at 3am with no actionable information. A notification that says "CPU on web-03 is at 94% — 35% above the Monday 3am baseline. A deploy of the auth service ran 12 minutes ago. Check for a memory leak in the new release." is dramatically more useful.

Use the LLM's explanation and recommended_action fields to generate the alert body, and format it for the target channel:

def format_alert_for_slack(
    assessment: AnomalyAssessment,
    candidate: AnomalyCandidate,
    system_context: dict,
) -> dict:
    """
    Format an anomaly assessment as a Slack Block Kit message.
    Returns a dict suitable for posting to the Slack API.
    """
    severity_emoji = {
        "low": ":yellow_circle:",
        "medium": ":orange_circle:",
        "high": ":red_circle:",
        "critical": ":rotating_light:",
    }.get(assessment.severity, ":white_circle:")

    recent_deploy = (
        system_context.get("recent_deployments", [{}])[0]
        if system_context.get("recent_deployments") else None
    )
    deploy_note = (
        f"*Recent deploy:* {recent_deploy['service']} "
        f"({recent_deploy['time_ago_minutes']}m ago)"
        if recent_deploy else ""
    )

    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{severity_emoji} {assessment.severity.upper()}: "
                            f"{candidate.point.metric_name}",
                },
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*What happened:* {assessment.explanation}\n"
                        f"*Probable cause:* {assessment.probable_cause}\n"
                        f"*Recommended action:* {assessment.recommended_action}\n"
                        f"{deploy_note}"
                    ),
                },
            },
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": (
                            f"Metric: `{candidate.point.metric_name}` | "
                            f"Value: `{candidate.point.value:.2f}` | "
                            f"z-score: `{candidate.z_score:.2f}` | "
                            f"Confidence: `{assessment.confidence:.0%}`"
                        ),
                    }
                ],
            },
        ]
    }

This makes the LLM's reasoning the primary content of the alert, not a footnote. Engineers respond faster to alerts that explain themselves.

Choosing the right statistical method by metric type

Z-scores work well for normally distributed, stationary metrics. Many real-world metrics are neither. Choosing the wrong detector for the metric type is the most common reason a monitoring system produces either too many false positives or too many misses.

For count metrics, a simple Poisson model outperforms z-scores at low counts where the normal approximation breaks down:

from scipy.stats import poisson

def detect_count_anomaly(
    current_count: int,
    baseline_rate: float,  # Expected events per interval
    p_threshold: float = 0.001,  # Alert if this unlikely under Poisson model
) -> bool:
    """
    Detect whether a count is anomalously high under a Poisson model.
    More accurate than z-score for low-count events like error spikes.
    """
    # P(X >= current_count | rate=baseline_rate)
    p_value = 1 - poisson.cdf(current_count - 1, baseline_rate)
    return p_value < p_threshold

Pass the detector type as part of the context to the LLM so it can adjust its reasoning. A z-score of 3.5 on a Gaussian metric and a p-value of 0.0005 on a Poisson metric are both statistically significant, but they have different practical implications for a metric like payment failures where the base rate is naturally low.

Putting it together: a production checklist

Before deploying LLM-assisted anomaly detection in production, verify these properties hold:

• Graceful LLM degradation: If the LLM API is unavailable, the statistical layer continues firing raw alerts rather than suppressing them. Never make your alerting system depend on an external AI API for basic functionality.
• Assessment audit trail: Store every LLM assessment with the full context sent, the model version, and the response. You need this to debug why an alert was or was not fired, especially after incidents.
• Human escalation path: Any "critical" severity assessment should page a human immediately. Reserve lower severities for ticket creation or Slack notifications that do not require immediate response.
• Baseline staleness detection: If the seasonal baseline is more than 7 days old and no update has run, fall back to a simple rolling window rather than silently using a stale model.
• Cost alerting: Set a budget alarm on your LLM API spend for the anomaly detection workload. An unexpected spike in anomaly candidates (itself a sign of a system problem) can drive an unexpected spike in API costs.

The combination of statistical speed, LLM reasoning depth, seasonal awareness, and feedback loops produces an alerting system that gets meaningfully better over time — unlike a static threshold configuration that silently degrades as your system evolves.