Drift Monitoring and Auto-Retraining Pipeline

Monitor a fraud model with Evidently and Prometheus, retrain on harm confirmed by mature labels via Airflow with approval, and roll back on guardrails.

Architecture Overview & Problem Statement

This project builds a drift-monitored fraud prediction service that retrains itself safely. A FastAPI service exports metrics to Prometheus and logs every prediction. A daily monitor uses Evidently for feature drift, PSI for prediction drift, and recall on complete, label-mature cohorts for real harm. A policy turns those numbers into one decision. An Airflow DAG freezes its exact inputs, retrains, backtests and registers a challenger, waits for an approval, and promotes it through a resumable state machine. Guardrails roll back to the previous champion if the new one misbehaves live. Grafana shows it all on one dashboard.

The problem

On day 6 of a simulated festive sale, acme-shop's payments change: bigger baskets, more electronics, and a new fraud ring using stolen old cards for small daytime grocery and fuel buys, which the model learned to treat as safe. Every server dashboard stays green. Drift alerts fire on day 6, but chargebacks take three days to confirm fraud, so nobody can yet say whether the model is wrong or the customers are different. This project answers that properly: alert on drift, retrain only on confirmed harm, ask a human, and watch the new model after it ships.

The loop

flowchart TD S[FastAPI service
metrics + prediction log] --> M[Daily monitor
Evidently, PSI, recall] M --> J[(state/monitoring.json)] J --> P{Retrain policy} P -->|INVESTIGATE| H([Page a human]) P -->|RETRAIN| A[Airflow DAG
train, backtest, register] A --> W{Human approves?} W -->|yes| R[Promote + reload] R --> G[Guardrail next day] G -->|breach| B([Roll back]) classDef svc fill:#326CE5,color:#FFFFFF,stroke:#326CE5 classDef data fill:#EDE9FE,color:#1E293B,stroke:#8B5CF6 classDef human fill:#DBEAFE,color:#1E293B,stroke:#326CE5 classDef bad fill:#EF4444,color:#FFFFFF,stroke:#EF4444 class S,M,A,R,G svc class J data class P,W,H human class B bad

Why labels decide, not drift

sequenceDiagram participant D6 as Day 6 traffic participant Mon as Monitor participant Lab as Chargebacks D6->>Mon: day 6: features drift Mon-->>Mon: INVESTIGATE (no labels yet) Lab->>Mon: day 9: day-6 labels mature Mon-->>Mon: below floor, 1 of 2 days Lab->>Mon: day 10: day-7 labels mature Mon-->>Mon: 2 days below floor: RETRAIN

Drift is a reason to look, not a verdict. Bigger baskets during a sale are drift; they are not harm. Harm is missed fraud, and you only know about it once labels arrive.

Repository layout

◈ DIAGRAM

acme-fraud-monitor/
├── params.yaml                  # traffic, monitoring, policy, retrain, gate, guardrail
├── requirements.txt
├── src/acme_fraud/              # data.py (regimes), model.py, registry.py (MLflow, SQLite)
├── service/app.py               # FastAPI + Prometheus metrics + prediction log + /admin/reload
├── scripts/
│   ├── bootstrap.py  replay.py  monitor.py  metrics_exporter.py  retrain_policy.py
│   ├── run_dir.py  freeze_params.py  frozen.py  train_window.py  backtest.py  register_challenger.py
│   └── approve.py  promote.py  service_ctl.py  rollback_monitor.py
├── dags/acme_fraud_retrain.py   # the Airflow DAG
├── monitoring/                  # prometheus.yml, alerts.yml, grafana/ (dashboard + provisioning)
├── compose.yaml                 # Prometheus + Grafana for the laptop
└── tests/                       # test_policy.py, test_lifecycle.py

Prerequisites (by module title)

Model Monitoring and Drift Detection with Evidently. This project writes state/monitoring.json with the fields that module defines (data_quality_ok, drift_share, prediction_psi, recall_drop, the last one null until labels mature). The module was not available when this project was written, so the monitor here is self-contained: if your module's monitor differs, keep these field names and meanings.
Automated Retraining and ML Incident Response. Script names follow that module (retrain_policy.py, freeze_params.py, train_window.py, backtest.py, rollback_monitor.py), written fresh here and run as shown.
Monitoring and Logging: Prometheus to Tracing, ML Training Pipelines with Airflow, MLflow Experiment Tracking and Model Registry.
Tools: Python 3.12, Docker with Compose (for Prometheus and Grafana), curl.

Versions used

Tool	Version	Checked
Evidently / pandas	0.7.23 / 2.3.3	PyPI, 5 Nov 2026
Airflow (own venv)	3.3.2 with official constraints	PyPI and constraints file, 5 Nov 2026
Prometheus / Grafana	v3.15.1 / 13.3.0	GitHub tags, 5 Nov 2026
MLflow / scikit-learn / FastAPI	3.18.0 / 1.9.1 / 0.162.0	PyPI, 5 Nov 2026

Security
Evidently 0.7.23 fails to import with pandas 3.x (ImportError: cannot import name 'infer_dtype_from_object'), found while building this project. requirements.txt pins pandas 2.3.3. Recheck when Evidently releases a fix.

Tip
What was run for this guide, in one clean run on Python 3.12 in a Linux sandbox: bootstrap, 14 replayed days through the live service, the monitor and policy each day, Prometheus v3.15.1 scraping the service and exporter with all six alert rules loaded and three firing on day 10, the Airflow 3.3.2 DAG end to end with airflow dags test (the approval record written by a script, see Milestone 6), promotion through its state machine with a verified service reload, the guardrail on three days, four break-it exercises, and 19 tests. Outputs marked "real run" are from that run. Grafana was not run (its download is blocked in the sandbox): the dashboard JSON was written by hand and its PromQL queries were run against the live Prometheus. The Docker Compose file was not run either.

Milestone 1: Bootstrap the Data, the Model and the Registry

This milestone creates the synthetic traffic generator with a normal and a festive regime, trains v1 on normal data, saves the drift reference, and makes v1 the champion in a local MLflow registry.

Step 1: Set up and write the parameters

Bash

mkdir acme-fraud-monitor && cd acme-fraud-monitor
python3.12 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
export PYTHONPATH=src:scripts MLFLOW_DISABLE_AGENT_HINT=1 MLFLOW_DISABLE_TELEMETRY=true \
       MLFLOW_ENABLE_ARTIFACTS_PROGRESS_BAR=false ADMIN_TOKEN="$(openssl rand -hex 16)"

requirements.txt:

TEXT

scikit-learn==1.9.1
pandas==2.3.3   # Evidently 0.7.23 fails to import with pandas 3.x (checked 5 Nov 2026)
numpy==2.5.4
joblib==1.6.0
pyyaml==6.0.3
evidently==0.7.23
prometheus-client==0.26.0
fastapi==0.162.0
uvicorn==0.54.0
mlflow==3.18.0
pytest==9.2.0

params.yaml (every threshold written before the first replay; all illustrative):

YAML

seed: 42
traffic:
  rows_per_day: 4000
  festive_from_day: 6          # the sale starts on day 6 and data shifts
  festive_to_day: 11           # ...and ends after day 11
  label_delay_days: 3          # chargebacks confirm fraud ~3 days later
monitoring:                    # written before any replay; illustrative
  drift_share_alert: 0.5       # share of features Evidently flags as drifted
  prediction_psi_alert: 0.2
  min_live_recall: 0.55        # business floor on mature, complete cohorts: the same yardstick in every regime
  recall_drop_alert: 0.10      # informational: drop vs the serving version's own baseline (regime-specific)
policy:
  min_mature_positives: 50     # do not retrain on fewer confirmed frauds
  min_label_coverage: 0.95     # a mature day counts only if >= 95% of its payments were scored and labelled
  consecutive_days: 2          # recall must be down on 2 mature days in a row (one bad day can be noise)
  cooldown_days: 3
retrain:
  max_leaf_nodes: 3            # little new-regime data: keep the challenger simple
  recall_target: 0.60
gate:
  min_pr_auc_gain: 0.0         # challenger must not be worse than champion on the same rows
  max_flag_rate: 0.40          # fraud ops can review at most 40% of payments
guardrail:
  min_predictions: 300
  hard_max_flag_rate: 0.60      # never acceptable, whatever the regime
  flagged_rate_band: [0.5, 2.0]   # live flagged rate must stay within 0.5x to 2x of the backtest rate

Step 2: Write the data, model and registry modules

src/acme_fraud/data.py:

Python

"""Synthetic payments with a normal regime and a festive-sale regime.
 
Festive regime = data drift (bigger amounts, more electronics) AND concept drift:
fraud rings switch to old cards, small grocery and fuel buys in daytime: all things v1 learned as "safe".
"""
import hashlib
import json
from pathlib import Path
 
import numpy as np
import pandas as pd
import yaml
 
FEATURES = ["amount", "hour", "merchant_category", "country_mismatch", "card_age_days", "txns_last_24h"]
CATEGORIES = ["grocery", "electronics", "travel", "fuel", "fashion", "gaming"]
 
 
def load_params(path: str = "params.yaml") -> dict:
    return yaml.safe_load(Path(path).read_text())
 
 
def make_payments(rows: int, seed: int, regime: str = "normal", fraud_rate: float = 0.03,
                  id_prefix: str = "txn") -> pd.DataFrame:
    rng = np.random.default_rng(seed)
    festive = regime == "festive"
    cat_p = [0.15, 0.40, 0.10, 0.10, 0.15, 0.10] if festive else None
    df = pd.DataFrame({
        "txn_id": [f"{id_prefix}_{i:06d}" for i in range(rows)],
        "amount": np.round(rng.lognormal(4.4 if festive else 3.5, 1.0, rows), 2),
        "hour": rng.integers(0, 24, rows),
        "merchant_category": rng.choice(CATEGORIES, rows, p=cat_p),
        "country_mismatch": rng.binomial(1, 0.08, rows),
        "card_age_days": rng.integers(1, 3650, rows),
        "txns_last_24h": rng.poisson(3.0 if festive else 2.0, rows),
    })
    if festive:   # new fraud pattern: stolen OLD cards, small domestic grocery/fuel buys in daytime
        score = (1.2 * (df["card_age_days"] > 1500) + 0.9 * df["merchant_category"].isin(["grocery", "fuel"])
                 + 0.5 * df["hour"].between(10, 16) + 0.3 * np.minimum(df["txns_last_24h"], 10) / 2
                 - 0.4 * np.log1p(df["amount"]) / 5 + rng.normal(0, 1.0, rows))
    else:
        score = (0.9 * np.log1p(df["amount"]) / 5 + 1.2 * df["country_mismatch"]
                 + 0.6 * df["merchant_category"].isin(["electronics", "gaming"])
                 + 0.5 * ((df["hour"] <= 4) | (df["hour"] >= 23))
                 + 0.25 * np.minimum(df["txns_last_24h"], 8) / 2
                 - 0.6 * (df["card_age_days"] > 365) + rng.normal(0, 1.0, rows))
    df["is_fraud"] = (score >= np.quantile(score, 1 - fraud_rate)).astype(int)
    return df
 
 
def day_traffic(params: dict, day: int) -> pd.DataFrame:
    t = params["traffic"]
    regime = "festive" if t["festive_from_day"] <= day <= t["festive_to_day"] else "normal"
    return make_payments(params["traffic"]["rows_per_day"], params["seed"] + day, regime, id_prefix=f"d{day:02d}")
 
 
def sha_of(path: str | Path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
 
 
def write_json(path: str | Path, data: dict) -> None:
    """Atomic write: readers never see half a file."""
    p = Path(path); p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_suffix(p.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2))
    tmp.replace(p)

src/acme_fraud/model.py:

Python

"""Model recipe and threshold choice, shared by bootstrap and retraining."""
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import recall_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder
 
 
def build_model(max_leaf_nodes: int, seed: int) -> Pipeline:
    return Pipeline([
        ("pre", ColumnTransformer([("cat", OneHotEncoder(handle_unknown="ignore"), ["merchant_category"])],
                                  remainder="passthrough")),
        ("clf", HistGradientBoostingClassifier(max_leaf_nodes=max_leaf_nodes, class_weight="balanced",
                                               random_state=seed)),
    ])
 
 
def pick_threshold(y_true, proba, recall_target: float) -> float:
    for t in np.round(np.arange(0.99, -0.001, -0.01), 2):
        if recall_score(y_true, proba >= t) >= recall_target:
            return float(t)
    raise ValueError(f"no threshold reaches recall {recall_target}")

src/acme_fraud/registry.py (exact versions, an expected-champion check, verification and an audit line on every alias move):

Python

"""MLflow registry helpers: exact versions, an expected-champion check, verify, audit.
 
The registry is a local SQLite file so the whole project runs on a laptop.
"""
import os
from datetime import datetime, timezone
from pathlib import Path
 
