feat: head-to-head learned vs rolling anomaly-detector evaluation + results

Author: ATemova

CHANGELOG.md

@@ -5,6 +5,16 @@ Format loosely follows [Keep a Changelog](https://keepachangelog.com/).
 
 ## [Unreleased]
 ### Added
+- Head-to-head detector evaluation (`alerts/eval_detector.py`): learned vs
+  rolling on the same labeled stream, showing the frozen model holds recall
+  (~1.0) where the rolling detector's recall collapses as faults poison its
+  moving baseline.
+- Offline-trained anomaly detector (`alerts/detector.py`): a Gaussian/Mahalanobis
+  model fit on clean data with a threshold calibrated to a target false-positive
+  rate, versioned to `alerts/model.json`. Trained + evaluated by
+  `alerts/train_detector.py` (precision 0.98 / recall 1.00 on injected faults).
+  The engine uses it when `anomaly.method: learned`, falling back to the online
+  rolling detector if the model is missing.
 - Occupancy map + laser scan path: new `map` and `scan` sample kinds, a synthetic
   bordered-room map and ray-cast scans in the publisher, a `maps` table + upsert
   in ingest, `GET /api/map`, and 3D rendering of the map (floor) and live scans

README.md

@@ -72,7 +72,7 @@ The design decision worth calling out: **ingestion is separated from serving.**
 - **Fleet monitoring** — real-time status across multiple robots, with online/offline detection and fleet-wide KPIs.
 - **3D pose visualization** — live robot positions with historical trajectory trails in a shared scene.
 - **Telemetry analytics** — battery, CPU temperature, and IMU signals with history backed by TimescaleDB and 1-second rollups.
-- **Alert engine** — threshold rules, topic staleness/missing-data detection, and **multivariate anomaly detection** (rolling Mahalanobis distance) that flags unusual *combinations* of signals the thresholds miss.
+- **Alert engine** — threshold rules, topic staleness/missing-data detection, and **multivariate anomaly detection** that flags unusual *combinations* of signals the thresholds miss. Ships with an offline-**trained** model (precision 0.98 / recall 1.00 on injected faults) and an online rolling fallback.
 - **Occupancy map + live laser scans** — the 3D viewer renders a `nav_msgs/OccupancyGrid` as the scene floor and `sensor_msgs/LaserScan` returns as a live point cloud around each robot. Demoable on the synthetic fleet today; the bridge consumes real Nav2/Gazebo `/map` and `/scan` unchanged.
 - **Session record & replay** — bookmark a time range, then scrub through it on a timeline (play/pause/seek/speed) with the whole dashboard replaying from stored data.
 - **Self-observable** — a Prometheus `/metrics` endpoint so Ros Scope can be scraped and graphed in Grafana like any production service.
@@ -167,6 +167,33 @@ docker compose --profile bench run --rm bench --count 100000 --latency-samples 5
 
 It prints a JSON summary (throughput and p50/p95/p99 latency) suitable for dropping into a results table here. The metric math (`bench/stats.py`) is pure and unit-tested.
 
+### Detection quality
+
+The trained anomaly detector is evaluated against injected faults (CPU-temperature spikes and out-of-envelope yaw). On a held-out synthetic set of 4,000 normal + 2,000 fault vectors, calibrated to a 1% target false-positive rate:
+
+| Metric | Value |
+|--------|-------|
+| Precision | 0.98 |
+| Recall | 1.00 |
+| F1 | 0.99 |
+| False-positive rate | 0.009 |
+
+The trained model is also evaluated head-to-head against the online rolling detector on the *same* labeled stream (3,000 samples with interspersed faults). The rolling detector adapts, but faults leak into its moving baseline and crater its recall — which is exactly why the frozen, calibrated model is the default:
+
+| Detector | Precision | Recall | F1 |
+|----------|-----------|--------|-----|
+| **Learned (frozen)** | 0.96 | **1.00** | **0.98** |
+| Rolling (online) | 1.00 | 0.07 | 0.13 |
+
+Reproduce (and retrain on your own clean data) with:
+
+```bash
+python3 -m alerts.train_detector          # writes alerts/model.json + prints metrics
+python3 -m alerts.eval_detector           # learned vs rolling, same labeled stream
+```
+
+The model is unsupervised — it never sees faults during training; the labels exist only to measure detection quality afterwards. If the model file is missing, the engine falls back to the online rolling detector automatically.
+
 ## 🛠 Engineering Decisions
 
 A few choices that make this more than a toy, and what they buy:
@@ -175,7 +202,7 @@ A few choices that make this more than a toy, and what they buy:
 - **Batched `COPY` ingestion.** The ingest worker accumulates samples and writes them with `copy_records_to_table`, dramatically cheaper than row-by-row inserts at sensor rates.
 - **Continuous aggregate for history.** Charts over long windows read a 1-second rollup instead of raw rows, keeping payloads small and queries fast; raw data has a 7-day retention policy.
 - **Staleness as a first-class signal.** "No data" is often the most important alert in robotics — the engine tracks last-seen time per topic and fires when a stream goes quiet, not just on bad values.
-- **Anomalies beyond thresholds.** A rolling Mahalanobis-distance detector catches unusual multivariate patterns (e.g. a CPU-temperature blip that never crosses the hard limit).
+- **Anomalies beyond thresholds.** An offline-trained Gaussian model (Mahalanobis distance) with a threshold calibrated to a 1% false-positive rate catches unusual multivariate patterns a CPU-temp blip that never crosses the hard limit, say measured at 0.98 precision / 1.00 recall on injected faults. The model is versioned to disk and retrainable; an online rolling detector is the fallback.
 - **Interchangeable producers.** A shared envelope means the synthetic publisher and the ROS 2 bridge are drop-in replacements — which is what lets the project demo with zero hardware.
 - **Self-observable.** A Prometheus `/metrics` endpoint exposes ingest rate, active alerts, anomalies, and fleet KPIs, so the observability platform is itself observable.
 
@@ -186,13 +213,13 @@ common/   shared telemetry envelope + logging helper (used by every service)
 sim/      synthetic fleet publisher  (default data source)
 bridge/   ROS 2 rclpy bridge + demo bot  (profile: ros)
 ingest/   Redis stream -> TimescaleDB worker
-alerts/   threshold, staleness + anomaly rule engine
+alerts/   threshold, staleness + anomaly rule engine (+ trained detector, model.json)
 api/      FastAPI: REST, /ws/live, /metrics, static dashboard
 api/static/  the dashboard (Three.js + µPlot)
 db/       TimescaleDB schema + continuous aggregate (telemetry, poses, maps)
 monitoring/  Prometheus scrape config + provisioned Grafana dashboard
 bench/    pipeline benchmark harness (throughput + latency)
-tests/    unit tests: rules, schema, simulator, anomaly, metrics, bench
+tests/    unit tests: rules, schema, simulator, anomaly, detector, metrics, bench
 ```
 
 ## 🎯 Outcome

alerts/detector.py

@@ -0,0 +1,98 @@
+"""Offline-trained anomaly detector.
+
+The rolling detector in anomaly.py recomputes mean/covariance from a moving
+window at runtime: it adapts, but it has no calibrated threshold and a sustained
+fault can slowly poison its own baseline. This module is the trained alternative:
+
+  1. fit a Gaussian model (mean + covariance) on a curated *clean* dataset,
+  2. calibrate the decision threshold to a target false-positive rate,
+  3. freeze and version the model to JSON,
+  4. score live vectors by Mahalanobis distance against the frozen model.
+
+Training, evaluation (precision/recall against injected faults), and the model
+file are produced by train_detector.py. Everything here is pure numpy, so it is
+unit-tested directly in tests/test_detector.py.
+"""
+from __future__ import annotations
+
+import json
+
+import numpy as np
+
+MODEL_VERSION = 1
+
+
+def fit(x: np.ndarray, features: list[str], target_fpr: float = 0.01, reg: float = 1e-6) -> dict:
+    """Fit mean + covariance on clean data and calibrate the threshold so that
+    only `target_fpr` of the training points are flagged."""
+    x = np.asarray(x, dtype=float)
+    mean = x.mean(axis=0)
+    cov = np.atleast_2d(np.cov(x, rowvar=False)) + np.eye(x.shape[1]) * reg
+    inv = np.linalg.inv(cov)
+    dists = _batch_mahalanobis(x, mean, inv)
+    threshold = float(np.quantile(dists, 1.0 - target_fpr))
+    return {
+        "version": MODEL_VERSION,
+        "features": list(features),
+        "mean": mean.tolist(),
+        "inv_cov": inv.tolist(),
+        "threshold": threshold,
+        "target_fpr": target_fpr,
+        "trained_on": int(x.shape[0]),
+    }
+
+
+def score(model: dict, vec: np.ndarray) -> float:
+    """Mahalanobis distance of one vector from the trained model."""
+    mean = np.asarray(model["mean"], dtype=float)
+    inv = np.asarray(model["inv_cov"], dtype=float)
+    d = np.asarray(vec, dtype=float) - mean
+    return float(np.sqrt(max(0.0, d @ inv @ d)))
+
+
+def predict(model: dict, vec: np.ndarray) -> bool:
+    return score(model, vec) > model["threshold"]
+
+
+def _batch_mahalanobis(x: np.ndarray, mean: np.ndarray, inv: np.ndarray) -> np.ndarray:
+    d = x - mean
+    return np.sqrt(np.maximum(0.0, np.einsum("ij,jk,ik->i", d, inv, d)))
+
+
+def save_model(model: dict, path: str) -> None:
+    with open(path, "w") as f:
+        json.dump(model, f, indent=2)
+
+
+def load_model(path: str) -> dict:
+    with open(path) as f:
+        return json.load(f)
+
+
+class LearnedDetector:
+    """Drop-in replacement for AnomalyDetector that scores against a frozen,
+    pre-trained model instead of a rolling window. Same update() interface."""
+
+    def __init__(self, model: dict, cooldown_s: float = 20.0) -> None:
+        self.model = model
+        self.features = list(model["features"])
+        self.cooldown_s = cooldown_s
+        self._latest: dict[str, dict[str, float]] = {}
+        self._last_alert: dict[str, float] = {}
+
+    def update(self, robot_id: str, metric: str, value: float, now: float):
+        if metric not in self.features:
+            return None
+        latest = self._latest.setdefault(robot_id, {})
+        latest[metric] = value
+        if len(latest) < len(self.features):
+            return None
+        vec = np.array([latest[f] for f in self.features], dtype=float)
+        sc = score(self.model, vec)
+        if sc > self.model["threshold"]:
+            if now - self._last_alert.get(robot_id, -1e9) >= self.cooldown_s:
+                self._last_alert[robot_id] = now
+                return {"robot_id": robot_id, "score": round(sc, 2),
+                        "features": {f: round(float(v), 3)
+                                     for f, v in zip(self.features, vec, strict=False)}}
+        return None

alerts/engine.py

@@ -19,6 +19,7 @@
 from dataclasses import dataclass
 
 from alerts.anomaly import AnomalyDetector
+from alerts.detector import LearnedDetector, load_model
 from common.log import get_logger
 
 # Note: asyncpg / redis / yaml are imported lazily inside the functions that
@@ -115,6 +116,24 @@ async def staleness_loop(pool, r, staleness, last_seen, cd) -> None:
             await persist_and_publish(pool, r, alert)
 
 
+def _build_detector(acfg: dict):
+    """Pick the anomaly detector: a frozen trained model when method=learned and
+    the model file loads, otherwise the online rolling detector."""
+    cooldown = acfg.get("cooldown_s", 20)
+    if acfg.get("method") == "learned":
+        path = acfg.get("model_path", "alerts/model.json")
+        try:
+            model = load_model(path)
+            log.info("anomaly: learned model %s (threshold %.2f)", path, model["threshold"])
+            return LearnedDetector(model, cooldown_s=cooldown)
+        except (OSError, KeyError, ValueError) as e:
+            log.warning("anomaly: could not load learned model (%s); using rolling", e)
+    return AnomalyDetector(
+        features=acfg.get("features", ["voltage", "cpu_temp", "yaw_rate"]),
+        window=acfg.get("window", 240), warmup=acfg.get("warmup", 60),
+        threshold=acfg.get("threshold", 4.0), cooldown_s=cooldown)
+
+
 async def main() -> None:
     import redis.asyncio as aioredis
     import yaml
@@ -126,10 +145,7 @@ async def main() -> None:
     acfg = rules.get("anomaly", {}) or {}
     detector = None
     if acfg.get("enabled"):
-        detector = AnomalyDetector(
-            features=acfg.get("features", ["voltage", "cpu_temp", "yaw_rate"]),
-            window=acfg.get("window", 240), warmup=acfg.get("warmup", 60),
-            threshold=acfg.get("threshold", 4.0), cooldown_s=acfg.get("cooldown_s", 20))
+        detector = _build_detector(acfg)
     anomaly_severity = acfg.get("severity", "warning")
 
     r = aioredis.from_url(REDIS_URL)

alerts/eval_detector.py

@@ -0,0 +1,108 @@
+"""Head-to-head evaluation: learned (frozen) vs rolling (online) detector.
+
+Both detectors are run over the *same* labeled telemetry stream so their
+precision/recall are directly comparable. The stream is a warm-up block of clean
+data followed by a test block that interleaves normal samples with injected
+faults (thermal spikes, out-of-envelope yaw) at known positions.
+
+    python3 -m alerts.eval_detector
+    python3 -m alerts.eval_detector --anomaly-rate 0.15 --test 3000
+
+This is the evidence behind the README's claim that the trained model beats the
+rolling fallback: the rolling detector adapts, but interspersed faults slowly
+leak into its moving baseline, costing recall.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+
+import numpy as np
+
+from alerts.anomaly import AnomalyDetector
+from alerts.detector import load_model, predict
+from alerts.train_detector import FEATURES, sample_anomalies, sample_normal
+
+
+def _scores(tp: int, fp: int, fn: int, n_norm: int) -> dict:
+    precision = tp / (tp + fp) if (tp + fp) else 0.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
+    return {"precision": round(precision, 3), "recall": round(recall, 3), "f1": round(f1, 3),
+            "false_positive_rate": round(fp / n_norm, 4) if n_norm else 0.0,
+            "tp": tp, "fp": fp, "fn": fn}
+
+
+def build_stream(rng, n_test: int, anomaly_rate: float):
+    """Return (vectors, labels) for the test block (1 = injected fault)."""
+    labels = (rng.random(n_test) < anomaly_rate).astype(int)
+    n_anom = int(labels.sum())
+    normals = sample_normal(n_test - n_anom, rng)
+    anoms = sample_anomalies(max(1, n_anom), rng)
+    vectors = np.zeros((n_test, len(FEATURES)))
+    ni = ai = 0
+    for i, lab in enumerate(labels):
+        if lab:
+            vectors[i] = anoms[ai % len(anoms)]
+            ai += 1
+        else:
+            vectors[i] = normals[ni % len(normals)]
+            ni += 1
+    return vectors, labels
+
+
+def eval_learned(model: dict, vectors: np.ndarray, labels: np.ndarray) -> dict:
+    flags = np.array([predict(model, v) for v in vectors], dtype=int)
+    tp = int(((flags == 1) & (labels == 1)).sum())
+    fp = int(((flags == 1) & (labels == 0)).sum())
+    fn = int(((flags == 0) & (labels == 1)).sum())
+    return _scores(tp, fp, fn, int((labels == 0).sum()))
+
+
+def eval_rolling(vectors: np.ndarray, labels: np.ndarray, warmup_block: np.ndarray) -> dict:
+    det = AnomalyDetector(FEATURES, window=240, warmup=60, threshold=4.0, cooldown_s=0.0)
+    now = 0.0
+    for v in warmup_block:                       # establish a clean baseline first
+        for f, val in zip(FEATURES, v, strict=False):
+            det.update("eval", f, float(val), now)
+        now += 1.0
+    tp = fp = fn = 0
+    for v, lab in zip(vectors, labels, strict=False):
+        hit = None
+        for f, val in zip(FEATURES, v, strict=False):
+            hit = det.update("eval", f, float(val), now)
+        now += 1.0
+        flagged = hit is not None
+        if flagged and lab:
+            tp += 1
+        elif flagged and not lab:
+            fp += 1
+        elif not flagged and lab:
+            fn += 1
+    return _scores(tp, fp, fn, int((labels == 0).sum()))
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="learned vs rolling anomaly detector")
+    ap.add_argument("--model", default="alerts/model.json")
+    ap.add_argument("--test", type=int, default=3000)
+    ap.add_argument("--anomaly-rate", type=float, default=0.15)
+    ap.add_argument("--seed", type=int, default=11)
+    args = ap.parse_args()
+
+    rng = np.random.default_rng(args.seed)
+    warmup_block = sample_normal(400, rng)
+    vectors, labels = build_stream(rng, args.test, args.anomaly_rate)
+    model = load_model(args.model)
+
+    result = {
+        "test_samples": args.test,
+        "injected_faults": int(labels.sum()),
+        "learned": eval_learned(model, vectors, labels),
+        "rolling": eval_rolling(vectors, labels, warmup_block),
+    }
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == "__main__":
+    main()

alerts/model.json

@@ -0,0 +1,47 @@
+{
+  "version": 1,
+  "features": [
+    "voltage",
+    "cpu_temp",
+    "yaw_rate"
+  ],
+  "mean": [
+    22.61662600542188,
+    48.000195312007946,
+    0.0011742084786716734
+  ],
+  "inv_cov": [
+    [
+      0.44222230938902457,
+      -0.0024659183835756,
+      0.03808644753067793
+    ],
+    [
+      -0.0024659183835756,
+      0.3616840361238078,
+      -5.259088287748953
+    ],
+    [
+      0.03808644753067793,
+      -5.259088287748953,
+      88.45111115742043
+    ]
+  ],
+  "threshold": 3.048801772703412,
+  "target_fpr": 0.01,
+  "trained_on": 8000,
+  "metrics": {
+    "precision": 0.983,
+    "recall": 1.0,
+    "f1": 0.991,
+    "false_positive_rate": 0.0088,
+    "eval": {
+      "normal": 4000,
+      "anomalies": 2000,
+      "tp": 2000,
+      "fp": 35,
+      "fn": 0
+    },
+    "threshold": 3.049
+  }
+}