Retraining Pipelines — Keeping Models Fresh

A Formula 1 pit stop team replacing a tyre during a race. They do not stop the car entirely (that would lose the race). They do not just bolt the new tyre on without checking it first (that would crash the car). They have a rehearsed procedure: jack the car, change the tyre, check it is secure, lower the car, driver goes. If anything is wrong, they abort and diagnose. The whole process takes 2 seconds because every step is practiced and every failure mode is handled. Retraining pipelines are the same — a procedure so well-engineered that updating a model in production takes minutes with zero downtime.

The champion model is the tyre that got the car this far. The challenger is the new tyre. You do not swap until you are certain the new tyre is at least as good. And you keep the old tyre nearby in case you need to switch back in a hurry.

import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
import json, pickle, os
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

np.random.seed(42)

# ── Champion baseline — stored at last promotion time ─────────────────
@dataclass
class ModelBaseline:
    """Stores the performance of the current production (champion) model."""
    model_version:   str
    promoted_at:     str
    test_mae:        float
    test_rmse:       float
    test_r2:         float
    feature_columns: list[str]
    training_cutoff: str
    n_training_rows: int

# ── Synthetic dataset ─────────────────────────────────────────────────
N = 6000
X = pd.DataFrame({
    'distance_km':    np.random.exponential(3.5, N),
    'is_peak_hour':   np.random.randint(0, 2, N).astype(float),
    'order_value':    np.random.exponential(400, N),
    'restaurant_avg': np.random.normal(35, 8, N),
    'driver_avg':     np.random.normal(32, 7, N),
})
y = (X['distance_km'] * 5.5 + X['is_peak_hour'] * 7 +
     X['restaurant_avg'] * 0.3 + np.random.normal(0, 5, N)).clip(10, 90)

# Fixed held-out test set — never used for training, never changes
# This is the standard against which every model is measured
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.15, random_state=999,   # fixed seed — same split always
)
X_train, X_val, y_train, y_val = train_test_split(
    X_train, y_train, test_size=0.15, random_state=42,
)

# ── Train champion (simulate existing production model) ────────────────
champion = GradientBoostingRegressor(
    n_estimators=150, max_depth=4, learning_rate=0.08, random_state=42,
)
champion.fit(X_train, y_train)
champ_test_mae  = mean_absolute_error(y_test, champion.predict(X_test))
champ_test_rmse = mean_squared_error(y_test, champion.predict(X_test), squared=False)
champ_test_r2   = r2_score(y_test, champion.predict(X_test))

champion_baseline = ModelBaseline(
    model_version='v2.1.0',
    promoted_at=datetime.now().isoformat(),
    test_mae=champ_test_mae,
    test_rmse=champ_test_rmse,
    test_r2=champ_test_r2,
    feature_columns=list(X.columns),
    training_cutoff='2026-03-01',
    n_training_rows=len(X_train),
)

print(f"Champion baseline (v{champion_baseline.model_version}):")
print(f"  MAE:  {champion_baseline.test_mae:.4f}")
print(f"  RMSE: {champion_baseline.test_rmse:.4f}")
print(f"  R²:   {champion_baseline.test_r2:.4f}")

# ── Train challenger (new retraining run) ─────────────────────────────
# Simulated: challenger uses more recent data + slightly better params
X_train_new = pd.concat([X_train, X_val])   # more data available now
y_train_new = pd.concat([y_train, y_val])

challenger = GradientBoostingRegressor(
    n_estimators=200, max_depth=4, learning_rate=0.06, random_state=42,
)
challenger.fit(X_train_new, y_train_new)