import mlflow
from mlflow import MlflowClient
from mlflow.exceptions import MlflowException
 
NAME = "acme-fraud"
STATE = Path(os.environ.get("ACME_STATE", "state"))
os.environ.setdefault("MLFLOW_TRACKING_URI", f"sqlite:///{(STATE / 'mlflow.db').resolve()}")
TRUSTED = ["sklearn.ensemble._hist_gradient_boosting.predictor.TreePredictor"]   # we trained it ourselves
 
 
def client() -> MlflowClient:
    return MlflowClient()
 
 
def champion():
    try:
        return client().get_model_version_by_alias(NAME, "champion")
    except MlflowException as err:
        if err.error_code in ("INVALID_PARAMETER_VALUE", "RESOURCE_DOES_NOT_EXIST"):
            return None
        raise
 
 
def register(model, tags: dict, run_name: str) -> str:
    exp = mlflow.set_experiment("acme-fraud-monitor")
    with mlflow.start_run(run_name=run_name):
        info = mlflow.sklearn.log_model(model, name="model", skops_trusted_types=TRUSTED)
    mv = mlflow.register_model(info.model_uri, NAME, tags={k: str(v) for k, v in tags.items()})
    return str(mv.version)
 
 
def load(version: str):
    return mlflow.sklearn.load_model(f"models:/{NAME}/{version}")
 
 
def move_champion(version: str, mode: str, approved_by: str, expect: str | None, reason: str = "") -> str:
    """Point @champion at an exact version. Refuses if the champion is not what the caller expected."""
    c = client()
    cur = champion()
    cur_v = str(cur.version) if cur else "none"
    if expect is not None and expect != cur_v:
        raise RuntimeError(f"champion is {cur_v}, caller expected {expect}: re-approve")
    c.set_registered_model_alias(NAME, "champion", version)
    if str(champion().version) != str(version):                      # verify, do not assume
        raise RuntimeError("alias did not move")
    c.set_model_version_tag(NAME, version, "was_champion", "true")
    stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with open(STATE / "audit.log", "a") as f:
        old = "none" if cur_v == "none" else f"v{cur_v}"
        f.write(f"{stamp} {mode} champion {old} -> v{version} by={approved_by} {reason}\n".rstrip() + "\n")
    return cur_v

Step 3: Bootstrap v1

scripts/bootstrap.py:

Python

"""Day 0: train v1 on normal-regime data, save the drift reference, make v1 champion."""
import pandas as pd
from sklearn.model_selection import train_test_split
 
from acme_fraud import registry
from acme_fraud.data import FEATURES, load_params, make_payments, write_json
from acme_fraud.model import build_model, pick_threshold
from sklearn.metrics import recall_score
 
params = load_params()
df = make_payments(20000, params["seed"], "normal", id_prefix="ref")
train, rest = train_test_split(df, test_size=0.4, stratify=df["is_fraud"], random_state=params["seed"])
cal, ref = train_test_split(rest, test_size=0.5, stratify=rest["is_fraud"], random_state=params["seed"])
 
model = build_model(31, params["seed"]).fit(train[FEATURES], train["is_fraud"])
thr = pick_threshold(cal["is_fraud"], model.predict_proba(cal[FEATURES])[:, 1], 0.75)
ref = ref.assign(score=model.predict_proba(ref[FEATURES])[:, 1])
ref_recall = recall_score(ref["is_fraud"], ref["score"] >= thr)
flag_rate = float((ref["score"] >= thr).mean())
 
registry.STATE.mkdir(exist_ok=True)
ref.to_csv(registry.STATE / "reference.csv", index=False)           # drift baseline: features + scores
train.to_csv(registry.STATE / "reference_train.csv", index=False)   # retraining starts from this
v = registry.register(model, {"threshold": thr, "reference_recall": round(ref_recall, 4),
                              "expected_flag_rate": round(flag_rate, 4), "trained_on": "reference",
                              "baseline_regime": "normal"}, "bootstrap-v1")
registry.move_champion(v, "bootstrap", "bootstrap", expect="none")
write_json(registry.STATE / "policy_state.json", {"last_retrain_day": 0})
print(f"v{v} champion: threshold={thr} reference_recall={ref_recall:.3f} flag_rate={flag_rate:.3f}")

Bash

python scripts/bootstrap.py

Expected output (real run):

TEXT

v1 champion: threshold=0.38 reference_recall=0.683 flag_rate=0.253
{"status":"ready","model_version":"1","threshold":0.38}

v1 flags about a quarter of payments to reach 0.68 recall: the fraud signal in this synthetic data is weak on purpose, so the numbers stay humble.

Acceptance check

state/reference.csv holds features and v1's scores; state/audit.log has one bootstrap line.

Milestone 2: Serve the Model with Metrics and a Prediction Log

This milestone runs a FastAPI service that loads @champion, exposes Prometheus metrics per model version, logs every prediction for the monitor, and can reload the champion after a promotion or rollback.

Step 1: Write the service

service/app.py:

Python

"""acme-fraud prediction service with Prometheus metrics and a prediction log.
 
GET  /metrics         Prometheus scrape target
GET  /ready           which registry version is serving
POST /predict         score one payment; logged for the drift monitor
POST /admin/reload    re-read @champion (needs X-Admin-Token); used after promote or rollback
"""
import csv
import os
import threading
import time
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Literal
 
import pandas as pd
from fastapi import FastAPI, Header, HTTPException, Response
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Gauge, Histogram, generate_latest
from pydantic import BaseModel, Field
 
from acme_fraud import registry
from acme_fraud.data import FEATURES
 
LOG = Path(os.environ.get("PREDICTION_LOG", "logs/predictions.csv"))
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "")
STATE: dict = {}
MODEL_LOCK = threading.Lock()   # model and threshold are read and swapped together
LOG_LOCK = threading.Lock()     # one writer for the CSV; kept apart so I/O never blocks a reload
 
PREDICTIONS = Counter("acme_predictions_total", "Predictions served", ["model_version"])
FLAGGED = Counter("acme_flagged_total", "Predictions above the threshold", ["model_version"])
SCORE = Histogram("acme_fraud_score", "Predicted fraud probability", ["model_version"],
                  buckets=[0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0])
LATENCY = Histogram("acme_predict_seconds", "Whole request handling, including the log write")
INFERENCE = Histogram("acme_inference_seconds", "Model scoring only")
SERVING = Gauge("acme_model_serving", "1 for the registry version currently loaded", ["model_version"])
 
 
class Payment(BaseModel):
    txn_id: str = Field(min_length=1, max_length=64)
    amount: float = Field(gt=0, le=1_000_000)
    hour: int = Field(ge=0, le=23)
    merchant_category: Literal["grocery", "electronics", "travel", "fuel", "fashion", "gaming"]
    country_mismatch: Literal[0, 1]
    card_age_days: int = Field(ge=0, le=36_500)
    txns_last_24h: int = Field(ge=0, le=1_000)
 
 
def load_champion() -> None:
    mv = registry.champion()
    if mv is None:
        raise RuntimeError("no @champion in the registry")
    model = registry.load(str(mv.version))
    with MODEL_LOCK:
        if "version" in STATE:
            SERVING.labels(STATE["version"]).set(0)
        STATE.update(model=model, version=str(mv.version), threshold=float(mv.tags["threshold"]))
        SERVING.labels(STATE["version"]).set(1)
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    load_champion()
    LOG.parent.mkdir(parents=True, exist_ok=True)
    yield
 
 
app = FastAPI(title="acme-fraud-monitor", lifespan=lifespan)
 
 
@app.get("/metrics")
def metrics() -> Response:
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
 
 
@app.get("/ready")
def ready() -> dict:
    return {"status": "ready", "model_version": STATE["version"], "threshold": STATE["threshold"]}
 
 
@app.post("/admin/reload")
def reload(x_admin_token: str = Header(default="")) -> dict:
    if not ADMIN_TOKEN or x_admin_token != ADMIN_TOKEN:
        raise HTTPException(status_code=403, detail="bad admin token")
    load_champion()
    return {"reloaded": True, "model_version": STATE["version"]}
 
 
@app.post("/predict")
def predict(p: Payment, x_replay_day: int = Header(default=0)) -> dict:
    start = time.perf_counter()
    with MODEL_LOCK:                             # take a consistent snapshot of model + threshold
        model, version, thr = STATE["model"], STATE["version"], STATE["threshold"]
    t0 = time.perf_counter()
    proba = float(model.predict_proba(pd.DataFrame([p.model_dump()])[FEATURES])[0, 1])
    INFERENCE.observe(time.perf_counter() - t0)
    flagged = proba >= thr
    with LOG_LOCK:
        new_file = not LOG.exists()
        with open(LOG, "a", newline="") as f:    # the drift monitor reads this log
            w = csv.writer(f)
            if new_file:
                w.writerow(["day", "txn_id", *FEATURES, "score", "flagged", "model_version"])
            w.writerow([x_replay_day, p.txn_id, *[getattr(p, c) for c in FEATURES], round(proba, 5), int(flagged), version])
    PREDICTIONS.labels(version).inc()
    FLAGGED.labels(version).inc(flagged)
    SCORE.labels(version).observe(proba)
    LATENCY.observe(time.perf_counter() - start)
    return {"txn_id": p.txn_id, "fraud_probability": round(proba, 4), "is_fraud": flagged, "model_version": version}

Step 2: Run it and the replay tool

scripts/replay.py sends one simulated day through the service and files that day's labels with the day they become known:

Python

"""Send one simulated day of payments through the live service; file its labels for later.
 
Idempotent: a day already in the prediction log is refused, so a rerun cannot duplicate it.
Labels cover every payment that HAPPENED that day (written with the day they become known),
even if an outage meant only part of the day reached the service. Coverage checks catch that.
Exit codes: 0 = sent, 3 = day already replayed.
"""
import argparse
import json
import sys
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
 
import pandas as pd
 
from acme_fraud.data import FEATURES, day_traffic, load_params
 
ap = argparse.ArgumentParser()
ap.add_argument("--day", type=int, required=True)
ap.add_argument("--url", default="http://127.0.0.1:8000/predict")
ap.add_argument("--fraction", type=float, default=1.0, help="break-it: only part of the day reaches the service")
a = ap.parse_args()
params = load_params()
log = Path("logs/predictions.csv")
if log.exists() and (pd.read_csv(log, usecols=["day"])["day"] == a.day).any():
    print(f"REFUSED: day {a.day} is already in {log}"); sys.exit(3)
 
df = day_traffic(params, a.day)
sent = df.sample(frac=a.fraction, random_state=0) if a.fraction < 1 else df
 
 
def send(row) -> int:
    body = json.dumps({"txn_id": row.txn_id, **{c: getattr(row, c) for c in FEATURES}}, default=int).encode()
    req = urllib.request.Request(a.url, data=body, headers={"Content-Type": "application/json",
                                                            "X-Replay-Day": str(a.day)})
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.status
 
 
with ThreadPoolExecutor(8) as pool:
    codes = list(pool.map(send, sent.itertuples(index=False)))
known_on = a.day + params["traffic"]["label_delay_days"]
Path("labels").mkdir(exist_ok=True)
df[["txn_id", "is_fraud"]].assign(known_on_day=known_on).to_csv(f"labels/day-{a.day:02d}.csv", index=False)
t = params["traffic"]
regime = "festive" if t["festive_from_day"] <= a.day <= t["festive_to_day"] else "normal"
print(f"day {a.day} ({regime}): {codes.count(200)}/{len(df)} payments scored, labels known on day {known_on}")

Bash

uvicorn service.app:app --port 8000 &
curl -s localhost:8000/ready

Security
/admin/reload refuses every call unless ADMIN_TOKEN is set and matches. Keep the token in your environment, never in the repo. The prediction log holds payment features: in production it needs the same protection as the payments database.

Acceptance check

curl -s localhost:8000/metrics | grep acme_model_serving shows model_version="1" at 1.

Milestone 3: Monitor Drift, Prediction Drift and Mature Recall

This milestone writes the daily monitor that produces state/monitoring.json: data quality, Evidently's share of drifted features, PSI of prediction scores, and recall on the newest day whose labels have matured, only when that day's cohort is complete.

Step 1: Write the monitor

What each number means:

drift_share: share of the six features Evidently's DataDriftPreset flags against the original normal-regime reference.
prediction_psi: PSI of today's scores against the current champion's expected scores. v1's come from the bootstrap reference; a retrained version's come from its backtest holdout.
live_recall: recall on the newest mature day, attributed to the version that served that day (mature_day_served_version), computed only if the cohort is complete: one serving version, no duplicate transactions, and at least 95 percent of that day's labelled payments found in the prediction log (mature_label_coverage).
recall_drop: the serving version's own baseline recall minus live_recall, for context only. Baselines are regime-specific (baseline_regime is normal for v1 and shifted for a version trained on sale data), so the policy does not use it; it judges every regime against one business floor, min_live_recall.

