Realtime ML Drift Detection

Realtime ML Drift Detection is a local-first monitoring project for model-serving pipelines. A producer streams prediction events into Kafka, a consumer evaluates feature drift with two complementary detection paths (windowed KS + PSI, and a consensus over the ADWIN, Page-Hinkley, and KSWIN online detectors), Postgres stores raw events, drift metrics, latency percentiles, and alerts, and Grafana visualizes the result.

The example pipeline uses the UCI Chronic Kidney Disease dataset and a scikit-learn classifier. The producer can inject controlled drift into selected features so the detectors can be exercised on demand instead of waiting for organic distribution shift.

What it includes

• Kafka in KRaft mode for event transport
• Confluent Schema Registry with Avro-encoded prediction events
• Kafka UI for browsing topics and messages during debugging
• A CKD training pipeline that writes a model bundle and frozen reference statistics, including reference distributions for the model's prediction and confidence outputs
• A producer that emits prediction events at a configurable rate
• Drift injection controls for mean shift, scaling, and additive noise
• A streaming detector that computes KS + PSI per feature on sliding windows
• Online change-point detection: ADWIN + Page-Hinkley + KSWIN per feature (z-score normalised against the reference) with consensus voting
• Bootstrap-calibrated per-feature PSI thresholds persisted with the reference artifact
• Buffered prediction writes via COPY + staging table
• Postgres tables for predictions, baselines, drift metrics, drift detector signals, latency metrics, and alerts
• A pre-provisioned Grafana dashboard with prediction volume, KS p-value and PSI per feature, a PSI heatmap, latency percentiles, prediction and confidence drift, alert summaries, and a recent alerts table
• Optional Slack-compatible webhook delivery for new alerts
• Multi-stage Dockerfile and an apps compose profile so the producer and consumer can run as containers
• Synthetic precision/recall, end-to-end latency, throughput, and Postgres ingest benchmarks
• GitHub Actions CI running ruff and the unit test suite on every push

Architecture

producer  ->  Kafka + Schema Registry  ->  consumer  ->  Postgres  ->  Grafana
                                              |
                                              +-> KS + PSI per feature window
                                              +-> ADWIN + Page-Hinkley + KSWIN (online)
                                              +-> optional alert webhook

Kafka, Schema Registry, Kafka UI, Postgres, and Grafana run through docker compose. The producer and consumer either run on the host (default, fast iteration) or as containers via the compose apps profile.

Quickstart

make install
make up
make topics
make train
make test
make demo

Grafana is available at http://localhost:3000/d/ml-drift-monitoring with admin / admin. Kafka UI is at http://localhost:8080.

The demo streams clean traffic first, then injects creatinine drift. During the second phase the sc feature shows a visible PSI jump, the model's output distribution shifts (so __prediction__ and __confidence__ also trigger), and three alert rows are produced — one per affected feature — rather than repeated duplicates inside the cooldown window.

Common commands

make produce        # clean stream
make produce-drift  # creatinine mean shift after 60s
make consume        # start the detector
make test           # run pytest (skips e2e if Docker is not on PATH)
make lint           # run ruff

make build-app      # build the producer/consumer container image
make up-apps        # bring up infra AND the containerized producer + consumer

make bench-accuracy # synthetic precision/recall for KS+PSI and online consensus
make bench-latency  # report end-to-end latency from recent predictions
make bench-load     # sustained-throughput sweep (consumer must be running)
make bench-ingest   # benchmark the Postgres write path in isolation

How detection works

Reference statistics are frozen at training time and written to artifacts/. For each tracked feature — the 14 input features plus two output pseudo-features (__prediction__ and __confidence__) — the project stores summary stats, raw reference values, and quantile-based histogram edges used for PSI.

The consumer keeps an event-time sliding window per feature. Window emission is driven by the event-timestamp watermark with a configurable lateness allowance. Once a window has enough samples, it compares current values with the frozen reference using two parallel detection paths:

• Windowed KS + PSI. KS uses scipy.stats.ks_2samp. PSI uses quantile-based decile bins with Laplace smoothing. Severity is ok, warn, or alert. Per-feature PSI thresholds can be calibrated from a bootstrap null distribution at training time.
• Online detectors. ADWIN, Page-Hinkley, and KSWIN run per feature and process events one at a time. Values are z-score normalised against the reference mean and stddev so a single set of detector hyperparameters applies uniformly across features with different natural scales. Consensus is True when ≥ 2 of 3 detectors fire in the same window.

