Data Drift, Model Monitoring & Retraining Triggers

Monitoring a model in production requires more than Prometheus metrics. The model can degrade silently — same CPU/memory, but wrong answers — because the world changed.

Types of Drift

graph TD
    subgraph Input["Input changes"]
        DD["Data Drift<br/>Feature distribution changed<br/>from training data<br/><br/>Example: avg order value was $50,<br/>now users spend $200"]
        CD["Covariate Shift<br/>Input distribution changed<br/>but relationship still holds<br/><br/>(subset of data drift)"]
    end

    subgraph Output["Output changes"]
        CONC["Concept Drift<br/>Relationship between input<br/>and correct output changed<br/><br/>Example: 'fast' meant quick delivery,<br/>now users mean fast website"]
        PD["Prediction Drift<br/>Model outputs shifted<br/>(proxy for concept drift)"]
    end

    DD & CD --> RETRAIN_DATA["Retrain with new data distribution"]
    CONC --> RETRAIN_LABELS["Retrain with new labels<br/>(harder — need human annotation)"]
    PD --> INVESTIGATE["Investigate: is input or output drifting?"]

Four named drift types, flip through them:

Evidently — Data and Model Monitoring

Evidently generates HTML reports and JSON metrics comparing a reference dataset (training) against current production data.

Install

pip install evidently

Generate a drift report

import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import ColumnDriftMetric

# Reference = training data (what the model was trained on)
reference_data = pd.read_parquet("s3://my-data/training/reference.parquet")

# Current = last 24h of production requests (with features logged)
current_data = pd.read_parquet("s3://my-data/production/last_24h.parquet")

# Generate report
report = Report(metrics=[
    DataDriftPreset(),           # checks all features for drift
    DataQualityPreset(),         # null rates, min/max, distribution
    ColumnDriftMetric(column_name="user_age"),     # specific feature
    ColumnDriftMetric(column_name="purchase_amount"),
])

report.run(reference_data=reference_data, current_data=current_data)
report.save_html("drift_report.html")

# Extract metrics programmatically
result = report.as_dict()
drift_score = result["metrics"][0]["result"]["dataset_drift"]
drifted_features = result["metrics"][0]["result"]["number_of_drifted_columns"]
print(f"Drift detected: {drift_score}, drifted features: {drifted_features}")

Run as a K8s CronJob

apiVersion: batch/v1
kind: CronJob
metadata:
  name: drift-monitor
  namespace: ml-monitoring
spec:
  schedule: "0 6 * * *"    # daily at 6am
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: drift-check
            image: myrepo/drift-monitor:v1.2
            env:
            - name: REFERENCE_PATH
              value: "s3://my-data/training/reference.parquet"
            - name: CURRENT_PATH
              value: "s3://my-data/production/yesterday.parquet"
            - name: DRIFT_THRESHOLD
              value: "0.3"
            - name: WEBHOOK_URL
              value: "http://pipeline-trigger/webhooks/drift-detected"
          restartPolicy: OnFailure

# monitor.py — inside the CronJob container
import sys, requests, pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

reference = pd.read_parquet(os.environ["REFERENCE_PATH"])
current = pd.read_parquet(os.environ["CURRENT_PATH"])
threshold = float(os.environ["DRIFT_THRESHOLD"])

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=reference, current_data=current)
result = report.as_dict()

drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]

if drift_share > threshold:
    print(f"DRIFT DETECTED: {drift_share:.2%} of features drifted")
    requests.post(os.environ["WEBHOOK_URL"], json={
        "drift_score": drift_share,
        "feature": "dataset",
        "timestamp": datetime.utcnow().isoformat(),
    })
    sys.exit(0)   # drift detected but handled — exit 0 (not a job failure)

print(f"No significant drift: {drift_share:.2%}")

The daily cycle that CronJob runs, step by step:

Statistical Tests Used by Evidently

Evidently auto-selects the right test based on column type. You can override:

from evidently.calculations.stattests import ks_stat_test, chi_stat_test

ColumnDriftMetric(
    column_name="user_age",
    stattest=ks_stat_test,
    stattest_threshold=0.05,   # p-value threshold
)

Shadow Scoring — Detecting Concept Drift

Data drift = inputs changed. Concept drift = correct answer for same input changed. You can only detect concept drift with ground truth labels — which arrive delayed (e.g., did the customer actually churn?).

sequenceDiagram
    participant REQ as Request
    participant PROD as Production Model
    participant SHADOW as Shadow Model (new version)
    participant LOG as Label Store

    REQ->>PROD: input features
    REQ->>SHADOW: same input (parallel, async)
    PROD-->>REQ: prediction (served to user)
    SHADOW-->>LOG: prediction logged (not served)
    Note over LOG: 7 days later...
    LOG->>LOG: ground truth arrives (did user churn?)
    LOG->>LOG: compare: production accuracy vs shadow accuracy
    Note over LOG: if shadow >> production: promote shadow

# Shadow scoring pattern in FastAPI
@app.post("/predict")
async def predict(features: dict):
    # Serve production model
    prod_prediction = production_model.predict(features)

    # Shadow: run new model async, log results but don't serve
    asyncio.create_task(
        log_shadow_prediction(shadow_model, features, prod_prediction)
    )

    return {"prediction": prod_prediction}

async def log_shadow_prediction(shadow_model, features, prod_pred):
    shadow_pred = shadow_model.predict(features)
    await metrics_store.log({
        "timestamp": datetime.utcnow(),
        "features_hash": hash(str(features)),
        "prod_prediction": prod_pred,
        "shadow_prediction": shadow_pred,
    })

A/B Testing Model Versions

Route a percentage of production traffic to the new model version:

# KServe canary — 10% to new model
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: churn-model
spec:
  predictor:
    canaryTrafficPercent: 10     # 10% to v2
    model:
      storageUri: "s3://models/churn-v1/"

# Track A/B metrics in Prometheus
from prometheus_client import Counter, Histogram

predictions = Counter("model_predictions_total", "Predictions by version",
                       labelnames=["model_version"])
accuracy = Histogram("model_accuracy", "Prediction accuracy by version",
                     labelnames=["model_version"])

# After ground truth arrives:
def record_outcome(version: str, correct: bool):
    predictions.labels(model_version=version).inc()
    accuracy.labels(model_version=version).observe(1.0 if correct else 0.0)

# A/B accuracy comparison
sum(rate(model_accuracy_sum[1h])) by (model_version)
/
sum(rate(model_accuracy_count[1h])) by (model_version)

The canary rollout, step by step:

Monitoring Stack Summary

graph TD
    A["Production request logging"] -->|daily batch| B["Feature store / data lake (S3)"]
    B --> C["Evidently CronJob"]
    C --> D["Drift report"]
    D -->|webhook if threshold exceeded| E["Training pipeline triggered"]
    E --> F["New model trained + evaluated"]
    F -->|if accuracy gate passes| G["Model registered in Staging"]
    G --> H["Shadow test against production"]
    H -->|if shadow accuracy > production| I["Promote to Production<br/>(via MLflow API or<br/>ArgoCD + model URI update)"]

The same pipeline, one stage at a time:

Key Metrics to Alert On