# ── Offline quality gate ───────────────────────────────────────────────
def offline_quality_gate(
    challenger_model,
    champion_baseline: ModelBaseline,
    X_test: pd.DataFrame,
    y_test: pd.Series,
    mae_tolerance:  float = 0.05,   # challenger can be at most 5% worse than champion
    rmse_tolerance: float = 0.05,
    min_r2:         float = 0.80,
) -> dict:
    """
    Compare challenger to champion on the fixed held-out test set.
    Returns gate result with detailed breakdown.
    """
    ch_preds = challenger_model.predict(X_test)
    ch_mae   = mean_absolute_error(y_test, ch_preds)
    ch_rmse  = mean_squared_error(y_test, ch_preds, squared=False)
    ch_r2    = r2_score(y_test, ch_preds)

    # Gate conditions
    mae_ok   = ch_mae  <= champion_baseline.test_mae  * (1 + mae_tolerance)
    rmse_ok  = ch_rmse <= champion_baseline.test_rmse * (1 + rmse_tolerance)
    r2_ok    = ch_r2   >= min_r2
    # Additional: challenger should not be drastically better (sanity check — might indicate leakage)
    mae_not_suspicious = ch_mae >= champion_baseline.test_mae * 0.5

    passed = mae_ok and rmse_ok and r2_ok and mae_not_suspicious

    result = {
        'passed': passed,
        'challenger': {'mae': ch_mae, 'rmse': ch_rmse, 'r2': ch_r2},
        'champion':   {'mae': champion_baseline.test_mae,
                       'rmse': champion_baseline.test_rmse,
                       'r2': champion_baseline.test_r2},
        'gates': {
            'mae_within_tolerance':   {'passed': mae_ok,  'value': ch_mae,  'threshold': champion_baseline.test_mae * (1 + mae_tolerance)},
            'rmse_within_tolerance':  {'passed': rmse_ok, 'value': ch_rmse, 'threshold': champion_baseline.test_rmse * (1 + rmse_tolerance)},
            'r2_above_minimum':       {'passed': r2_ok,   'value': ch_r2,   'threshold': min_r2},
            'mae_not_suspicious':     {'passed': mae_not_suspicious, 'value': ch_mae, 'threshold': champion_baseline.test_mae * 0.5},
        },
        'improvement_pct': (champion_baseline.test_mae - ch_mae) / champion_baseline.test_mae * 100,
    }
    return result

gate_result = offline_quality_gate(
    challenger, champion_baseline, X_test, y_test,
)

