MLOps on Cloud — CI/CD for ML

A car manufacturer runs two quality checks. The design check happens when engineers submit a blueprint change — does the new design meet safety standards on paper? The production check happens on the assembly line — does the manufactured car meet standards in reality? ML CI/CD is the same: code-change tests (does the training script work correctly on a small sample?) and data-change tests (does the model trained on new production data beat the current production model?). Neither check alone is sufficient. Together they ensure nothing bad reaches customers.

The key shift in mindset: in ML, the model is not just code — it is code plus data plus hyperparameters. CI/CD must test all three dimensions simultaneously. A code change that looks fine in unit tests might produce a degraded model when combined with production data. The full pipeline test is the only reliable check.

# .github/workflows/ml-ci.yml
# Runs on every PR that touches ML code

name: ML Code CI

on:
  pull_request:
    paths:
      - 'src/**'
      - 'data/**'
      - 'requirements*.txt'
      - 'params.yaml'

env:
  PYTHON_VERSION: '3.11'

jobs:
  # ── Job 1: Unit tests ─────────────────────────────────────────────
  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: pip

      - name: Install dependencies
        run: pip install -r requirements-dev.txt

      - name: Run unit tests
        run: pytest src/ -v --tb=short --junit-xml=test-results.xml

      - name: Upload test results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: unit-test-results
          path: test-results.xml

  # ── Job 2: Training smoke test (no cloud — small data) ────────────
  training-smoke-test:
    name: Training Smoke Test
    runs-on: ubuntu-latest
    needs: unit-tests

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: pip

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Generate small test dataset
        run: python tests/generate_test_data.py --n-rows 500 --output /tmp/test_data.csv

      - name: Run training on sample data
        run: |
          python src/train.py \
            --data-path /tmp/test_data.csv \
            --output-dir /tmp/model-output \
            --n-estimators 10 \
            --max-depth 3

      - name: Verify model output exists
        run: |
          python -c "
          import pickle, os, json
          assert os.path.exists('/tmp/model-output/model.pkl'), 'model.pkl missing'
          with open('/tmp/model-output/model.pkl', 'rb') as f:
              art = pickle.load(f)
          assert 'model' in art, 'model key missing from artifact'
          assert 'feature_cols' in art, 'feature_cols key missing'
          print('Model artifact validation passed')
          print(f'Features: {art[\"feature_cols\"]}')
          "

      - name: Verify prediction schema
        run: |
          python -c "
          import pickle, numpy as np, json
          with open('/tmp/model-output/model.pkl', 'rb') as f:
              art = pickle.load(f)
          model        = art['model']
          feature_cols = art['feature_cols']
          test_input   = np.zeros((3, len(feature_cols)))
          preds        = model.predict(test_input)
          assert preds.shape == (3,), f'Wrong output shape: {preds.shape}'
          assert all(5 <= p <= 120 for p in preds), f'Predictions out of range: {preds}'
          print(f'Schema test passed: {preds}')
          "

      - name: Check metrics are reasonable
        run: |
          python -c "
          import json
          with open('/tmp/model-output/metrics.json') as f:
              m = json.load(f)
          # On tiny 500-row sample, MAE should be computable
          assert 'train_mae' in m, 'train_mae missing from metrics'
          assert m['train_mae'] > 0, 'train_mae must be positive'
          assert m['train_mae'] < 50, f'train_mae suspiciously high: {m["train_mae"]}'
          print(f'Metrics check passed: {m}')
          "

  # ── Job 3: Code quality ───────────────────────────────────────────
  code-quality:
    name: Code Quality
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}' }

      - run: pip install ruff mypy
      - run: ruff check src/
      - run: mypy src/ --ignore-missing-imports

# .github/workflows/ml-retrain.yml
# Runs on schedule AND when triggered manually or by drift alert

name: ML Retraining Pipeline

