Case 5 — Design a Retraining Pipeline for a Fraud Model

# Case 5 — Design a Retraining Pipeline for a Fraud Model > *"Your fraud model has been deployed for 3 months. Drift detection alerts fire every few weeks. Design a retraining pipeline: cadence, validation gates, deployment process."* --- ## Step 1 — Pick the cadence Three retraining options: ### Scheduled monthly Simple, predictable. Most teams start here. ### Triggered by drift signals Retrain when monitoring fires. Responsive but unpredictable. ### Hybrid: scheduled + triggered Monthly baseline + on-demand if drift is severe. For fraud (adversarial domain, drift can be sudden): - **Scheduled weekly** for baseline freshness. - **Triggered** if PR-AUC drops more than 5% or feature drift is severe. Document the rule: > "Retraining triggers: (a) every Sunday at 02:00 UTC, (b) when 7-day rolling PR-AUC drops below 0.35 (baseline 0.45), (c) when PSI on any top-5 feature exceeds 0.30." ## Step 2 — Build the retraining pipeline `retrain.py`: ```python import argparse import json import os from datetime import datetime, timedelta import mlflow import mlflow.sklearn import pandas as pd from sklearn.metrics import roc_auc_score, average_precision_score from xgboost import XGBClassifier from features.feature_engineering import build_features from data.loaders import load_training_window from validation.tests import validate_candidate def main(args): mlflow.set_experiment('fraud-detection-retraining') with mlflow.start_run(run_name=f'retrain_{datetime.utcnow():%Y%m%d_%H%M}'): mlflow.set_tag('triggered_by', args.trigger) mlflow.set_tag('data_version', args.data_version) mlflow.set_tag('schedule_or_drift', args.cadence) # Load fresh training data — last 180 days, time-based split end_date = datetime.utcnow().date() start_date = end_date - timedelta(days=180) train_end = end_date - timedelta(days=30) val_end = end_date - timedelta(days=14) data = load_training_window(start_date, end_date) train = data[data.date = train_end) & (data.date = val_end] mlflow.log_metric('train_rows', len(train)) mlflow.log_metric('val_rows', len(val)) mlflow.log_metric('test_rows', len(test)) # Feature engineering (shared with serving) X_train = train.apply(build_features, axis=1) X_val = val.apply(build_features, axis=1) X_test = test.apply(build_features, axis=1) # Train candidate candidate = XGBClassifier( n_estimators=10000, learning_rate=0.03, max_depth=6, scale_pos_weight=(train.is_fraud == 0).sum() / (train.is_fraud == 1).sum(), eval_metric='aucpr', early_stopping_rounds=100, random_state=42, ) candidate.fit(X_train, train.is_fraud, eval_set=[(X_val, val.is_fraud)]) # Evaluate candidate test_pred = candidate.predict_proba(X_test)[:, 1] candidate_auc = roc_auc_score(test.is_fraud, test_pred) candidate_pr_auc = average_precision_score(test.is_fraud, test_pred) mlflow.log_metric('candidate_test_auc', candidate_auc) mlflow.log_metric('candidate_test_pr_auc', candidate_pr_auc) # Load current production model and compare prod_model = mlflow.sklearn.load_model('models:/fraud_detector/Production') prod_pred = prod_model.predict_proba(X_test)[:, 1] prod_auc = roc_auc_score(test.is_fraud, prod_pred) prod_pr_auc = average_precision_score(test.is_fraud, prod_pred) mlflow.log_metric('current_prod_test_auc', prod_auc) mlflow.log_metric('current_prod_test_pr_auc', prod_pr_auc) mlflow.log_metric('pr_auc_improvement', candidate_pr_auc - prod_pr_auc) # Save candidate mlflow.sklearn.log_model(candidate, 'model') # Validation gates validation = validate_candidate(candidate, prod_model, X_test, test, args.thresholds) mlflow.log_dict(validation, 'validation.json') if validation['passed']: # Register, but don't auto-promote mlflow.register_model( f'runs:/{mlflow.active_run().info.run_id}/model', name='fraud_detector' ) print(f'New candidate registered. Manual review for Production promotion.') else: print(f'Candidate failed validation: {validation["failed_checks"]}') print(f'No new model registered. Investigate.') # Alert send_slack_alert(f"Retraining FAILED validation: {validation['failed_checks']}") if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--data-version', default='auto') parser.add_argument('--trigger', default='scheduled') parser.add_argument('--cadence', default='weekly') parser.add_argument('--thresholds', default='configs/retrain