Model Monitoring — Drift Detection and Retraining

A doctor trained in the 1990s using the medical knowledge of that era. Data drift: the patient population has changed — more diabetes, more sedentary lifestyle, new drug interactions. Concept drift: the same symptoms now indicate different conditions due to new pathogens. A good doctor keeps learning. A monitoring system is the mechanism that tells the doctor which patients they are getting wrong, so they know what to study next.

Monitoring without ground truth labels is like checking a patient's vital signs without doing bloodwork. You can detect that something is wrong (features look unusual) but not what is wrong (model is wrong) without comparing predictions to actual outcomes. Both layers — leading indicators and lagging indicators — are needed.

import numpy as np
import pandas as pd
from scipy import stats

np.random.seed(42)

# ── Simulate training reference data (January) ────────────────────────
N_REF = 10000
ref_distance  = np.random.exponential(3.2, N_REF)                 # avg 3.2 km
ref_order_val = np.random.exponential(400, N_REF)                  # avg $400
ref_peak_hour = np.random.binomial(1, 0.35, N_REF)                # 35% peak
ref_category  = np.random.choice(['bike', 'cycle', 'walk'], N_REF,
                                   p=[0.6, 0.3, 0.1])

# ── Simulate current production data (June — monsoon) ─────────────────
N_CUR = 2000
cur_distance  = np.random.exponential(5.8, N_CUR)   # DRIFTED: avg 5.8 km
cur_order_val = np.random.exponential(410, N_CUR)   # stable
cur_peak_hour = np.random.binomial(1, 0.33, N_CUR)  # stable
cur_category  = np.random.choice(['bike', 'cycle', 'walk'], N_CUR,
                                   p=[0.45, 0.40, 0.15])  # DRIFTED: fewer bikes

# ── KS Test — for continuous features ────────────────────────────────
def ks_drift_test(reference: np.ndarray, current: np.ndarray,
                   alpha: float = 0.05) -> dict:
    """
    Kolmogorov-Smirnov test for distribution shift.
    Returns statistic, p-value, and whether drift was detected.
    """
    stat, p_value = stats.ks_2samp(reference, current)
    drifted = p_value < alpha
    return {
        'test':      'KS',
        'statistic': round(stat, 4),
        'p_value':   round(p_value, 6),
        'drifted':   drifted,
        'severity':  'high' if stat > 0.2 else 'medium' if stat > 0.1 else 'low',
    }

# ── PSI — Population Stability Index ─────────────────────────────────
def psi(reference: np.ndarray, current: np.ndarray,
         n_bins: int = 10) -> dict:
    """
    PSI for continuous features.
    Bins reference data, computes proportions, measures shift.
    """
    # Bin using reference data quantiles
    breakpoints = np.percentile(reference, np.linspace(0, 100, n_bins + 1))
    breakpoints[0]  -= 1e-8   # ensure all values included
    breakpoints[-1] += 1e-8

    ref_counts = np.histogram(reference, bins=breakpoints)[0]
    cur_counts = np.histogram(current,   bins=breakpoints)[0]

    ref_pct = ref_counts / len(reference)
    cur_pct = cur_counts / len(current)

    # Avoid log(0)
    ref_pct = np.where(ref_pct == 0, 1e-8, ref_pct)
    cur_pct = np.where(cur_pct == 0, 1e-8, cur_pct)

    psi_val = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
    severity = 'high' if psi_val > 0.2 else 'medium' if psi_val > 0.1 else 'low'

    return {
        'test':      'PSI',
        'psi':       round(psi_val, 4),
        'drifted':   psi_val > 0.1,
        'severity':  severity,
        'threshold': '0.10 minor / 0.20 significant',
    }

# ── Chi-squared test — for categorical features ───────────────────────
def chi2_drift_test(reference: np.ndarray, current: np.ndarray) -> dict:
    """Chi-squared test for categorical feature drift."""
    categories  = np.unique(np.concatenate([reference, current]))
    ref_counts  = np.array([np.sum(reference == c) for c in categories])
    cur_counts  = np.array([np.sum(current == c)   for c in categories])

    # Scale reference to same size as current
    expected = ref_counts * len(current) / len(reference)
    stat, p_value = stats.chisquare(cur_counts, f_exp=expected)

    return {
        'test':      'Chi-squared',
        'statistic': round(stat, 4),
        'p_value':   round(p_value, 6),
        'drifted':   p_value < 0.05,
        'severity':  'high' if stat > 50 else 'medium' if stat > 10 else 'low',
    }