on:
  schedule:
    - cron: '30 20 * * 0'   # Every Sunday 20:30 UTC = Monday 02:00 IST
  workflow_dispatch:          # Manual trigger from GitHub UI
    inputs:
      reason:
        description: 'Reason for manual retraining'
        required: true
        default: 'drift-detected'
      force_promote:
        description: 'Force promote even if challenger is not better'
        type: boolean
        default: false
  repository_dispatch:        # Triggered by external system (drift monitor)
    types: [drift-detected, scheduled-retrain]

env:
  PYTHON_VERSION: '3.11'
  MODEL_NAME:     freshmart-delivery-time
  REGISTRY_NAME:  FreshMartDeliveryTime

jobs:
  # ── Stage 1: Unit tests (fast, no cloud) ─────────────────────────
  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }
      - run: pip install -r requirements-dev.txt
      - run: pytest src/ -v --tb=short

  # ── Stage 2: Cloud training job ───────────────────────────────────
  cloud-training:
    name: Cloud Training Job
    runs-on: ubuntu-latest
    needs: unit-tests
    outputs:
      job_name:      ${{ steps.submit.outputs.job_name }}
      challenger_mae: ${{ steps.metrics.outputs.mae }}
      champion_mae:   ${{ steps.compare.outputs.champion_mae }}
      should_promote: ${{ steps.compare.outputs.should_promote }}

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }

      - name: Install cloud SDK
        run: pip install azure-ai-ml azure-identity

      # Azure credentials injected from GitHub Secrets
      - name: Azure login
        uses: azure/login@v2
        with:
          creds: ${{ secrets.AZURE_CREDENTIALS }}

      - name: Submit training job
        id: submit
        run: |
          python ci/submit_training_job.py \
            --subscription-id ${{ secrets.AZURE_SUBSCRIPTION_ID }} \
            --resource-group  rg-freshmart-ml \
            --workspace       ws-freshmart-ml \
            --cutoff-date     $(date +%Y-%m-%d) \
            --output-job-name /tmp/job_name.txt

          JOB_NAME=$(cat /tmp/job_name.txt)
          echo "job_name=${JOB_NAME}" >> $GITHUB_OUTPUT
          echo "Submitted: ${JOB_NAME}"

      - name: Wait for training job
        run: |
          python ci/wait_for_job.py \
            --job-name ${{ steps.submit.outputs.job_name }} \
            --timeout  3600   # 1 hour max

      - name: Retrieve job metrics
        id: metrics
        run: |
          python ci/get_job_metrics.py \
            --job-name ${{ steps.submit.outputs.job_name }} \
            --output   /tmp/challenger_metrics.json

          MAE=$(python -c "import json; print(json.load(open('/tmp/challenger_metrics.json'))['val_mae'])")
          echo "mae=${MAE}" >> $GITHUB_OUTPUT
          echo "Challenger MAE: ${MAE}"

      - name: Compare challenger to champion
        id: compare
        run: |
          python ci/compare_models.py \
            --challenger-metrics /tmp/challenger_metrics.json \
            --registry-name      ${{ env.REGISTRY_NAME }} \
            --tolerance          0.05 \
            --output             /tmp/comparison.json

          CHAMP_MAE=$(python -c "import json; print(json.load(open('/tmp/comparison.json'))['champion_mae'])")
          PROMOTE=$(python -c "import json; print(json.load(open('/tmp/comparison.json'))['should_promote'])")
          echo "champion_mae=${CHAMP_MAE}" >> $GITHUB_OUTPUT
          echo "should_promote=${PROMOTE}" >> $GITHUB_OUTPUT

      - name: Register challenger model
        run: |
          python ci/register_model.py \
            --job-name    ${{ steps.submit.outputs.job_name }} \
            --model-name  ${{ env.MODEL_NAME }} \
            --stage       Staging

      - name: Post metrics to PR / Slack
        run: |
          python ci/notify.py \
            --challenger-mae ${{ steps.metrics.outputs.mae }} \
            --champion-mae   ${{ steps.compare.outputs.champion_mae }} \
            --should-promote ${{ steps.compare.outputs.should_promote }} \
            --slack-webhook  ${{ secrets.SLACK_WEBHOOK_ML_ALERTS }}

  # ── Stage 3: Staging tests ────────────────────────────────────────
  staging-tests:
    name: Staging Deployment Tests
    runs-on: ubuntu-latest
    needs: cloud-training
    if: needs.cloud-training.outputs.should_promote == 'True'

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }
      - run: pip install requests pytest

      - name: Deploy to staging endpoint
        run: |
          python ci/deploy_to_staging.py \
            --model-name ${{ env.MODEL_NAME }} \
            --endpoint   freshmart-delivery-staging

      - name: Run integration tests against staging
        run: |
          pytest tests/integration/ \
            --endpoint freshmart-delivery-staging \
            --n-samples 100 \
            -v

      - name: Load test staging endpoint
        run: |
          python ci/load_test.py \
            --endpoint      freshmart-delivery-staging \
            --rps            50 \
            --duration-s    30 \
            --max-p99-ms    500 \
            --max-error-pct 1.0

  # ── Stage 4: Production promotion (with optional approval gate) ───
  production-promote:
    name: Promote to Production
    runs-on: ubuntu-latest
    needs: [cloud-training, staging-tests]
    environment:
      name: production          # GitHub environment — can require manual approval
      url:  https://freshmart-delivery-endpoint.azurewebsites.net/health

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }

      - name: Gradual traffic promotion
        run: |
          python ci/gradual_promote.py \
            --model-name    ${{ env.MODEL_NAME }} \
            --endpoint      freshmart-delivery-endpoint \
            --traffic-steps "10,25,50,100" \
            --step-wait-s   600 \  # 10 min at each step
            --max-p99-ms    500 \
            --max-error-pct 1.0

      - name: Archive old production model
        run: |
          python ci/archive_old_model.py \
            --model-name ${{ env.MODEL_NAME }} \
            --keep-versions 3

      - name: Notify success
        run: |
          python ci/notify.py \
            --event promoted \
            --model-name    ${{ env.MODEL_NAME }} \
            --challenger-mae ${{ needs.cloud-training.outputs.challenger_mae }} \
            --champion-mae   ${{ needs.cloud-training.outputs.champion_mae }} \
            --slack-webhook ${{ secrets.SLACK_WEBHOOK_ML_ALERTS }}

