Author(s): EMMANUEL NWANGUMA

Originally published on Towards AI.

There’s a category of problem where being right tomorrow is the same as being wrong.

A fraudulent transaction clears. A server starts throwing errors at 2pm and nobody notices until the morning report. A sensor drifts out of spec and the machine it’s attached to grinds itself apart over six hours. In every one of those cases the detection logic might be perfect — but if it runs as a nightly batch job, the answer arrives after the damage.

So I built the opposite: a streaming pipeline where events flow in continuously, get scored the moment they arrive, and turn into a Slack alert in under two seconds. It handles three genuinely different data types — card transactions, server metrics, and IoT sensor readings — on one pipeline.

Along the way I found two bugs that had my LSTM detector performing at 4% recall, and the fix for the second one had nothing to do with the model at all. More on that below.

Why batch is the wrong shape for this problem

The instinct is to treat anomaly detection as a data science problem: get data, train model, evaluate, ship. But in production it’s mostly a systems problem. Three things matter more than the model:

1. Latency — how long between the event happening and a human knowing.
2. Noise — whether the alerts are still worth reading after a week.
3. Adaptability — whether you can change the detector without taking the system down.

A batch job fails all three. It’s slow by construction, it dumps a pile of findings with no grouping, and updating it means a redeploy.

The pipeline

Data sources: transactions, server metrics, IoT sensors
 │
 ▼
 Redpanda topics (Kafka API)
 anomaly.fraud / anomaly.metrics / anomaly.iot
 │
 ▼
 Faust stream processor
 ├── rolling windows (1m / 5m / 1h, per entity)
 ├── route event type → detector(s)
 └── real-time inference
 │
 ┌─────────────────┴─────────────────┐
 ▼ ▼
 Detection models TimescaleDB
 ├── Isolation Forest (fraud) (events + flags,
 ├── LSTM Autoencoder (IoT) hypertables)
 └── Z-score / EWMA (metrics) │
 │ ▼
 ▼ Grafana dashboard
 Alert engine
 ├── severity scoring
 ├── deduplication
 └── Slack + email

Redpanda gives me the Kafka API without the JVM. Faust does the stream processing in Python. TimescaleDB stores everything as hypertables so time-bucketed queries stay fast. Grafana reads both TimescaleDB and Prometheus.

Rolling windows, and why they’re per-entity

A single event usually isn’t enough to judge anything. A £2,000 transaction is unremarkable — unless that card has already made eleven transactions in the last hour. So the stream keeps rolling windows (1 minute, 5 minutes, 1 hour) and derives counts, means, standard deviations, and deltas on top of the raw fields.

The subtle part is the key. Windows are kept per source:entity, not per entity:

window_key = f"{source}:{event.entity_id}"
self._features.add(window_key, event)

I found this the hard way. My first version keyed windows by entity_id alone, and a test that reused the same ID across two source types blew up with a KeyError. A server's window had been filled with fraud features. Scoping by source makes the collision structurally impossible rather than merely unlikely.

Three detectors, three different jobs

Routing is per source type:

ROUTING = {
 "fraud": ["isolation_forest", "zscore"],
 "metrics": ["zscore", "ewma", "isolation_forest"],
 "iot": ["lstm_autoencoder", "zscore"],
}

Z-score / EWMA for server metrics. They track a running mean and standard deviation per feature and flag deviations. No training run, no model file, cheap enough to run inline on every event. For high-volume metrics where “normal” is a stable band, this is genuinely hard to beat.

Isolation Forest for fraud. Fraud rarely looks wrong on any single dimension — it’s the combination that’s off. A large amount is fine. A foreign transaction is fine. A 3am transaction is fine. All three together on a card that’s already been used eleven times this hour is not. Isolation Forest handles that interaction; a per-feature threshold never will.

LSTM Autoencoder for IoT. Sensors produce sequences, and the anomaly is often a pattern rather than a value — a temperature that’s climbing at the wrong rate is a problem long before it crosses any single threshold. The autoencoder learns to reconstruct a window of normal readings; when reconstruction error spikes, the pattern is off.

