model-retraining

name	model-retraining
description	Design automated ML model retraining pipelines triggered by drift, schedule, or performance degradation. Outputs retraining orchestration, data validation, automated promotion logic, and rollback mechanisms.
argument-hint	["model type","retraining frequency","data volume","deployment target","approval process"]
allowed-tools	Read, Write, Bash

Model Retraining Pipeline

Models are not static artifacts — they need to be retrained as the world changes. Design automated pipelines that retrain safely, validate rigorously, and deploy only when the new model is provably better.

Process

1. Define retraining triggers — scheduled, drift-based, performance-based, data-volume-based.
2. Design data pipeline — fetch recent data, validate quality, compute features.
3. Train with same config — reproduce hyperparameters from the winning experiment.
4. Evaluate challenger vs. champion — hold-out test set comparison.
5. Automated promotion logic — go/no-go rules that prevent regression.
6. Shadow deployment — run challenger alongside champion before full promotion.
7. Rollback mechanism — instant revert if production metrics degrade.
8. Audit trail — every retrain logged with data snapshot, metrics, approver.

Output Format

Retraining Orchestration (Prefect)

# retraining/pipeline.py
from prefect import flow, task, get_run_logger
from prefect.deployments import Deployment
from prefect.blocks.system import Secret
from datetime import datetime, timezone, timedelta
import pandas as pd
import numpy as np
import mlflow
import json

@task(retries=3, retry_delay_seconds=60, name="fetch_training_data")
def fetch_training_data(
    lookback_days: int = 90,
    min_samples: int = 10000
) -> pd.DataFrame:
    logger = get_run_logger()
    
    end_date = datetime.now(timezone.utc)
    start_date = end_date - timedelta(days=lookback_days)
    
    logger.info(f"Fetching training data from {start_date} to {end_date}")
    
    df = pd.read_parquet(
        "s3://ml-data/features/",
        filters=[
            ("date", ">=", start_date.date()),
            ("date", "<", end_date.date()),
        ]
    )
    
    if len(df) < min_samples:
        raise ValueError(
            f"Insufficient training data: {len(df)} samples (min: {min_samples})"
        )
    
    logger.info(f"Fetched {len(df):,} training samples")
     df



 () -> :
    logger = get_run_logger()
    
    issues = []
    
    
    null_rates = df.isnull().mean()
    high_null = null_rates[null_rates > ]
      high_null.empty:
        issues.append()
    
    
       df.columns:
        positive_rate = df[].mean()
         positive_rate <   positive_rate > :
            issues.append()
    
    
       df.columns:
        issues.append()
    
    
       df.columns:
        date_range = df[].agg([, ])
        logger.info()
    
    quality_report = {
        : (df),
        : null_rates.to_dict(),
        : issues,
        : (issues) == ,
    }
    
      quality_report[]:
        logger.warning()
    
     quality_report



 () -> pd.DataFrame:
    
     feature_store  FeatureTransformer
    
    transformer = FeatureTransformer.load()
     transformer.transform(df)



 () -> :
    logger = get_run_logger()
    
    
    client = mlflow.tracking.MlflowClient()
    champion_run = client.get_run(champion_run_id)
    params = champion_run.data.params
    
    logger.info()
    
    
    split_date = df[].quantile()
    train_df = df[df[] <= split_date]
    val_df = df[df[] > split_date]
    
     mlflow.start_run(
        run_name=,
        tags={: , : champion_run_id}
    )  run:
        mlflow.log_params({
            **params,
            : (train_df[].()),
            : (train_df[].()),
            : (train_df),
            : (val_df),
        })
        
        
         training.trainer  train_model
        model, metrics = train_model(
            train_df,
            val_df,
            params=params
        )
        
        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, )
        
        run_id = run.info.run_id
        logger.info()
    
     run_id



 () -> :
    logger = get_run_logger()
    
    
    champion_model = mlflow.pyfunc.load_model()
    challenger_model = mlflow.pyfunc.load_model()
    
    X_test = test_df.drop(columns=[, ])
    y_test = test_df[].values
    
    
     sklearn.metrics  roc_auc_score, f1_score
    
    champ_prob = champion_model.predict(X_test)
    chall_prob = challenger_model.predict(X_test)
    
    champ_auc = roc_auc_score(y_test, champ_prob)
    chall_auc = roc_auc_score(y_test, chall_prob)
    
    improvement = chall_auc - champ_auc
    
    comparison = {
        : champ_auc,
        : chall_auc,
        : improvement,
        : improvement / champ_auc * ,
        : improvement >= improvement_threshold,
        :   improvement >= improvement_threshold  ,
    }
    
    logger.info(
        
        
    )
    
    
     mlflow.start_run(challenger_run_id, nested=):
        mlflow.log_metrics({
            : champ_auc,
            : chall_auc,
            : improvement,
        })
    
     comparison



 () -> :
    logger = get_run_logger()
    
      comparison[]:
        logger.info()
         {: , : }
    
     require_approval:
        
        send_approval_request(model_name, comparison)
        logger.info()
         {: , : , : comparison}
    
    
    client = mlflow.tracking.MlflowClient()
    
    
    model_uri = 
    result = mlflow.register_model(model_uri, model_name)
    
    
    prod_versions = client.get_latest_versions(model_name, stages=[])
     v  prod_versions:
        client.transition_model_version_stage(
            name=model_name,
            version=v.version,
            stage=
        )
    
    
    client.transition_model_version_stage(
        name=model_name,
        version=result.version,
        stage=
    )
    
    logger.info()
     {
        : ,
        : result.version,
        : comparison,
    }



 ():
    logger = get_run_logger()
    logger.info()
    
    
    raw_df = fetch_training_data(lookback_days=lookback_days)
    quality_report = validate_data_quality(raw_df)
    
      quality_report[]:
         ValueError()
    
    
    features_df = run_feature_engineering(raw_df)
    
    
    client = mlflow.tracking.MlflowClient()
    champion_versions = client.get_latest_versions(model_name, stages=[])
    champion_run_id = champion_versions[].run_id  champion_versions  
    
    
    challenger_run_id = train_challenger_model(features_df, champion_run_id)
    
    
    test_df = pd.read_parquet()
    test_features = run_feature_engineering(test_df)
    
    comparison = evaluate_challenger_vs_champion(
        challenger_run_id,
        model_name,
        test_features,
        improvement_threshold
    )
    
    
    result = promote_challenger_to_production(
        challenger_run_id,
        model_name,
        comparison,
        require_approval
    )
    
    logger.info()
     result