# ── Run all tests ─────────────────────────────────────────────────────
features = {
    'distance_km':  (ref_distance,  cur_distance,  'continuous'),
    'order_value':  (ref_order_val, cur_order_val, 'continuous'),
    'is_peak_hour': (ref_peak_hour, cur_peak_hour, 'continuous'),
    'vehicle_type': (ref_category,  cur_category,  'categorical'),
}

print(f"{'Feature':<18} {'Test':<12} {'Stat':>8} {'p-value':>10} {'Drifted':>8} {'Severity'}")
print("─" * 65)
for feat, (ref, cur, ftype) in features.items():
    if ftype == 'categorical':
        result = chi2_drift_test(ref, cur)
        stat   = f"{result['statistic']:.4f}"
        pval   = f"{result['p_value']:.6f}"
    else:
        ks   = ks_drift_test(ref, cur)
        psi_ = psi(ref, cur)
        result = ks   # primary test
        stat   = f"{ks['statistic']:.4f} / PSI={psi_['psi']:.3f}"
        pval   = f"{ks['p_value']:.6f}"

    mark = '⚠ YES' if result['drifted'] else '  no'
    print(f"  {feat:<18} {result['test']:<12} {stat:>8} {pval:>10} "
          f"{mark:>8}  {result['severity']}")

# pip install evidently

import pandas as pd
import numpy as np
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DataDriftTable,
    ColumnDriftMetric,
    ColumnSummaryMetric,
)

np.random.seed(42)

# ── Build reference and current datasets ──────────────────────────────
N_REF = 5000
N_CUR = 1000

reference_data = pd.DataFrame({
    'distance_km':    np.random.exponential(3.2, N_REF),
    'order_value':    np.random.exponential(400, N_REF),
    'is_peak_hour':   np.random.binomial(1, 0.35, N_REF).astype(float),
    'restaurant_avg': np.random.normal(35, 8, N_REF),
    'driver_avg':     np.random.normal(32, 7, N_REF),
    'prediction':     np.random.normal(35, 10, N_REF).clip(10, 90),
})

# Monsoon period — distance has drifted, others stable
current_data = pd.DataFrame({
    'distance_km':    np.random.exponential(5.8, N_CUR),  # drifted
    'order_value':    np.random.exponential(415, N_CUR),  # stable
    'is_peak_hour':   np.random.binomial(1, 0.33, N_CUR).astype(float),
    'restaurant_avg': np.random.normal(38, 9, N_CUR),     # slight shift
    'driver_avg':     np.random.normal(32, 7, N_CUR),
    'prediction':     np.random.normal(42, 12, N_CUR).clip(10, 90),  # drifted
})

# ── Generate drift report ─────────────────────────────────────────────
report = Report(metrics=[
    DataDriftPreset(drift_share_threshold=0.3),   # alert if >30% features drifted
    DataQualityPreset(),
    ColumnDriftMetric(column_name='distance_km'),
    ColumnDriftMetric(column_name='prediction'),
    ColumnSummaryMetric(column_name='prediction'),
])

report.run(reference_data=reference_data, current_data=current_data)

# Save HTML report (human-readable dashboard)
report.save_html('/tmp/drift_report.html')

# Get JSON results for programmatic parsing
result_dict = report.as_dict()

# ── Parse drift results ────────────────────────────────────────────────
def parse_evidently_results(result: dict) -> dict:
    """Extract key metrics from Evidently report for alerting."""
    metrics = result.get('metrics', [])
    summary = {'drifted_features': [], 'stable_features': [], 'share_drifted': 0}

    for m in metrics:
        if m.get('metric') == 'DatasetDriftMetric':
            summary['share_drifted'] = m['result'].get('share_of_drifted_columns', 0)
            summary['dataset_drifted'] = m['result'].get('dataset_drift', False)

        if m.get('metric') == 'ColumnDriftMetric':
            col    = m['result'].get('column_name', '')
            drifted = m['result'].get('drift_detected', False)
            if drifted:
                summary['drifted_features'].append(col)
            else:
                summary['stable_features'].append(col)

    return summary