# ci/submit_training_job.py
# Called by GitHub Actions to submit training to cloud platform

import argparse, sys, json
from datetime import datetime

def submit_azure_ml(args) -> str:
    """Submit training job to Azure ML and return job name."""
    from azure.ai.ml import MLClient, command, Input
    from azure.ai.ml.constants import AssetTypes
    from azure.identity import DefaultAzureCredential

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=args.subscription_id,
        resource_group_name=args.resource_group,
        workspace_name=args.workspace,
    )

    job = command(
        name=f'ci-retrain-{datetime.now().strftime("%Y%m%d-%H%M%S")}',
        code='./src',
        command=(
            f'python train.py '
            f'--cutoff-date {args.cutoff_date} '
            f'--output-dir ${{{{outputs.model_output}}}}'
        ),
        environment='freshmart-training-env:latest',
        compute='cpu-cluster',
        experiment_name='freshmart-delivery-ci',
        tags={'triggered_by': 'github-actions', 'cutoff': args.cutoff_date},
    )

    submitted = ml_client.jobs.create_or_update(job)
    print(f"Job submitted: {submitted.name}")
    print(f"Studio URL:    {submitted.studio_url}")

    with open(args.output_job_name, 'w') as f:
        f.write(submitted.name)

    return submitted.name

# ci/wait_for_job.py
def wait_for_azure_job(job_name: str, timeout_s: int = 3600):
    """Poll job status until complete or timeout."""
    from azure.ai.ml import MLClient
    from azure.identity import DefaultAzureCredential
    import time

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
        resource_group_name='rg-freshmart-ml',
        workspace_name='ws-freshmart-ml',
    )

    TERMINAL_STATES = {'Completed', 'Failed', 'Canceled'}
    start = time.time()

    while time.time() - start < timeout_s:
        job    = ml_client.jobs.get(job_name)
        status = job.status
        elapsed = int(time.time() - start)
        print(f"[{elapsed:>4}s] Status: {status}")

        if status in TERMINAL_STATES:
            if status != 'Completed':
                print(f"Job {status}: {job_name}")
                sys.exit(1)
            print(f"Job completed in {elapsed}s")
            return

        time.sleep(30)   # poll every 30 seconds

    print(f"Timeout after {timeout_s}s")
    sys.exit(1)