Deployment Configuration

# deploy_retraining.py
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

# Scheduled retraining (weekly)
weekly_deployment = Deployment.build_from_flow(
    flow=retraining_pipeline,
    name="weekly-retrain",
    schedule=CronSchedule(cron="0 2 * * 0"),  # Sunday 2 AM
    parameters={
        "model_name": "order-propensity",
        "lookback_days": 90,
        "require_approval": True,  # Require human sign-off for scheduled
    },
    tags=["ml", "retraining", "scheduled"],
    work_pool_name="ml-workers"
)

# Drift-triggered retraining (auto-approve for urgent cases)
drift_deployment = Deployment.build_from_flow(
    flow=retraining_pipeline,
    name="drift-triggered-retrain",
    parameters={
        "model_name": "order-propensity",
        "lookback_days": 60,
        "improvement_threshold": 0.0,  # Any improvement is acceptable
        "require_approval": False,      # Auto-promote on drift
    },
    tags=["ml", "retraining", "drift-triggered"],
    work_pool_name="ml-workers"
)

weekly_deployment.apply()
drift_deployment.apply()

Rollback Mechanism

# rollback.py
def rollback_to_previous(model_name: str, reason: str):
    """Instantly revert to the previously archived model version."""
    client = mlflow.tracking.MlflowClient()
    
    # Find most recent Archived version
    archived = client.get_latest_versions(model_name, stages=["Archived"])
    if not archived:
        raise ValueError("No archived version to roll back to")
    
    prev_version = sorted(archived, key=lambda v: v.version, reverse=True)[0]
    
    # Archive current Production
    prod_versions = client.get_latest_versions(model_name, stages=["Production"])
    for v in prod_versions:
        client.transition_model_version_stage(
            name=model_name,
            version=v.version,
            stage="Archived"
        )
        client.update_model_version(
            name=model_name,
            version=v.version,
            description=f"Archived due to rollback: {reason}"
        )
    
    # Restore previous version
    client.transition_model_version_stage(
        name=model_name,
        version=prev_version.version,
        stage="Production"
    )
    
    logger.info(
        f"Rolled back {model_name} from v{prod_versions[0].version} "
        f"to v{prev_version.version}. Reason: {reason}"
    )

Rules

• Never retrain on the same test set used for promotion — maintain a truly held-out evaluation set.
• Champion vs. challenger on identical data — the comparison must be apples-to-apples.
• Minimum improvement threshold — don't promote a model that's only marginally better (noise territory).
• Validate data quality before training — bad data silently produces bad models.
• Log everything — data snapshot hash, git commit, hyperparameters, evaluation metrics, promoter.
• Keep archived models for at least 90 days — rollback window must be real.
• Test the rollback — actually execute rollback in staging quarterly.
• Shadow mode before full promotion — run challenger in shadow for high-stakes models.
• Retraining ≠ full re-architecture — hyperparameter changes require separate experiments, not automatic retraining.
• Alert on failed retraining jobs — a silently failing retraining pipeline is worse than no pipeline.