summary = parse_evidently_results(result_dict)
print("Evidently drift report summary:")
print(f"  Dataset drifted:    {summary.get('dataset_drifted', 'unknown')}")
print(f"  Share drifted:      {summary.get('share_drifted', 0):.1%}")
print(f"  Drifted features:   {summary['drifted_features']}")
print(f"  Stable features:    {summary['stable_features']}")
print(f"
Full HTML report saved to /tmp/drift_report.html")

# ── Scheduled monitoring job ──────────────────────────────────────────
def daily_monitoring_job(
    model_name: str,
    reference_df: pd.DataFrame,
    current_df: pd.DataFrame,
    alert_threshold: float = 0.3,
    send_alert=None,
) -> dict:
    """
    Run daily drift monitoring.
    Call from Airflow DAG at 6 AM every day.
    """
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_df, current_data=current_df)
    summary = parse_evidently_results(report.as_dict())

    share_drifted = summary.get('share_drifted', 0)
    needs_alert   = share_drifted > alert_threshold

    if needs_alert and send_alert:
        send_alert(
            subject=f'[DRIFT ALERT] {model_name}: {share_drifted:.0%} features drifted',
            body=(
                f"Model: {model_name}
"
                f"Drifted features: {summary['drifted_features']}
"
                f"Action required: investigate data pipeline and consider retraining"
            ),
        )
    return {
        'model':          model_name,
        'share_drifted':  share_drifted,
        'alert_sent':     needs_alert,
        'drifted':        summary['drifted_features'],
    }