That last one is where things got interesting.

Bug #1: the autoencoder that couldn’t detect anything

My first backtest of the LSTM came back with 4% recall. It was catching essentially nothing.

The cause was in one line of my training setup: I was training the autoencoder on the full labeled dataset — which included the anomalies.

An autoencoder detects anomalies by learning to reconstruct normal data well, then flagging inputs it reconstructs badly. The detection threshold is set at, say, the 99th percentile of reconstruction error observed during training. But if 5% of your training data is anomalous, those anomalies produce the largest reconstruction errors, and they drag the 99th-percentile threshold up to their own level. You end up with a threshold that only the most extreme anomalies could ever exceed.

The fix is one line, and it’s a methodological rule rather than a tuning trick:

# Autoencoders must train on NORMAL data only — training on the
# contaminated set pushes the reconstruction-error threshold up to the
# anomalies themselves and collapses recall.
normal_events = [e for e, lab in zip(events, y_true) if lab == 0]
det = train_lstm_autoencoder(normal_events, fn, seq_len=seq_len, epochs=10)

Recall went from 0.04 to 1.00.

Bug #2: the model was fine, my evaluation was wrong

With recall fixed, precision came back at 0.14. The detector was now flagging roughly seven times more windows than there were anomalies.

I nearly started tuning the threshold. Then I looked at how I was scoring it.

The autoencoder consumes a window of 10 events and produces one verdict about that window. I was comparing that verdict against the label of the last event in the window only. With a 5% anomaly rate and a 10-event window, roughly 40% of windows contain an anomaly somewhere. So every time the detector correctly flagged a window because an anomaly sat at position 3, my evaluation counted it as a false positive.

The model was right. The measurement was wrong.

# A sequence detector judges the whole window, so it must be scored
# against a window-level label: anomalous if ANY event in the window
# is anomalous. Comparing a window verdict to only the last event's
# label counts correct detections as false positives.
y_window.append(int(max(y_true[i - seq_len + 1:i + 1])))

Precision went from 0.14 to 0.98. Final confusion matrix: 302 true positives, 0 false negatives, 6 false positives out of 489 negatives.

The lesson generalises well beyond LSTMs: when a model scores a window but your labels describe a point, you are not measuring what you think you are measuring. Check the evaluation before you touch the model.

The alert engine — the least glamorous, most important part

Here is the thing that decides whether anyone actually uses your system.

One failing server produces a burst of anomalies. In my demo run, the metrics source generated 50 anomaly flags. If each one fires a Slack message, you’ve sent 50 pings about a single incident, and by next week the channel is muted — including for the one alert that mattered.

So anomalies get grouped by source:entity:severity within a configurable window. The first raises an alert and notifies; subsequent ones increment a count and stay quiet, unless the severity escalates into a band that needs a new channel.

low → log only
medium → Slack
high → Slack + email
critical → Slack + email

360 events processed, 32 actionable alerts. Every alert carries its own count, so a burst is visible as one incident with a magnitude rather than fifty separate pings.

Notifiers also degrade gracefully — leave the Slack webhook and SMTP host empty and alerts are logged instead of sent, so the whole system runs end to end with zero external configuration.

Hot-reloading models without stopping the stream

You will want to change a detector while the pipeline is live. A background timer re-reads the active model version from the registry and swaps the in-memory instance:

async def hot_reload(self, source_type, detector):
 mv = await self._store.get_active_model(detector, source_type)
 cached = self._cache.get(key)
 if cached is not None and cached[0] == mv.version:
 return cached[1] # unchanged
 obj = _load_detector(detector, mv.storage_path)
 self._cache[key] = (mv.version, obj) # atomic swap
 return obj

Replacing a dict value is atomic in CPython, so an in-flight score() never sees a half-swapped model. Activate a new version through the API and the running worker picks it up within the timer interval — no restart, no dropped events.

Performance: the bottleneck wasn’t the models