The prediction log is de-duplicated on (day, txn_id) before anything is computed, and duplicates found fail that day's data quality.

scripts/monitor.py:

Python

"""Daily monitor: data quality, feature drift (Evidently), prediction drift (PSI), and recall
on the newest day whose labels have matured. Writes state/monitoring.json.
 
Drift is a reason to look, not a verdict: this script only measures. The retrain policy
decides what the numbers mean.
Exit codes: 0 = written, 4 = no predictions logged for that day.
"""
import argparse
import json
import warnings
from pathlib import Path
 
import numpy as np
import pandas as pd
from sklearn.metrics import recall_score
 
from acme_fraud import registry
from acme_fraud.data import CATEGORIES, FEATURES, load_params, write_json
 
warnings.filterwarnings("ignore", category=SyntaxWarning)
from evidently import DataDefinition, Dataset, Report  # noqa: E402
from evidently.presets import DataDriftPreset  # noqa: E402
 
NUMERIC = ["amount", "hour", "card_age_days", "txns_last_24h"]
CATEGORICAL = ["merchant_category", "country_mismatch"]
 
 
def load_log(path: str = "logs/predictions.csv") -> tuple[pd.DataFrame, int]:
    """Prediction log with one row per (day, txn_id). Returns the log and how many duplicates were dropped."""
    log = pd.read_csv(path)
    clean = log.drop_duplicates(subset=["day", "txn_id"], keep="last")
    return clean, len(log) - len(clean)
 
 
def psi(ref: np.ndarray, cur: np.ndarray, bins: int = 10) -> float:
    """Population Stability Index of cur against ref, on reference-quantile bins."""
    ref, cur = np.asarray(ref, float), np.asarray(cur, float)
    if len(ref) == 0 or len(cur) == 0:
        raise ValueError("PSI needs non-empty reference and current samples")
    edges = np.unique(np.quantile(ref, np.linspace(0, 1, bins + 1)))
    if len(edges) < 3:                       # (near) constant reference: "same value" vs "anything else"
        v = edges[0]
        edges = np.array([-np.inf, v - 1e-9, v + 1e-9, np.inf])
    else:
        edges[0], edges[-1] = -np.inf, np.inf
    r = np.histogram(ref, edges)[0] / len(ref)
    c = np.histogram(cur, edges)[0] / len(cur)
    r, c = np.clip(r, 1e-4, None), np.clip(c, 1e-4, None)
    r, c = r / r.sum(), c / c.sum()          # renormalise after clipping
    return float(np.sum((c - r) * np.log(c / r)))
 
 
def quality(cur: pd.DataFrame, expected_rows: int) -> list[str]:
    problems = []
    if cur[FEATURES].isna().any().any():
        problems.append("missing values")
    if (cur["amount"] <= 0).any() or not cur["hour"].between(0, 23).all():
        problems.append("values out of range")
    if not set(cur["merchant_category"]).issubset(CATEGORIES):
        problems.append("unknown merchant_category")
    if len(cur) < 0.5 * expected_rows:
        problems.append(f"only {len(cur)} rows, expected about {expected_rows}")
    return problems
 
 
def mature_recall(log: pd.DataFrame, labels: pd.DataFrame, mature_day: int, today: int,
                  min_coverage: float) -> dict:
    """Recall of whichever version served `mature_day`, only if the labelled cohort is complete."""
    lab = labels[labels["known_on_day"] <= today].drop_duplicates("txn_id")
    served = log[log["day"] == mature_day]
    joined = served.merge(lab, on="txn_id", how="inner", validate="one_to_one")
    coverage = len(joined) / len(lab) if len(lab) else 0.0
    out = {"mature_day": mature_day, "label_rows": int(len(lab)), "scored_rows": int(len(served)),
           "coverage": round(coverage, 4), "positives": int(joined["is_fraud"].sum()),
           "served_version": None, "recall": None, "cohort_ok": False}
    if joined.empty:
        return out
    versions = joined["model_version"].astype(str).unique()
    out["served_version"] = versions[0] if len(versions) == 1 else "mixed"
    out["cohort_ok"] = coverage >= min_coverage and len(versions) == 1
    out["recall"] = float(recall_score(joined["is_fraud"], joined["flagged"]))
    return out
 
 
def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--day", type=int, required=True)
    a = ap.parse_args()
    p = load_params()
    log, dupes = load_log()
    cur = log[log["day"] == a.day]
    if cur.empty:
        print(f"ERROR: no predictions logged for day {a.day}"); return 4
    ref = pd.read_csv(registry.STATE / "reference.csv")
    champ_v = str(registry.champion().version)
 
    problems = quality(cur, p["traffic"]["rows_per_day"]) + ([f"{dupes} duplicate log rows"] if dupes else [])
    dd = DataDefinition(numerical_columns=NUMERIC, categorical_columns=CATEGORICAL)
    snap = Report([DataDriftPreset()]).run(Dataset.from_pandas(cur[FEATURES], data_definition=dd),
                                           Dataset.from_pandas(ref[FEATURES], data_definition=dd))
    Path("reports").mkdir(exist_ok=True)
    snap.save_html(f"reports/drift-day-{a.day:02d}.html")
    metrics = snap.dict()["metrics"]
    drift_share = next(m["value"]["share"] for m in metrics if m["metric_name"].startswith("DriftedColumnsCount"))
    drifted = [m["config"]["column"] for m in metrics
               if m["metric_name"].startswith("ValueDrift") and m["value"] < m["config"]["threshold"]]
 
    # PSI: today's scores vs the CURRENT champion's own expected scores
    own = registry.STATE / f"reference_scores_v{champ_v}.csv"
    score_ref = (pd.read_csv(own) if own.exists() else ref)["score"].to_numpy()
 
    # recall on the newest mature day, attributed to the version that SERVED it
    md = a.day - p["traffic"]["label_delay_days"]
    mat = {"mature_day": None, "cohort_ok": False, "recall": None, "served_version": None,
           "coverage": None, "positives": 0}
    recall_drop, comparable, baseline_regime = None, None, None
    lab_file = Path(f"labels/day-{md:02d}.csv")
    if md >= 1 and lab_file.exists():
        mat = mature_recall(log, pd.read_csv(lab_file), md, a.day, p["policy"]["min_label_coverage"])
        if mat["cohort_ok"]:
            tags = registry.client().get_model_version(registry.NAME, mat["served_version"]).tags
            hist_file = registry.STATE / "history.jsonl"
            history = {h["day"]: h for h in map(json.loads, open(hist_file))} if hist_file.exists() else {}
            day_drifted = history.get(md, {}).get("drift_share", 0) >= p["monitoring"]["drift_share_alert"]
            baseline_regime = tags["baseline_regime"]              # "normal" or "shifted"
            comparable = (baseline_regime == "shifted") == day_drifted   # compare like with like
            recall_drop = round(float(tags["reference_recall"]) - mat["recall"], 4)
 
    out = {
        "day": a.day, "rows": int(len(cur)), "current_champion_version": champ_v,
        "data_quality_ok": not problems, "data_quality_problems": problems,
        "drift_share": round(float(drift_share), 4), "drifted_columns": drifted,
        "prediction_psi": round(psi(score_ref, cur["score"].to_numpy()), 4),
        "flag_rate": round(float(cur["flagged"].mean()), 4),
        "mature_day": mat["mature_day"], "mature_day_served_version": mat["served_version"],
        "mature_label_coverage": mat["coverage"], "mature_cohort_ok": mat["cohort_ok"],
        "mature_positives": mat["positives"], "baseline_regime": baseline_regime,
        "recall_comparable": comparable,
        "live_recall": None if mat["recall"] is None else round(mat["recall"], 4),
        "recall_drop": recall_drop,
    }
    write_json(registry.STATE / "monitoring.json", out)
    with open(registry.STATE / "history.jsonl", "a") as f:
        f.write(json.dumps(out) + "\n")
    print(f"day {a.day}: quality_ok={out['data_quality_ok']} drift_share={out['drift_share']} "
          f"psi={out['prediction_psi']} flag_rate={out['flag_rate']} | mature day {out['mature_day']}: "
          f"served v{out['mature_day_served_version']}, coverage {out['mature_label_coverage']}, "
          f"live_recall={out['live_recall']} (baseline drop {out['recall_drop']}, comparable={out['recall_comparable']})")
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(main())

Tip
Feature drift is always measured against the original normal-regime reference. That is deliberate: after the sale, drift falls back to zero when traffic returns to normal. Refresh this reference on a schedule, with a human signing off.

Step 2: Write the retrain policy

scripts/retrain_policy.py:

Python

"""Decide what the monitoring numbers mean. Writes state/decision.json.
 
A mature day counts as HARMED only if its labelled cohort is complete (coverage, one serving
version) and its live recall is below the business floor `min_live_recall`. The floor is the same
in every regime, so normal and festive days are judged by one yardstick. recall_drop against the
serving version's own baseline is reported, but that baseline is regime-specific, so it does not decide.
 
RETRAIN      the last `consecutive_days` mature days are all HARMED, consecutive, enough frauds
INVESTIGATE  drift or PSI high, harm not yet confirmed, or the newest cohort is incomplete
BLOCKED      today's data failed quality checks: never act on broken data
COOLDOWN     a retrain happened too recently
NONE         nothing to do
Exit codes: 0 = RETRAIN, 99 = no retrain (Airflow marks the task skipped), 1 = BLOCKED.
"""
import argparse
import json
 
from acme_fraud import registry
from acme_fraud.data import load_params, write_json
 
 
def harmed(h: dict, mon: dict) -> bool:
    return (h.get("mature_cohort_ok") is True and h.get("live_recall") is not None
            and h["live_recall"] < mon["min_live_recall"])
 
 
def decide(m: dict, p: dict, last_retrain_day: int, day: int, history: list[dict]) -> tuple[str, list[str]]:
    mon, pol = p["monitoring"], p["policy"]
    if not m["data_quality_ok"]:
        return "BLOCKED", m["data_quality_problems"]
    by_mature = {h["mature_day"]: h for h in history if h.get("mature_day") is not None}   # keyed by MATURE day
    need = pol["consecutive_days"]
    if m.get("mature_day") is not None:
        wanted = list(range(m["mature_day"] - need + 1, m["mature_day"] + 1))      # e.g. [6, 7]
        streak = [d for d in wanted if d in by_mature and harmed(by_mature[d], mon)]
        if harmed(m, mon):
            reasons = [f"live recall {m['live_recall']} < floor {mon['min_live_recall']} on mature day "
                       f"{m['mature_day']} (served by v{m['mature_day_served_version']}, "
                       f"coverage {m['mature_label_coverage']})"]
            if len(streak) < need:
                return "INVESTIGATE", reasons + [f"mature days harmed: {streak}, need {wanted}"]
            if sum(by_mature[d]["mature_positives"] for d in wanted) < pol["min_mature_positives"]:
                return "INVESTIGATE", reasons + ["too few confirmed frauds"]
            if day - last_retrain_day < pol["cooldown_days"]:
                return "COOLDOWN", reasons + [f"last retrain on day {last_retrain_day}"]
            return "RETRAIN", reasons + [f"harmed mature days {wanted}"]
        if m.get("mature_cohort_ok") is False:
            return "INVESTIGATE", [f"mature day {m['mature_day']} cohort incomplete "
                                   f"(coverage {m['mature_label_coverage']}): recall not trusted"]
    if m["drift_share"] >= mon["drift_share_alert"] or m["prediction_psi"] >= mon["prediction_psi_alert"]:
        return "INVESTIGATE", [f"drift_share {m['drift_share']}, psi {m['prediction_psi']}; "
                               "mature labels show no recall drop (yet)"]
    return "NONE", ["all within limits"]
 
 
def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--day", type=int, required=True)
    a = ap.parse_args()
    m = json.loads((registry.STATE / "monitoring.json").read_text())
    if m["day"] != a.day:
        print(f"ERROR: monitoring.json is for day {m['day']}, not {a.day}"); return 1
    last = json.loads((registry.STATE / "policy_state.json").read_text())["last_retrain_day"]
    lines = (registry.STATE / "history.jsonl").read_text().splitlines()
    history = list({h["day"]: h for h in map(json.loads, lines)}.values())   # last record per monitor day
    action, reasons = decide(m, load_params(), last, a.day, history)
    write_json(registry.STATE / "decision.json", {"day": a.day, "action": action, "reasons": reasons,
                                                   "champion_version": m["current_champion_version"]})
    print(f"day {a.day}: {action}: " + "; ".join(reasons))
    return {"RETRAIN": 0, "BLOCKED": 1}.get(action, 99)
 
 
if __name__ == "__main__":
    raise SystemExit(main())