result = daily_monitoring_job(
    'doordash-delivery-time',
    reference_data, current_data,
)
print(f"
Daily monitoring result: {result}")

import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from collections import deque

# ── Prediction store — log every prediction for later joining ─────────
@dataclass
class PredictionLog:
    prediction_id:  str
    timestamp:      datetime
    features:       dict
    prediction:     float
    actual:         float = None       # filled in later when label arrives
    label_received: bool  = False

class PredictionStore:
    """
    In production: backed by a database (Postgres/BigQuery).
    Stores predictions and joins with delayed ground truth labels.
    """
    def __init__(self):
        self.predictions: dict[str, PredictionLog] = {}

    def log_prediction(self, prediction_id: str,
                        features: dict, prediction: float) -> None:
        self.predictions[prediction_id] = PredictionLog(
            prediction_id=prediction_id,
            timestamp=datetime.now(),
            features=features,
            prediction=prediction,
        )

    def receive_label(self, prediction_id: str, actual: float) -> None:
        """Called when ground truth arrives (e.g. actual delivery time)."""
        if prediction_id in self.predictions:
            self.predictions[prediction_id].actual         = actual
            self.predictions[prediction_id].label_received = True

    def get_labelled_window(self, hours: int = 24) -> pd.DataFrame:
        """Get all predictions with labels from the last N hours."""
        cutoff = datetime.now() - timedelta(hours=hours)
        rows   = [
            {
                'prediction_id': p.prediction_id,
                'timestamp':     p.timestamp,
                'prediction':    p.prediction,
                'actual':        p.actual,
                **p.features,
            }
            for p in self.predictions.values()
            if p.label_received and p.timestamp > cutoff
        ]
        return pd.DataFrame(rows) if rows else pd.DataFrame()

# ── Performance metrics over rolling windows ─────────────────────────
def compute_performance_metrics(df: pd.DataFrame,
                                  prediction_col: str = 'prediction',
                                  actual_col:     str = 'actual') -> dict:
    if df.empty or actual_col not in df.columns:
        return {}

    errors   = df[prediction_col] - df[actual_col]
    abs_err  = errors.abs()

    return {
        'n_predictions': len(df),
        'mae':           round(abs_err.mean(), 3),
        'rmse':          round(np.sqrt((errors**2).mean()), 3),
        'mape':          round((abs_err / df[actual_col].clip(1)).mean() * 100, 2),
        'within_5min':   round((abs_err <= 5).mean() * 100, 1),   # % within 5 min
        'within_10min':  round((abs_err <= 10).mean() * 100, 1),
        'mean_prediction': round(df[prediction_col].mean(), 2),
        'mean_actual':     round(df[actual_col].mean(), 2),
        'bias':            round(errors.mean(), 3),  # positive = over-estimating
    }

# ── Simulate a week of predictions and labels ─────────────────────────
store = PredictionStore()
np.random.seed(42)

# Week 1 (Jan): good model performance
for i in range(500):
    pid     = f'PRED_{i:04d}'
    actual  = float(np.random.normal(35, 10))
    pred    = actual + np.random.normal(0, 5)   # small error
    store.log_prediction(pid, {'distance_km': 3.5}, pred)
    store.receive_label(pid, actual)

# Week 2 (June monsoon): model starts drifting
for i in range(500, 1000):
    pid    = f'PRED_{i:04d}'
    actual = float(np.random.normal(48, 12))    # longer actual times (monsoon)
    pred   = float(np.random.normal(35, 10))    # model still predicts Jan times
    store.log_prediction(pid, {'distance_km': 5.8}, pred)
    store.receive_label(pid, actual)

# Compute metrics for each week
jan_df   = pd.DataFrame([
    {'prediction': store.predictions[f'PRED_{i:04d}'].prediction,
     'actual':     store.predictions[f'PRED_{i:04d}'].actual}
    for i in range(500)
])
june_df  = pd.DataFrame([
    {'prediction': store.predictions[f'PRED_{i:04d}'].prediction,
     'actual':     store.predictions[f'PRED_{i:04d}'].actual}
    for i in range(500, 1000)
])

jan_metrics  = compute_performance_metrics(jan_df)
june_metrics = compute_performance_metrics(june_df)

print("Performance degradation over time:")
print(f"  {'Metric':<20} {'January':>10} {'June':>10} {'Change':>10}")
print("  " + "─" * 52)
for metric in ['mae', 'rmse', 'mape', 'within_5min', 'bias']:
    jan_v  = jan_metrics.get(metric, 0)
    june_v = june_metrics.get(metric, 0)
    delta  = june_v - jan_v
    flag   = ' ⚠' if abs(delta) > jan_v * 0.3 else ''
    print(f"  {metric:<20} {jan_v:>10.2f} {june_v:>10.2f} "
          f"{delta:>+10.2f}{flag}")

import numpy as np
from dataclasses import dataclass
from typing import Callable
from datetime import datetime

# ── Retraining trigger conditions ─────────────────────────────────────
@dataclass
class RetriggerCondition:
    name:        str
    check_fn:    Callable[..., bool]
    priority:    str   # 'critical', 'high', 'medium'
    description: str

def check_performance_drop(current_mae: float,
                             baseline_mae: float,
                             threshold_pct: float = 20.0) -> bool:
    """Trigger if MAE increased by more than threshold_pct%."""
    if baseline_mae == 0:
        return False
    pct_change = (current_mae - baseline_mae) / baseline_mae * 100
    return pct_change > threshold_pct

def check_drift_threshold(share_drifted: float,
                            threshold: float = 0.30) -> bool:
    """Trigger if more than threshold fraction of features have drifted."""
    return share_drifted > threshold

def check_prediction_distribution(current_mean: float,
                                    baseline_mean: float,
                                    current_std: float,
                                    baseline_std: float) -> bool:
    """Trigger if prediction distribution has shifted significantly."""
    mean_shift = abs(current_mean - baseline_mean) / (baseline_std + 1e-8)
    std_ratio  = max(current_std, baseline_std) / (min(current_std, baseline_std) + 1e-8)
    return mean_shift > 2.0 or std_ratio > 2.0

def check_sample_size(n_new_samples: int, min_samples: int = 5000) -> bool:
    """Trigger on schedule when enough new data has accumulated."""
    return n_new_samples >= min_samples

# ── Retraining orchestrator ────────────────────────────────────────────
class RetrainingOrchestrator:
    """
    Checks all trigger conditions and fires the training pipeline
    when any critical or combination of high-priority triggers fire.
    """

    def __init__(self, training_pipeline_fn: Callable,
                  alert_fn: Callable = None):
        self.training_pipeline = training_pipeline_fn
        self.alert             = alert_fn or print
        self.last_retrain      = None
        self.min_retrain_interval_hours = 24   # at most once per day

    def evaluate_triggers(self, monitoring_data: dict) -> list[dict]:
        """
        Evaluate all trigger conditions.
        Returns list of fired triggers with priority.
        """
        fired = []

        # Performance degradation (requires labels)
        if 'current_mae' in monitoring_data and 'baseline_mae' in monitoring_data:
            if check_performance_drop(
                monitoring_data['current_mae'],
                monitoring_data['baseline_mae'],
                threshold_pct=25.0,
            ):
                fired.append({
                    'trigger':  'performance_drop',
                    'priority': 'critical',
                    'details':  f"MAE {monitoring_data['baseline_mae']:.2f} → {monitoring_data['current_mae']:.2f}",
                })

        # Feature drift
        if 'share_drifted' in monitoring_data:
            if check_drift_threshold(monitoring_data['share_drifted'], 0.4):
                fired.append({
                    'trigger':  'feature_drift',
                    'priority': 'high',
                    'details':  f"{monitoring_data['share_drifted']:.0%} of features drifted",
                })

        # Prediction distribution shift
        if all(k in monitoring_data for k in
               ['pred_mean', 'baseline_pred_mean', 'pred_std', 'baseline_pred_std']):
            if check_prediction_distribution(
                monitoring_data['pred_mean'],
                monitoring_data['baseline_pred_mean'],
                monitoring_data['pred_std'],
                monitoring_data['baseline_pred_std'],
            ):
                fired.append({
                    'trigger':  'prediction_shift',
                    'priority': 'high',
                    'details':  f"Prediction mean {monitoring_data['baseline_pred_mean']:.1f} → {monitoring_data['pred_mean']:.1f}",
                })

        # Scheduled data accumulation trigger
        if monitoring_data.get('n_new_samples', 0) >= 10000:
            fired.append({
                'trigger':  'scheduled_data_volume',
                'priority': 'medium',
                'details':  f"{monitoring_data['n_new_samples']:,} new samples available",
            })

        return fired

    def should_retrain(self, fired_triggers: list[dict]) -> bool:
        """Decide whether to retrain based on fired triggers."""
        if not fired_triggers:
            return False

        # Retrain on any critical trigger
        has_critical = any(t['priority'] == 'critical' for t in fired_triggers)
        if has_critical:
            return True

        # Retrain if 2+ high-priority triggers
        high_count = sum(1 for t in fired_triggers if t['priority'] == 'high')
        if high_count >= 2:
            return True

        return False

    def run(self, monitoring_data: dict) -> dict:
        fired   = self.evaluate_triggers(monitoring_data)
        retrain = self.should_retrain(fired)

        if retrain:
            # Check cooldown
            if self.last_retrain:
                hours_since = (datetime.now() - self.last_retrain).seconds / 3600
                if hours_since < self.min_retrain_interval_hours:
                    return {'action': 'cooldown', 'triggers': fired,
                             'hours_until_eligible': self.min_retrain_interval_hours - hours_since}

            self.alert(f"[RETRAIN] Triggering retraining: {[t['trigger'] for t in fired]}")
            result = self.training_pipeline(triggered_by=fired)
            self.last_retrain = datetime.now()
            return {'action': 'retrained', 'triggers': fired, 'result': result}

        if fired:
            self.alert(f"[MONITOR] Triggers fired but not retraining yet: {fired}")
            return {'action': 'monitored', 'triggers': fired}

        return {'action': 'no_action', 'triggers': []}

# ── Simulate monitoring cycle ─────────────────────────────────────────
def mock_training_pipeline(triggered_by=None):
    print(f"    → Training pipeline started. Triggers: {[t['trigger'] for t in triggered_by or []]}")
    return {'status': 'success', 'new_model_mae': 6.2}

orchestrator = RetrainingOrchestrator(mock_training_pipeline)

test_scenarios = [
    {
        'name': 'January (stable)',
        'data': {'current_mae': 5.8, 'baseline_mae': 5.5, 'share_drifted': 0.1,
                  'pred_mean': 35.2, 'baseline_pred_mean': 35.0,
                  'pred_std': 10.1, 'baseline_pred_std': 10.0, 'n_new_samples': 2000},
    },
    {
        'name': 'June monsoon (performance drop + drift)',
        'data': {'current_mae': 9.8, 'baseline_mae': 5.5, 'share_drifted': 0.5,
                  'pred_mean': 42.0, 'baseline_pred_mean': 35.0,
                  'pred_std': 12.0, 'baseline_pred_std': 10.0, 'n_new_samples': 3000},
    },
    {
        'name': 'Scheduled (enough new data)',
        'data': {'current_mae': 5.9, 'baseline_mae': 5.5, 'share_drifted': 0.15,
                  'pred_mean': 35.5, 'baseline_pred_mean': 35.0,
                  'pred_std': 10.2, 'baseline_pred_std': 10.0, 'n_new_samples': 12000},
    },
]

print("Retraining orchestrator simulation:")
for scenario in test_scenarios:
    result = orchestrator.run(scenario['data'])
    print(f"
  Scenario: {scenario['name']}")
    print(f"  Action:   {result['action']}")
    print(f"  Triggers: {[t['trigger'] for t in result.get('triggers', [])]}")

Accuracy requires a ground truth label, and for most production models labels arrive late (delivery time: thirty to sixty minutes; fraud chargebacks: a week to a month) or in some cases never arrive at scale at all. If monitoring only meant tracking accuracy, a model could be silently broken for weeks before anyone noticed, simply because the labels needed to compute accuracy had not shown up yet. Real monitoring leans on leading indicators that need no labels — input feature drift, prediction distribution shift — precisely so problems surface before the lagging, label-dependent metric ever catches up.

Running a statistical test costs nothing computationally, but making its output useful is ongoing work. Thresholds need tuning against real production noise or every model gets flooded with false alarms. The reference distribution the current data is compared against goes stale and needs periodic refreshing, or the test starts flagging drift against a world that no longer resembles today's baseline anyway. New features added to the model later have to be deliberately wired into the monitoring job — they do not appear there automatically just because they exist in the model. None of that is a one-time cost.

A monitoring configuration that was well-tuned at launch degrades on its own as the business changes: normal seasonal patterns look like drift to a threshold set before the team had seen a full year of data; a model retrain shifts what "normal" prediction output even looks like, so the old baseline is now wrong; a new feature or a new market segment needs its own reference distribution before monitoring can say anything useful about it. Treating the initial monitoring setup as finished work is exactly how a monitoring system quietly stops being trustworthy without anyone deciding that it should.

Data drift only checks whether the input feature distributions have moved. Concept drift — where the relationship between the same features and the correct outcome has changed — can happen with the input distribution looking completely unchanged. A fraud model can see the exact same feature ranges it always has while an entirely new fraud scheme, invisible to those features, is bypassing it undetected. Data drift monitoring alone would report a perfectly clean, stable-looking dashboard the entire time this is happening, which is precisely why performance monitoring against delayed ground truth cannot be skipped just because the drift dashboard looks calm.

A monitoring system that fires constantly trains its own team to stop reading it — alert fatigue is a real failure mode, not a minor annoyance. Statistical significance is sensitive to sample size, so at high production volume even a practically meaningless shift can produce a technically significant p-value and trigger an alert every single day. A smaller number of well-calibrated alerts, routed to the right person at the right urgency, catches real problems faster than a dashboard so noisy that a genuine critical alert gets lost in the same channel as twenty routine ones from that week.

Data drift is a shift in the input feature distribution — it is detectable immediately, without any ground truth label, using statistical tests like KS or PSI directly on incoming features. Concept drift is a change in the relationship between those features and the correct outcome — the inputs can look completely unchanged while the correct prediction for them has shifted, and detecting it requires ground truth labels, which are usually delayed. The distinction matters operationally because it determines what tooling can even see the problem: a monitoring stack that only watches feature distributions will report a clean bill of health during a concept drift event, so performance monitoring against delayed labels has to run in parallel, not as a replacement.

I would alert on anything that is both statistically meaningful and actionable right now — a confirmed performance regression past a defined threshold, a majority of features drifting at once, or a business KPI moving in the wrong direction. I would only log, not alert, on individual feature drift below a severity threshold, small fluctuations that fall within normal week-to-week noise, or anything that requires a human to gather more context before it is even clear whether action is needed. The test I would apply to any candidate alert: if this fired right now, is there a specific person who should do something today, or does it just want an audit trail in case it becomes relevant later?

I would rely on leading indicators that need no labels at all in the interim: input feature drift, prediction distribution shift, and for a classification model, the distribution of predicted confidence scores — a sudden rise in low-confidence predictions is often an early sign of trouble long before labels confirm it. I would also set up a small label-sampling programme where a subset of predictions gets manually reviewed or fast-tracked to a confirmed outcome within a day, giving an early, lower-volume performance signal well before the full delayed label set arrives at scale. Neither approach replaces eventual ground truth evaluation — they buy time to react sooner.

It comes down to the feature's type and how the result needs to be interpreted. For a continuous feature I would default to the KS test when I want a clean statistical significance answer with a p-value, since it directly compares empirical distributions with no binning decisions to second-guess. I would use PSI when I want a single interpretable severity score with industry-standard thresholds — it is the default in credit risk and fraud teams specifically because everyone already agrees on what a PSI above 0.2 means. For a categorical feature, neither KS nor PSI directly applies the same way, so chi-squared, which compares observed to expected category frequencies, is the natural choice.

I would treat this as a monitoring design problem, not a data problem. First, I would switch from pure statistical significance to effect-size thresholds — requiring a KS statistic above a meaningful cutoff, not just a p-value below 0.05, since large sample sizes make almost any tiny shift statistically significant. Second, I would require drift to persist across several consecutive checks before alerting, to filter out single-day noise. Third, I would re-route by severity so only genuinely critical signals reach a page, with the rest going to a digest a human actually has time to read. The underlying goal is restoring trust in the alerts that do fire, since a monitoring system nobody reads is equivalent to having no monitoring at all.