print(f"
Offline quality gate result: {'✓ PASSED' if gate_result['passed'] else '✗ FAILED'}")
print(f"  Challenger MAE: {gate_result['challenger']['mae']:.4f}")
print(f"  Champion MAE:   {gate_result['champion']['mae']:.4f}")
print(f"  Improvement:    {gate_result['improvement_pct']:+.2f}%")
print(f"
  Gate breakdown:")
for gate, detail in gate_result['gates'].items():
    mark = '✓' if detail['passed'] else '✗'
    print(f"    {mark} {gate}: {detail['value']:.4f} (threshold: {detail['threshold']:.4f})")

import asyncio
import time
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from collections import defaultdict
from sklearn.metrics import mean_absolute_error

# ── Shadow prediction logger ──────────────────────────────────────────
@dataclass
class ShadowLog:
    request_id:          str
    timestamp:           float
    features:            dict
    champion_prediction: float
    challenger_prediction: float
    actual:              float = None   # filled in when label arrives

class ShadowDeployment:
    """
    Routes every request to both champion and challenger.
    Returns champion prediction to caller.
    Logs both predictions for comparison.
    """
    def __init__(self, champion, challenger, feature_columns: list[str]):
        self.champion         = champion
        self.challenger       = challenger
        self.feature_columns  = feature_columns
        self.logs: list[ShadowLog] = []
        self.request_count    = 0

    def predict(self, features: dict, request_id: str = None) -> float:
        """
        Returns champion prediction. Logs challenger prediction silently.
        In production: challenger.predict() runs in a background thread.
        """
        import uuid
        request_id = request_id or str(uuid.uuid4())[:8]

        X = np.array([features.get(c, 0) for c in self.feature_columns]).reshape(1, -1)

        # Champion — this is what the user gets
        champ_pred = float(self.champion.predict(X)[0])

        # Challenger — silent, no user impact
        try:
            chal_pred = float(self.challenger.predict(X)[0])
        except Exception:
            chal_pred = None

        self.logs.append(ShadowLog(
            request_id=request_id,
            timestamp=time.time(),
            features=features,
            champion_prediction=champ_pred,
            challenger_prediction=chal_pred,
        ))
        self.request_count += 1
        return champ_pred   # ← only champion prediction returned to user

    def receive_label(self, request_id: str, actual: float):
        """Join actual label back to shadow log entry."""
        for log in self.logs:
            if log.request_id == request_id:
                log.actual = actual
                break

    def shadow_gate(self, mae_tolerance: float = 0.10) -> dict:
        """
        Compare challenger to champion on labelled shadow requests.
        Gate passes if challenger MAE <= champion MAE × (1 + tolerance).
        """
        labelled = [l for l in self.logs if l.actual is not None
                     and l.challenger_prediction is not None]

        if len(labelled) < 100:
            return {'passed': None, 'reason': f'Insufficient labelled data: {len(labelled)} < 100'}

        actuals    = np.array([l.actual for l in labelled])
        champ_preds = np.array([l.champion_prediction for l in labelled])
        chal_preds  = np.array([l.challenger_prediction for l in labelled])

        champ_mae = mean_absolute_error(actuals, champ_preds)
        chal_mae  = mean_absolute_error(actuals, chal_preds)
        threshold = champ_mae * (1 + mae_tolerance)
        passed    = chal_mae <= threshold

        # Prediction divergence — how different are the two models?
        divergence = np.abs(champ_preds - chal_preds).mean()

        return {
            'passed':        passed,
            'n_labelled':    len(labelled),
            'champion_mae':  round(champ_mae, 4),
            'challenger_mae': round(chal_mae, 4),
            'improvement_pct': round((champ_mae - chal_mae) / champ_mae * 100, 2),
            'prediction_divergence': round(divergence, 4),
            'threshold':     round(threshold, 4),
        }

# ── Simulate shadow deployment ─────────────────────────────────────────
shadow = ShadowDeployment(champion, challenger, list(X.columns))

# Simulate 500 live requests with slight distribution shift (monsoon)
np.random.seed(123)
n_requests = 500
for i in range(n_requests):
    features = {
        'distance_km':    float(np.random.exponential(5.2)),   # slightly shifted
        'is_peak_hour':   float(np.random.randint(0, 2)),
        'order_value':    float(np.random.exponential(420)),
        'restaurant_avg': float(np.random.normal(37, 9)),
        'driver_avg':     float(np.random.normal(33, 7)),
    }
    pred = shadow.predict(features, request_id=f'REQ{i:04d}')

    # Simulate labels arriving for 70% of requests
    if np.random.random() < 0.7:
        actual = (features['distance_km'] * 5.5 +
                   features['is_peak_hour'] * 7 +
                   features['restaurant_avg'] * 0.3 +
                   np.random.normal(0, 5))
        actual = float(np.clip(actual, 10, 90))
        shadow.receive_label(f'REQ{i:04d}', actual)

gate = shadow.shadow_gate(mae_tolerance=0.10)
print(f"Shadow deployment gate ({gate['n_labelled']} labelled requests):")
print(f"  Champion MAE:    {gate['champion_mae']}")
print(f"  Challenger MAE:  {gate['challenger_mae']}")
print(f"  Improvement:     {gate['improvement_pct']:+.2f}%")
print(f"  Divergence:      {gate['prediction_divergence']} min avg")
print(f"  Gate result:     {'✓ PASSED' if gate['passed'] else '✗ FAILED'}")

import numpy as np
from scipy import stats
from dataclasses import dataclass, field
from collections import defaultdict
import hashlib

@dataclass
class ABExperiment:
    """
    Routes traffic between champion (control) and challenger (treatment).
    Uses consistent hashing on request_id so each user always goes
    to the same model — prevents a user seeing mixed predictions.
    """
    champion_model:   object
    challenger_model: object
    feature_columns:  list[str]
    challenger_pct:   float = 0.10   # 10% to challenger
    experiment_id:    str   = 'exp_001'

    champion_errors:   list = field(default_factory=list)
    challenger_errors: list = field(default_factory=list)
    champion_preds:    list = field(default_factory=list)
    challenger_preds:  list = field(default_factory=list)

    def _route(self, user_id: str) -> str:
        """
        Consistent hash routing — same user always goes to same model.
        Prevents seeing different estimates for the same order.
        """
        h     = int(hashlib.md5(f"{self.experiment_id}:{user_id}".encode()).hexdigest(), 16)
        bucket = (h % 100) / 100.0
        return 'challenger' if bucket < self.challenger_pct else 'champion'

    def predict(self, features: dict, user_id: str) -> dict:
        X     = np.array([features.get(c, 0) for c in self.feature_columns]).reshape(1, -1)
        group = self._route(user_id)

        if group == 'challenger':
            pred = float(self.challenger_model.predict(X)[0])
            self.challenger_preds.append(pred)
        else:
            pred = float(self.champion_model.predict(X)[0])
            self.champion_preds.append(pred)

        return {'prediction': pred, 'group': group, 'user_id': user_id}

    def record_outcome(self, user_id: str, actual: float):
        """Record actual outcome for a user — joins to their prediction."""
        group = self._route(user_id)
        # Find the most recent prediction for this user
        # In production: join via prediction_id stored in prediction log
        if group == 'challenger' and self.challenger_preds:
            # Simplified — in production join by prediction_id
            error = abs(self.challenger_preds[-1] - actual)
            self.challenger_errors.append(error)
        elif self.champion_preds:
            error = abs(self.champion_preds[-1] - actual)
            self.champion_errors.append(error)

    def analyse(self, min_samples: int = 200) -> dict:
        """
        Statistical analysis of A/B test results.
        Uses Welch's t-test (unequal variances, unequal sample sizes).
        """
        ce = np.array(self.champion_errors)
        he = np.array(self.challenger_errors)

        if len(ce) < min_samples or len(he) < min_samples:
            return {
                'status':  'insufficient_data',
                'champion_n':   len(ce),
                'challenger_n': len(he),
                'min_required': min_samples,
            }

        champ_mae = ce.mean()
        chal_mae  = he.mean()

        # Welch's t-test — does not assume equal variance
        t_stat, p_value = stats.ttest_ind(ce, he, equal_var=False)
        # One-sided: is challenger significantly BETTER (lower errors)?
        p_one_sided = p_value / 2 if t_stat > 0 else 1 - p_value / 2

        # Effect size (Cohen's d)
        pooled_std = np.sqrt((ce.std()**2 + he.std()**2) / 2)
        cohens_d   = (champ_mae - chal_mae) / (pooled_std + 1e-8)

        significant = p_one_sided < 0.05 and chal_mae < champ_mae

        return {
            'status':          'complete',
            'champion_mae':    round(champ_mae, 4),
            'challenger_mae':  round(chal_mae, 4),
            'improvement_pct': round((champ_mae - chal_mae) / champ_mae * 100, 2),
            'p_value':         round(p_one_sided, 4),
            'cohens_d':        round(cohens_d, 4),
            'significant':     significant,
            'champion_n':      len(ce),
            'challenger_n':    len(he),
            'promote':         significant and chal_mae < champ_mae * 0.99,
        }

# ── Simulate A/B experiment ───────────────────────────────────────────
ab = ABExperiment(
    champion_model=champion,
    challenger_model=challenger,
    feature_columns=list(X.columns),
    challenger_pct=0.10,
)

np.random.seed(456)
for i in range(3000):
    user_id  = f'USER{np.random.randint(0, 1000):04d}'
    features = {
        'distance_km':    float(np.random.exponential(3.5)),
        'is_peak_hour':   float(np.random.randint(0, 2)),
        'order_value':    float(np.random.exponential(400)),
        'restaurant_avg': float(np.random.normal(35, 8)),
        'driver_avg':     float(np.random.normal(32, 7)),
    }
    result = ab.predict(features, user_id)

    # 80% of requests get labels (delivery completed)
    if np.random.random() < 0.8:
        actual = float(np.clip(
            features['distance_km'] * 5.5 + features['is_peak_hour'] * 7
            + features['restaurant_avg'] * 0.3 + np.random.normal(0, 5), 10, 90,
        ))
        ab.record_outcome(user_id, actual)

analysis = ab.analyse(min_samples=100)
print("A/B test analysis:")
for k, v in analysis.items():
    if k == 'status': continue
    print(f"  {k:<22}: {v}")

decision = 'PROMOTE challenger' if analysis.get('promote') else 'KEEP champion'
print(f"
Decision: {decision}")

import time
import numpy as np
from dataclasses import dataclass, field
from datetime import datetime
from typing import Callable

@dataclass
class TrafficSplit:
    champion_pct:  float   # 0.0 to 1.0
    challenger_pct: float  # 1 - champion_pct
    updated_at:    str = ''

    def __post_init__(self):
        assert abs(self.champion_pct + self.challenger_pct - 1.0) < 1e-6

class GradualPromotion:
    """
    Manages progressive traffic shifting from champion to challenger.
    Monitors real-time metrics and auto-rolls back on degradation.
    """
    PROMOTION_STEPS = [0.10, 0.25, 0.50, 1.00]   # challenger traffic fraction
    STEP_DURATION_S = 3600   # 1 hour per step in production (use seconds for demo)

    def __init__(self,
                  champion_model,
                  challenger_model,
                  feature_columns: list[str],
                  alert_fn: Callable = print):
        self.champion    = champion_model
        self.challenger  = challenger_model
        self.features    = feature_columns
        self.alert       = alert_fn
        self.split       = TrafficSplit(champion_pct=1.0, challenger_pct=0.0)
        self.step_idx    = -1
        self.errors      = {'champion': [], 'challenger': []}
        self.latencies   = {'champion': [], 'challenger': []}
        self.promoted    = False
        self.rolled_back = False
        self.history     = []

    def _serve(self, features: dict, use_challenger: bool) -> tuple[float, float]:
        X        = np.array([features.get(c, 0) for c in self.features]).reshape(1, -1)
        t_start  = time.time()
        model    = self.challenger if use_challenger else self.champion
        pred     = float(model.predict(X)[0])
        latency  = (time.time() - t_start) * 1000
        return pred, latency

    def predict(self, features: dict, user_id: str) -> dict:
        """Route request based on current traffic split."""
        import hashlib
        h            = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
        use_challenger = (h % 100) / 100.0 < self.split.challenger_pct

        pred, latency = self._serve(features, use_challenger)
        group         = 'challenger' if use_challenger else 'champion'
        self.latencies[group].append(latency)
        return {'prediction': pred, 'group': group}

    def record_error(self, group: str, error: float):
        self.errors[group].append(error)

    def check_rollback_conditions(self) -> dict:
        """Check if challenger metrics have degraded enough to roll back."""
        ce  = self.errors.get('champion', [])
        he  = self.errors.get('challenger', [])
        cl  = self.latencies.get('champion', [])
        hl  = self.latencies.get('challenger', [])

        if len(he) < 50:
            return {'should_rollback': False, 'reason': 'insufficient_data'}

        champ_mae = np.mean(ce) if ce else 0
        chal_mae  = np.mean(he)

        # Rollback condition 1: MAE regression
        mae_regression = (chal_mae > champ_mae * 1.15) if ce else False

        # Rollback condition 2: p99 latency spike
        chal_p99  = np.percentile(hl, 99) if hl else 0
        champ_p99 = np.percentile(cl, 99) if cl else 0
        latency_regression = chal_p99 > champ_p99 * 1.5 and chal_p99 > 500

        should_rollback = mae_regression or latency_regression
        return {
            'should_rollback':    should_rollback,
            'mae_regression':     mae_regression,
            'latency_regression': latency_regression,
            'challenger_mae':     round(chal_mae, 3),
            'champion_mae':       round(champ_mae, 3) if ce else None,
            'challenger_p99_ms':  round(chal_p99, 1),
        }

    def advance_step(self) -> bool:
        """Move to the next traffic split step if conditions are met."""
        if self.promoted or self.rolled_back:
            return False

        rb = self.check_rollback_conditions()
        if rb['should_rollback']:
            self._rollback(rb)
            return False

        self.step_idx += 1
        if self.step_idx >= len(self.PROMOTION_STEPS):
            self._complete_promotion()
            return False

        new_chal = self.PROMOTION_STEPS[self.step_idx]
        self.split = TrafficSplit(
            champion_pct=1 - new_chal,
            challenger_pct=new_chal,
            updated_at=datetime.now().isoformat(),
        )
        self.history.append({'step': self.step_idx, 'split': new_chal,
                              'at': datetime.now().isoformat()})
        print(f"  → Step {self.step_idx+1}: {new_chal:.0%} challenger / "
              f"{1-new_chal:.0%} champion")
        return True

    def _rollback(self, reason: dict):
        self.split       = TrafficSplit(champion_pct=1.0, challenger_pct=0.0)
        self.rolled_back = True
        self.alert(f"[ROLLBACK] Auto-rollback triggered: {reason}")

    def _complete_promotion(self):
        self.split    = TrafficSplit(champion_pct=0.0, challenger_pct=1.0)
        self.promoted = True
        self.alert("[PROMOTION] Challenger fully promoted to production ✓")

# ── Simulate gradual promotion ─────────────────────────────────────────
promo = GradualPromotion(champion, challenger, list(X.columns))
np.random.seed(789)

print("Gradual promotion simulation:")
for step in range(5):
    advanced = promo.advance_step()
    if not advanced:
        break

    # Simulate 200 requests at this traffic level
    for _ in range(200):
        uid      = f'USER{np.random.randint(0, 500):03d}'
        features = {c: float(np.random.randn()) for c in X.columns}
        result   = promo.predict(features, uid)

        # Simulate actual delivery time arriving
        actual = 35.0 + np.random.normal(0, 8)
        error  = abs(result['prediction'] - actual)
        promo.record_error(result['group'], error)

    rb = promo.check_rollback_conditions()
    print(f"    Metrics: challenger_mae={rb['challenger_mae']} "
          f"p99={rb['challenger_p99_ms']}ms "
          f"rollback={rb['should_rollback']}")

print(f"
Final state: promoted={promo.promoted}  rolled_back={promo.rolled_back}")
print(f"Traffic: {promo.split.challenger_pct:.0%} challenger / "
      f"{promo.split.champion_pct:.0%} champion")

Every retrain has a real cost beyond compute: a new model version means a new set of quirks to debug if something goes wrong, and it becomes harder to tell which of several recent changes caused a regression when the model underneath keeps shifting weekly or daily. Retraining on a schedule that outpaces how fast the underlying world actually changes mostly means chasing statistical noise in the training data rather than real drift. A model that is stable and well-validated is often worth more than one that is marginally fresher but has been swapped five times this month — freshness is a means to better predictions, not a goal by itself.

A pipeline finishing without throwing an exception only confirms the training job did not crash — it says nothing at all about whether the resulting model is any good. A model trained on corrupted, leaked, or simply worse data will complete training successfully and produce a confident-looking artifact that is quietly wrong. That is exactly why this module's five stages exist as separate checks after training completes: the offline quality gate, the shadow deployment comparison, and the champion-challenger A/B test all evaluate something the training job itself cannot see — whether the new model is actually better, not just whether it finished running.

A held-out test set is a fixed snapshot from whenever it was created, and production traffic keeps moving further from that snapshot every day it is not refreshed. A challenger can beat the champion cleanly on that stale test set and still perform worse on the live distribution, because the test set no longer represents what today's traffic actually looks like. This is precisely why the pipeline does not stop at the offline gate — shadow deployment checks the challenger against real live traffic before any user sees its prediction, and gradual promotion limits the blast radius even after that, instead of trusting one offline number to justify a full instant cutover.

Retraining adjusts the weights a model assigns to the features it already has — it cannot invent a signal that was never captured in the first place. If the features genuinely lack the information needed to predict the target, or the one feature that actually matters was dropped or never collected, no cadence of retraining closes that gap; more frequent retraining on the same insufficient features just produces a new model that is confidently wrong in a slightly different way. That is a feature engineering and data collection problem, and it gets solved by adding or fixing features, not by running the same pipeline more often.

A model promotion often ships alongside other changes — an updated feature schema, a new feature-store column, a changed preprocessing step, a config value the new model depends on. Reverting only the model artifact while those dependent changes stay in place can leave the old model receiving inputs it was never trained to handle, which is sometimes worse than the failure the rollback was meant to fix. A safe rollback plan accounts for everything that shipped together, not just the model file, which is exactly why keeping the previous deployment fully intact (rather than deleting it) until the new version is proven is worth the extra resource cost during promotion.

I would start from how quickly the model's world actually changes and how costly a stale prediction is, not from an arbitrary schedule. A demand or delivery-time model exposed to strong seasonality needs a tighter cadence or drift-based triggers, while a model whose relationships are fairly stable might only need monthly retraining. I would also weigh the cost of a bad promotion: for a low-stakes ranking model, a fast, largely automated cadence gated by offline and shadow checks is reasonable; for a lending or fraud model, I would favour a slower cadence with mandatory human review, since the cost of promoting a subtly worse model there is much higher than the cost of a few extra days of staleness.

I would run it through staged gates rather than trusting any single check. First, an offline quality gate comparing the challenger to the champion on a fixed, recently-refreshed held-out test set. Second, shadow deployment — running the challenger on live traffic in parallel with zero user impact, to catch cases where production data has drifted away from the test set. Third, a champion-challenger A/B test that sends a small percentage of real traffic to the challenger and requires a statistically significant improvement, not just a favourable-looking average. Only after all three pass would I begin a gradual traffic ramp with automated rollback triggers watching latency and error metrics at every step.

I would avoid full automatic retrain-and-promote when the model is high-stakes enough that a bad version reaching all traffic would be costly or hard to reverse quickly — regulated domains like credit or healthcare are the clearest examples. I would also hold off on auto-retraining when the trigger itself is ambiguous, such as drift that might just be a temporary seasonal blip rather than a genuine permanent shift — retraining on a fluke can bake noise into the new model. And I would not auto-retrain immediately after a major upstream change, like a new feature pipeline or a business rule change, until a human has confirmed the new data is actually correct, since a pipeline bug upstream would otherwise get trained into the model silently.

The offline gate compares the challenger to the champion on a fixed held-out test set that never changes — fast, cheap, and a reasonable first filter, but only as representative of production as that test set currently is. The shadow deployment gate compares the two models on live, current production traffic instead, catching cases where the test set's distribution has quietly diverged from what the system is actually seeing today. Relying only on the offline gate risks promoting a model that looks great on old data and mediocre on today's; relying only on shadow deployment means waiting longer and spending more compute before catching an obviously bad candidate that the cheap offline gate would have rejected immediately.

First, I would ask when the held-out test set was last refreshed — metrics that look "too good" are a classic signal of a stale test set or, worse, temporal leakage where the model saw information from after the prediction point during training. Second, I would ask whether it has been run through shadow deployment on live traffic yet, since an offline win does not guarantee a live one. Third, I would ask what actually changed between the two models — new features, more training data, a different architecture — because "better on paper" for the wrong reason (a subtle leak) looks identical to "better for the right reason" until you dig into why the numbers moved.