The streak is checked on mature days, not monitor runs: mature days 6 and 7 must both be complete and below the floor, and a missing day breaks the streak. If the monitor misses a day, backfill it with monitor.py --day N before trusting the next decision.

Step 3: Write the tests

tests/test_policy.py:

Python

"""Retrain policy and PSI, tested without a service."""
import sys
from pathlib import Path
 
import numpy as np
import pytest
 
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
from monitor import psi  # noqa: E402
from retrain_policy import decide  # noqa: E402
 
from acme_fraud.data import load_params  # noqa: E402
 
P = load_params()
 
 
def mon(day, drift=0.0, psi_=0.0, drop=None, ok=True, cohort=True, comparable=True, pos=120):
    """drop is expressed against a 0.68 baseline, so live_recall = 0.68 - drop."""
    return {"day": day, "data_quality_ok": ok, "data_quality_problems": [] if ok else ["missing values"],
            "drift_share": drift, "prediction_psi": psi_, "recall_drop": drop, "mature_day": day - 3,
            "mature_cohort_ok": cohort if drop is not None else False, "recall_comparable": comparable,
            "mature_label_coverage": 1.0, "mature_positives": pos, "mature_day_served_version": "1",
            "baseline_regime": "normal", "live_recall": None if drop is None else round(0.68 - drop, 4)}
 
 
def test_drift_without_confirmed_harm_is_investigate():
    m = mon(6, drift=0.5, psi_=0.34, drop=-0.09)
    assert decide(m, P, 0, 6, [m])[0] == "INVESTIGATE"
 
 
def test_one_harmed_mature_day_is_not_enough():
    m = mon(9, drop=0.32)
    assert decide(m, P, 0, 9, [mon(8, drop=-0.05), m])[0] == "INVESTIGATE"
 
 
def test_two_consecutive_harmed_mature_days_retrain():
    m = mon(10, drop=0.32)
    assert decide(m, P, 0, 10, [mon(9, drop=0.23), m])[0] == "RETRAIN"
 
 
def test_gap_in_mature_days_is_not_a_streak():
    m = mon(10, drop=0.32)                       # mature day 7; day 6 never monitored
    assert decide(m, P, 0, 10, [mon(8, drop=0.30), m])[0] == "INVESTIGATE"   # mature 5 and 7, not 6 and 7
 
 
def test_incomplete_cohort_cannot_trigger_retrain():
    m = mon(10, drop=0.32, cohort=False)
    assert decide(m, P, 0, 10, [mon(9, drop=0.23), m])[0] == "INVESTIGATE"
 
 
def test_floor_is_the_yardstick_even_when_the_baseline_regime_differs():
    m = mon(10, drop=0.32, comparable=False)       # live recall 0.36, below 0.55 in any regime
    assert decide(m, P, 0, 10, [mon(9, drop=0.23, comparable=False), m])[0] == "RETRAIN"
 
 
def test_small_drop_above_the_floor_is_not_harm():
    m = mon(10, drop=0.10)                         # live recall 0.58 >= 0.55
    assert decide(m, P, 0, 10, [mon(9, drop=0.10), m])[0] == "NONE"
 
 
def test_broken_data_blocks_everything():
    m = mon(10, drop=0.5, ok=False)
    assert decide(m, P, 0, 10, [mon(9, drop=0.5), m])[0] == "BLOCKED"
 
 
def test_cooldown_after_a_recent_retrain():
    m = mon(11, drop=0.3)
    assert decide(m, P, 10, 11, [mon(10, drop=0.3), m])[0] == "COOLDOWN"
 
 
def test_psi_same_vs_shifted():
    rng = np.random.default_rng(0)
    a = rng.normal(0, 1, 5000)
    assert psi(a, rng.normal(0, 1, 5000)) < 0.02
    assert psi(a, rng.normal(1, 1, 5000)) > 0.2
 
 
def test_psi_constant_reference_and_repeated_values():
    assert psi(np.full(1000, 0.3), np.full(500, 0.3)) == pytest.approx(0.0, abs=1e-9)
    assert psi(np.full(1000, 0.3), np.full(500, 0.9)) > 1.0
    assert np.isfinite(psi(np.repeat([0.1, 0.2], 500), np.repeat([0.1, 0.2, 0.9], 300)))
 
 
def test_psi_extreme_shift_is_finite_and_large():
    v = psi(np.linspace(0, 1, 1000), np.full(1000, 50.0))
    assert np.isfinite(v) and v > 5
 
 
def test_psi_rejects_empty_input():
    with pytest.raises(ValueError):
        psi(np.array([]), np.array([1.0]))

tests/test_lifecycle.py (cohort coverage, duplicates, immature labels, a promotion that resumes after a failed reload, and an edited frozen input):

Python

"""Data integrity and promotion recovery, without the real service."""
import json
import os
import subprocess
import sys
import textwrap
from pathlib import Path
 
import pandas as pd
import pytest
 
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
from monitor import mature_recall  # noqa: E402
 
 
def cohort(n=100, served=100, dup=False):
    labels = pd.DataFrame({"txn_id": [f"t{i}" for i in range(n)], "is_fraud": [i % 10 == 0 for i in range(n)],
                           "known_on_day": 9})
    log = pd.DataFrame({"day": 6, "txn_id": [f"t{i}" for i in range(served)], "flagged": [i % 20 == 0 for i in range(served)],
                        "model_version": "1"})
    if dup:
        log = pd.concat([log, log.head(5)])
    return log, labels
 
 
def test_complete_cohort_gives_recall():
    out = mature_recall(*cohort(), mature_day=6, today=9, min_coverage=0.95)
    assert out["cohort_ok"] and out["coverage"] == 1.0 and out["recall"] == 0.5
 
 
def test_partial_cohort_is_not_trusted():
    out = mature_recall(*cohort(served=30), mature_day=6, today=9, min_coverage=0.95)
    assert not out["cohort_ok"] and out["coverage"] == 0.3
 
 
def test_duplicate_log_rows_are_rejected_not_double_counted():
    with pytest.raises(pd.errors.MergeError):          # callers must dedupe first (load_log does)
        mature_recall(*cohort(dup=True), mature_day=6, today=9, min_coverage=0.95)
 
 
def test_immature_labels_are_ignored():
    log, labels = cohort()
    out = mature_recall(log, labels.assign(known_on_day=12), mature_day=6, today=9, min_coverage=0.95)
    assert out["label_rows"] == 0 and not out["cohort_ok"]
 
 
HARNESS = textwrap.dedent('''
    import json, os, sys
    from pathlib import Path
    sys.path.insert(0, "scripts")
    import mlflow
    from sklearn.linear_model import LogisticRegression
    from acme_fraud import registry
    from acme_fraud.data import write_json
    m = LogisticRegression().fit([[0], [1]], [0, 1])
    for v in ("1", "2"):
        registry.register(m, {"threshold": 0.5}, f"t{v}")
    registry.move_champion("1", "bootstrap", "test", expect="none")
    d = Path("runs/test"); d.mkdir(parents=True)
    write_json(d / "manifest.json", {"champion_at_start": "1", "day": 10, "frozen_files": {}})
    write_json(d / "registered.json", {"version": "2"})
    write_json(d / "approval.json", {"version": "2", "approved_by_claimed": "alice"})
    import promote
    served = {"v": "1"}
    def bad_reload(): raise ConnectionError("service down")
    def good_reload(): served["v"] = str(registry.champion().version)
    print("first", promote.main(reload=bad_reload, serving=lambda: served["v"]),
          json.load(open(d / "promotion.json"))["state"])
    print("second", promote.main(reload=good_reload, serving=lambda: served["v"]),
          json.load(open(d / "promotion.json"))["state"],
          registry.client().get_model_version("acme-fraud", "2").tags["previous_champion"])
''')
 
 
def test_promotion_resumes_after_a_failed_reload(tmp_path):
    (tmp_path / "state").mkdir()
    env = {**os.environ, "ACME_STATE": str(tmp_path / "state"), "RUN_ID": "test",
           "PYTHONPATH": f"{ROOT / 'src'}:{ROOT / 'scripts'}", "MLFLOW_DISABLE_TELEMETRY": "true"}
    env.pop("MLFLOW_TRACKING_URI", None)
    r = subprocess.run([sys.executable, "-c", HARNESS], cwd=tmp_path, env=env, capture_output=True, text=True)
    lines = [l for l in r.stdout.splitlines() if l.startswith(("first", "second"))]
    assert lines == ["first 2 ALIAS_MOVED", "second 0 VERIFIED 1"], r.stdout + r.stderr
 
 
def test_frozen_input_edit_is_detected(tmp_path):
    from acme_fraud.data import sha_of, write_json
    from frozen import verified
    (tmp_path / "train.csv").write_text("a\n1\n")
    write_json(tmp_path / "manifest.json", {"frozen_files": {"train.csv": sha_of(tmp_path / "train.csv")}})
    assert verified(tmp_path, "train.csv")
    (tmp_path / "train.csv").write_text("a\n2\n")
    with pytest.raises(SystemExit):
        verified(tmp_path, "train.csv")

Acceptance check

pytest -q tests shows 19 passed.

Milestone 4: Export to Prometheus, Alert, and Build the Grafana Dashboard

This milestone exposes the monitor's numbers as Prometheus gauges, loads example alert rules, and provisions a Grafana dashboard. Request rates, latency and the serving version come from the service itself.

Step 1: Write the exporter

scripts/metrics_exporter.py:

Python

"""Expose state/monitoring.json as Prometheus gauges on :9101/metrics.
 
* recall_drop carries the version that SERVED the mature day; the other gauges carry the
  current champion. A missing recall (labels not mature, cohort incomplete) is not exported.
* acme_monitor_json_valid and acme_monitor_age_seconds tell you when the file is broken or stale.
"""
import json
import time
from pathlib import Path
 
from prometheus_client import start_http_server
from prometheus_client.core import REGISTRY, GaugeMetricFamily
 
STATE = Path("state/monitoring.json")
REQUIRED = {"day", "current_champion_version", "data_quality_ok", "drift_share", "prediction_psi",
            "flag_rate", "recall_drop", "mature_day_served_version"}
 
 
class MonitoringCollector:
    def collect(self):
        valid = GaugeMetricFamily("acme_monitor_json_valid", "1 if monitoring.json parsed and has every field")
        try:
            m = json.loads(STATE.read_text())
            ok = REQUIRED.issubset(m)
        except (OSError, ValueError):
            m, ok = {}, False
        valid.add_metric([], 1.0 if ok else 0.0)
        yield valid
        if not ok:
            return
        yield GaugeMetricFamily("acme_monitor_age_seconds", "Seconds since monitoring.json was written",
                                value=time.time() - STATE.stat().st_mtime)
        champ = m["current_champion_version"]
        for name, key, doc in [("acme_drift_share", "drift_share", "Share of features drifted (Evidently)"),
                               ("acme_prediction_psi", "prediction_psi", "PSI of scores vs the champion's baseline"),
                               ("acme_flag_rate", "flag_rate", "Share of payments flagged")]:
            g = GaugeMetricFamily(name, doc, labels=["model_version"])
            g.add_metric([champ], float(m[key]))
            yield g
        if m.get("live_recall") is not None:
            g = GaugeMetricFamily("acme_live_recall", "Recall on a complete mature day",
                                  labels=["model_version", "mature_day"])
            g.add_metric([str(m["mature_day_served_version"]), str(m["mature_day"])], float(m["live_recall"]))
            yield g
        if m["recall_drop"] is not None:
            g = GaugeMetricFamily("acme_recall_drop", "Baseline minus live recall on a complete mature day",
                                  labels=["model_version", "mature_day", "comparable"])
            g.add_metric([str(m["mature_day_served_version"]), str(m["mature_day"]),
                          str(m["recall_comparable"]).lower()], float(m["recall_drop"]))
            yield g
        yield GaugeMetricFamily("acme_data_quality_ok", "1 if the day passed data checks",
                                value=1.0 if m["data_quality_ok"] else 0.0)
        yield GaugeMetricFamily("acme_monitor_day", "Last monitored day", value=float(m["day"]))
 
 
if __name__ == "__main__":
    REGISTRY.register(MonitoringCollector())
    start_http_server(9101)
    print("exporter on :9101/metrics")
    while True:
        time.sleep(3600)

Step 2: Configure Prometheus and alerts

monitoring/prometheus.yml:

YAML

global:
  scrape_interval: 15s
  evaluation_interval: 15s
rule_files:
  - alerts.yml
scrape_configs:
  - job_name: acme-fraud-service        # requests, latency, scores, which version serves
    static_configs:
      - targets: ["host.docker.internal:8000"]
  - job_name: acme-fraud-monitor        # drift, PSI, recall drop from state/monitoring.json
    static_configs:
      - targets: ["host.docker.internal:9101"]

monitoring/alerts.yml (examples; tune on your own traffic):

YAML