Alerts are written when severity is warn/alert and no other alert for the same feature+severity is in its cooldown window (suppressed_until). When configured, alerts also POST to ALERT_WEBHOOK_URL (Slack-compatible body shape).

Default thresholds:

variable	default	meaning
WINDOW_SECONDS	300	window width
SLIDE_SECONDS	60	emission interval
ALLOWED_LATENESS_SECONDS	5	event-time lateness allowance
KS_PVALUE_WARN	0.05	KS warn threshold
KS_PVALUE_ALERT	0.01	KS alert threshold
PSI_WARN	0.10	PSI warn threshold
PSI_ALERT	0.20	PSI alert threshold
ALERT_WEBHOOK_URL	(empty)	optional Slack-compatible webhook

Drift injection

The producer accepts one or more drift specs in the form feature:transform.

Supported transforms:

• mean+2sigma
• mean-1.5sigma
• scale1.4
• noise2x

Examples:

uv run python -m drift.producer --rate 50 --drift sc:mean+2sigma
uv run python -m drift.producer --rate 50 --drift hemo:mean-1.5sigma
uv run python -m drift.producer --rate 50 --drift bgr:scale1.4
uv run python -m drift.producer --rate 50 --drift "sc:mean+2sigma,hemo:scale0.7" --start-after 60

Friendly aliases include creatinine -> sc, glucose -> bgr, hemoglobin -> hemo, potassium -> pot, and sodium -> sod.

Configuration

Configuration is environment-driven. Copy .env.example to .env if you want to override defaults.

The most commonly adjusted values are:

• Kafka connection and topic settings
• Schema Registry URL
• Postgres host, port, and credentials
• Window size and slide interval
• KS and PSI thresholds
• PSI threshold calibration settings
• Prediction batch size and flush interval
• Artifact directory, model name, and model version
• ALERT_WEBHOOK_URL for Slack-compatible alert delivery

By default Postgres is exposed on host port 55432 to avoid collisions with an existing local Postgres instance on 5432 or 5433.

Benchmarks

Recorded results live in benchmarks/RESULTS.md. On a developer laptop:

• Synthetic 2σ drift detection: precision 0.98, recall 1.00, FPR 0.02 for both windowed KS+PSI and online consensus.
• End-to-end latency at 50 events/sec: p50 134 ms, p95 254 ms, p99 301 ms (batching-bounded by PREDICTION_FLUSH_INTERVAL_MS).
• Postgres ingest at batch size 1000: ~43 k rows/sec via the COPY + staging-table path.

Repo layout

drift/
  config.py            environment-backed settings
  data.py              CKD dataset loading and cleaning
  train.py             model training + reference + threshold calibration
  producer.py          event producer and drift injection CLI
  consumer.py          streaming detector
  detectors.py         windowed KS + PSI logic
  online_detectors.py  ADWIN + Page-Hinkley + KSWIN with consensus
  events.py            event schema
  serde.py             Avro serialization helpers
  db.py                Postgres persistence helpers
  webhook.py           optional alert webhook POST
  kafka_admin.py       Kafka topic management
  benchmark_ingest.py  Postgres write benchmark
benchmarks/
  accuracy_suite.py    precision/recall on synthetic streams
  latency_report.py    e2e latency from existing data
  load_test.py         throughput sweep
  RESULTS.md           recorded numbers
demo/
  run_drift_demo.sh
infra/
  docker-compose.yml
  postgres/init.sql
  grafana/
docs/
  JUNIOR_ENGINEER_GUIDE.md
  V2_ROADMAP.md
tests/
Dockerfile
.github/workflows/ci.yml

For a deeper walkthrough — every module, the math, the libraries, and the cross-cutting CS concepts — see docs/JUNIOR_ENGINEER_GUIDE.md.

Future work

• Categorical / chi-square drift (the example dataset is fully numeric)
• Cloud deployment with managed Kafka + Postgres
• A real model registry integration (currently we only enforce a bundle ↔ reference identity check at consumer startup)