My first end-to-end demo ran at 2 events per second, which was embarrassing for something advertising sub-two-second latency.

The models weren’t the problem. Every persisted row was doing its own commit() plus a refresh() — two network round-trips each, and with a Windows→container Postgres connection each round-trip cost around 70ms. One event with three detectors meant roughly eight round-trips.

Detection is in-memory and fast; only persistence needed batching:

verdicts = await router.detect(e) # in-memory; no DB round-trips
...
await store.bulk_insert_events(event_rows)
await store.bulk_insert_flags(flag_rows)

Two commits instead of hundreds. Throughput went to 12–30 events/sec, and measured p95 detection latency landed at roughly 900ms for metrics and 485ms for IoT — comfortably inside the two-second target.

The honest part: simulated vs real accuracy

My simulated benchmark reports precision and recall of ~1.00 across fraud and metrics. You should not be impressed by that number, and I’m not going to pretend you should be.

The simulators inject anomalies with known labels, which is the only reason a backtest is possible at all. But those injected anomalies are cleanly separable: a fraud anomaly is a large amount and foreign and an odd hour, simultaneously; a metrics anomaly is 90%+ CPU against a 15–45% baseline. There is almost no overlap between the two distributions. Any competent detector scores near-perfectly on that.

Those numbers measure the pipeline, not real-world detection ability.

Real anomaly detection is much harder:

 Simulated Reality
Class overlap clean separation anomalies look almost normal
Base rate 2–10% often 0.1% or lower
Labels perfect usually none
"Normal" fixed drifts constantly
Training data clean contaminated with undetected anomalies
Adversaries none fraudsters actively adapt

The base rate is the one that kills you. At a 0.1% anomaly rate, even a 1% false-positive rate produces roughly ten false alarms per true one — precision collapses no matter how good your recall is.

Which is exactly why this project spends as much effort on severity scoring and deduplication as on the models. Those aren’t decoration; at realistic base rates they’re the difference between a system people use and a muted channel.

For an honest number, the backtester also runs against the real Kaggle Credit Card Fraud dataset (284k transactions, ~0.17% fraud, real labels):

python -m backtest.replay --csv data/creditcard.csv --label Class

Expect materially lower precision and recall than the simulated run. That gap is the point.

What the browser found that code review didn’t

One last thing worth mentioning. After everything passed its tests, I opened the Grafana dashboard in an actual browser — and found four problems that no amount of unit testing would have caught:

• panel titles rendered as mojibake (anomalies â€" metrics) from em-dash encoding
• three of six panels showed No data, because they queried Prometheus metrics that required scrape wiring I hadn’t done
• the severity bar gauge collapsed three severities into a single bar
• the time-series was one vertical spike, because the demo inserted every event at nearly the same timestamp

All four are invisible in code and obvious in a screenshot. Two of the panels became TimescaleDB SQL queries so the dashboard now works standalone after import, and the demo spreads simulated arrivals across a 25-minute window so the charts show a realistic stream.

If your project produces a UI, look at it before you say it works.

How it fits the platform

This is the fourth piece of a connected MLOps platform:

1. ml-feature-store — define a feature once, serve it consistently for training and live inference.
2. pipeline-lineage-tracker — trace any prediction back to the exact data and code that produced it.
3. ml-canary-deploy — ship model updates to a slice of traffic with automatic rollback.
4. realtime-anomaly-detection — watch the live streams those systems run on and react in seconds.

The code

Everything is open source: github.com/Emart29/realtime-anomaly-detection

examples/demo.py runs all three use cases end to end — trains the detectors, streams simulated traffic through the real detection path, and raises deduplicated alerts — against a live Redpanda + TimescaleDB stack.

Next: a causal inference engine — because knowing that two things move together tells you nothing about which one to change.

Join thousands of data leaders on the AI newsletter. Join over 80,000 subscribers and keep up to date with the latest developments in AI. From research to projects and ideas. If you are building an AI startup, an AI-related product, or a service, we invite you to consider becoming a sponsor.

Published via Towards AI