# Example alert rules. Thresholds mirror params.yaml; tune them on your own traffic.
groups:
  - name: acme-fraud-ml
    rules:
      - alert: FeatureDriftHigh
        expr: max(acme_drift_share) >= 0.5
        labels: {severity: ticket}
        annotations:
          summary: "Half or more of the features drifted. Investigate; drift is not proof of harm."
      - alert: PredictionDriftHigh
        expr: max(acme_prediction_psi) >= 0.2
        labels: {severity: ticket}
        annotations:
          summary: "Score distribution moved (PSI {{ $value }}). Check flag volume with the fraud ops team."
      - alert: RecallBelowFloor
        expr: min(acme_live_recall) < 0.55
        labels: {severity: page}
        annotations:
          summary: "Live recall {{ $value }} on a complete mature day is below the 0.55 floor. Retrain policy will evaluate."
      - alert: DataQualityFailed
        expr: acme_data_quality_ok == 0
        labels: {severity: page}
        annotations:
          summary: "Today's traffic failed data checks. Retraining is blocked until fixed."
      - alert: MonitoringStale
        expr: acme_monitor_json_valid == 0 or acme_monitor_age_seconds > 26 * 3600
        labels: {severity: ticket}
        annotations:
          summary: "monitoring.json is broken or older than 26 hours: the daily monitor did not run."
      - alert: FlagRateSpike
        expr: >
          sum(rate(acme_flagged_total[10m])) / sum(rate(acme_predictions_total[10m])) > 0.6
        for: 5m
        labels: {severity: page}
        annotations:
          summary: "More than 60% of payments flagged for 5 minutes. Check the serving version and threshold."

Step 3: Provision Grafana

compose.yaml:

YAML

# Prometheus and Grafana for the laptop. The service (8000) and exporter (9101) run on the host.
services:
  prometheus:
    image: prom/prometheus:v3.15.1
    command: ["--config.file=/etc/prometheus/prometheus.yml"]
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - ./monitoring/alerts.yml:/etc/prometheus/alerts.yml:ro
    extra_hosts: ["host.docker.internal:host-gateway"]   # reach the host from the container (Linux too)
    ports: ["9090:9090"]
  grafana:
    image: grafana/grafana:13.3.0
    environment:
      GF_SECURITY_ADMIN_PASSWORD: change-me-locally     # local lab only
    volumes:
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro
      - ./monitoring/grafana/dashboard.json:/etc/grafana/dashboards/acme.json:ro
    ports: ["3000:3000"]

monitoring/grafana/provisioning/datasources/prometheus.yml:

YAML

apiVersion: 1
datasources:
  - name: Prometheus
    uid: prom
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true

monitoring/grafana/provisioning/dashboards/acme.yml:

YAML

apiVersion: 1
providers:
  - name: acme-fraud
    folder: MLOps
    type: file
    options:
      path: /etc/grafana/dashboards

monitoring/grafana/dashboard.json (8 panels: serving version, data quality, requests by version, drift share, PSI, recall drop, flag rate, p95 latency):

JSON

{
  "uid": "acme-fraud-ml",
  "title": "acme-fraud: model health",
  "schemaVersion": 41,
  "time": {
    "from": "now-1h",
    "to": "now"
  },
  "refresh": "30s",
  "tags": [
    "mlops",
    "acme-fraud"
  ],
  "panels": [
    {
      "id": 1,
      "type": "stat",
      "title": "Serving version",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 0,
        "y": 0,
        "w": 6,
        "h": 4
      },
      "targets": [
        {
          "refId": "A",
          "expr": "acme_model_serving == 1",
          "legendFormat": "v{{model_version}}",
          "instant": true
        }
      ],
      "options": {
        "textMode": "name",
        "reduceOptions": {
          "calcs": [
            "lastNotNull"
          ]
        }
      }
    },
    {
      "id": 2,
      "type": "stat",
      "title": "Data quality",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 6,
        "y": 0,
        "w": 6,
        "h": 4
      },
      "targets": [
        {
          "refId": "A",
          "expr": "acme_data_quality_ok",
          "instant": true
        }
      ],
      "fieldConfig": {
        "defaults": {
          "mappings": [
            {
              "type": "value",
              "options": {
                "0": {
                  "text": "FAILED",
                  "color": "red"
                },
                "1": {
                  "text": "OK",
                  "color": "green"
                }
              }
            }
          ]
        },
        "overrides": []
      }
    },
    {
      "id": 3,
      "type": "timeseries",
      "title": "Requests per second by version",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 12,
        "y": 0,
        "w": 12,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "sum by (model_version) (rate(acme_predictions_total[5m]))",
          "legendFormat": "v{{model_version}}"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "reqps"
        },
        "overrides": []
      }
    },
    {
      "id": 4,
      "type": "timeseries",
      "title": "Feature drift share (Evidently)",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 0,
        "y": 8,
        "w": 12,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "max(acme_drift_share)",
          "legendFormat": "drift share"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "percentunit",
          "thresholds": {
            "mode": "absolute",
            "steps": [
              {
                "color": "green",
                "value": null
              },
              {
                "color": "red",
                "value": 0.5
              }
            ]
          },
          "custom": {
            "thresholdsStyle": {
              "mode": "line"
            }
          }
        },
        "overrides": []
      }
    },
    {
      "id": 5,
      "type": "timeseries",
      "title": "Prediction PSI vs reference",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 12,
        "y": 8,
        "w": 12,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "max(acme_prediction_psi)",
          "legendFormat": "PSI"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "none",
          "thresholds": {
            "mode": "absolute",
            "steps": [
              {
                "color": "green",
                "value": null
              },
              {
                "color": "red",
                "value": 0.2
              }
            ]
          },
          "custom": {
            "thresholdsStyle": {
              "mode": "line"
            }
          }
        },
        "overrides": []
      }
    },
    {
      "id": 6,
      "type": "timeseries",
      "title": "Live recall on complete mature days (floor 0.55)",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 0,
        "y": 16,
        "w": 12,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "min(acme_live_recall)",
          "legendFormat": "v{{model_version}}"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "percentunit",
          "thresholds": {
            "mode": "absolute",
            "steps": [
              {
                "color": "red",
                "value": null
              },
              {
                "color": "green",
                "value": 0.55
              }
            ]
          },
          "custom": {
            "thresholdsStyle": {
              "mode": "line"
            }
          }
        },
        "overrides": []
      }
    },
    {
      "id": 7,
      "type": "timeseries",
      "title": "Flag rate (live)",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 12,
        "y": 16,
        "w": 12,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "sum(rate(acme_flagged_total[10m])) / sum(rate(acme_predictions_total[10m]))",
          "legendFormat": "flagged"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "percentunit",
          "thresholds": {
            "mode": "absolute",
            "steps": [
              {
                "color": "green",
                "value": null
              },
              {
                "color": "red",
                "value": 0.6
              }
            ]
          },
          "custom": {
            "thresholdsStyle": {
              "mode": "line"
            }
          }
        },
        "overrides": []
      }
    },
    {
      "id": 8,
      "type": "timeseries",
      "title": "p95 latency",
      "datasource": {
        "type": "prometheus",
        "uid": "prom"
      },
      "gridPos": {
        "x": 0,
        "y": 24,
        "w": 24,
        "h": 8
      },
      "targets": [
        {
          "refId": "A",
          "expr": "histogram_quantile(0.95, sum by (le) (rate(acme_predict_seconds_bucket[5m])))",
          "legendFormat": "p95"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "unit": "s"
        },
        "overrides": []
      }
    }
  ]
}

Bash

python scripts/metrics_exporter.py &
docker compose up -d
promtool check config monitoring/prometheus.yml     # or: docker compose exec prometheus promtool check config /etc/prometheus/prometheus.yml

Open Grafana at http://localhost:3000 (admin / the password in compose.yaml) and find MLOps, acme-fraud: model health.

Acceptance check

Prometheus Status, Targets shows both jobs UP, and acme_monitor_json_valid is 1.
The dashboard's "Serving version" panel shows v1.

Common Mistake
Exporting recall as 0 while labels are immature. A zero looks like "all good" on a dashboard. The exporter skips recall when it is null, so the panel shows a gap. Recall is labelled with the version that served the mature day, not whatever is champion today.

Milestone 5: Replay the Festive Sale and Watch the Policy

This milestone replays ten days, five normal and five festive, running the monitor and the policy each day, and checks what Prometheus sees on day 10.

Step 1: Replay ten days

Bash

for d in $(seq 1 10); do
  python scripts/replay.py --day $d
  python scripts/monitor.py --day $d
  python scripts/retrain_policy.py --day $d
done

Expected output (real run; replay lines for days 1 to 9 trimmed):

TEXT

day 1: quality_ok=True drift_share=0.0 psi=0.0044 flag_rate=0.2565 | mature day None: served vNone, coverage None, live_recall=None (baseline drop None, comparable=None)
   day 1: NONE: all within limits
day 2: quality_ok=True drift_share=0.0 psi=0.0039 flag_rate=0.2595 | mature day None: served vNone, coverage None, live_recall=None (baseline drop None, comparable=None)
   day 2: NONE: all within limits
day 3: quality_ok=True drift_share=0.0 psi=0.0034 flag_rate=0.2575 | mature day None: served vNone, coverage None, live_recall=None (baseline drop None, comparable=None)
   day 3: NONE: all within limits
day 4: quality_ok=True drift_share=0.0 psi=0.0045 flag_rate=0.2605 | mature day 1: served v1, coverage 1.0, live_recall=0.7417 (baseline drop -0.0584, comparable=True)
   day 4: NONE: all within limits
day 5: quality_ok=True drift_share=0.0 psi=0.0026 flag_rate=0.2572 | mature day 2: served v1, coverage 1.0, live_recall=0.7083 (baseline drop -0.025, comparable=True)
   day 5: NONE: all within limits
day 6: quality_ok=True drift_share=0.5 psi=0.3431 flag_rate=0.494 | mature day 3: served v1, coverage 1.0, live_recall=0.775 (baseline drop -0.0917, comparable=True)
   day 6: INVESTIGATE: drift_share 0.5, psi 0.3431; mature labels show no recall drop (yet)
day 7: quality_ok=True drift_share=0.5 psi=0.2953 flag_rate=0.4855 | mature day 4: served v1, coverage 1.0, live_recall=0.6667 (baseline drop 0.0166, comparable=True)
   day 7: INVESTIGATE: drift_share 0.5, psi 0.2953; mature labels show no recall drop (yet)
day 8: quality_ok=True drift_share=0.5 psi=0.3147 flag_rate=0.4895 | mature day 5: served v1, coverage 1.0, live_recall=0.7333 (baseline drop -0.05, comparable=True)
   day 8: INVESTIGATE: drift_share 0.5, psi 0.3147; mature labels show no recall drop (yet)
day 9: quality_ok=True drift_share=0.5 psi=0.3461 flag_rate=0.5015 | mature day 6: served v1, coverage 1.0, live_recall=0.45 (baseline drop 0.2333, comparable=False)
   day 9: INVESTIGATE: live recall 0.45 < floor 0.55 on mature day 6 (served by v1, coverage 1.0); mature days harmed: [6], need [5, 6]
day 10 (festive): 4000/4000 payments scored, labels known on day 13
day 10: quality_ok=True drift_share=0.5 psi=0.3474 flag_rate=0.5085 | mature day 7: served v1, coverage 1.0, live_recall=0.3667 (baseline drop 0.3166, comparable=False)
   day 10: RETRAIN: live recall 0.3667 < floor 0.55 on mature day 7 (served by v1, coverage 1.0); harmed mature days [6, 7]

Read it day by day:

Days 4 and 5: recall is measured for the first time (day 1 and 2 labels matured), on complete cohorts, above the 0.55 floor.
Days 6 to 8: half the features drift and PSI is far above 0.2, but the mature labels are still from normal days. The policy says INVESTIGATE: look, do not retrain.
Day 9: day-6 labels mature: recall 0.45, below the floor. One day can be noise, so still INVESTIGATE.
Day 10: mature days 6 and 7 are both complete and below the floor (0.45 and 0.37). RETRAIN.

comparable=False on days 9 and 10 says v1's normal-regime baseline is the wrong yardstick for sale traffic, which is why the decision uses the floor instead.

Step 2: Check what Prometheus sees on day 10

Bash

for q in 'max(acme_drift_share)' 'max(acme_prediction_psi)' 'min(acme_live_recall)' 'acme_model_serving'; do
  curl -s localhost:9090/api/v1/query --data-urlencode "query=$q"; echo
done
curl -s localhost:9090/api/v1/alerts

Expected output (real run, summarised):

◈ DIAGRAM

max(acme_drift_share) => [('', '0.5')]
max(acme_prediction_psi) => [('', '0.3474')]
min(acme_live_recall) => [('', '0.3667')]
acme_model_serving => [('1', '1')]
[('FeatureDriftHigh', 'firing'), ('PredictionDriftHigh', 'firing'), ('RecallBelowFloor', 'firing')]