# ci/compare_models.py
def compare_models(challenger_metrics_path: str,
                    registry_name: str,
                    tolerance: float = 0.05) -> dict:
    """
    Compare challenger metrics to current production champion.
    Returns whether challenger should be promoted.
    """
    import json
    from azure.ai.ml import MLClient
    from azure.identity import DefaultAzureCredential

    with open(challenger_metrics_path) as f:
        challenger = json.load(f)
    challenger_mae = challenger['val_mae']

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
        resource_group_name='rg-freshmart-ml',
        workspace_name='ws-freshmart-ml',
    )

    # Get champion metrics from model registry
    try:
        versions = ml_client.models.list(name=registry_name)
        prod_versions = [v for v in versions if v.stage == 'Production']
        if prod_versions:
            champion_run  = ml_client.jobs.get(prod_versions[0].run_id)
            champion_mae  = champion_run.outputs.get('val_mae', 999.0)
        else:
            champion_mae  = 999.0   # no champion — always promote
            print("No production champion found — challenger will be promoted")
    except Exception as e:
        champion_mae = 999.0
        print(f"Could not retrieve champion metrics: {e}")

    improvement   = (champion_mae - challenger_mae) / champion_mae * 100
    should_promote = challenger_mae <= champion_mae * (1 + tolerance)

    result = {
        'challenger_mae':  round(challenger_mae, 4),
        'champion_mae':    round(champion_mae, 4),
        'improvement_pct': round(improvement, 2),
        'should_promote':  should_promote,
        'tolerance':       tolerance,
    }

    print(f"Champion MAE:    {champion_mae:.4f}")
    print(f"Challenger MAE:  {challenger_mae:.4f}")
    print(f"Improvement:     {improvement:+.2f}%")
    print(f"Should promote:  {should_promote}")

    with open('/tmp/comparison.json', 'w') as f:
        json.dump(result, f, indent=2)

    return result

# ci/load_test.py
def load_test_endpoint(endpoint_url: str, api_key: str,
                        rps: int = 50, duration_s: int = 30,
                        max_p99_ms: float = 500.0,
                        max_error_pct: float = 1.0) -> bool:
    """
    Simple load test using concurrent requests.
    Returns True if endpoint meets SLO targets.
    """
    import threading, time, requests, statistics

    latencies = []
    errors    = []
    lock      = threading.Lock()

    sample_payload = {
        'distance_km': 3.5, 'is_peak_hour': 1,
        'order_value': 450.0, 'restaurant_avg': 33.2, 'driver_avg': 29.8,
    }

    def make_request():
        start = time.time()
        try:
            r = requests.post(
                endpoint_url,
                json=sample_payload,
                headers={'Authorization': f'Bearer {api_key}'},
                timeout=2.0,
            )
            latency = (time.time() - start) * 1000
            with lock:
                if r.status_code == 200:
                    latencies.append(latency)
                else:
                    errors.append(r.status_code)
        except Exception as e:
            with lock:
                errors.append(str(e))

    interval = 1.0 / rps
    end_time = time.time() + duration_s

    while time.time() < end_time:
        t = threading.Thread(target=make_request)
        t.daemon = True
        t.start()
        time.sleep(interval)

    time.sleep(2)   # wait for in-flight requests

    if not latencies:
        print("No successful requests — endpoint unavailable")
        return False

    p50  = statistics.median(latencies)
    p99  = sorted(latencies)[int(0.99 * len(latencies))]
    err_pct = len(errors) / (len(latencies) + len(errors)) * 100

    print(f"Load test results ({rps} RPS, {duration_s}s):")
    print(f"  Total requests:  {len(latencies) + len(errors)}")
    print(f"  p50 latency:     {p50:.1f}ms")
    print(f"  p99 latency:     {p99:.1f}ms  (max: {max_p99_ms}ms)")
    print(f"  Error rate:      {err_pct:.2f}%  (max: {max_error_pct}%)")

    passed = p99 <= max_p99_ms and err_pct <= max_error_pct
    if not passed:
        print("LOAD TEST FAILED — SLO violation")
        sys.exit(1)
    print("Load test passed")
    return True