Acceptance check

state/decision.json says RETRAIN for day 10, with the reason.
Three alerts are firing: two tickets (drift) and one page (RecallBelowFloor).

Milestone 6: Retrain Through Airflow with a Human Approval

This milestone runs an Airflow 3.3.2 DAG that re-checks the policy, freezes every input the run will use, trains a challenger on harmed mature days, backtests it against the champion on the same frozen rows, registers it, waits for an approval record, and promotes it through a resumable state machine that ends only when the service is verified.

Step 1: Write the retraining scripts

scripts/run_dir.py:

Python

"""Shared helper: every retraining run gets its own folder, named from RUN_ID."""
import os
import re
from pathlib import Path
 
 
def run_dir() -> Path:
    run_id = os.environ.get("RUN_ID")
    if not run_id:
        raise SystemExit("ERROR: set RUN_ID (Airflow passes {{ run_id }} through env)")
    d = Path("runs") / re.sub(r"[^A-Za-z0-9_.-]", "_", run_id)   # run ids contain ':' and '+'
    d.mkdir(parents=True, exist_ok=True)
    return d

scripts/freeze_params.py (freezes params, the decision, the exact training and holdout rows, checksums of every source file, the git commit and the installed packages):

Python

"""Freeze EVERYTHING this run will use, before any training starts.
 
Writes runs/<run>/: params.yaml, decision.json, train.csv, holdout.csv and manifest.json
(checksums, transaction counts, source-file checksums, git commit, installed packages).
Later tasks read only these files and verify the checksums first.
 
Training window = mature days the policy counted as HARMED (complete cohort, comparable recall,
drop above the alert) and whose own traffic passed data checks. Feature drift is not a criterion.
Holdout = the newest harmed mature day. Exit codes: 0 = frozen, 1 = refused, 3 = frozen copy edited.
"""
import json
import os
import shutil
import subprocess
import sys
 
import pandas as pd
 
from acme_fraud.data import load_params, sha_of, write_json
from monitor import load_log
from retrain_policy import harmed
from run_dir import run_dir
 
d = run_dir()
if (d / "manifest.json").exists():
    man = json.load(open(d / "manifest.json"))
    ok = all(sha_of(d / f) == h for f, h in man["frozen_files"].items())
    print("already frozen, checksums match" if ok else "ERROR: a frozen file was edited"); sys.exit(0 if ok else 3)
 
decision = json.load(open("state/decision.json"))
if decision["action"] != "RETRAIN":
    print(f"REFUSED: decision is {decision['action']}"); sys.exit(1)
p = load_params()
history = list({h["day"]: h for h in map(json.loads, open("state/history.jsonl"))}.values())
quality_by_day = {h["day"]: h["data_quality_ok"] for h in history}
harmed_days = sorted({h["mature_day"] for h in history
                      if harmed(h, p["monitoring"]) and quality_by_day.get(h["mature_day"], False)})
if len(harmed_days) < 2:
    print(f"REFUSED: need at least 2 harmed mature days (one to train, one to hold out), have {harmed_days}")
    sys.exit(1)
train_days, holdout_day = harmed_days[:-1], harmed_days[-1]
 
log, _ = load_log()                                        # one row per (day, txn_id)
labels = pd.concat(pd.read_csv(f"labels/day-{k:02d}.csv") for k in harmed_days)
labels = labels[labels["known_on_day"] <= decision["day"]].drop_duplicates("txn_id")
rows = log.merge(labels, on="txn_id", validate="one_to_one")
shutil.copy("params.yaml", d / "params.yaml")
write_json(d / "decision.json", decision)
rows[rows["day"].isin(train_days)].to_csv(d / "train.csv", index=False)
rows[rows["day"] == holdout_day].to_csv(d / "holdout.csv", index=False)
 
git = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True)
pkgs = subprocess.run([sys.executable, "-m", "pip", "freeze"], capture_output=True, text=True).stdout
(d / "packages.txt").write_text(pkgs)
frozen = ["params.yaml", "decision.json", "train.csv", "holdout.csv", "packages.txt"]
write_json(d / "manifest.json", {
    "run_id": os.environ["RUN_ID"], "day": decision["day"], "champion_at_start": decision["champion_version"],
    "train_days": train_days, "holdout_day": holdout_day,
    "train_rows": int(rows["day"].isin(train_days).sum()), "holdout_rows": int((rows["day"] == holdout_day).sum()),
    "frozen_files": {f: sha_of(d / f) for f in frozen},
    "source_files": {f: sha_of(f) for f in ["logs/predictions.csv"] + [f"labels/day-{k:02d}.csv" for k in harmed_days]},
    "git_commit": git.stdout.strip() if git.returncode == 0 else "not a git repository",
})
print(f"frozen {d.name}: train days {train_days} ({int(rows['day'].isin(train_days).sum())} rows), "
      f"holdout day {holdout_day}, params sha {sha_of(d / 'params.yaml')[:12]}")

scripts/frozen.py (every later task reads frozen files through this, which refuses an edited file):

Python

"""Read a frozen run input, refusing if it no longer matches the manifest."""
import json
from pathlib import Path
 
import pandas as pd
 
from acme_fraud.data import load_params, sha_of
 
 
def manifest(d: Path) -> dict:
    return json.load(open(d / "manifest.json"))
 
 
def verified(d: Path, name: str) -> Path:
    if sha_of(d / name) != manifest(d)["frozen_files"][name]:
        raise SystemExit(f"ERROR: {name} changed after it was frozen")
    return d / name
 
 
def params(d: Path) -> dict:
    return load_params(verified(d, "params.yaml"))
 
 
def table(d: Path, name: str) -> pd.DataFrame:
    return pd.read_csv(verified(d, name))

scripts/train_window.py (trains on the frozen rows only):

Python

"""Train the challenger on the FROZEN training rows only.
Exit codes: 0 = trained, 1 = not enough confirmed frauds.
"""
import sys
 
import joblib
from sklearn.model_selection import train_test_split
 
from acme_fraud.data import FEATURES, write_json
from acme_fraud.model import build_model, pick_threshold
from frozen import manifest, params, table
from run_dir import run_dir
 
d = run_dir()
p, man, train = params(d), manifest(d), table(d, "train.csv")
if train["is_fraud"].sum() < p["policy"]["min_mature_positives"]:
    print(f"REFUSED: {train['is_fraud'].sum()} frauds in the frozen training rows"); sys.exit(1)
fit, cal = train_test_split(train, test_size=0.25, stratify=train["is_fraud"], random_state=p["seed"])
model = build_model(p["retrain"]["max_leaf_nodes"], p["seed"]).fit(fit[FEATURES], fit["is_fraud"])
thr = pick_threshold(cal["is_fraud"], model.predict_proba(cal[FEATURES])[:, 1], p["retrain"]["recall_target"])
joblib.dump(model, d / "challenger.joblib")
write_json(d / "train.json", {"train_days": man["train_days"], "holdout_day": man["holdout_day"],
                              "rows": int(len(train)), "frauds": int(train["is_fraud"].sum()), "threshold": thr})
print(f"challenger trained on frozen days {man['train_days']} ({len(train)} rows, "
      f"{int(train['is_fraud'].sum())} frauds), threshold {thr}")

scripts/backtest.py (the gate: better ranking, enough recall, and a flag rate the review team can handle; decisions use full-precision metrics):

Python

"""Champion vs challenger on the SAME frozen holdout rows. The gate.
 
Decisions use full-precision metrics; rounding happens only when printing.
Exit codes: 0 = challenger passes, 1 = it does not.
"""
import json
import sys
 
import joblib
import pandas as pd
from sklearn.metrics import average_precision_score, recall_score
 
from acme_fraud import registry
from acme_fraud.data import FEATURES, write_json
from frozen import manifest, params, table
from run_dir import run_dir
 
d = run_dir()
p, man, hold = params(d), manifest(d), table(d, "holdout.csv")
thr_new = float(json.load(open(d / "train.json"))["threshold"])
champ_v = man["champion_at_start"]
champ_thr = float(registry.client().get_model_version(registry.NAME, champ_v).tags["threshold"])
 
 
def score(model, thr):
    proba = model.predict_proba(hold[FEATURES])[:, 1]
    return proba, {"pr_auc": float(average_precision_score(hold["is_fraud"], proba)),
                   "recall": float(recall_score(hold["is_fraud"], proba >= thr)),
                   "flag_rate": float((proba >= thr).mean())}
 
 
_, c = score(registry.load(champ_v), champ_thr)
chall_scores, n = score(joblib.load(d / "challenger.joblib"), thr_new)
pd.DataFrame({"score": chall_scores}).to_csv(d / "challenger_scores.csv", index=False)   # PSI baseline if promoted
checks = {
    "better ranking on the same rows": n["pr_auc"] >= c["pr_auc"] + p["gate"]["min_pr_auc_gain"],
    "catches enough fraud": n["recall"] >= p["retrain"]["recall_target"],
    "flag rate within review capacity": n["flag_rate"] <= p["gate"]["max_flag_rate"],
}
ok = all(checks.values())
write_json(d / "backtest.json", {"holdout_day": man["holdout_day"], "rows": int(len(hold)),
                                 "frauds": int(hold["is_fraud"].sum()), "champion": {"version": champ_v, **c},
                                 "challenger": n, "checks": checks, "passed": ok})
r = lambda m: {k: round(v, 4) for k, v in m.items()}   # noqa: E731  (display only)
print(f"holdout day {man['holdout_day']} ({len(hold)} rows, {int(hold['is_fraud'].sum())} frauds)")
print(f"  champion v{champ_v}: {r(c)}")
print(f"  challenger : {r(n)}")
for name, passed in checks.items():
    print(f"  {'PASS' if passed else 'FAIL'}  {name}")
print("GATE PASSED" if ok else "GATE FAILED")
sys.exit(0 if ok else 1)

scripts/register_challenger.py:

Python

"""Register the challenger that passed the backtest as an exact registry version.
Its baseline (recall, flag rate, score distribution) comes from the frozen holdout, which is
post-shift traffic, so it is tagged baseline_regime=shifted. Idempotent per run.
Exit codes: 0 = registered, 1 = gate not passed.
"""
import json
import os
import shutil
import sys
 
import joblib
 
from acme_fraud import registry
from acme_fraud.data import write_json
from frozen import manifest
from run_dir import run_dir
 
d = run_dir()
if (d / "registered.json").exists():
    print(f"already registered: v{json.load(open(d / 'registered.json'))['version']}"); sys.exit(0)
bt, tr, man = json.load(open(d / "backtest.json")), json.load(open(d / "train.json")), manifest(d)
if not bt["passed"]:
    print("REFUSED: backtest did not pass"); sys.exit(1)
v = registry.register(joblib.load(d / "challenger.joblib"), {
    "threshold": tr["threshold"], "run_id": os.environ["RUN_ID"],
    "params_sha256": man["frozen_files"]["params.yaml"], "train_days": man["train_days"],
    "backtest_pr_auc": bt["challenger"]["pr_auc"],
    "reference_recall": bt["challenger"]["recall"],          # baseline for recall_drop on comparable days
    "expected_flag_rate": bt["challenger"]["flag_rate"],     # the guardrail's yardstick
    "baseline_regime": "shifted", "baseline_day": man["holdout_day"],
}, f"retrain-{d.name}")
shutil.copy(d / "challenger_scores.csv", registry.STATE / f"reference_scores_v{v}.csv")
write_json(d / "registered.json", {"version": v})
print(f"registered challenger as v{v} (baseline: shifted regime, day {man['holdout_day']}); waiting for approval")

scripts/approve.py (a manual approval simulation: it writes an approval record for one exact version and does not authenticate anyone):

Python

"""MANUAL APPROVAL SIMULATION for the laptop lab.
 
Writes an approval record for one exact version of one run, after printing the backtest.
It does NOT authenticate anyone: --by is whatever you type, and the OS user and host are
recorded beside it only as a hint. In production, use Airflow's HITL ApprovalOperator or an
approval UI behind real login, and record the verified identity from there.
"""
import argparse
import getpass
import json
import socket
import sys
from datetime import datetime, timezone
from pathlib import Path
 
from acme_fraud.data import write_json
 
ap = argparse.ArgumentParser()
ap.add_argument("--run-dir", required=True)
ap.add_argument("--version", required=True)
ap.add_argument("--by", required=True)
a = ap.parse_args()
d = Path(a.run_dir)
reg = json.load(open(d / "registered.json"))
if str(reg["version"]) != a.version:
    print(f"REFUSED: this run registered v{reg['version']}, not v{a.version}"); sys.exit(1)
print(json.dumps(json.load(open(d / "backtest.json")), indent=2))   # what you are approving
write_json(d / "approval.json", {"version": a.version, "approved_by_claimed": a.by,
                                 "os_user": getpass.getuser(), "host": socket.gethostname(),
                                 "authenticated": False,
                                 "approved_at": datetime.now(timezone.utc).isoformat(timespec="seconds")})
print(f"approval record written for v{a.version} (claimed by {a.by}, not authenticated)")

scripts/service_ctl.py:

Python

"""Talk to the live service: ask it to reload @champion, and ask what it actually serves."""
import json
import os
import urllib.request
 
BASE = os.environ.get("SERVICE_URL", "http://127.0.0.1:8000")
 
 
def reload() -> None:
    req = urllib.request.Request(f"{BASE}/admin/reload", method="POST",
                                 headers={"X-Admin-Token": os.environ.get("ADMIN_TOKEN", "")})
    urllib.request.urlopen(req, timeout=60).read()
 
 
def serving() -> str:
    return str(json.load(urllib.request.urlopen(f"{BASE}/ready", timeout=10))["model_version"])

scripts/promote.py (a resumable state machine: APPROVED, PROMOTION_STARTED, ALIAS_MOVED, SERVICE_RELOADED, VERIFIED; the previous champion is recorded before the alias moves, and policy state changes only after verification):

Python

"""Promote the approved exact version as a resumable state machine.
 
APPROVED -> PROMOTION_STARTED -> ALIAS_MOVED -> SERVICE_RELOADED -> VERIFIED
State lives in runs/<run>/promotion.json. The previous champion is recorded BEFORE the alias
moves. Rerunning after any failure continues from the last completed state. Nothing reports
success, and policy state is not updated, until the service is verified serving the new version.
Exit codes: 0 = VERIFIED, 1 = refused, 2 = stopped part way (rerun to finish, or roll back).
"""
import json
import sys
 
from acme_fraud import registry
from acme_fraud.data import write_json
from frozen import manifest
from run_dir import run_dir
import service_ctl
 
 
def main(reload=service_ctl.reload, serving=service_ctl.serving) -> int:
    d = run_dir()
    reg, man = json.load(open(d / "registered.json")), manifest(d)
    v = str(reg["version"])
    if not (d / "approval.json").exists():
        print("REFUSED: no approval record for this run"); return 1
    appr = json.load(open(d / "approval.json"))
    if str(appr["version"]) != v:
        print(f"REFUSED: approval is for v{appr['version']}, run registered v{v}"); return 1
    sf = d / "promotion.json"
    st = json.load(open(sf)) if sf.exists() else {"state": "APPROVED", "version": v}
 
    def save(state: str, **extra) -> None:
        st.update(state=state, **extra)
        write_json(sf, st)
 
    if st["state"] == "APPROVED":
        cur = registry.champion()
        cur_v = str(cur.version) if cur else "none"
        if cur_v not in (man["champion_at_start"], v):
            print(f"REFUSED: champion is v{cur_v}, approver saw v{man['champion_at_start']}: re-approve"); return 1
        registry.client().set_model_version_tag(registry.NAME, v, "previous_champion", man["champion_at_start"])
        save("PROMOTION_STARTED", previous=man["champion_at_start"])
    if st["state"] == "PROMOTION_STARTED":
        if str(registry.champion().version) != v:            # a crash right after the move is fine too
            registry.move_champion(v, "promote", appr["approved_by_claimed"], expect=st["previous"],
                                   reason=f"run={d.name} approval=simulated")
        save("ALIAS_MOVED")
    if st["state"] == "ALIAS_MOVED":
        try:
            reload()
        except Exception as err:
            print(f"STOPPED at ALIAS_MOVED: @champion is v{v} but the service did not reload ({err}). "
                  "Rerun promote to finish, or roll back."); return 2
        save("SERVICE_RELOADED")
    if st["state"] == "SERVICE_RELOADED":
        now = serving()
        if now != v:
            print(f"STOPPED at SERVICE_RELOADED: service serves v{now}, expected v{v}. Rerun promote."); return 2
        write_json("state/policy_state.json", {"last_retrain_day": man["day"], "promoted_version": v})
        save("VERIFIED")
    print(f"promoted: champion v{st['previous']} -> v{v}, state {st['state']}, service verified serving v{v}")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())

Tip
The backtest gate has a flag-rate ceiling for a reason found while building this project. A first challenger with 15 leaves and a 0.75 recall target passed "better PR-AUC" and "enough recall" with a threshold of 0.01, flagging 60 percent of all payments. Recall alone is easy to buy by flagging everything. The parameters were then changed to 3 leaves, 0.60 recall and a 40 percent ceiling, and the whole run was repeated from day 1.

Step 2: Write the DAG

dags/acme_fraud_retrain.py:

Python

"""Retrain acme-fraud when the policy says so, then wait for a human before promoting.
 
Trigger:  airflow dags trigger acme_fraud_retrain --conf '{"day": 10}'
Every task runs the project's own scripts in the project's own venv, so the
Airflow environment and the ML libraries never fight over versions.
"""
from datetime import timedelta
from pathlib import Path
 
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.sensors.python import PythonSensor
from airflow.sdk import DAG, Param
 
PROJECT = Path(__file__).resolve().parents[1]
ENV = {
    "PATH": f"{PROJECT}/.venv/bin:/usr/bin:/bin",
    "PYTHONPATH": "src:scripts",
    "RUN_ID": "{{ run_id }}",            # templated values reach scripts as env vars, never pasted into code
    "DAY": "{{ params.day }}",
    "MLFLOW_DISABLE_AGENT_HINT": "1",
    "MLFLOW_DISABLE_TELEMETRY": "true",
}
 
 
def step(task_id: str, cmd: str, **kw) -> BashOperator:
    return BashOperator(task_id=task_id, bash_command=cmd, cwd=str(PROJECT), env=ENV,
                        append_env=True, retries=0, **kw)   # bad data and bad models are not retried
 
 
def approved(run_id: str) -> bool:
    import re
    return (PROJECT / "runs" / re.sub(r"[^A-Za-z0-9_.-]", "_", run_id) / "approval.json").exists()
 
 
with DAG(
    dag_id="acme_fraud_retrain",
    schedule=None,                        # triggered by the monitor, not by the clock
    catchup=False,
    max_active_runs=1,                    # one retrain at a time
    params={"day": Param(0, type="integer", minimum=1)},
    tags=["mlops", "acme-fraud"],
) as dag:
    policy = step("check_policy", 'python scripts/retrain_policy.py --day "$DAY"')   # exit 99 = skip the rest
    freeze = step("freeze_params", "python scripts/freeze_params.py")
    train = step("train_challenger", "python scripts/train_window.py")
    backtest = step("backtest_gate", "python scripts/backtest.py")
    register = step("register_challenger", "python scripts/register_challenger.py")
    wait = PythonSensor(
        task_id="wait_for_approval",
        python_callable=approved,
        op_kwargs={"run_id": "{{ run_id }}"},
        mode="reschedule",                # frees the worker slot while waiting
        poke_interval=10,
        timeout=timedelta(days=2).total_seconds(),
    )
    promote = step("promote", "python scripts/promote.py")
 
    policy >> freeze >> train >> backtest >> register >> wait >> promote

Design choices:

Two environments. Airflow runs in its own venv with its official constraints file; every task calls the project's venv. Airflow 3.3.2's constraints pin pandas 3.0.5, which Evidently cannot import, so sharing one environment would break the monitor.
Skip, do not fail. retrain_policy.py exits 99 when there is nothing to do, which BashOperator treats as skipped.
No retries. Bad data and bad models are not transient; retrying them only hides the failure.
Approval as a sensor. wait_for_approval waits for approval.json in reschedule mode, so it holds no worker while someone reads the backtest. This is a lab simulation: the record says who claims to approve, not who did. In production use Airflow 3.3's Human-in-the-loop ApprovalOperator (airflow.providers.standard.operators.hitl) or an approval UI behind real login, and record the verified identity. The file sensor is used here because a script can drive it inside airflow dags test.

Step 3: Run the DAG

Airflow lives in its own venv, created with the constraints file:

Bash

python3.12 -m venv ../airflow-venv
../airflow-venv/bin/pip install "apache-airflow==3.3.2" \
  -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt
export AIRFLOW_HOME=$PWD/../airflow-home AIRFLOW__CORE__DAGS_FOLDER=$PWD/dags AIRFLOW__CORE__LOAD_EXAMPLES=False
../airflow-venv/bin/airflow db migrate
../airflow-venv/bin/airflow dags test acme_fraud_retrain -c '{"day": 10}'

When register_challenger finishes, the run waits. In a second terminal, read the backtest and write the approval record for the exact version:

Bash

python scripts/approve.py --run-dir runs/<the run folder> --version 2 --by "$(git config user.name)"

Expected task output (real run, from the Airflow log, timestamps removed; the approval record was written by a script):

◈ DIAGRAM

day 10: RETRAIN: live recall 0.3667 < floor 0.55 on mature day 7 (served by v1, coverage 1.0); harmed mature days [6, 7]
frozen manual__2026-11-05T06_07_38.898430_00_00: train days [6] (4000 rows), holdout day 7, params sha b5b7342435b3
challenger trained on frozen days [6] (4000 rows, 120 frauds), threshold 0.37
holdout day 7 (4000 rows, 120 frauds)
  champion v1: {'pr_auc': 0.0265, 'recall': 0.3667, 'flag_rate': 0.4855}
  challenger : {'pr_auc': 0.1378, 'recall': 0.8167, 'flag_rate': 0.3422}
  PASS  better ranking on the same rows
  PASS  catches enough fraud
  PASS  flag rate within review capacity
GATE PASSED
Created version '2' of model 'acme-fraud'.
registered challenger as v2 (baseline: shifted regime, day 7); waiting for approval
approval record written for v2 (claimed by kushagra, not authenticated)
promoted: champion v1 -> v2, state VERIFIED, service verified serving v2
DAG run: 7 of 7 tasks succeeded, state=success

Day 10's backtest compares both models on the frozen day-7 rows, which neither trained on: the champion's PR-AUC there is barely above the 3 percent fraud base rate, the challenger's is about five times higher, and it flags fewer payments.

Acceptance check

All 7 tasks succeeded, and runs/<run>/manifest.json lists checksums for params.yaml, decision.json, train.csv, holdout.csv and packages.txt, the source files and the git commit.
runs/<run>/promotion.json says VERIFIED, approval.json says "authenticated": false, and /ready shows version 2.

Common Mistake
Choosing training days by feature drift. Concept drift can happen with no feature drift at all, and drifted days can still have broken data. The window here is the harmed mature days (complete cohort, recall below the floor) whose own traffic passed data checks. Training on everything since day 1 is the other trap: the normal days drown the new pattern.

Milestone 7: Watch the New Champion with Guardrails

This milestone runs two guardrails after each new day and rolls back to the recorded previous champion, with a verified service reload, if either fires.

Step 1: Write the guardrails

Operational, same day. The flag rate must stay under a hard ceiling (60 percent) in every regime, and within 0.5 to 2 times the backtest flag rate when today's regime matches the baseline's. This catches broken thresholds and serving bugs. It does not measure model quality: a model can flag a normal share and still miss fraud.
Performance, days later. Any complete mature day served by the champion with recall below the floor also triggers a rollback.

scripts/rollback_monitor.py:

Python

"""Guardrails after a promotion, two kinds.
 
1. Operational (same day): the flag rate must stay under a hard ceiling in every regime, and
   within a band around the backtest flag rate when today's regime matches the baseline's
   (drifted vs not). This catches broken thresholds and serving bugs. It does NOT measure model
   quality: a model can flag a normal share and still miss fraud.
2. Performance (days later): any complete mature day served by the champion with live recall
   below the business floor also triggers rollback.
On breach: move @champion back to the recorded previous champion, reload, verify.
Exit codes: 0 = healthy, 1 = rolled back and verified, 2 = not enough data yet,
            3 = alias rolled back but service not verified (rerun or fix by hand).
"""
import argparse
import json
import sys
 
from acme_fraud import registry
from acme_fraud.data import load_params
from monitor import load_log
from retrain_policy import harmed
import service_ctl
 
ap = argparse.ArgumentParser()
ap.add_argument("--day", type=int, required=True, help="judge this day's live traffic")
day = ap.parse_args().day
P = load_params()
g = P["guardrail"]
champ = registry.champion()
v, prev = str(champ.version), champ.tags.get("previous_champion")
log, _ = load_log()
live = log[(log["model_version"].astype(str) == v) & (log["day"] == day)]
 