# ci/cloud_adapter.py
# Platform-agnostic interface — one file per cloud

import os
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class JobResult:
    job_name:   str
    status:     str   # 'Completed', 'Failed'
    val_mae:    float
    val_r2:     float
    model_uri:  str   # path to model artifact

class CloudMLAdapter(ABC):
    @abstractmethod
    def submit_training_job(self, cutoff_date: str, **hyperparams) -> str:
        """Submit a training job and return the job name/ID."""

    @abstractmethod
    def wait_for_job(self, job_name: str, timeout_s: int = 3600) -> JobResult:
        """Wait for a job to complete and return results."""

    @abstractmethod
    def register_model(self, job_result: JobResult,
                        model_name: str, stage: str) -> str:
        """Register a trained model and return its version."""

    @abstractmethod
    def get_champion_metrics(self, model_name: str) -> dict:
        """Return metrics of the current production champion model."""

    @abstractmethod
    def deploy_to_endpoint(self, model_name: str, version: str,
                            endpoint: str, traffic_pct: int) -> None:
        """Deploy a model version to an endpoint with given traffic %."""

# ── Azure ML implementation ───────────────────────────────────────────
class AzureMLAdapter(CloudMLAdapter):
    def __init__(self):
        from azure.ai.ml import MLClient
        from azure.identity import DefaultAzureCredential
        self.client = MLClient(
            credential=DefaultAzureCredential(),
            subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
            resource_group_name=os.environ['AZURE_RESOURCE_GROUP'],
            workspace_name=os.environ['AZURE_WORKSPACE'],
        )

    def submit_training_job(self, cutoff_date: str, **params) -> str:
        from azure.ai.ml import command
        job = command(
            name=f'ci-{cutoff_date}',
            code='./src',
            command=f'python train.py --cutoff-date {cutoff_date}',
            environment='freshmart-training-env:latest',
            compute='cpu-cluster',
        )
        return self.client.jobs.create_or_update(job).name

    def wait_for_job(self, job_name: str, timeout_s=3600) -> JobResult:
        import time
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            job = self.client.jobs.get(job_name)
            if job.status in ('Completed', 'Failed', 'Canceled'):
                mae = job.outputs.get('val_mae', 999.0)
                return JobResult(
                    job_name=job_name, status=job.status,
                    val_mae=float(mae), val_r2=0.0,
                    model_uri=f'azureml://jobs/{job_name}/outputs/model',
                )
            time.sleep(30)
        raise TimeoutError(f"Job {job_name} did not complete in {timeout_s}s")

    def get_champion_metrics(self, model_name: str) -> dict:
        try:
            versions = list(self.client.models.list(name=model_name))
            prod = [v for v in versions if getattr(v, 'stage', '') == 'Production']
            if not prod:
                return {'val_mae': 999.0}
            # In practice: retrieve metrics from the run that produced this model
            return {'val_mae': float(prod[0].tags.get('val_mae', 999.0))}
        except Exception:
            return {'val_mae': 999.0}

    def register_model(self, job_result, model_name, stage='Staging') -> str:
        from azure.ai.ml.entities import Model
        from azure.ai.ml.constants import AssetTypes
        model = self.client.models.create_or_update(Model(
            name=model_name,
            path=job_result.model_uri,
            type=AssetTypes.MLFLOW_MODEL,
            tags={'val_mae': str(job_result.val_mae), 'stage': stage},
        ))
        return str(model.version)

    def deploy_to_endpoint(self, model_name, version, endpoint, traffic_pct):
        print(f"AML: deploying {model_name}:{version} to {endpoint} ({traffic_pct}%)")