history = [json.loads(line) for line in open(registry.STATE / "history.jsonl")]
today = {h["day"]: h for h in history}.get(day, {})
breaches = []
if len(live) >= g["min_predictions"]:
    expected, seen = float(champ.tags["expected_flag_rate"]), float(live["flagged"].mean())
    lo, hi = g["flagged_rate_band"][0] * expected, g["flagged_rate_band"][1] * expected
    today_shifted = today.get("drift_share", 0) >= P["monitoring"]["drift_share_alert"]
    same_regime = (champ.tags["baseline_regime"] == "shifted") == today_shifted
    print(f"day {day}, v{v}: flag rate {seen:.3f} on {len(live)} payments; hard ceiling {g['hard_max_flag_rate']}; "
          f"band {lo:.3f} to {hi:.3f} ({champ.tags['baseline_regime']} baseline, "
          f"{'applies' if same_regime else 'not applied: today is a different regime'})")
    if seen > g["hard_max_flag_rate"]:
        breaches.append(f"flag rate {seen:.3f} above hard ceiling")
    elif same_regime and not lo <= seen <= hi:
        breaches.append(f"flag rate {seen:.3f} outside backtest band")
else:
    print(f"day {day}, v{v}: only {len(live)} live predictions, operational check needs {g['min_predictions']}")
bad = [h["mature_day"] for h in history if h.get("mature_day_served_version") == v and harmed(h, P["monitoring"])]
if bad:
    breaches.append(f"live recall below floor on mature days {sorted(set(bad))}")
 
if not breaches:
    enough = len(live) >= g["min_predictions"]
    print("GUARDRAIL OK" if enough else "WAIT"); sys.exit(0 if enough else 2)
if not prev:
    print(f"GUARDRAIL BREACHED ({'; '.join(breaches)}) and no previous champion: needs a human"); sys.exit(1)
registry.move_champion(prev, "rollback", "guardrail", expect=v, reason="; ".join(breaches))
try:
    service_ctl.reload()
    now = service_ctl.serving()
except Exception as err:
    print(f"rollback initiated (alias moved to v{prev}), service NOT verified: {err}"); sys.exit(3)
if now != prev:
    print(f"rollback initiated (alias moved to v{prev}), but service serves v{now}"); sys.exit(3)
print(f"GUARDRAIL BREACHED ({'; '.join(breaches)}): rolled back v{v} -> v{prev}, service verified serving v{now}")
print("No page was sent: this lab has no alert routing. Wire the exit code or the audit log to Alertmanager.")
sys.exit(1)

Step 2: Day 11 (still the sale) and day 12 (the sale ends)

Bash

for d in 11 12; do
  python scripts/replay.py --day $d && python scripts/monitor.py --day $d
  python scripts/retrain_policy.py --day $d
  python scripts/rollback_monitor.py --day $d
done

Expected output (real run):

TEXT

day 11 (festive): 4000/4000 payments scored, labels known on day 14
day 11: quality_ok=True drift_share=0.5 psi=0.0028 flag_rate=0.342 | mature day 8: served v1, coverage 1.0, live_recall=0.4417 (baseline drop 0.2416, comparable=False)
day 11: COOLDOWN: live recall 0.4417 < floor 0.55 on mature day 8 (served by v1, coverage 1.0); last retrain on day 10
day 11, v2: flag rate 0.342 on 4000 payments; hard ceiling 0.6; band 0.171 to 0.684 (shifted baseline, applies)
GUARDRAIL OK
exit=0
day 12 (normal): 4000/4000 payments scored, labels known on day 15
day 12: quality_ok=True drift_share=0.0 psi=0.0224 flag_rate=0.3623 | mature day 9: served v1, coverage 1.0, live_recall=0.4417 (baseline drop 0.2416, comparable=False)
day 12: COOLDOWN: live recall 0.4417 < floor 0.55 on mature day 9 (served by v1, coverage 1.0); last retrain on day 10
day 12, v2: flag rate 0.362 on 4000 payments; hard ceiling 0.6; band 0.171 to 0.684 (shifted baseline, not applied: today is a different regime)
GUARDRAIL OK
exit=0

Three things to notice:

The mature days on 11 and 12 (days 8 and 9) were served by v1, and the monitor says so. Their low recall is v1's, not v2's. The policy answers COOLDOWN because v2 was just promoted; v2's own recall becomes measurable from day 14, when day-11 labels mature.
PSI is now near zero because it compares against v2's own expected scores.
On day 12 the sale has ended, so v2's festive baseline band is not applied; only the hard ceiling is. Comparing normal traffic with a festive baseline would be apples to oranges.

Acceptance check

The guardrail printed which checks applied and GUARDRAIL OK on both days.

Break It on Purpose

Each exercise breaks a real input the pipeline trusts. All four ran for this guide.

Step 1: An outage: only 30 percent of the day reaches the service

Bash

python scripts/replay.py --day 13 --fraction 0.3
python scripts/monitor.py --day 13
python scripts/retrain_policy.py --day 13; echo "exit=$?"

Expected output (real run):

TEXT

day 13 (normal): 1200/4000 payments scored, labels known on day 16
day 13: quality_ok=False drift_share=0.0 psi=0.0184 flag_rate=0.3633 | mature day 10: served v1, coverage 1.0, live_recall=0.3917 (baseline drop 0.2916, comparable=False)
day 13: BLOCKED: only 1200 rows, expected about 4000
exit=1

Never act on broken data: exit 1 fails the Airflow task loudly instead of skipping it. Labels for day 13 cover all 4,000 payments, so when they mature on day 16 the cohort coverage will be 0.3 and that day's recall will not be trusted either.

Step 2: Approve the wrong version

Bash

python scripts/approve.py --run-dir runs/<the run folder> --version 1 --by someone; echo "exit=$?"

Expected output (real run):

TEXT

REFUSED: this run registered v2, not v1
exit=1

Step 3: Someone edits the threshold in the registry by hand

A teammate "tunes" v2's threshold tag to 0.02 directly in the registry and reloads the service. No pipeline ran, no gate saw it.

Bash

python -c "from acme_fraud import registry; registry.client().set_model_version_tag(registry.NAME, '2', 'threshold', '0.02')"
curl -s -X POST -H "X-Admin-Token: $ADMIN_TOKEN" localhost:8000/admin/reload
python scripts/replay.py --day 14
python scripts/rollback_monitor.py --day 14; echo "exit=$?"
curl -s localhost:8000/ready; cat state/audit.log

Expected output (real run):

◈ DIAGRAM

v2 threshold tag edited by hand to 0.02
{"reloaded":true,"model_version":"2"}
day 14 (normal): 4000/4000 payments scored, labels known on day 17
day 14, v2: flag rate 0.960 on 4000 payments; hard ceiling 0.6; band 0.171 to 0.684 (shifted baseline, not applied: today is a different regime)
GUARDRAIL BREACHED (flag rate 0.960 above hard ceiling): rolled back v2 -> v1, service verified serving v1
No page was sent: this lab has no alert routing. Wire the exit code or the audit log to Alertmanager.
exit=1
{"status":"ready","model_version":"1","threshold":0.38}
2026-11-05T06:01:50+00:00 bootstrap champion none -> v1 by=bootstrap
2026-11-05T06:08:15+00:00 promote champion v1 -> v2 by=kushagra run=manual__2026-11-05T06_07_38.898430_00_00 approval=simulated
2026-11-05T06:10:39+00:00 rollback champion v2 -> v1 by=guardrail flag rate 0.960 above hard ceiling

The band did not apply (normal traffic, festive baseline), but the hard ceiling did. Roll back first, investigate second, and fix the cause: registry tags should be writable only by the pipeline, with every change audited.

Step 4: Replay a day twice

Bash

python scripts/replay.py --day 14; echo "exit=$?"

Expected output (real run):

TEXT

REFUSED: day 14 is already in logs/predictions.csv
exit=3

Duplicates would inflate drift samples, flag rates and recall. The replay refuses, and the monitor de-duplicates and fails data quality if duplicates ever appear.

Validation & Testing

Check the project end to end, then tear down. Everything ran locally for free.

Step 1: Final checks

Bash

pytest -q tests                                 # 19 passed
cat state/audit.log                             # bootstrap, promote, rollback
ls reports/                                     # one Evidently HTML report per day
cat runs/*/manifest.json runs/*/promotion.json  # the frozen inputs and the promotion state

Expected test output (real run):

TEXT

19 passed in 15.56s

Definition of done

state/monitoring.json has data_quality_ok, drift_share, prediction_psi and recall_drop (null until labels mature), plus live_recall, mature_label_coverage and mature_day_served_version.
Prometheus scrapes the service and the exporter; six alert rules load; drift and RecallBelowFloor fired on day 10.
The policy said INVESTIGATE for drift without confirmed harm, and RETRAIN only after two consecutive complete mature days below the recall floor.
The retraining run froze its exact rows, checksums, git commit and packages, and every later task verified them.
Promotion reached VERIFIED; the approval record is clearly marked as not authenticated.
A guardrail rollback was observed and verified against /ready; a double replay was refused.
The README states that Grafana and Docker Compose were not executed during this guide's validation (run them yourself and add a screenshot).

Step 2: Tear down

Bash

docker compose down -v
kill %1 %2                 # the service and the exporter, if started in this shell
rm -rf state logs labels reports runs ../airflow-home

Present This Project in an Interview

The 60-second version

"I built a fraud service that watches itself. Every prediction is logged and exported to Prometheus. A daily monitor computes Evidently feature drift, PSI on scores, and recall on the newest day whose labels have matured, because chargebacks arrive three days late. In a simulated festive sale, drift alerts fired on day 6, but the policy only said investigate. When recall on two consecutive complete mature days fell below the 0.55 floor, 0.45 then 0.37, it triggered an Airflow DAG: freeze the exact rows and checksums, train on the harmed days, backtest champion and challenger on the same held-out day, register, and wait for an approval record. Promotion is a resumable state machine that ends only when the service is verified. When a hand-edited threshold made the model flag 96 percent of payments, the guardrail rolled back to v1 and verified the service switched."

Questions to expect

"Why not retrain as soon as drift fires?" Drift says the inputs changed, not that the model is wrong. Retraining on unconfirmed drift chases seasonality and can make things worse. I page a human on drift and retrain on confirmed harm.
"Why two consecutive days, and why a floor?" 120 frauds a day means each fraud is almost a point of recall, so one bad day is easily noise. And a floor is the same in every regime, while a baseline from normal traffic is the wrong yardstick during a sale.
"How is the challenger compared fairly?" Both models score the same held-out mature day, which neither trained on, with each model's own threshold, and the gate also caps the flag rate.
"What if the guardrail fires during the night?" In this lab it rolls back automatically, because the previous champion is recorded and the action is reversible, and it exits non-zero with an audit line. It does not page anyone: notification needs Alertmanager or similar wired to that exit code. Promotion is the step that needs a human; rollback is not.
"What is missing for production?" Authenticated approvals (Airflow HITL or an approval UI), alert routing, immutable daily datasets instead of append-only CSVs, a reference refresh process, locked-down registry tags, and a registry server instead of SQLite.

What to show on screen

The day-by-day policy output, INVESTIGATE turning into RETRAIN.
The Airflow run with all tasks green and the backtest numbers.
The Grafana dashboard around the rollback, and state/audit.log.

Found a mistake, an outdated step or a broken link?

Drift Monitoring and Auto-Retraining Pipeline

Blueprint Walkthrough

Architecture Overview & Problem Statement

The problem

The loop

Why labels decide, not drift

Repository layout

Prerequisites (by module title)

Versions used

Milestone 1: Bootstrap the Data, the Model and the Registry

Step 1: Set up and write the parameters

Step 2: Write the data, model and registry modules

Step 3: Bootstrap v1

Acceptance check

Milestone 2: Serve the Model with Metrics and a Prediction Log

Step 1: Write the service

Step 2: Run it and the replay tool

Acceptance check

Milestone 3: Monitor Drift, Prediction Drift and Mature Recall

Step 1: Write the monitor

Step 2: Write the retrain policy

Step 3: Write the tests

Acceptance check

Milestone 4: Export to Prometheus, Alert, and Build the Grafana Dashboard

Step 1: Write the exporter

Step 2: Configure Prometheus and alerts

Step 3: Provision Grafana

Acceptance check

Milestone 5: Replay the Festive Sale and Watch the Policy

Step 1: Replay ten days

Step 2: Check what Prometheus sees on day 10

Acceptance check

Milestone 6: Retrain Through Airflow with a Human Approval

Step 1: Write the retraining scripts

Step 2: Write the DAG

Step 3: Run the DAG

Acceptance check

Milestone 7: Watch the New Champion with Guardrails

Step 1: Write the guardrails

Step 2: Day 11 (still the sale) and day 12 (the sale ends)

Acceptance check

Break It on Purpose

Step 1: An outage: only 30 percent of the day reaches the service

Step 2: Approve the wrong version

Step 3: Someone edits the threshold in the registry by hand

Step 4: Replay a day twice

Validation & Testing

Step 1: Final checks

Definition of done

Step 2: Tear down

Present This Project in an Interview

The 60-second version

Questions to expect

What to show on screen