# ── SageMaker implementation ──────────────────────────────────────────
class SageMakerAdapter(CloudMLAdapter):
    def __init__(self):
        import boto3, sagemaker
        self.sm        = boto3.client('sagemaker')
        self.session   = sagemaker.Session()
        self.role      = sagemaker.get_execution_role()
        self.bucket    = self.session.default_bucket()

    def submit_training_job(self, cutoff_date: str, **params) -> str:
        from sagemaker.sklearn.estimator import SKLearn
        est = SKLearn(
            entry_point='train.py', source_dir='./src',
            framework_version='1.2-1', instance_type='ml.m5.large',
            role=self.role, hyperparameters={'cutoff-date': cutoff_date},
            output_path=f's3://{self.bucket}/model-output/',
            base_job_name='ci-retrain',
        )
        est.fit(inputs={'train': f's3://{self.bucket}/training-data/'}, wait=False)
        return est.latest_training_job.name

    def wait_for_job(self, job_name: str, timeout_s=3600) -> JobResult:
        import time
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            desc   = self.sm.describe_training_job(TrainingJobName=job_name)
            status = desc['TrainingJobStatus']
            if status in ('Completed', 'Failed', 'Stopped'):
                model_uri = desc.get('ModelArtifacts', {}).get('S3ModelArtifacts', '')
                mae = float(desc.get('FinalMetricDataList', [{}])[0].get('Value', 999.0))
                return JobResult(job_name, status, mae, 0.0, model_uri)
            time.sleep(30)
        raise TimeoutError(f"SageMaker job {job_name} timed out")

    def get_champion_metrics(self, model_name: str) -> dict:
        try:
            pkgs = self.sm.list_model_packages(
                ModelPackageGroupName=model_name,
                ModelApprovalStatus='Approved',
            )['ModelPackageSummaryList']
            if not pkgs:
                return {'val_mae': 999.0}
            return {'val_mae': 999.0}   # retrieve from package metadata
        except Exception:
            return {'val_mae': 999.0}

    def register_model(self, job_result, model_name, stage='Staging') -> str:
        print(f"SageMaker: registering {model_name} from {job_result.model_uri}")
        return '1'

    def deploy_to_endpoint(self, model_name, version, endpoint, traffic_pct):
        print(f"SageMaker: deploying {model_name}:{version} to {endpoint} ({traffic_pct}%)")

# ── Factory — choose adapter from environment variable ────────────────
def get_adapter() -> CloudMLAdapter:
    platform = os.environ.get('ML_PLATFORM', 'azure').lower()
    if platform == 'azure':
        return AzureMLAdapter()
    elif platform == 'sagemaker':
        return SageMakerAdapter()
    else:
        raise ValueError(f"Unknown ML_PLATFORM: {platform}")

# ── Unified CI script — platform selected by env var ─────────────────
if __name__ == '__main__':
    import sys
    adapter     = get_adapter()
    cutoff_date = sys.argv[1] if len(sys.argv) > 1 else '2026-03-28'

    job_name   = adapter.submit_training_job(cutoff_date)
    job_result = adapter.wait_for_job(job_name)

    if job_result.status != 'Completed':
        print(f"Training failed: {job_result.status}")
        sys.exit(1)

    champion   = adapter.get_champion_metrics('freshmart-delivery-time')
    promote    = job_result.val_mae <= champion['val_mae'] * 1.05

    print(f"Challenger: {job_result.val_mae:.4f}")
    print(f"Champion:   {champion['val_mae']:.4f}")
    print(f"Promote:    {promote}")

    if promote:
        version = adapter.register_model(job_result, 'freshmart-delivery-time')
        adapter.deploy_to_endpoint('freshmart-delivery-time', version,
                                    'freshmart-delivery-staging', 100)
    sys.exit(0 if promote else 2)   # exit 2 = no improvement, not a failure