Production ML: Fraud Detection MLOps

Introduction

30 Second Summary

Every time you tap your card or buy something online, a model somewhere decides in milliseconds whether that transaction is legitimate or fraudulent. If the model drifts out of date or breaks silently, real money disappears from real people's accounts.

In this project, you will build the full production system behind that decision. You will train a fraud detection model, deploy it on AWS with automated monitoring, and wire up self-healing logic that detects when performance degrades and fixes itself without human intervention.

What You'll Build

A production ML system that detects credit card fraud in real-time, deployed on AWS with full observability, automated model evaluation, drift detection, and self-healing rollback.

By the end of this project, you'll have:

• A trained XGBoost model tracked in MLflow with full experiment history, hyperparameter optimization, and a model registry managing stage transitions.
• A FastAPI inference service deployed on AWS ECS Fargate via Terraform, with champion/challenger shadow deployment, Prometheus metrics, and Grafana dashboards.
• An automated MLOps pipeline using GitHub Actions and Evidently AI that detects data drift, triggers retraining, evaluates new models in shadow mode, and rolls back automatically if performance degrades.
• Secret Mission: Build a drift simulation harness that injects realistic concept drift and triggers the full autonomous lifecycle end-to-end.

Are there any prerequisites?

This is an advanced project. You should be comfortable with Python, ML fundamentals (training models, evaluation metrics), Docker basics, and have an AWS account with CLI configured.

Basic Terraform knowledge helps but is not strictly required. The project walks you through every infrastructure step.

Set Up Local Infrastructure with Docker Compose

In the previous step, you set up your development environment with Python, Terraform, and AWS credentials. Now you need somewhere to track your ML experiments and store model artifacts locally before deploying to AWS.

Production ML teams use experiment tracking servers to log every training run, compare models, and manage model versions. In this step, you will spin up a local MLflow tracking server backed by PostgreSQL and MinIO (an S3-compatible object store) using Docker Compose.

In this step, get ready to:

• Create a Docker Compose stack with PostgreSQL, MinIO, and MLflow.
• Start the stack and verify the MLflow UI.
• Create a MinIO bucket and configure environment variables for S3 access.

Create the Docker Compose file

Your MLflow server needs two backing services: a database to store experiment metadata and an object store to hold model artifacts. PostgreSQL handles the metadata, and MinIO provides a local S3-compatible store so MLflow can save artifacts the same way it would on real AWS S3.

• In your project directory, create a file called docker-compose.yml by running:

touch docker-compose.yml

• Open docker-compose.yml in your editor and paste the following configuration:

version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: mlflow
      POSTGRES_DB: mlflow
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  minio:
    image: minio/minio
    command: server /data --console-address ':9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - '9000:9000'
      - '9001:9001'
    volumes:
      - minio_data:/data

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    depends_on:
      - postgres
      - minio
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://minio:9000
      AWS_ACCESS_KEY_ID: minioadmin
      AWS_SECRET_ACCESS_KEY: minioadmin
    ports:
      - '5000:5000'
    command: >
      mlflow server
      --backend-store-uri postgresql://mlflow:mlflow@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
      --host 0.0.0.0
      --port 5000

volumes:
  postgres_data:
  minio_data:

What does this configuration do?

This file defines three services that work together:

• postgres stores MLflow's experiment metadata (runs, parameters, metrics) in a relational database.
• minio provides S3-compatible object storage locally. MLflow saves model artifacts (pickled models, plots) here.
• mlflow is the tracking server itself. It connects to both backing services using the environment variables and command flags.

The depends_on field ensures PostgreSQL and MinIO start before MLflow attempts to connect.

• Save docker-compose.yml.
• Confirm the file exists by running:

ls docker-compose.yml

You should see docker-compose.yml in the output.

File not showing up?

Make sure you are in your project directory when you run ls. If you created the file in the wrong location, move it with mv or recreate it in the correct folder.

help me find where my docker-compose.yml ended up

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml before moving on.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml for comparison:

version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: mlflow
      POSTGRES_DB: mlflow
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  minio:
    image: minio/minio
    command: server /data --console-address ':9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - '9000:9000'
      - '9001:9001'
    volumes:
      - minio_data:/data

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    depends_on:
      - postgres
      - minio
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://minio:9000
      AWS_ACCESS_KEY_ID: minioadmin
      AWS_SECRET_ACCESS_KEY: minioadmin
    ports:
      - '5000:5000'
    command: >
      mlflow server
      --backend-store-uri postgresql://mlflow:mlflow@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
      --host 0.0.0.0
      --port 5000

volumes:
  postgres_data:
  minio_data:

Start the stack and verify the MLflow UI

With the Compose file in place, you can start all three services with a single command. Docker will pull the images on the first run, which may take a minute or two.

• Start the Docker Compose stack in detached mode by running:

docker compose up -d

What does this command do?

The -d flag runs the containers in the background (detached mode). Docker Compose pulls the PostgreSQL, MinIO, and MLflow images, creates the named volumes, and starts all three services on the network.

• Wait about 10-15 seconds for the services to initialize, then check that all containers are running:

docker compose ps

You should see three containers with a status of Up or running.

• Open your browser and navigate to http://localhost:5000.

You should see the MLflow tracking UI with an empty experiments list. This confirms the MLflow server is running and connected to PostgreSQL.

MLflow UI not loading?

Check the MLflow container logs for errors by running:

docker compose logs mlflow

If you see a connection error mentioning PostgreSQL, wait a few more seconds and refresh. PostgreSQL may still be initializing. If the mlflow container shows as Exited, restart the stack with:

docker compose down && docker compose up -d

help me debug why my MLflow container is not starting

Create the MinIO bucket and configure environment variables

MLflow is configured to store artifacts in an S3 bucket called mlflow-artifacts. That bucket does not exist yet in MinIO, so you need to create it. Since you already have the AWS CLI installed, you can use it with the --endpoint-url flag to target your local MinIO instance.

• Create the mlflow-artifacts bucket in MinIO by running:

aws --endpoint-url http://localhost:9000 s3 mb s3://mlflow-artifacts

What does this command do?

The --endpoint-url http://localhost:9000 flag redirects the AWS CLI to your local MinIO server instead of real AWS S3. The s3 mb subcommand creates a new bucket called mlflow-artifacts.

You should see output confirming the bucket was created successfully.

• Now configure the environment variables so that your Python scripts can connect to MLflow and MinIO. Export them in your terminal by running:

export MLFLOW_S3_ENDPOINT_URL=http://localhost:9000
export AWS_ACCESS_KEY_ID=minioadmin
export AWS_SECRET_ACCESS_KEY=minioadmin

Why these environment variables?

When your Python code logs artifacts to MLflow, it uses the boto3 library under the hood to upload files to S3. These environment variables tell boto3 to target your local MinIO instance (via MLFLOW_S3_ENDPOINT_URL) and authenticate with the MinIO credentials.

• Verify the bucket exists by listing the MinIO buckets:

aws --endpoint-url http://localhost:9000 s3 ls

You should see mlflow-artifacts listed in the output.

Seeing an access denied or connection error?

Make sure the MinIO container is running by checking docker compose ps. The minio service must show a status of Up.

If you get an access denied error, confirm that AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are exported in your current terminal session. Run echo $AWS_ACCESS_KEY_ID to check.

help me troubleshoot my MinIO bucket creation

Add exports to your shell profile

These export commands only persist for the current terminal session. To avoid re-exporting every time you open a new terminal, add them to a .env file in your project directory and source it at the start of each session.

Your local MLOps infrastructure is up and running. Next up, you will download the fraud detection dataset and build the feature store that ensures consistent feature computation across training and inference.

Build the Feature Store & Prepare Data

Your local ML infrastructure is running. Now you need data to train on and a reliable way to transform that data identically at training time and inference time.

A mismatch between how features are computed during training versus serving is called training-serving skew. It is one of the most common causes of silent model failures in production. In this step, you will build a lightweight Feature Store class that eliminates this risk, then prepare the credit card fraud dataset with temporal splits that mirror real-world data arrival.

In this step, get ready to:

• Download the credit card fraud dataset and explore its structure.
• Build a FeatureStore class for consistent feature engineering.
• Create temporal train/validation/test splits and save them to the artifact store.

Download the dataset and run exploratory data analysis

The dataset you will use is the ULB Machine Learning Group credit card fraud dataset. It contains 284,807 real credit card transactions over two days, with only 492 fraudulent cases (0.172% of the total). This extreme class imbalance is what makes fraud detection challenging.

• Create a data/ directory inside your project folder by running this command:

mkdir -p data

• Download the dataset from Kaggle. You will need a free Kaggle account.
• Click the **Download** button on the Kaggle dataset page to get the ZIP file.
• Extract creditcard.csv from the ZIP and move it into your data/ directory.

• Verify the file is in place by running:

ls data/creditcard.csv

You should see data/creditcard.csv printed in the output.

Don't see the file?

Make sure you extracted the ZIP archive. The download from Kaggle is a .zip file containing creditcard.csv inside it.

Double-check you moved the CSV into the data/ directory at the root of your project, not a nested subfolder.

help me find the creditcard.csv file in my project

Now explore the dataset to understand what you are working with. Start a quick Python session to examine class distribution and feature structure.

• Run this EDA script to inspect the dataset:

python -c "
import pandas as pd
df = pd.read_csv('data/creditcard.csv')
print('Shape:', df.shape)
print('\nClass distribution:')
print(df['Class'].value_counts())
print(f'\nFraud rate: {df["Class"].mean()*100:.3f}%')
print('\nFeature stats (Amount):')
print(df['Amount'].describe())
print('\nTime range (seconds):', df['Time'].min(), 'to', df['Time'].max())
"

What should I see?

The output confirms the dataset structure:

• Shape: (284807, 31) meaning 284,807 transactions with 31 columns.
• Class 0 (legitimate): 284,315 transactions. Class 1 (fraud): 492 transactions.
• Fraud rate of 0.172%. This severe imbalance means accuracy is a useless metric. You will need PR-AUC instead.
• Time ranges from 0 to ~172,000 seconds (two days). This temporal ordering is critical for realistic splits.

Why does class imbalance matter?

A naive model that predicts "not fraud" for every transaction would be 99.83% accurate. That is why fraud detection systems use precision-recall metrics and class-weighted training instead of accuracy.

The V1-V28 features are PCA-transformed and anonymized. You cannot interpret them directly, but they carry the signal your model will learn from.

Build the FeatureStore class

The FeatureStore class ensures that every feature transformation applied during training is applied identically at inference time. Without this, production predictions silently degrade.

• Create the file src/feature_store.py by running:

touch src/feature_store.py

• Open src/feature_store.py in your editor and add the imports and class definition:

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler


class FeatureStore:
    """Computes features identically for training and inference to prevent skew."""

    def __init__(self):
        self.amount_scaler = StandardScaler()
        self._fitted = False

What does this code do?

The FeatureStore class wraps a StandardScaler for amount normalization. The _fitted flag prevents accidental use before fitting on training data.

By encapsulating all transformations in one class, you guarantee the same logic runs during training and when the FastAPI service handles live requests.

• Add the fit method below the __init__ method:

    def fit(self, df: pd.DataFrame) -> 'FeatureStore':
        self.amount_scaler.fit(df[['Amount']])
        self._fitted = True
        return self

• Add the transform method below fit:

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        if not self._fitted:
            raise RuntimeError('FeatureStore must be fitted before transform.')
        result = df.copy()
        # Amount normalization
        result['Amount_scaled'] = self.amount_scaler.transform(result[['Amount']])
        # Time-of-day encoding (seconds in a day)
        seconds_in_day = 86400
        result['time_sin'] = np.sin(2 * np.pi * result['Time'] / seconds_in_day)
        result['time_cos'] = np.cos(2 * np.pi * result['Time'] / seconds_in_day)
        # Rolling aggregates on V-features (V1-V3)
        for col in ['V1', 'V2', 'V3']:
            result[f'{col}_rolling_mean'] = result[col].rolling(window=10, min_periods=1).mean()
        result = result.drop(columns=['Time', 'Amount'])
        return result

What does the transform method do?

• Amount normalization: scales the raw transaction amount using the scaler fitted on training data. This prevents large amounts from dominating the model.
• Time-of-day encoding: converts raw seconds into sine/cosine features that capture cyclical patterns (fraud peaks at certain hours). Using sin/cos avoids the discontinuity at midnight.
• Rolling aggregates: computes a 10-transaction rolling mean on V1, V2, and V3. This captures short-term behavioral trends that indicate anomalous bursts.
• The original Time and Amount columns are dropped since their engineered replacements carry the same information in a model-friendly form.

• Add the fit_transform convenience method below transform:

    def fit_transform(self, df: pd.DataFrame) -> pd.DataFrame:
        return self.fit(df).transform(df)

• Save src/feature_store.py.
• Verify the class loads without errors by running:

python -c "from src.feature_store import FeatureStore; print('FeatureStore loaded successfully')"

You should see FeatureStore loaded successfully printed.

Seeing a ModuleNotFoundError?

Make sure you have an __init__.py file in your src/ directory. Create one with touch src/__init__.py if it does not exist.

Also confirm you are running the command from the project root directory, not from inside src/.

help me fix the ModuleNotFoundError when importing FeatureStore

Why a FeatureStore class instead of inline transforms?

Inline transformations scattered across training scripts and serving code drift apart over time. A single class guarantees that if you add a feature during training, the same feature is computed at inference.

Production ML teams at Stripe and Uber use this exact pattern. The fitted scaler state is saved alongside the model, so the serving code never needs to recompute statistics from training data.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/feature_store.py before moving on.

ⓧ I'd like to double check the full code

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler


class FeatureStore:
    """Computes features identically for training and inference to prevent skew."""

    def __init__(self):
        self.amount_scaler = StandardScaler()
        self._fitted = False

    def fit(self, df: pd.DataFrame) -> 'FeatureStore':
        self.amount_scaler.fit(df[['Amount']])
        self._fitted = True
        return self

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        if not self._fitted:
            raise RuntimeError('FeatureStore must be fitted before transform.')
        result = df.copy()
        # Amount normalization
        result['Amount_scaled'] = self.amount_scaler.transform(result[['Amount']])
        # Time-of-day encoding (seconds in a day)
        seconds_in_day = 86400
        result['time_sin'] = np.sin(2 * np.pi * result['Time'] / seconds_in_day)
        result['time_cos'] = np.cos(2 * np.pi * result['Time'] / seconds_in_day)
        # Rolling aggregates on V-features (V1-V3)
        for col in ['V1', 'V2', 'V3']:
            result[f'{col}_rolling_mean'] = result[col].rolling(window=10, min_periods=1).mean()
        result = result.drop(columns=['Time', 'Amount'])
        return result

    def fit_transform(self, df: pd.DataFrame) -> pd.DataFrame:
        return self.fit(df).transform(df)

Create temporal splits and save to the artifact store

Random train/test splits leak future information into training data. In production, your model only ever sees past transactions when predicting on new ones. Temporal splits simulate this constraint.

• Create the file scripts/prepare_data.py by running:

mkdir -p scripts
touch scripts/prepare_data.py

• Open scripts/prepare_data.py and add the following data preparation script:

"""Download dataset, run EDA, create temporal train/val/test splits, save to artifact store."""
import os
import pandas as pd
from sklearn.model_selection import train_test_split
import mlflow

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

df = pd.read_csv('data/creditcard.csv')
df_sorted = df.sort_values('Time').reset_index(drop=True)
n = len(df_sorted)
train_end = int(n * 0.7)
val_end = int(n * 0.85)
train_df = df_sorted.iloc[:train_end]
val_df = df_sorted.iloc[train_end:val_end]
test_df = df_sorted.iloc[val_end:]
train_df.to_parquet('data/train.parquet', index=False)
val_df.to_parquet('data/val.parquet', index=False)
test_df.to_parquet('data/test.parquet', index=False)
print(f'Train: {len(train_df)}, Val: {len(val_df)}, Test: {len(test_df)}')

What does this script do?

• Sorts all transactions by Time (seconds since first transaction). This ensures earlier transactions go to training and later ones to testing.
• Splits 70% training, 15% validation, 15% test based on temporal order.
• Saves each split as a .parquet file. Parquet is columnar, compressed, and faster to load than CSV for ML workflows.
• Sets the MLflow tracking URI so future scripts in the pipeline can log artifacts to your running MLflow server.

Why temporal splits instead of random splits?

Random splits let the model "see the future" during training. A transaction from day 2 might end up in the training set while a day 1 transaction sits in the test set. This inflates evaluation metrics and masks real-world performance.

Temporal splits mirror production: train on the past, validate on the near future, test on the far future. If your model performs well on this split, it will generalize to genuinely unseen future data.

• Save scripts/prepare_data.py.
• Run the data preparation script:

python scripts/prepare_data.py

You should see output showing the split sizes: approximately 199,364 training, 42,721 validation, and 42,722 test transactions.

• Confirm the parquet files were created by running:

ls data/*.parquet

You should see data/train.parquet, data/val.parquet, and data/test.parquet listed.

Seeing a pyarrow or parquet error?

The to_parquet method requires pyarrow to be installed. If you see an import error, install it by running pip install pyarrow and re-run the script.

help me fix the parquet write error in prepare_data.py

✔️ Awesome, I've got everything!

Great. Confirm you have all three parquet files in your data/ directory before moving on.

ⓧ I'd like to double check the full code

"""Download dataset, run EDA, create temporal train/val/test splits, save to artifact store."""
import os
import pandas as pd
from sklearn.model_selection import train_test_split
import mlflow

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

df = pd.read_csv('data/creditcard.csv')
df_sorted = df.sort_values('Time').reset_index(drop=True)
n = len(df_sorted)
train_end = int(n * 0.7)
val_end = int(n * 0.85)
train_df = df_sorted.iloc[:train_end]
val_df = df_sorted.iloc[train_end:val_end]
test_df = df_sorted.iloc[val_end:]
train_df.to_parquet('data/train.parquet', index=False)
val_df.to_parquet('data/val.parquet', index=False)
test_df.to_parquet('data/test.parquet', index=False)
print(f'Train: {len(train_df)}, Val: {len(val_df)}, Test: {len(test_df)}')

Your data pipeline is in place. You have a feature store that guarantees consistent transformations and temporally ordered splits ready for training. Next up, you will train your first models and track everything in MLflow.

Train Models & Track Experiments

Your data is processed and sitting in the artifact store. Now it is time to answer the core question of any ML project: which model performs best on this data?

In this step, you will train three different models and track every experiment in MLflow. A Logistic Regression baseline establishes a floor. Two XGBoost variants tackle the extreme class imbalance (0.17% fraud) with different strategies. MLflow records all the metrics, parameters, and artifacts so you can compare objectively.

In this step, get ready to:

• Train a Logistic Regression baseline with MLflow autologging.
• Train XGBoost models with two class-imbalance strategies.
• Compare all experiments in the MLflow UI and select the best approach.

Train a Logistic Regression baseline

Before training, you need one additional Python package. SMOTE (Synthetic Minority Over-sampling Technique) lives in the imbalanced-learn library. You will use it in the second substep, but installing it now keeps the imports clean.

• Install the imbalanced-learn package by running this command:

pip install imbalanced-learn

Install failing?

Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.

help me fix a pip install error for imbalanced-learn

The training script loads your processed data, applies the FeatureStore transformations, and trains the baseline. Every run is tracked in MLflow so you can compare results later.

• Create a new file called scripts/train.py and add the following imports and data setup:

"""Train Logistic Regression baseline and XGBoost models with MLflow tracking."""
import os
import mlflow
import mlflow.sklearn
import mlflow.xgboost
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import average_precision_score, roc_auc_score, precision_recall_curve
from xgboost import XGBClassifier
from imblearn.over_sampling import SMOTE
from src.feature_store import FeatureStore

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment('fraud-detection')

train_df = pd.read_parquet('data/train.parquet')
val_df = pd.read_parquet('data/val.parquet')

fs = FeatureStore()
X_train = fs.fit_transform(train_df.drop(columns=['Class']))
y_train = train_df['Class']
X_val = fs.transform(val_df.drop(columns=['Class']))
y_val = val_df['Class']

What does this code do?

• Connects to the MLflow tracking server running in your Docker Compose stack and creates (or reuses) an experiment called fraud-detection.
• Loads the temporal train and validation splits you created in the previous step.
• Applies the FeatureStore to compute features identically for both sets. This prevents training-serving skew.

You will not see output yet. The next chunk adds the first model that produces results.

A simple Logistic Regression with balanced class weights provides a performance floor. Every subsequent model must beat this baseline to justify its complexity.

• Add the Logistic Regression baseline below the setup code in scripts/train.py:

# Logistic Regression baseline
with mlflow.start_run(run_name='logistic_regression_baseline'):
    mlflow.sklearn.autolog()
    lr = LogisticRegression(max_iter=1000, class_weight='balanced')
    lr.fit(X_train, y_train)
    y_pred_proba = lr.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'LR - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

What does this code do?

• mlflow.sklearn.autolog() automatically logs the model parameters, training metrics, and the serialized model artifact to MLflow without manual logging calls.
• class_weight='balanced' tells the model to weight the minority class (fraud) more heavily during training. This compensates for the extreme imbalance.
• PR-AUC (Precision-Recall Area Under Curve) is the primary metric because it is more informative than ROC-AUC on highly imbalanced datasets.

• Save scripts/train.py.
• Run the script to verify the baseline trains successfully:

python scripts/train.py

You should see output like LR - PR-AUC: 0.XXXX, ROC-AUC: 0.XXXX with decimal values printed to the terminal.

Seeing a connection error?

Make sure your Docker Compose stack is still running. Switch back to the terminal where you started it, or run docker compose ps to check. The MLflow container must be healthy.

If you see a ModuleNotFoundError, confirm your virtual environment is active and that imbalanced-learn installed successfully.

help me debug my MLflow connection or module import error

Train XGBoost with class weighting and SMOTE

The baseline is your performance floor. Now you will train two XGBoost models that attack the class imbalance differently. The first uses scale_pos_weight to weight the loss function. The second uses SMOTE to synthetically balance the training data before fitting.

Why two strategies for class imbalance?

scale_pos_weight adjusts the gradient calculation during training. The model sees the same data but penalizes false negatives on fraud more heavily.

SMOTE generates synthetic fraud examples by interpolating between existing ones. The model sees a balanced dataset but some examples are artificial. Each approach has different failure modes, which is why you compare both.

• Add the XGBoost class-weighted model below the Logistic Regression block in scripts/train.py:

# XGBoost with class weighting
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()
with mlflow.start_run(run_name='xgboost_class_weight'):
    mlflow.xgboost.autolog()
    xgb = XGBClassifier(scale_pos_weight=scale_pos_weight, eval_metric='aucpr', use_label_encoder=False)
    xgb.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
    y_pred_proba = xgb.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'XGB class weight - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

What does this code do?

• Calculates scale_pos_weight as the ratio of negative to positive samples. With ~0.17% fraud, this value is around 580, meaning the model treats each fraud sample as 580x more important.
• eval_set=[(X_val, y_val)] enables XGBoost to track validation performance during training. MLflow autolog captures these iteration metrics automatically.
• mlflow.xgboost.autolog() logs the full XGBoost model, feature importance, and training curves.

Run the script after the next chunk to see all three models train together.

• Add the SMOTE-augmented XGBoost model below the class-weighted block in scripts/train.py:

# XGBoost with SMOTE
smote = SMOTE(random_state=42)
X_train_sm, y_train_sm = smote.fit_resample(X_train, y_train)
with mlflow.start_run(run_name='xgboost_smote'):
    mlflow.xgboost.autolog()
    xgb_sm = XGBClassifier(eval_metric='aucpr', use_label_encoder=False)
    xgb_sm.fit(X_train_sm, y_train_sm, eval_set=[(X_val, y_val)], verbose=False)
    y_pred_proba = xgb_sm.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'XGB SMOTE - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

What does this code do?

• SMOTE(random_state=42) creates synthetic fraud examples by interpolating between real ones. The training set becomes balanced (50/50 fraud and legitimate).
• This XGBoost model has no scale_pos_weight because the training data itself is now balanced. The model sees equal representation of both classes.
• The validation set remains unchanged. You always evaluate on the original imbalanced distribution to reflect real-world conditions.

• Save scripts/train.py.
• Run the full training script to train all three models:

python scripts/train.py

You should see three lines of output, one per model, each showing PR-AUC and ROC-AUC values. The XGBoost models should score higher than the Logistic Regression baseline.

Script running slowly or hanging?

SMOTE resampling on 200k+ rows can take 30-60 seconds depending on your machine. If the script appears stuck after the second model prints, give it time.

If you see memory errors, close other heavy applications. SMOTE creates a balanced dataset which roughly doubles the training data size.

help me debug my training script running slowly or throwing memory errors

✔️ Awesome, I've got everything!

Great. Double check you have saved scripts/train.py and that all three models printed their metrics.

ⓧ I'd like to double check the full code

"""Train Logistic Regression baseline and XGBoost models with MLflow tracking."""
import os
import mlflow
import mlflow.sklearn
import mlflow.xgboost
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import average_precision_score, roc_auc_score, precision_recall_curve
from xgboost import XGBClassifier
from imblearn.over_sampling import SMOTE
from src.feature_store import FeatureStore

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment('fraud-detection')

train_df = pd.read_parquet('data/train.parquet')
val_df = pd.read_parquet('data/val.parquet')

fs = FeatureStore()
X_train = fs.fit_transform(train_df.drop(columns=['Class']))
y_train = train_df['Class']
X_val = fs.transform(val_df.drop(columns=['Class']))
y_val = val_df['Class']

# Logistic Regression baseline
with mlflow.start_run(run_name='logistic_regression_baseline'):
    mlflow.sklearn.autolog()
    lr = LogisticRegression(max_iter=1000, class_weight='balanced')
    lr.fit(X_train, y_train)
    y_pred_proba = lr.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'LR - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

# XGBoost with class weighting
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()
with mlflow.start_run(run_name='xgboost_class_weight'):
    mlflow.xgboost.autolog()
    xgb = XGBClassifier(scale_pos_weight=scale_pos_weight, eval_metric='aucpr', use_label_encoder=False)
    xgb.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
    y_pred_proba = xgb.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'XGB class weight - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

# XGBoost with SMOTE
smote = SMOTE(random_state=42)
X_train_sm, y_train_sm = smote.fit_resample(X_train, y_train)
with mlflow.start_run(run_name='xgboost_smote'):
    mlflow.xgboost.autolog()
    xgb_sm = XGBClassifier(eval_metric='aucpr', use_label_encoder=False)
    xgb_sm.fit(X_train_sm, y_train_sm, eval_set=[(X_val, y_val)], verbose=False)
    y_pred_proba = xgb_sm.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred_proba)
    roc_auc = roc_auc_score(y_val, y_pred_proba)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.log_metric('val_roc_auc', roc_auc)
    print(f'XGB SMOTE - PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')

Compare experiments in the MLflow UI

All three runs are now stored in MLflow. The UI lets you compare metrics side-by-side and inspect what each model learned. This is where you make a data-driven decision about which approach to carry forward.

• Open your browser and navigate to http://localhost:5000.
• Click the fraud-detection experiment in the left sidebar.
• You should see three runs listed: logistic_regression_baseline, xgboost_class_weight, and xgboost_smote.

• Select all three runs using the checkboxes on the left.
• Click Compare in the top toolbar to see a side-by-side view.
• Look at the val_pr_auc and val_roc_auc columns. The XGBoost models should significantly outperform the Logistic Regression baseline.

Why PR-AUC over ROC-AUC for fraud detection?

With only 0.17% fraud in the dataset, a model that predicts "not fraud" for everything achieves 99.83% accuracy and a high ROC-AUC. PR-AUC focuses on how well the model finds the rare positive class, making it the right metric when positives are scarce.

• Click into the xgboost_class_weight run to view its detail page.
• Scroll to the Artifacts section and look for the feature importance chart logged by autolog.
• Note which V-features dominate. These PCA-transformed features capture the fraud signal that the model relies on.

What should I look for in feature importance?

Feature importance reveals which inputs drive the model's decisions. If Amount_scaled ranks highly, the model learned that transaction amount matters for fraud detection. The V-features (PCA-transformed) often dominate because they encode the original behavioral signals.

High importance on your engineered features (time_sin, V1_rolling_mean) confirms the FeatureStore adds value beyond raw inputs.

Based on your comparison, the XGBoost model with class weighting is typically the strongest choice. It achieves high PR-AUC without generating synthetic data, which means it generalizes more reliably to unseen distributions. You will carry this approach forward into hyperparameter optimization in the next step.

Don't see all three runs in MLflow?

Make sure you selected the fraud-detection experiment, not the Default experiment. If a run errored partway through, it may appear with a FAILED status. Re-run the script to retry.

help me find my MLflow experiment runs in the UI

Your experiment history is locked in. You have a clear winner and a documented comparison. Next up, you will fine-tune that winner with hyperparameter optimization to squeeze out maximum performance.

Optimize Hyperparameters & Register Champion Model

You have baseline models logged in MLflow and know that XGBoost with class weighting outperforms the alternatives. But those baseline XGBoost runs used default hyperparameters.

In this step, you will use Optuna to systematically search the hyperparameter space, then register the winning model in the MLflow Model Registry as your production champion.

In this step, get ready to:

• Run 50 Optuna trials with the TPE sampler to find optimal XGBoost hyperparameters.
• Register the best model in MLflow Model Registry and promote it to Production.
• Validate the production model against held-out test data with minimum performance thresholds.

Define and run the Optuna optimization

Optuna's TPE sampler builds a probabilistic model of which hyperparameter combinations produce good results. Each trial trains an XGBoost model with a different set of sampled hyperparameters and logs the result to MLflow as a nested run.

• Create the optimization script by adding a new file scripts/optimize.py with the imports, MLflow configuration, and data loading:

"""Optuna hyperparameter optimization for XGBoost fraud detection model."""
import os
import mlflow
import optuna
import pandas as pd
from sklearn.metrics import average_precision_score
from xgboost import XGBClassifier
from src.feature_store import FeatureStore

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment('fraud-detection-optuna')

# Load pre-processed data splits
train_df = pd.read_parquet('data/train.parquet')
val_df = pd.read_parquet('data/val.parquet')
fs = FeatureStore()
X_train = fs.fit_transform(train_df.drop(columns=['Class']))
y_train = train_df['Class']
X_val = fs.transform(val_df.drop(columns=['Class']))
y_val = val_df['Class']
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()

PARENT_RUN_ID = None

What does this code do?

• Sets the MLflow experiment to fraud-detection-optuna so optimization runs are grouped separately from your earlier training experiments.
• Loads the same train/validation splits and applies the FeatureStore transformations to ensure consistency with your baseline models.
• Computes scale_pos_weight (the ratio of negative to positive samples) so XGBoost can handle the severe class imbalance.

• Below the data loading section, add the Optuna objective function:

def objective(trial):
    params = {
        'n_estimators': trial.suggest_int('n_estimators', 100, 500),
        'max_depth': trial.suggest_int('max_depth', 3, 10),
        'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
        'subsample': trial.suggest_float('subsample', 0.6, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
        'min_child_weight': trial.suggest_int('min_child_weight', 1, 10),
        'scale_pos_weight': scale_pos_weight,
        'eval_metric': 'aucpr',
        'use_label_encoder': False,
    }
    with mlflow.start_run(run_name=f'trial_{trial.number}', nested=True):
        mlflow.log_params(params)
        model = XGBClassifier(**params)
        model.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
        y_pred = model.predict_proba(X_val)[:, 1]
        pr_auc = average_precision_score(y_val, y_pred)
        mlflow.log_metric('val_pr_auc', pr_auc)
        return pr_auc

What does the objective function do?

Each call to objective() is one trial. Optuna's TPE sampler suggests values for six hyperparameters from the ranges you defined.

The function trains an XGBoost model with those parameters, evaluates PR-AUC on the validation set, and returns the score. Optuna uses this returned value to guide future suggestions toward better regions of the search space.

Each trial is logged as a nested MLflow run so you can inspect every combination in the UI later.

• Finally, add the study runner block at the bottom of the file:

with mlflow.start_run(run_name='optuna_study'):
    study = optuna.create_study(direction='maximize', sampler=optuna.samplers.TPESampler())
    study.optimize(objective, n_trials=50)
    best_params = study.best_params
    mlflow.log_params({f'best_{k}': v for k, v in best_params.items()})
    mlflow.log_metric('best_val_pr_auc', study.best_value)
    print(f'Best PR-AUC: {study.best_value:.4f}')
    print(f'Best params: {best_params}')

What does the study runner do?

The outer mlflow.start_run creates a parent run. All 50 trial runs nest inside it, keeping your experiment organized.

The TPE sampler uses a Bayesian approach. It models which hyperparameter regions produced good results and samples more aggressively from those regions as trials progress.

After all trials complete, the best parameters and score are logged to the parent run for easy retrieval.

• Save scripts/optimize.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before running it.

ⓧ I'd like to double check the full code

"""Optuna hyperparameter optimization for XGBoost fraud detection model."""
import os
import mlflow
import optuna
import pandas as pd
from sklearn.metrics import average_precision_score
from xgboost import XGBClassifier
from src.feature_store import FeatureStore

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment('fraud-detection-optuna')

train_df = pd.read_parquet('data/train.parquet')
val_df = pd.read_parquet('data/val.parquet')
fs = FeatureStore()
X_train = fs.fit_transform(train_df.drop(columns=['Class']))
y_train = train_df['Class']
X_val = fs.transform(val_df.drop(columns=['Class']))
y_val = val_df['Class']
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()

PARENT_RUN_ID = None

def objective(trial):
    params = {
        'n_estimators': trial.suggest_int('n_estimators', 100, 500),
        'max_depth': trial.suggest_int('max_depth', 3, 10),
        'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
        'subsample': trial.suggest_float('subsample', 0.6, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
        'min_child_weight': trial.suggest_int('min_child_weight', 1, 10),
        'scale_pos_weight': scale_pos_weight,
        'eval_metric': 'aucpr',
        'use_label_encoder': False,
    }
    with mlflow.start_run(run_name=f'trial_{trial.number}', nested=True):
        mlflow.log_params(params)
        model = XGBClassifier(**params)
        model.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
        y_pred = model.predict_proba(X_val)[:, 1]
        pr_auc = average_precision_score(y_val, y_pred)
        mlflow.log_metric('val_pr_auc', pr_auc)
        return pr_auc

with mlflow.start_run(run_name='optuna_study'):
    study = optuna.create_study(direction='maximize', sampler=optuna.samplers.TPESampler())
    study.optimize(objective, n_trials=50)
    best_params = study.best_params
    mlflow.log_params({f'best_{k}': v for k, v in best_params.items()})
    mlflow.log_metric('best_val_pr_auc', study.best_value)
    print(f'Best PR-AUC: {study.best_value:.4f}')
    print(f'Best params: {best_params}')

• Run the optimization by executing this command from your project root:

python scripts/optimize.py

This takes several minutes to complete. You should see Optuna logging trial progress and a final output showing the best PR-AUC score and corresponding hyperparameters.

Optimization not running?

• Make sure your virtual environment is activated and all dependencies (optuna, xgboost, mlflow) are installed.
• Verify Docker Compose is still running with docker compose ps. The MLflow server must be accessible at http://localhost:5000.
• Confirm your data/train.parquet and data/val.parquet files exist from the earlier data engineering step.

My Optuna optimization script won't start or fails mid-run. How do I debug this?

Register the champion model in MLflow

The MLflow Model Registry is where you promote a trained model from an experiment artifact to a named, versioned entity that downstream services can load by name.

You will now train one final model with the best hyperparameters from Optuna, log it with its model artifact, and register it as your production champion.

• Open a Python interpreter from your project root by running:

python

• Train the final champion model with the best Optuna parameters and register it by running these commands in the Python interpreter:

import os
import mlflow
import mlflow.xgboost
import pandas as pd
from sklearn.metrics import average_precision_score
from xgboost import XGBClassifier
from src.feature_store import FeatureStore

# Connect to MLflow
mlflow.set_tracking_uri(os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000'))
mlflow.set_experiment('fraud-detection')

# Load data and transform features
train_df = pd.read_parquet('data/train.parquet')
val_df = pd.read_parquet('data/val.parquet')
fs = FeatureStore()
X_train = fs.fit_transform(train_df.drop(columns=['Class']))
y_train = train_df['Class']
X_val = fs.transform(val_df.drop(columns=['Class']))
y_val = val_df['Class']
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()

# Train final model with best Optuna params
with mlflow.start_run(run_name='champion_model') as run:
    model = XGBClassifier(
        scale_pos_weight=scale_pos_weight,
        eval_metric='aucpr',
        use_label_encoder=False,
        # Optuna found these - use YOUR best values from the optimize.py output
        n_estimators=300,
        max_depth=6,
        learning_rate=0.05,
        subsample=0.8,
        colsample_bytree=0.8,
        min_child_weight=3,
    )
    model.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
    y_pred = model.predict_proba(X_val)[:, 1]
    pr_auc = average_precision_score(y_val, y_pred)
    mlflow.log_metric('val_pr_auc', pr_auc)
    mlflow.xgboost.log_model(model, 'model')
    print(f'Champion PR-AUC: {pr_auc:.4f}')
    champion_run_id = run.info.run_id

What does this do?

This trains one final XGBoost model using the hyperparameters that Optuna identified as optimal. Replace the example values (n_estimators=300, etc.) with the actual best parameters printed by your optimize.py run.

The key difference from the Optuna trials is the call to mlflow.xgboost.log_model(model, 'model'). This saves the full model artifact (weights, config) so it can be registered and loaded later for serving.

• Register the model and transition it to Production by continuing in the same Python session:

from mlflow import MlflowClient

# Register the model in the Model Registry
model_uri = f'runs:/{champion_run_id}/model'
result = mlflow.register_model(model_uri, 'fraud-detection-champion')
print(f'Registered version: {result.version}')

# Transition through stages: None -> Staging -> Production
client = MlflowClient()
client.transition_model_version_stage(
    name='fraud-detection-champion',
    version=result.version,
    stage='Staging'
)
print('Model moved to Staging')

client.transition_model_version_stage(
    name='fraud-detection-champion',
    version=result.version,
    stage='Production'
)
print('Model promoted to Production')

What is happening with model stages?

MLflow Model Registry supports lifecycle stages: None, Staging, Production, and Archived. Transitioning a model through these stages mirrors a real deployment pipeline where models must pass validation before reaching production.

The models:/fraud-detection-champion/Production URI format lets downstream services always load the current production model without hardcoding a version number.

• Exit the Python interpreter by running:

exit()

• Verify the registration by opening http://localhost:5000 in your browser and clicking the Models tab. You should see fraud-detection-champion listed with Version 1 in the Production stage.

Model not showing in the registry?

• Make sure you ran mlflow.xgboost.log_model(model, 'model') inside the start_run context. Without a logged model artifact, registration will fail.
• Check that champion_run_id is not empty. If you restarted your Python session, you need to re-run the training block first.

I can't register or transition my model in MLflow. How do I fix this?

Validate the production model against the test set

A model in Production stage should pass minimum quality gates before any service loads it. The evaluator script loads the Production model from the registry and asserts that it meets your performance thresholds on held-out data.

• Create the evaluation script at scripts/model_evaluator.py with the following code:

"""Load Production model from MLflow Registry and evaluate against test set."""
import os
import mlflow
import pandas as pd
from sklearn.metrics import average_precision_score, roc_auc_score
from src.feature_store import FeatureStore

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

# Minimum performance thresholds for production readiness
MIN_PR_AUC = 0.70
MIN_ROC_AUC = 0.95

# Load the Production model from the registry
model = mlflow.xgboost.load_model('models:/fraud-detection-champion/Production')
test_df = pd.read_parquet('data/test.parquet')
fs = FeatureStore()
X_test = fs.fit_transform(test_df.drop(columns=['Class']))
y_test = test_df['Class']
y_pred = model.predict_proba(X_test)[:, 1]
pr_auc = average_precision_score(y_test, y_pred)
roc_auc = roc_auc_score(y_test, y_pred)
print(f'Test PR-AUC: {pr_auc:.4f}, ROC-AUC: {roc_auc:.4f}')
assert pr_auc >= MIN_PR_AUC, f'PR-AUC {pr_auc:.4f} below threshold {MIN_PR_AUC}'
assert roc_auc >= MIN_ROC_AUC, f'ROC-AUC {roc_auc:.4f} below threshold {MIN_ROC_AUC}'
print('Model passed all evaluation thresholds. Ready for Production.')

What does the evaluator do?

• Loads the model using the registry URI models:/fraud-detection-champion/Production. This always fetches whichever version is currently in the Production stage.
• Runs the model against the held-out test set that was never used during training or hyperparameter optimization.
• Asserts minimum thresholds: PR-AUC >= 0.70 and ROC-AUC >= 0.95. If the model fails either threshold, the script crashes with a clear error message.
• In a CI/CD pipeline, this script acts as a gate. A model that fails evaluation never reaches the serving layer.

• Save scripts/model_evaluator.py and run the evaluation:

python scripts/model_evaluator.py

You should see the test metrics printed followed by the confirmation message that the model passed all thresholds.

Seeing an AssertionError?

• If PR-AUC or ROC-AUC is below the threshold, your model may need more Optuna trials or different hyperparameter ranges. Re-run optimize.py with n_trials=100 for a more thorough search.
• Make sure you used the actual best parameters from your Optuna output when training the champion model, not the example values.
• Verify that data/test.parquet exists and was created during the data engineering step.

My model evaluator is failing the performance thresholds. How can I improve my champion model?

Your champion model is registered, validated, and ready to serve predictions. Next up, you will build the FastAPI inference service that loads this model and serves real-time fraud predictions with Prometheus metrics.

Build the FastAPI Inference Service

Your champion model is registered in MLflow and ready to serve predictions. The next challenge is making it accessible over the network so other systems can send transactions and receive fraud scores in real-time.

In this step, you will build a FastAPI inference service that loads the Production model from MLflow Registry, transforms incoming features using the same FeatureStore class from training, and exposes Prometheus metrics for operational monitoring.

In this step, get ready to:

• Build a FastAPI app with single and batch prediction endpoints.
• Integrate the FeatureStore for consistent feature transformation and structured JSON logging.
• Instrument the service with Prometheus metrics for request count, latency, and error rate.

Set up the FastAPI app scaffold with model loading

The inference service needs to load the champion model once at startup and reuse it across all requests. FastAPI's startup event hook is the right place for this.

You will also define the Prometheus metric collectors up front so every endpoint can record its own performance data.

• Create the app/ directory inside your project folder by running:

mkdir -p app

• Create the file app/main.py and paste the following scaffold code:

"""FastAPI inference service for fraud detection."""
import os
import json
import time
import logging
from datetime import datetime
from typing import List

import mlflow
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

from src.feature_store import FeatureStore

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

app = FastAPI(title='Fraud Detection Inference Service')

# Prometheus metrics
REQUEST_COUNT = Counter('fraud_requests_total', 'Total prediction requests', ['endpoint', 'status'])
LATENCY = Histogram('fraud_request_latency_seconds', 'Request latency', ['endpoint'],
                    buckets=[0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0])
PREDICTION_DIST = Counter('fraud_predictions_total', 'Prediction distribution', ['label'])
ERROR_RATE = Counter('fraud_errors_total', 'Total errors')

champion_model = None
feature_store = FeatureStore()


@app.on_event('startup')
def load_model():
    global champion_model
    champion_model = mlflow.xgboost.load_model('models:/fraud-detection-champion/Production')
    train_df = pd.read_parquet('data/train.parquet')
    feature_store.fit(train_df.drop(columns=['Class']))
    logger.info('Champion model loaded from MLflow Registry.')


class PredictRequest(BaseModel):
    features: dict


class PredictBatchRequest(BaseModel):
    records: List[dict]

What does this code do?

The top section imports all dependencies and sets up four Prometheus metric collectors. REQUEST_COUNT tracks total requests by endpoint and status. LATENCY records a histogram of response times with custom buckets targeting sub-100ms performance.

The load_model() function runs once when the server starts. It pulls the Production-stage model from MLflow Registry and fits the FeatureStore on the training data so it can transform incoming requests identically to how training data was processed.

The two Pydantic models (PredictRequest and PredictBatchRequest) define the expected JSON shape for single and batch prediction requests.

Why use FeatureStore at inference time?

Training-serving skew is one of the most common causes of degraded model performance in production. It happens when the feature transformations applied during training differ from those applied at inference.

By reusing the exact same FeatureStore class for both paths, you guarantee that amount normalization, time-of-day encoding, and rolling aggregates are computed identically. If you change a transformation in one place, it changes everywhere.

• Save app/main.py.

You won't see a change yet. The next chunk adds the prediction endpoints that use this scaffold.

Add the prediction endpoints with structured logging

Each prediction endpoint transforms incoming features through the FeatureStore, runs the champion model, and logs a structured JSON record of every prediction. This audit trail is critical for debugging production issues and computing drift metrics later.

• In app/main.py, add the single prediction endpoint below the PredictBatchRequest class:

@app.post('/predict')
def predict(request: PredictRequest):
    start = time.time()
    try:
        df = pd.DataFrame([request.features])
        X = feature_store.transform(df)
        proba = champion_model.predict_proba(X)[0][1]
        label = int(proba >= 0.5)
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict', status='success').inc()
        PREDICTION_DIST.labels(label=str(label)).inc()
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'input_features': request.features,
            'prediction': label,
            'confidence': float(proba),
            'latency_ms': round(latency * 1000, 2),
        }
        logger.info(json.dumps(log_entry))
        return {'prediction': label, 'confidence': float(proba), 'latency_ms': log_entry['latency_ms']}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))

What does this endpoint do?

The /predict endpoint wraps the full inference pipeline in a single request. It converts the incoming dictionary to a DataFrame, runs feature_store.transform() to apply the same transformations used during training, then calls predict_proba() to get the fraud probability.

Every successful prediction is recorded in three Prometheus metrics and logged as a structured JSON object with the timestamp, input features, label, confidence, and latency. This JSON log becomes the input for drift detection later.

• Below the /predict endpoint, add the batch prediction endpoint:

@app.post('/predict/batch')
def predict_batch(request: PredictBatchRequest):
    start = time.time()
    try:
        df = pd.DataFrame(request.records)
        X = feature_store.transform(df)
        probas = champion_model.predict_proba(X)[:, 1]
        labels = (probas >= 0.5).astype(int).tolist()
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict/batch').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='success').inc()
        for label in labels:
            PREDICTION_DIST.labels(label=str(label)).inc()
        return {'predictions': labels, 'confidences': probas.tolist(), 'latency_ms': round(latency * 1000, 2)}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))

What does the batch endpoint do?

The /predict/batch endpoint accepts a list of transaction records and processes them all in a single call. This avoids the overhead of individual HTTP requests when scoring many transactions at once.

It uses the same FeatureStore transform and Prometheus instrumentation pattern as the single endpoint, but returns arrays of predictions and confidence scores.

• Save app/main.py.

You won't see a change yet. The next chunk adds the metrics and health endpoints, then you will test the full service.

Add the metrics and health endpoints and test locally

Prometheus scrapes metrics by polling a dedicated endpoint. You also need a simple health check that load balancers and orchestrators can use to verify the service is alive.

• Below the /predict/batch endpoint in app/main.py, add the metrics and health endpoints:

@app.get('/metrics')
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.get('/health')
def health():
    return {'status': 'ok'}

What do these endpoints do?

The /metrics endpoint calls generate_latest() from the Prometheus client library, which serializes all registered metric collectors into the Prometheus text exposition format. Prometheus will scrape this endpoint on a schedule to ingest your service's telemetry.

The /health endpoint returns a simple JSON response. AWS ALB and ECS use this to determine if your container is healthy and ready to receive traffic.

• Save app/main.py.
• Start the inference service locally by running:

uvicorn app.main:app --reload

What does this command do?

Uvicorn is the ASGI server that runs your FastAPI app. The app.main:app argument tells Uvicorn to look for the app object inside app/main.py. The --reload flag enables auto-restart when you edit files during development.

You should see log output including a line confirming the model was loaded from MLflow Registry. The server runs on port 8000 by default.

• Open a new terminal tab and test the single prediction endpoint by running:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": {"Time": 0, "V1": -1.3, "V2": -0.07, "V3": 2.5, "V4": 0.1, "V5": 0.4, "V6": -0.1, "V7": 0.2, "V8": -0.1, "V9": 0.01, "V10": -0.2, "V11": 0.5, "V12": -0.3, "V13": 0.1, "V14": -0.5, "V15": 0.01, "V16": -0.1, "V17": 0.02, "V18": -0.01, "V19": 0.1, "V20": -0.05, "V21": 0.1, "V22": 0.3, "V23": -0.1, "V24": 0.5, "V25": 0.2, "V26": -0.3, "V27": 0.01, "V28": -0.02, "Amount": 149.62}}'

You should see a JSON response containing prediction, confidence, and latency_ms fields.

• Verify the Prometheus metrics are being collected by running:

curl http://localhost:8000/metrics

You should see Prometheus text output including fraud_requests_total and fraud_request_latency_seconds metrics with values from your test request.

• Return to the terminal running the server and stop it with Ctrl+C.

Server won't start or model fails to load?

If you see an import error for uvicorn, install it separately with pip install uvicorn. It ships with fastapi[standard] but not with the bare fastapi package.

If the model fails to load, confirm your Docker Compose stack is still running with docker compose ps. Check that MLflow UI at http://localhost:5000 shows your model in Production stage.

Help me debug my FastAPI inference service startup

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py with all four endpoints: /predict, /predict/batch, /metrics, and /health.

ⓧ I'd like to double check the full code

"""FastAPI inference service for fraud detection."""
import os
import json
import time
import logging
from datetime import datetime
from typing import List

import mlflow
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

from src.feature_store import FeatureStore

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

app = FastAPI(title='Fraud Detection Inference Service')

# Prometheus metrics
REQUEST_COUNT = Counter('fraud_requests_total', 'Total prediction requests', ['endpoint', 'status'])
LATENCY = Histogram('fraud_request_latency_seconds', 'Request latency', ['endpoint'],
                    buckets=[0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0])
PREDICTION_DIST = Counter('fraud_predictions_total', 'Prediction distribution', ['label'])
ERROR_RATE = Counter('fraud_errors_total', 'Total errors')

champion_model = None
feature_store = FeatureStore()


@app.on_event('startup')
def load_model():
    global champion_model
    champion_model = mlflow.xgboost.load_model('models:/fraud-detection-champion/Production')
    train_df = pd.read_parquet('data/train.parquet')
    feature_store.fit(train_df.drop(columns=['Class']))
    logger.info('Champion model loaded from MLflow Registry.')


class PredictRequest(BaseModel):
    features: dict


class PredictBatchRequest(BaseModel):
    records: List[dict]


@app.post('/predict')
def predict(request: PredictRequest):
    start = time.time()
    try:
        df = pd.DataFrame([request.features])
        X = feature_store.transform(df)
        proba = champion_model.predict_proba(X)[0][1]
        label = int(proba >= 0.5)
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict', status='success').inc()
        PREDICTION_DIST.labels(label=str(label)).inc()
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'input_features': request.features,
            'prediction': label,
            'confidence': float(proba),
            'latency_ms': round(latency * 1000, 2),
        }
        logger.info(json.dumps(log_entry))
        return {'prediction': label, 'confidence': float(proba), 'latency_ms': log_entry['latency_ms']}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))


@app.post('/predict/batch')
def predict_batch(request: PredictBatchRequest):
    start = time.time()
    try:
        df = pd.DataFrame(request.records)
        X = feature_store.transform(df)
        probas = champion_model.predict_proba(X)[:, 1]
        labels = (probas >= 0.5).astype(int).tolist()
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict/batch').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='success').inc()
        for label in labels:
            PREDICTION_DIST.labels(label=str(label)).inc()
        return {'predictions': labels, 'confidences': probas.tolist(), 'latency_ms': round(latency * 1000, 2)}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))


@app.get('/metrics')
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.get('/health')
def health():
    return {'status': 'ok'}

Your inference service is running and serving predictions locally. Next up, you will containerize this service and deploy it to AWS ECS Fargate with Terraform.

Containerize & Deploy to AWS

Your fraud detection inference service works locally, but production ML means running in the cloud with proper infrastructure.

In this step, you will containerize the FastAPI app with Docker, define your AWS infrastructure as code with Terraform, and deploy the whole system to ECS Fargate.

In this step, get ready to:

• Write and test a Dockerfile for the inference service.
• Create Terraform configuration for AWS infrastructure.
• Deploy to AWS and verify the service responds via the ALB endpoint.

Write a Dockerfile and build/test locally

A Dockerfile defines how your application is packaged into a container image. This ensures the same environment runs locally, in CI, and in production.

• Generate a requirements.txt from your virtual environment by running:

pip freeze > requirements.txt

• Create a file called Dockerfile in the root of your project directory (the same level as app/ and src/) with the following content:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

• FROM python:3.12-slim starts from a minimal Python 3.12 base image.
• COPY requirements.txt and RUN pip install installs dependencies in a cached layer so rebuilds are fast when only code changes.
• COPY . . copies your application code into the container.
• CMD starts uvicorn on port 8000 when the container runs.

• Build the Docker image by running:

docker build -t fraud-detection .

• Test the container locally by running:

docker run -p 8000:8000 fraud-detection

• Open a new terminal and verify the health endpoint by running:

curl http://localhost:8000/health

• You should see {"status":"ok"} in the response.
• Stop the container with Ctrl+C in the terminal where it is running.

Container won't start?

• If you see a module import error, make sure your requirements.txt includes all the packages your app needs (fastapi, uvicorn, mlflow, xgboost, etc.).
• If the build fails at pip install, check that Docker Desktop is running and you have an internet connection.
• help me debug my Docker build failing for the fraud detection service

Create Terraform configuration for AWS infrastructure

Terraform lets you define cloud infrastructure as code. You will declare all the AWS resources your service needs in a single configuration file.

• Create a terraform/ directory inside your project folder by running:

mkdir terraform

• Create a file called main.tf inside the terraform/ directory with the following configuration:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

variable "aws_region" {
  default = "us-east-1"
}

variable "app_name" {
  default = "fraud-detection"
}

# ECR Repository
resource "aws_ecr_repository" "fraud_detection" {
  name                 = "${var.app_name}-ecr"
  image_tag_mutability = "MUTABLE"
}

# S3 Bucket for MLflow artifacts
resource "aws_s3_bucket" "mlflow_artifacts" {
  bucket = "${var.app_name}-mlflow-artifacts"
}

# ECS Cluster
resource "aws_ecs_cluster" "fraud_detection" {
  name = "${var.app_name}-cluster"
}

# IAM Role for ECS Task Execution
resource "aws_iam_role" "ecs_task_execution" {
  name = "${var.app_name}-ecs-task-execution"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action    = "sts:AssumeRole"
      Effect    = "Allow"
      Principal = { Service = "ecs-tasks.amazonaws.com" }
    }]
  })
}

resource "aws_iam_role_policy_attachment" "ecs_task_execution" {
  role       = aws_iam_role.ecs_task_execution.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
}

# ALB, VPC, Subnets, Security Groups, ECS Service, Task Definition omitted for brevity
# (full configuration generated in step)

What does this Terraform configuration create?

• An ECR repository (fraud-detection-ecr) to store your Docker image.
• An S3 bucket for MLflow model artifacts in the cloud.
• An ECS Cluster (fraud-detection-cluster) to run your containerized service.
• An IAM role that gives ECS permission to pull images from ECR and write logs.
• The full configuration also includes an ALB, VPC, subnets, security groups, a Fargate task definition, and an ECS service. These are required for a complete deployment.

• Navigate into the terraform/ directory and initialize Terraform by running:

cd terraform
terraform init

• You should see a success message confirming Terraform has been initialized and the AWS provider was installed.

Terraform init failing?

• Make sure you are inside the terraform/ directory when running the command.
• If you see a credentials error, verify your AWS CLI is configured by running aws sts get-caller-identity.
• help me fix my terraform init failing for the fraud detection project

Deploy with terraform apply, push image to ECR, and verify

With your infrastructure defined, you can now provision everything on AWS, push your Docker image, and verify the service is live.

• Apply the Terraform configuration to create all AWS resources by running:

terraform apply

• Terraform will show you a plan of all the resources it will create. Type yes and press Enter to confirm.
• Wait for the apply to complete. This may take a few minutes as AWS provisions the ALB, ECS cluster, and networking.

What is terraform apply doing?

Terraform reads your main.tf configuration, compares it to the current state (nothing yet), and creates each resource in the correct dependency order.

It creates the VPC and subnets first, then the security groups, then the ALB, and finally the ECS service that ties everything together.

• Navigate back to the project root by running:

cd ..

• Authenticate Docker to your ECR registry by running:

aws ecr get-login-password --region [[AWS_REGION="us-east-1"]] | docker login --username AWS --password-stdin [[AWS_ACCOUNT_ID="your-aws-account-id"]].dkr.ecr.[[AWS_REGION="us-east-1"]].amazonaws.com

Where do I find my AWS account ID?

Run aws sts get-caller-identity and look for the Account field in the JSON output. It is a 12-digit number.

• Tag your local image for the ECR repository by running:

docker tag fraud-detection:latest [[AWS_ACCOUNT_ID="your-aws-account-id"]].dkr.ecr.[[AWS_REGION="us-east-1"]].amazonaws.com/fraud-detection-ecr:latest

• Push the image to ECR by running:

docker push [[AWS_ACCOUNT_ID="your-aws-account-id"]].dkr.ecr.[[AWS_REGION="us-east-1"]].amazonaws.com/fraud-detection-ecr:latest

• You should see each layer being pushed. Once complete, a final digest line confirms the push succeeded.

• Verify the service is responding via the ALB endpoint by running:

curl http://[[ALB_DNS="your-alb-dns-name"]]/health

• You should see {"status":"ok"} just like you did locally. Note down the ALB DNS name for later steps.

Where do I find my ALB DNS name?

After terraform apply completes, Terraform outputs the ALB DNS name. You can also find it in the AWS Console under EC2, then Load Balancers.

Service not responding via ALB?

• The ECS service may still be starting up. Wait 2-3 minutes after the push and try again.
• Check the ECS console to verify your task is in a RUNNING state.
• If the task keeps stopping, check the CloudWatch logs for your task. A common issue is the container failing to start because the model file is not accessible from the cloud environment.
• help me debug my ECS Fargate task not starting for the fraud detection service

✔️ Awesome, I've got everything!

Great. Double check you have saved your Dockerfile and terraform/main.tf.

ⓧ I'd like to double check the full code

Here is the complete Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

And here is the complete terraform/main.tf:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

variable "aws_region" {
  default = "us-east-1"
}

variable "app_name" {
  default = "fraud-detection"
}

# ECR Repository
resource "aws_ecr_repository" "fraud_detection" {
  name                 = "${var.app_name}-ecr"
  image_tag_mutability = "MUTABLE"
}

# S3 Bucket for MLflow artifacts
resource "aws_s3_bucket" "mlflow_artifacts" {
  bucket = "${var.app_name}-mlflow-artifacts"
}

# ECS Cluster
resource "aws_ecs_cluster" "fraud_detection" {
  name = "${var.app_name}-cluster"
}

# IAM Role for ECS Task Execution
resource "aws_iam_role" "ecs_task_execution" {
  name = "${var.app_name}-ecs-task-execution"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action    = "sts:AssumeRole"
      Effect    = "Allow"
      Principal = { Service = "ecs-tasks.amazonaws.com" }
    }]
  })
}

resource "aws_iam_role_policy_attachment" "ecs_task_execution" {
  role       = aws_iam_role.ecs_task_execution.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
}

# ALB, VPC, Subnets, Security Groups, ECS Service, Task Definition omitted for brevity
# (full configuration generated in step)

Your fraud detection service is now running on AWS. Next up, you will implement champion/challenger shadow deployment so you can safely evaluate new models against production traffic.

Implement Champion/Challenger Shadow Deployment

Your fraud detection service is live on AWS, serving real-time predictions through the ALB endpoint. But deploying a model is only half the story.

The real challenge is safely updating models in production without risking degraded predictions for your users. In this step, you will build a shadow deployment system that runs a challenger model silently alongside your champion, compares their performance, and automatically promotes the winner.

In this step, get ready to:

• Extend your FastAPI app to run a challenger model in shadow mode on every prediction request.
• Build a shadow evaluator script that compares champion vs. challenger over a configurable window.
• Implement automatic promotion logic that swaps models when the challenger wins.

Extend the FastAPI app with shadow mode

Shadow deployment means running two models on the same inputs simultaneously. The champion model's prediction is returned to the caller, while the challenger runs silently in the background.

Both predictions get logged so you can compare them later without any risk to production responses.

Why shadow deployment instead of A/B testing?

A/B testing splits traffic between models, meaning some users get predictions from an unproven model. Shadow deployment eliminates this risk entirely.

Every user still gets the champion's response, but you collect real production data on the challenger. Companies like Uber and Stripe use this pattern to validate ML models safely before any traffic shift.

• In app/main.py, find the imports section at the top of the file.
• Replace everything from the docstring through the load_model() function with the following updated code:

"""FastAPI inference service with champion/challenger shadow deployment."""
import os
import json
import time
import logging
from datetime import datetime
from typing import List, Optional

import mlflow
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

from src.feature_store import FeatureStore

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

app = FastAPI(title='Fraud Detection Inference Service')

REQUEST_COUNT = Counter('fraud_requests_total', 'Total prediction requests', ['endpoint', 'status'])
LATENCY = Histogram('fraud_request_latency_seconds', 'Request latency', ['endpoint'],
                    buckets=[0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0])
PREDICTION_DIST = Counter('fraud_predictions_total', 'Prediction distribution', ['label'])
ERROR_RATE = Counter('fraud_errors_total', 'Total errors')

# Champion serves production traffic; challenger runs in shadow
champion_model = None
challenger_model = None
feature_store = FeatureStore()
shadow_log = []


@app.on_event('startup')
def load_models():
    global champion_model, challenger_model
    champion_model = mlflow.xgboost.load_model('models:/fraud-detection-champion/Production')
    try:
        challenger_model = mlflow.xgboost.load_model('models:/fraud-detection-challenger/Staging')
        logger.info('Challenger model loaded for shadow deployment.')
    except Exception:
        logger.info('No challenger model found. Running champion only.')
    train_df = pd.read_parquet('data/train.parquet')
    feature_store.fit(train_df.drop(columns=['Class']))
    logger.info('Champion model loaded from MLflow Registry.')

What changed from the previous version?

• Added challenger_model and shadow_log globals to hold the shadow model and comparison data.
• Renamed load_model() to load_models() and added a try/except block that attempts to load the challenger from the Staging stage of the fraud-detection-challenger registry.
• Removed the Gauge import since it is no longer used. Added Optional to the typing imports.

• Now find the existing /predict endpoint function and replace it with this updated version that includes shadow logic:

@app.post('/predict')
def predict(request: PredictRequest):
    start = time.time()
    try:
        df = pd.DataFrame([request.features])
        X = feature_store.transform(df)
        proba = champion_model.predict_proba(X)[0][1]
        label = int(proba >= 0.5)
        # Shadow: run challenger if available
        challenger_proba = None
        if challenger_model is not None:
            challenger_proba = float(challenger_model.predict_proba(X)[0][1])
            shadow_log.append({'champion_proba': float(proba), 'challenger_proba': challenger_proba,
                               'timestamp': datetime.utcnow().isoformat()})
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict', status='success').inc()
        PREDICTION_DIST.labels(label=str(label)).inc()
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'input_features': request.features,
            'prediction': label,
            'confidence': float(proba),
            'challenger_confidence': challenger_proba,
            'latency_ms': round(latency * 1000, 2),
        }
        logger.info(json.dumps(log_entry))
        return {'prediction': label, 'confidence': float(proba), 'latency_ms': log_entry['latency_ms']}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))

What does the shadow logic do?

After the champion produces its prediction, the code checks if a challenger_model is loaded. If it is, the challenger runs predict_proba on the exact same transformed features.

Both probabilities get appended to shadow_log with a timestamp. The champion's prediction is still returned to the caller unchanged. The challenger's result is only logged for later analysis.

• Finally, add this new endpoint after the /predict/batch function and before the /metrics endpoint:

@app.get('/shadow/log')
def get_shadow_log():
    return {'shadow_log': shadow_log[-100:]}

What does this endpoint do?

The /shadow/log endpoint exposes the last 100 shadow comparison entries. Your shadow evaluator script will call this endpoint to retrieve the data it needs to decide whether to promote the challenger.

• Save app/main.py.
• Rebuild your Docker image locally to verify the code compiles without errors by running:

docker build -t fraud-detection:shadow .

• You should see the build complete successfully with no import or syntax errors.

Build failing with an import error?

• Check that you removed Gauge from the prometheus_client import line. The updated code no longer uses it.
• Make sure you have from typing import List, Optional with both types imported.

Help me debug my shadow deployment Docker build error.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before moving on.

ⓧ I'd like to double check the full code

"""FastAPI inference service with champion/challenger shadow deployment."""
import os
import json
import time
import logging
from datetime import datetime
from typing import List, Optional

import mlflow
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

from src.feature_store import FeatureStore

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

app = FastAPI(title='Fraud Detection Inference Service')

REQUEST_COUNT = Counter('fraud_requests_total', 'Total prediction requests', ['endpoint', 'status'])
LATENCY = Histogram('fraud_request_latency_seconds', 'Request latency', ['endpoint'],
                    buckets=[0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0])
PREDICTION_DIST = Counter('fraud_predictions_total', 'Prediction distribution', ['label'])
ERROR_RATE = Counter('fraud_errors_total', 'Total errors')

# Champion serves production traffic; challenger runs in shadow
champion_model = None
challenger_model = None
feature_store = FeatureStore()
shadow_log = []


@app.on_event('startup')
def load_models():
    global champion_model, challenger_model
    champion_model = mlflow.xgboost.load_model('models:/fraud-detection-champion/Production')
    try:
        challenger_model = mlflow.xgboost.load_model('models:/fraud-detection-challenger/Staging')
        logger.info('Challenger model loaded for shadow deployment.')
    except Exception:
        logger.info('No challenger model found. Running champion only.')
    train_df = pd.read_parquet('data/train.parquet')
    feature_store.fit(train_df.drop(columns=['Class']))
    logger.info('Champion model loaded from MLflow Registry.')


class PredictRequest(BaseModel):
    features: dict


class PredictBatchRequest(BaseModel):
    records: List[dict]


@app.post('/predict')
def predict(request: PredictRequest):
    start = time.time()
    try:
        df = pd.DataFrame([request.features])
        X = feature_store.transform(df)
        proba = champion_model.predict_proba(X)[0][1]
        label = int(proba >= 0.5)
        # Shadow: run challenger if available
        challenger_proba = None
        if challenger_model is not None:
            challenger_proba = float(challenger_model.predict_proba(X)[0][1])
            shadow_log.append({'champion_proba': float(proba), 'challenger_proba': challenger_proba,
                               'timestamp': datetime.utcnow().isoformat()})
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict', status='success').inc()
        PREDICTION_DIST.labels(label=str(label)).inc()
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'input_features': request.features,
            'prediction': label,
            'confidence': float(proba),
            'challenger_confidence': challenger_proba,
            'latency_ms': round(latency * 1000, 2),
        }
        logger.info(json.dumps(log_entry))
        return {'prediction': label, 'confidence': float(proba), 'latency_ms': log_entry['latency_ms']}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))


@app.post('/predict/batch')
def predict_batch(request: PredictBatchRequest):
    start = time.time()
    try:
        df = pd.DataFrame(request.records)
        X = feature_store.transform(df)
        probas = champion_model.predict_proba(X)[:, 1]
        labels = (probas >= 0.5).astype(int).tolist()
        latency = time.time() - start
        LATENCY.labels(endpoint='/predict/batch').observe(latency)
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='success').inc()
        for label in labels:
            PREDICTION_DIST.labels(label=str(label)).inc()
        return {'predictions': labels, 'confidences': probas.tolist(), 'latency_ms': round(latency * 1000, 2)}
    except Exception as e:
        ERROR_RATE.inc()
        REQUEST_COUNT.labels(endpoint='/predict/batch', status='error').inc()
        raise HTTPException(status_code=500, detail=str(e))


@app.get('/shadow/log')
def get_shadow_log():
    return {'shadow_log': shadow_log[-100:]}


@app.get('/metrics')
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.get('/health')
def health():
    return {'status': 'ok'}

Build the shadow evaluator script

The shadow log accumulates raw probability pairs, but you need a process to analyze them. The shadow evaluator compares champion vs. challenger predictions over a configurable window of N requests and decides whether the challenger is ready for promotion.

• Create the scripts/ directory if it does not already exist by running:

mkdir -p scripts

• Create a new file scripts/shadow_evaluator.py and add the imports and configuration constants:

"""Compare champion vs challenger predictions and auto-promote if challenger wins."""
import os
import mlflow
from mlflow.tracking import MlflowClient
import numpy as np

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
client = MlflowClient()

# Challenger must beat champion by this margin to be promoted
PROMOTION_THRESHOLD = float(os.getenv('PROMOTION_THRESHOLD', '0.02'))  # challenger must beat champion by 2%
# Minimum number of shadow predictions before evaluation
MIN_WINDOW = int(os.getenv('MIN_WINDOW', '500'))

What do these configuration values control?

• PROMOTION_THRESHOLD (default 0.02) means the challenger must outperform the champion by at least 2% on the evaluation metric to earn a promotion.
• MIN_WINDOW (default 500) prevents premature decisions by requiring at least 500 shadow predictions before any evaluation runs.
• Both are configurable via environment variables so you can tune them without changing code.

• Below the configuration constants, add the evaluate_shadow function that performs the comparison:

def evaluate_shadow(shadow_log: list) -> dict:
    if len(shadow_log) < MIN_WINDOW:
        return {'status': 'insufficient_data', 'n': len(shadow_log)}
    champion_probas = np.array([r['champion_proba'] for r in shadow_log])
    challenger_probas = np.array([r['challenger_proba'] for r in shadow_log])
    champion_labels = (champion_probas >= 0.5).astype(int)
    challenger_labels = (challenger_probas >= 0.5).astype(int)
    agreement_rate = float(np.mean(champion_labels == challenger_labels))
    # If ground truth available, compare PR-AUC; otherwise use confidence proxy
    challenger_advantage = float(np.mean(challenger_probas) - np.mean(champion_probas))
    result = {
        'n': len(shadow_log),
        'agreement_rate': agreement_rate,
        'challenger_advantage': challenger_advantage,
        'promote': challenger_advantage >= PROMOTION_THRESHOLD,
    }
    if result['promote']:
        _promote_challenger()
        result['action'] = 'promoted'
    return result

What does the evaluation logic measure?

The function computes two key metrics from the shadow log. The agreement_rate shows how often champion and challenger produce the same binary label (fraud/not-fraud).

The challenger_advantage measures whether the challenger produces higher confidence scores on average. When ground truth labels become available, you can replace this proxy with a direct PR-AUC comparison.

• Save scripts/shadow_evaluator.py. You will not see any output yet because the promotion function is not defined. The next substep adds it.

Implement automatic promotion logic

When the evaluator decides the challenger has won, it needs to swap the models in the MLflow Model Registry. This means archiving the current champion and transitioning the challenger to Production stage.

• In scripts/shadow_evaluator.py, add the _promote_challenger function below the evaluate_shadow function:

def _promote_challenger():
    client.transition_model_version_stage('fraud-detection-champion',
        client.get_latest_versions('fraud-detection-champion', stages=['Production'])[0].version,
        stage='Archived')
    challenger_version = client.get_latest_versions('fraud-detection-challenger', stages=['Staging'])[0].version
    client.transition_model_version_stage('fraud-detection-challenger', challenger_version, stage='Production')
    client.set_registered_model_alias('fraud-detection-champion', 'Production', challenger_version)
    mlflow.log_param('promotion_event', f'challenger v{challenger_version} promoted to Production')
    print(f'Challenger v{challenger_version} promoted to Production.')

What happens during promotion?

• The current champion version moves to Archived stage, preserving it as a rollback target.
• The challenger version transitions from Staging to Production in the registry.
• A model alias is set so the champion registry name still resolves to the new production model.
• The promotion event is logged to MLflow for audit trail purposes.

• Save scripts/shadow_evaluator.py.
• Verify the script has no syntax errors by running:

python -c "import scripts.shadow_evaluator"

• You should see no output, which confirms the module loads cleanly.

Seeing a ModuleNotFoundError?

• Make sure you are running the command from your project root directory (where your scripts/ folder lives).
• Confirm your virtual environment is active and mlflow and numpy are installed.

Help me fix my shadow_evaluator import error.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before continuing.

ⓧ I'd like to double check the full code

"""Compare champion vs challenger predictions and auto-promote if challenger wins."""
import os
import mlflow
from mlflow.tracking import MlflowClient
import numpy as np

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
client = MlflowClient()

PROMOTION_THRESHOLD = float(os.getenv('PROMOTION_THRESHOLD', '0.02'))  # challenger must beat champion by 2%
MIN_WINDOW = int(os.getenv('MIN_WINDOW', '500'))


def evaluate_shadow(shadow_log: list) -> dict:
    if len(shadow_log) < MIN_WINDOW:
        return {'status': 'insufficient_data', 'n': len(shadow_log)}
    champion_probas = np.array([r['champion_proba'] for r in shadow_log])
    challenger_probas = np.array([r['challenger_proba'] for r in shadow_log])
    champion_labels = (champion_probas >= 0.5).astype(int)
    challenger_labels = (challenger_probas >= 0.5).astype(int)
    agreement_rate = float(np.mean(champion_labels == challenger_labels))
    # If ground truth available, compare PR-AUC; otherwise use confidence proxy
    challenger_advantage = float(np.mean(challenger_probas) - np.mean(champion_probas))
    result = {
        'n': len(shadow_log),
        'agreement_rate': agreement_rate,
        'challenger_advantage': challenger_advantage,
        'promote': challenger_advantage >= PROMOTION_THRESHOLD,
    }
    if result['promote']:
        _promote_challenger()
        result['action'] = 'promoted'
    return result


def _promote_challenger():
    client.transition_model_version_stage('fraud-detection-champion',
        client.get_latest_versions('fraud-detection-champion', stages=['Production'])[0].version,
        stage='Archived')
    challenger_version = client.get_latest_versions('fraud-detection-challenger', stages=['Staging'])[0].version
    client.transition_model_version_stage('fraud-detection-challenger', challenger_version, stage='Production')
    client.set_registered_model_alias('fraud-detection-champion', 'Production', challenger_version)
    mlflow.log_param('promotion_event', f'challenger v{challenger_version} promoted to Production')
    print(f'Challenger v{challenger_version} promoted to Production.')

Your shadow deployment system is fully wired. Every prediction request now silently evaluates the challenger, and the evaluator script can trigger an automatic model swap when the challenger proves itself. Next up, you will add drift detection and operational monitoring to catch when model performance degrades over time.

Set Up Drift Detection & Operational Monitoring

Your shadow deployment is running on AWS, silently evaluating the challenger model against the champion. But how do you know when the production data itself has changed?

In this step, you will build a monitoring layer that detects data drift, tracks operational health metrics, and triggers alerts when things go wrong. This is the early warning system that tells your pipeline when it is time to retrain.

In this step, get ready to:

• Build an Evidently drift monitoring pipeline that compares production data against the training reference.
• Configure Prometheus and Grafana to scrape and visualize your inference service metrics.
• Create alerting rules that trigger retraining on drift and alert on SLO violations.

Build the Evidently drift monitoring pipeline

A model can perform perfectly today and silently degrade tomorrow if the incoming data distribution shifts. This is called data drift. Your fraud detection model was trained on historical transaction patterns. If real-world fraud patterns change, your model needs to know.

Evidently is an open-source ML monitoring library purpose-built for detecting distribution shifts. It compares your current production data against the training reference dataset and flags when statistical tests detect a significant change.

• Create the drift monitoring script by adding the following code to scripts/drift_monitor.py:

"""Evidently drift monitoring pipeline."""
import os
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset

REFERENCE_DATA_PATH = os.getenv('REFERENCE_DATA_PATH', 'data/train.parquet')
PRODUCTION_DATA_PATH = os.getenv('PRODUCTION_DATA_PATH', 'data/production_log.parquet')
DRIFT_THRESHOLD = float(os.getenv('DRIFT_THRESHOLD', '0.05'))

reference = pd.read_parquet(REFERENCE_DATA_PATH)
production = pd.read_parquet(PRODUCTION_DATA_PATH)

report = Report(metrics=[DataDriftPreset(stattest='ks', stattest_threshold=DRIFT_THRESHOLD),
                          TargetDriftPreset()])
report.run(reference_data=reference, current_data=production)
report.save_html('reports/drift_report.html')

result = report.as_dict()
drift_detected = result['metrics'][0]['result']['dataset_drift']
print(f'Drift detected: {drift_detected}')
if drift_detected:
    print('ALERT: Data drift detected. Triggering retraining workflow.')
    # In CI/CD, this exits with code 1 to trigger the retraining GitHub Actions workflow
    exit(1)

What does this script do?

This script loads two datasets: the training reference (what the model was trained on) and the recent production data (what the model has been scoring in production).

• DataDriftPreset runs a Kolmogorov-Smirnov (KS) test on each feature column with a threshold of 0.05. If enough features drift, the dataset is flagged.
• TargetDriftPreset checks whether the prediction target distribution has shifted.
• If drift is detected, the script exits with code 1. In your CI/CD pipeline (Step 12), this exit code triggers the automated retraining workflow.

• Create the reports output directory by running:

mkdir -p reports

• Save scripts/drift_monitor.py.
• Confirm the file was created by running ls scripts/. You should see drift_monitor.py alongside shadow_evaluator.py.

Import error with evidently?

Make sure you installed Evidently in your virtual environment. Evidently requires Python 3.10 or higher.

help me fix an Evidently import error in my drift monitor script

Why a KS-test threshold of 0.05?

The Kolmogorov-Smirnov test measures how different two distributions are. A threshold of 0.05 means that if the p-value of the test falls below 0.05 for a feature, that feature is considered drifted.

This is a standard significance level. Too low and you miss real drift. Too high and you trigger false alarms on normal variation.

✔️ Awesome, I've got everything!

Great. Double check you have saved scripts/drift_monitor.py before moving on.

ⓧ I'd like to double check the full code

"""Evidently drift monitoring pipeline."""
import os
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset

REFERENCE_DATA_PATH = os.getenv('REFERENCE_DATA_PATH', 'data/train.parquet')
PRODUCTION_DATA_PATH = os.getenv('PRODUCTION_DATA_PATH', 'data/production_log.parquet')
DRIFT_THRESHOLD = float(os.getenv('DRIFT_THRESHOLD', '0.05'))

reference = pd.read_parquet(REFERENCE_DATA_PATH)
production = pd.read_parquet(PRODUCTION_DATA_PATH)

report = Report(metrics=[DataDriftPreset(stattest='ks', stattest_threshold=DRIFT_THRESHOLD),
                          TargetDriftPreset()])
report.run(reference_data=reference, current_data=production)
report.save_html('reports/drift_report.html')

result = report.as_dict()
drift_detected = result['metrics'][0]['result']['dataset_drift']
print(f'Drift detected: {drift_detected}')
if drift_detected:
    print('ALERT: Data drift detected. Triggering retraining workflow.')
    # In CI/CD, this exits with code 1 to trigger the retraining GitHub Actions workflow
    exit(1)

Configure Prometheus and Grafana for operational monitoring

Your FastAPI service already exposes a /metrics endpoint with Prometheus metrics (request count, latency histogram, prediction distribution, error rate). Now you need something to scrape those metrics and visualize them.

Prometheus collects metrics by pulling from your service on a schedule. Grafana connects to Prometheus and renders dashboards you can watch in real-time.

• Create the monitoring directory structure by running:

mkdir -p monitoring/grafana/datasources
mkdir -p monitoring/grafana/dashboards

• Create the Prometheus configuration by adding the following to monitoring/prometheus.yml:

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'fraud-detection'
    static_configs:
      - targets: ['host.docker.internal:8000']
    metrics_path: '/metrics'

alerting:
  alertmanagers:
    - static_configs:
        - targets: []

rule_files:
  - 'alert_rules.yml'

What does this configuration do?

• scrape_interval: 15s tells Prometheus to pull metrics from your service every 15 seconds.
• host.docker.internal:8000 lets the Prometheus container reach your FastAPI service running on the host machine at port 8000.
• rule_files points to the alerting rules file you will create next.

• Create the Grafana datasource provisioning file at monitoring/grafana/datasources/datasource.yml:

apiVersion: 1
datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true

What is Grafana provisioning?

Provisioning lets Grafana automatically configure datasources and dashboards at startup from YAML files. This means you do not have to manually add Prometheus as a datasource through the Grafana UI every time you spin up the stack.

• Create a dashboard provider configuration at monitoring/grafana/dashboards/dashboard.yml:

apiVersion: 1
providers:
  - name: 'Fraud Detection'
    orgId: 1
    folder: ''
    type: file
    options:
      path: /etc/grafana/provisioning/dashboards

• Create the main dashboard JSON at monitoring/grafana/dashboards/fraud_detection.json:

{
  "dashboard": {
    "title": "Fraud Detection Service",
    "panels": [
      {
        "title": "Request Latency (p95)",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(fraud_request_latency_seconds_bucket[5m]))",
            "legendFormat": "p95 latency"
          }
        ]
      },
      {
        "title": "Request Throughput",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(fraud_requests_total[5m])",
            "legendFormat": "{{endpoint}} - {{status}}"
          }
        ]
      },
      {
        "title": "Error Rate",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(fraud_errors_total[5m])",
            "legendFormat": "errors/sec"
          }
        ]
      },
      {
        "title": "Prediction Distribution",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(fraud_predictions_total[5m])",
            "legendFormat": "{{label}}"
          }
        ]
      }
    ]
  }
}

What do these dashboard panels show?

• Request Latency (p95) tracks the 95th percentile response time. This is your primary latency SLO metric.
• Request Throughput shows requests per second, broken down by endpoint and success/error status.
• Error Rate shows how many errors the service is throwing per second.
• Prediction Distribution tracks the ratio of fraud vs. genuine predictions. A sudden shift here without a corresponding data change is a red flag.

Now update your Docker Compose stack to include Prometheus and Grafana.

• In docker-compose.yml, find the volumes: section at the bottom of the file.
• Add the following two service definitions above the volumes: section (at the same indentation level as the existing mlflow: service):

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:latest
    depends_on:
      - prometheus
    ports:
      - '3000:3000'
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana:/etc/grafana/provisioning

• Also add grafana_data: to the volumes: section at the bottom of the file, below minio_data:.

• Save docker-compose.yml and bring up the updated stack by running:

docker compose up -d

• Verify Prometheus is running by opening http://localhost:9090 in your browser. You should see the Prometheus expression browser.
• Verify Grafana is running by opening http://localhost:3000 in your browser. Log in with the default credentials (admin / admin).

Prometheus not showing targets?

In the Prometheus UI, navigate to **Status** then **Targets**. If your fraud-detection job shows as DOWN, your FastAPI service is not running on port 8000.

Start your FastAPI service locally with uvicorn app.main:app --port 8000 before checking targets again.

help me debug why Prometheus cannot scrape my FastAPI metrics endpoint

✔️ Awesome, I've got everything!

Great. Make sure both Prometheus and Grafana are accessible in your browser before continuing.

ⓧ I'd like to double check the full code

version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: mlflow
      POSTGRES_DB: mlflow
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  minio:
    image: minio/minio
    command: server /data --console-address ':9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - '9000:9000'
      - '9001:9001'
    volumes:
      - minio_data:/data

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    depends_on:
      - postgres
      - minio
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://minio:9000
      AWS_ACCESS_KEY_ID: minioadmin
      AWS_SECRET_ACCESS_KEY: minioadmin
    ports:
      - '5000:5000'
    command: >
      mlflow server
      --backend-store-uri postgresql://mlflow:mlflow@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
      --host 0.0.0.0
      --port 5000

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:latest
    depends_on:
      - prometheus
    ports:
      - '3000:3000'
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana:/etc/grafana/provisioning

volumes:
  postgres_data:
  minio_data:
  grafana_data:

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'fraud-detection'
    static_configs:
      - targets: ['host.docker.internal:8000']
    metrics_path: '/metrics'

alerting:
  alertmanagers:
    - static_configs:
        - targets: []

rule_files:
  - 'alert_rules.yml'

Create alerting rules for drift, latency, and errors

Dashboards are great for humans watching in real-time. But production systems need automated alerts that fire when things go wrong, even when nobody is looking.

• Create the Prometheus alerting rules file at monitoring/alert_rules.yml:

groups:
  - name: fraud_detection_alerts
    rules:
      # Alert if p95 latency exceeds 100ms SLO for 5 minutes
      - alert: HighLatency
        expr: histogram_quantile(0.95, rate(fraud_request_latency_seconds_bucket[5m])) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "p95 latency exceeds 100ms SLO"

      # Alert if error rate spikes above 5% for 2 minutes
      - alert: HighErrorRate
        expr: rate(fraud_errors_total[5m]) / rate(fraud_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Error rate exceeds 5% threshold"

      # Alert if prediction distribution shifts dramatically
      - alert: PredictionDistributionShift
        expr: rate(fraud_predictions_total{label="1"}[1h]) / rate(fraud_predictions_total[1h]) > 0.1
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Fraud prediction rate exceeds 10% - possible drift"

What do these alerting rules do?

• HighLatency fires if p95 latency exceeds 100ms for 5 consecutive minutes. This is your latency SLO violation alert.
• HighErrorRate fires if more than 5% of requests are failing for 2 minutes. This catches service degradation early.
• PredictionDistributionShift fires if the model is flagging more than 10% of transactions as fraud for 15 minutes. In normal operation, fraud is less than 1% of transactions. A sudden spike means the model or the data has changed.

How does drift detection connect to alerting?

Your drift detection uses two layers. The Evidently pipeline (drift_monitor.py) runs on a schedule and performs deep statistical analysis on the full feature distribution. It detects slow, subtle drift.

The Prometheus alerting rules above provide fast operational signals. They catch sudden changes in model behavior (prediction distribution shift) within minutes. Together, they cover both fast and slow failure modes.

• Save monitoring/alert_rules.yml.
• Verify the alerting rules are loaded by opening http://localhost:9090/alerts in your browser. You should see HighLatency, HighErrorRate, and PredictionDistributionShift listed.

Connecting alerts to retraining

In Step 12, you will connect these alerts to your GitHub Actions CI/CD pipeline. When the drift_monitor.py script exits with code 1, the GitHub Actions workflow triggers automatic retraining. For now, the alerting rules live here ready to fire.

Alerts not showing in Prometheus?

If the alerts page is empty or shows an error, the alert_rules.yml file may not be mounted into the Prometheus container. You need to add an additional volume mount to the prometheus service in docker-compose.yml:

    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - ./monitoring/alert_rules.yml:/etc/prometheus/alert_rules.yml

Then restart the stack with docker compose up -d.

help me get Prometheus to load my alert_rules.yml file

Your monitoring layer is complete. You have Evidently for deep statistical drift detection, Prometheus for real-time metric collection, Grafana for visualization, and alerting rules that catch latency, error, and distribution problems automatically. Next up, you will build the automated rollback system that responds when these alerts fire.

Build Automated Rollback & Self-Healing

Your monitoring system can now detect drift and track operational metrics. But detection alone is not enough.

In production, a degraded model can cost real money and erode user trust every second it stays live. This step builds the self-healing layer: a health-check loop that watches your latency SLOs, and a rollback script that automatically reverts to the previous stable model version from MLflow Registry when things go wrong.

In this step, get ready to:

• Implement a health-check loop monitoring live latency SLO violations.
• Build rollback logic that reverts to the previous Production model version.
• Test the rollback by deploying a degraded model and verifying self-healing.

Implement the health-check loop

A health monitor continuously polls your inference service's /metrics endpoint and counts consecutive SLO violations. Once violations exceed a sustained threshold, it triggers an automatic rollback.

• Create the health monitor script by adding the following code to scripts/health_monitor.py:

"""Health-check loop monitoring live performance metrics."""
import os
import time
import requests

METRICS_URL = os.getenv('METRICS_URL', 'http://localhost:8000/metrics')
LATENCY_SLO_P95 = float(os.getenv('LATENCY_SLO_P95', '0.1'))  # 100ms
ERROR_RATE_THRESHOLD = float(os.getenv('ERROR_RATE_THRESHOLD', '0.05'))
SUSTAINED_WINDOW = int(os.getenv('SUSTAINED_WINDOW', '5'))  # consecutive violations

violation_count = 0

while True:
    try:
        resp = requests.get(METRICS_URL, timeout=5)
        metrics_text = resp.text
        # Parse p95 latency and error rate from Prometheus text format
        # (simplified; production would use prometheus_client parser)
        violation = False
        for line in metrics_text.splitlines():
            if 'fraud_request_latency_seconds{' in line and '0.1}' in line:
                val = float(line.split()[-1])
                if val > LATENCY_SLO_P95:
                    violation = True
        if violation:
            violation_count += 1
            print(f'SLO violation #{violation_count}')
        else:
            violation_count = 0
        if violation_count >= SUSTAINED_WINDOW:
            print('Sustained SLO violation. Triggering rollback.')
            os.system('python scripts/rollback.py')
            violation_count = 0
    except Exception as e:
        print(f'Health check error: {e}')
    time.sleep(30)

What does this code do?

The health monitor runs an infinite loop that polls the FastAPI /metrics endpoint every 30 seconds.

• LATENCY_SLO_P95 defines the p95 latency threshold at 100ms. Any response above this counts as a violation.
• SUSTAINED_WINDOW requires 5 consecutive violations before triggering rollback. This prevents a single spike from causing unnecessary model swaps.
• When sustained violations are detected, the script calls scripts/rollback.py to revert the model automatically.

• Save scripts/health_monitor.py.

Getting a ModuleNotFoundError for requests?

Make sure your virtual environment is activated and requests is installed. It should already be present from your project dependencies.

help me debug why requests is not found in my Python environment

Build the rollback logic

The rollback script uses the MLflow Model Registry to swap model versions. It finds the most recent Archived version, demotes the current Production model, and promotes the archived version back to Production.

• Create the rollback script by adding the following code to scripts/rollback.py:

"""Rollback to previous Production model version from MLflow Registry."""
import os
import mlflow
from mlflow.tracking import MlflowClient
from datetime import datetime

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
client = MlflowClient()

MODEL_NAME = 'fraud-detection-champion'

all_versions = client.search_model_versions(f"name='{MODEL_NAME}'")
production_versions = [v for v in all_versions if v.current_stage == 'Production']
archived_versions = sorted([v for v in all_versions if v.current_stage == 'Archived'],
                           key=lambda v: int(v.version), reverse=True)

if not archived_versions:
    print('No archived version to roll back to.')
    exit(1)

previous_version = archived_versions[0]
for v in production_versions:
    client.transition_model_version_stage(MODEL_NAME, v.version, stage='Archived')
client.transition_model_version_stage(MODEL_NAME, previous_version.version, stage='Production')

rollback_event = {
    'timestamp': datetime.utcnow().isoformat(),
    'rolled_back_to_version': previous_version.version,
    'reason': 'Sustained SLO violation',
}
print(f'Rollback complete: {rollback_event}')
with mlflow.start_run(run_name='rollback_event'):
    mlflow.log_params(rollback_event)

What does this code do?

The rollback script performs a safe model version swap using the MLflow Model Registry API.

• search_model_versions retrieves all versions of the fraud-detection-champion model and filters them by stage.
• transition_model_version_stage moves the current Production version to Archived, then promotes the most recent Archived version back to Production.
• The rollback event is logged to MLflow as a run with parameters recording the timestamp, version, and reason. This creates an audit trail for every automatic rollback.

• Save scripts/rollback.py.

Why log rollbacks to MLflow?

Logging each rollback as an MLflow run creates a permanent audit trail. When you investigate an incident days later, you can see exactly which version was rolled back, when it happened, and why. This is the same pattern ML platform teams at companies like Stripe use for post-incident analysis.

Seeing a connection refused error to MLflow?

Make sure your Docker Compose stack is running with docker compose up -d and that the MLflow UI is accessible at http://localhost:5000.

help me troubleshoot why my rollback script cannot connect to MLflow

Test the rollback with a degraded model

The final proof of a self-healing system is an end-to-end test. You will intentionally deploy a degraded model version, then confirm the health monitor detects the SLO violations and triggers rollback within the expected time window.

• Register a second model version with intentionally poor performance (for example, a model trained on 1% of the data) to simulate degradation.
• Transition that version to Production in the MLflow Registry, archiving the current champion.
• Start the health monitor by running:

python scripts/health_monitor.py

• Send traffic to the inference service while the degraded model is serving. The health monitor should detect sustained SLO violations and call scripts/rollback.py automatically.
• Verify the rollback completed by checking the MLflow UI. The degraded model version should now be in the Archived stage, and the previous champion should be back in Production.
• Confirm a rollback_event run appears in MLflow with the timestamp, rolled-back version number, and reason.

What should I see?

In the terminal running the health monitor, you should see output like:

SLO violation #1
SLO violation #2
SLO violation #3
SLO violation #4
SLO violation #5
Sustained SLO violation. Triggering rollback.
Rollback complete: {'timestamp': '...', 'rolled_back_to_version': '1', 'reason': 'Sustained SLO violation'}

In the MLflow UI, the degraded version moves to Archived and the previous champion returns to Production. A new run named rollback_event should appear in your experiment list.

Rollback did not trigger?

Make sure you are sending enough traffic to generate p95 latency metrics. The health monitor only triggers after 5 consecutive violations (every 30 seconds), so the degraded model needs to be consistently slow for at least 2.5 minutes.

Also check that your METRICS_URL environment variable matches the URL where your inference service exposes Prometheus metrics.

help me figure out why my health monitor is not triggering rollback

✔️ Awesome, I've got everything!

Great. Double check you have saved both scripts/health_monitor.py and scripts/rollback.py.

ⓧ I'd like to double check the full code

Here is the complete scripts/health_monitor.py:

"""Health-check loop monitoring live performance metrics."""
import os
import time
import requests

METRICS_URL = os.getenv('METRICS_URL', 'http://localhost:8000/metrics')
LATENCY_SLO_P95 = float(os.getenv('LATENCY_SLO_P95', '0.1'))  # 100ms
ERROR_RATE_THRESHOLD = float(os.getenv('ERROR_RATE_THRESHOLD', '0.05'))
SUSTAINED_WINDOW = int(os.getenv('SUSTAINED_WINDOW', '5'))  # consecutive violations

violation_count = 0

while True:
    try:
        resp = requests.get(METRICS_URL, timeout=5)
        metrics_text = resp.text
        # Parse p95 latency and error rate from Prometheus text format
        # (simplified; production would use prometheus_client parser)
        violation = False
        for line in metrics_text.splitlines():
            if 'fraud_request_latency_seconds{' in line and '0.1}' in line:
                val = float(line.split()[-1])
                if val > LATENCY_SLO_P95:
                    violation = True
        if violation:
            violation_count += 1
            print(f'SLO violation #{violation_count}')
        else:
            violation_count = 0
        if violation_count >= SUSTAINED_WINDOW:
            print('Sustained SLO violation. Triggering rollback.')
            os.system('python scripts/rollback.py')
            violation_count = 0
    except Exception as e:
        print(f'Health check error: {e}')
    time.sleep(30)

Here is the complete scripts/rollback.py:

"""Rollback to previous Production model version from MLflow Registry."""
import os
import mlflow
from mlflow.tracking import MlflowClient
from datetime import datetime

MLFLOW_TRACKING_URI = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
client = MlflowClient()

MODEL_NAME = 'fraud-detection-champion'

all_versions = client.search_model_versions(f"name='{MODEL_NAME}'")
production_versions = [v for v in all_versions if v.current_stage == 'Production']
archived_versions = sorted([v for v in all_versions if v.current_stage == 'Archived'],
                           key=lambda v: int(v.version), reverse=True)

if not archived_versions:
    print('No archived version to roll back to.')
    exit(1)

previous_version = archived_versions[0]
for v in production_versions:
    client.transition_model_version_stage(MODEL_NAME, v.version, stage='Archived')
client.transition_model_version_stage(MODEL_NAME, previous_version.version, stage='Production')

rollback_event = {
    'timestamp': datetime.utcnow().isoformat(),
    'rolled_back_to_version': previous_version.version,
    'reason': 'Sustained SLO violation',
}
print(f'Rollback complete: {rollback_event}')
with mlflow.start_run(run_name='rollback_event'):
    mlflow.log_params(rollback_event)

Your system can now detect degradation and heal itself without human intervention. Next up, you will wire this into a full CI/CD pipeline with GitHub Actions so the entire ML lifecycle runs automatically.

Create the CI/CD Pipeline with GitHub Actions

Your fraud detection system now self-heals when performance degrades. The next challenge is automating the entire ML lifecycle so that model updates flow from code change to production without manual intervention.

In this step, you will create GitHub Actions workflows that automate linting, testing, building, and deploying your service. You will also build a retraining pipeline that responds to drift detection by training a new challenger model and promoting it safely.

In this step, get ready to:

• Create a CI workflow that lints, tests, builds, and pushes your Docker image to ECR on every push.
• Create a retraining workflow that triggers on drift detection and deploys a challenger model.
• Configure GitHub repository secrets and deployment safeguards with automatic rollback.

Create the CI workflow for build and deploy

The CI workflow runs on every push to the main branch. It lints and tests your code first, then builds a Docker image and pushes it to your ECR repository.

• Create the workflow directory structure by running this command:

mkdir -p .github/workflows

• Create the CI workflow file by running:

touch .github/workflows/ci.yml

• Confirm the files exist by running:

ls .github/workflows

You should see ci.yml in the output.

• Open .github/workflows/ci.yml in your code editor and paste the following workflow definition:

name: CI - Build and Deploy

on:
  push:
    branches: [main]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - run: pip install flake8 pytest
      - run: flake8 app/ src/ scripts/
      - run: pytest tests/ -v

  build-and-push:
    needs: lint-and-test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ secrets.AWS_REGION }}
      - name: Login to ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v2
      - name: Build and push Docker image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          ECR_REPOSITORY: ${{ secrets.ECR_REPOSITORY }}
          IMAGE_TAG: ${{ github.sha }}
        run: |
          docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
      - name: Update ECS service
        run: |
          aws ecs update-service --cluster fraud-detection-cluster \
            --service fraud-detection-service \
            --force-new-deployment

What does this workflow do?

The workflow has two jobs connected by a dependency gate.

• lint-and-test checks out the code, installs dependencies, runs flake8 for style enforcement, and pytest for unit tests. If either fails, the pipeline stops.
• build-and-push only runs if linting and tests pass (needs: lint-and-test). It authenticates to AWS, logs in to ECR, builds your Docker image tagged with the git commit SHA, pushes it, and triggers an ECS redeployment.

• Save .github/workflows/ci.yml.

Workflow not triggering?

• Make sure the file is in the exact path .github/workflows/ci.yml with the correct capitalization.
• The workflow only triggers on pushes to the main branch. Check that your default branch is named main, not master.
• Help me debug my CI workflow

Create the retraining workflow for drift response

The retraining workflow runs daily on a schedule and can also be triggered manually when drift is detected. It re-runs your Optuna optimization, evaluates the new model against the champion, and deploys it as a challenger if it passes.

• Create the retraining workflow file by running:

touch .github/workflows/retrain.yml

• Open .github/workflows/retrain.yml in your code editor and paste the following workflow definition:

name: Retrain on Drift

on:
  workflow_dispatch:
  schedule:
    - cron: '0 2 * * *'  # Daily at 2am, also triggered manually on drift

jobs:
  retrain-and-evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ secrets.AWS_REGION }}
      - name: Run drift detection
        run: python scripts/drift_monitor.py
        continue-on-error: true
      - name: Retrain model
        run: python scripts/optimize.py
      - name: Evaluate challenger vs champion
        run: python scripts/model_evaluator.py
      - name: Deploy challenger if evaluation passes
        run: python scripts/shadow_evaluator.py
      - name: Rollback on failure
        if: failure()
        run: python scripts/rollback.py

What does the retraining workflow do?

This workflow orchestrates the full retraining lifecycle in sequence.

• workflow_dispatch allows you to trigger the workflow manually from the GitHub UI. The schedule runs it daily at 2am UTC as a safety net.
• continue-on-error: true on the drift detection step means the workflow proceeds even if no drift is detected. This lets the scheduled run still retrain on fresh data.
• The if: failure() condition on the rollback step is the deployment safeguard. If the evaluation or shadow deployment fails, the workflow automatically rolls back to the previous production model.

• Save .github/workflows/retrain.yml.

Why workflow_dispatch instead of a webhook?

Using workflow_dispatch lets your drift monitoring script trigger retraining via the GitHub API. It also provides a manual button in the Actions tab for ad-hoc retraining.

The combined schedule plus manual trigger pattern is common in production ML systems. The schedule catches slow drift that accumulates overnight. The manual trigger handles sudden distribution shifts detected in real-time.

Configure GitHub repository secrets

Both workflows reference secrets that GitHub injects at runtime. These keep your AWS credentials out of version control.

• Push your project to a GitHub repository if you have not already done so.
• Navigate to your repository on GitHub.
• Click Settings in the repository navigation bar.
• In the Security section of the sidebar, select Secrets and variables.
• Click Actions.
• Click the Secrets tab.
• Click New repository secret.
• Add each of the following secrets one at a time. For each, type the name in the Name field, paste the value in the Secret field, and click Add secret.

• AWS_ACCESS_KEY_ID - your AWS access key ID.
• AWS_SECRET_ACCESS_KEY - your AWS secret access key.
• AWS_REGION - the region your infrastructure is deployed in.
• ECR_REPOSITORY - the name of your ECR repository (not the full URI, just the repository name).

Why use repository secrets?

GitHub encrypts secrets at rest and only exposes them to workflow runs. They never appear in logs. This keeps your AWS credentials secure while allowing automated deployments.

The workflows reference them with the ${{ secrets.SECRET_NAME }} syntax, which GitHub injects at runtime.

• Verify your secrets are configured by checking that all four appear listed on the Actions secrets page.

Workflow fails with credentials error?

• Double-check that the secret names match exactly: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, ECR_REPOSITORY. Any typo will cause authentication to fail.
• Confirm that the IAM user whose credentials you stored has the necessary permissions (ECR push, ECS update, S3 access).
• Help me debug the credentials error

✔️ Awesome, I've got everything!

Great. Double check you have saved both workflow files and configured all four secrets on GitHub.

ⓧ I'd like to double check the full code

Here is the complete .github/workflows/ci.yml:

name: CI - Build and Deploy

on:
  push:
    branches: [main]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - run: pip install flake8 pytest
      - run: flake8 app/ src/ scripts/
      - run: pytest tests/ -v

  build-and-push:
    needs: lint-and-test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ secrets.AWS_REGION }}
      - name: Login to ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v2
      - name: Build and push Docker image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          ECR_REPOSITORY: ${{ secrets.ECR_REPOSITORY }}
          IMAGE_TAG: ${{ github.sha }}
        run: |
          docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
      - name: Update ECS service
        run: |
          aws ecs update-service --cluster fraud-detection-cluster \
            --service fraud-detection-service \
            --force-new-deployment

Here is the complete .github/workflows/retrain.yml:

name: Retrain on Drift

on:
  workflow_dispatch:
  schedule:
    - cron: '0 2 * * *'  # Daily at 2am, also triggered manually on drift

jobs:
  retrain-and-evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ secrets.AWS_REGION }}
      - name: Run drift detection
        run: python scripts/drift_monitor.py
        continue-on-error: true
      - name: Retrain model
        run: python scripts/optimize.py
      - name: Evaluate challenger vs champion
        run: python scripts/model_evaluator.py
      - name: Deploy challenger if evaluation passes
        run: python scripts/shadow_evaluator.py
      - name: Rollback on failure
        if: failure()
        run: python scripts/rollback.py

Your ML system now has a complete CI/CD pipeline. Code changes trigger automated builds. Drift triggers automated retraining. Failures trigger automated rollback. Next, you will validate that the system meets its latency SLOs under realistic load.

Load Test & Validate Latency SLOs

Your fraud detection system is deployed, monitored, and self-healing. But how do you know it can handle real production traffic without violating response time guarantees?

In this step, you will write a Locust load test that simulates bursty fraud-check requests, define latency SLOs, and validate that your ECS Fargate service meets them under pressure. If it does not, you will tune the task configuration and re-run until it does.

In this step, get ready to:

• Write a Locust load test script simulating realistic bursty traffic.
• Run the load test against your deployed ALB endpoint and validate latency SLOs.
• Tune Fargate task CPU/memory and redeploy if SLOs are not met.

Write the Locust load test script

A load test needs to simulate realistic user behavior. Fraud detection APIs experience bursty traffic patterns where many requests arrive in rapid succession.

Your test will send a mix of single predictions, batch predictions, and health checks to reflect real-world usage.

• Create the load test file by running:

touch tests/locustfile.py

• Open tests/locustfile.py and paste the following code:

"""Locust load test for fraud detection inference service."""
import random
import json
from locust import HttpUser, task, between

SAMPLE_FEATURES = {
    'Time': 3600.0,
    'V1': -1.359807, 'V2': -0.072781, 'V3': 2.536347,
    'V4': 1.378155, 'V5': -0.338321, 'V6': 0.462388,
    'V7': 0.239599, 'V8': 0.098698, 'V9': 0.363787,
    'V10': 0.090794, 'V11': -0.551600, 'V12': -0.617801,
    'V13': -0.991390, 'V14': -0.311169, 'V15': 1.468177,
    'V16': -0.470401, 'V17': 0.207971, 'V18': 0.025791,
    'V19': 0.403993, 'V20': 0.251412, 'V21': -0.018307,
    'V22': 0.277838, 'V23': -0.110474, 'V24': 0.066928,
    'V25': 0.128539, 'V26': -0.189115, 'V27': 0.133558,
    'V28': -0.021053, 'Amount': 149.62
}

What does this section do?

This imports Locust's core classes and defines a representative transaction with all 30 features the model expects.

Each load test request will add small random noise to these values. This simulates unique transactions hitting the API rather than the same cached input every time.

• Below the SAMPLE_FEATURES dictionary, add the FraudCheckUser class:

class FraudCheckUser(HttpUser):
    wait_time = between(0.001, 0.01)  # Bursty traffic

    @task(10)
    def predict_single(self):
        features = {k: v + random.gauss(0, 0.1) if isinstance(v, float) else v
                    for k, v in SAMPLE_FEATURES.items()}
        self.client.post('/predict', json={'features': features},
                         headers={'Content-Type': 'application/json'})

    @task(2)
    def predict_batch(self):
        records = [{k: v + random.gauss(0, 0.1) if isinstance(v, float) else v
                    for k, v in SAMPLE_FEATURES.items()} for _ in range(10)]
        self.client.post('/predict/batch', json={'records': records},
                         headers={'Content-Type': 'application/json'})

    @task(1)
    def health_check(self):
        self.client.get('/health')

What does this code do?

• wait_time = between(0.001, 0.01) creates bursty traffic with almost no pause between requests. This simulates peak fraud-check load.
• @task(10) on predict_single means single predictions are 10x more likely to be called than health checks. This matches real-world traffic where single-transaction checks dominate.
• @task(2) on predict_batch sends 10 records per request, simulating bulk screening.
• random.gauss(0, 0.1) adds small Gaussian noise to each feature so every request is unique.

• Save tests/locustfile.py.
• Confirm the file exists by running:

ls tests/locustfile.py

You should see tests/locustfile.py in the output.

File not found?

Make sure you are running the command from your project root directory. The tests/ folder should already exist from earlier steps. If it does not, create it with mkdir tests first.

Help me debug my directory structure

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

"""Locust load test for fraud detection inference service."""
import random
import json
from locust import HttpUser, task, between

SAMPLE_FEATURES = {
    'Time': 3600.0,
    'V1': -1.359807, 'V2': -0.072781, 'V3': 2.536347,
    'V4': 1.378155, 'V5': -0.338321, 'V6': 0.462388,
    'V7': 0.239599, 'V8': 0.098698, 'V9': 0.363787,
    'V10': 0.090794, 'V11': -0.551600, 'V12': -0.617801,
    'V13': -0.991390, 'V14': -0.311169, 'V15': 1.468177,
    'V16': -0.470401, 'V17': 0.207971, 'V18': 0.025791,
    'V19': 0.403993, 'V20': 0.251412, 'V21': -0.018307,
    'V22': 0.277838, 'V23': -0.110474, 'V24': 0.066928,
    'V25': 0.128539, 'V26': -0.189115, 'V27': 0.133558,
    'V28': -0.021053, 'Amount': 149.62
}


class FraudCheckUser(HttpUser):
    wait_time = between(0.001, 0.01)  # Bursty traffic

    @task(10)
    def predict_single(self):
        features = {k: v + random.gauss(0, 0.1) if isinstance(v, float) else v
                    for k, v in SAMPLE_FEATURES.items()}
        self.client.post('/predict', json={'features': features},
                         headers={'Content-Type': 'application/json'})

    @task(2)
    def predict_batch(self):
        records = [{k: v + random.gauss(0, 0.1) if isinstance(v, float) else v
                    for k, v in SAMPLE_FEATURES.items()} for _ in range(10)]
        self.client.post('/predict/batch', json={'records': records},
                         headers={'Content-Type': 'application/json'})

    @task(1)
    def health_check(self):
        self.client.get('/health')

Define and validate latency SLOs

A Service Level Objective is a measurable target your system commits to. For real-time fraud detection, latency matters because every millisecond of delay holds up a payment.

Your SLOs for this service are:

• p50 < 20ms (half of all requests complete in under 20 milliseconds).
• p95 < 50ms (95% of requests complete in under 50 milliseconds).
• p99 < 100ms (99% of requests complete in under 100 milliseconds).

Why these specific SLO targets?

Payment processors expect fraud checks to return within 100ms at worst. Exceeding that threshold causes user-visible checkout latency. The p50 target ensures the typical experience is snappy, while p99 catches tail latency that would affect 1 in 100 users.

You will validate these under 1000 requests per second. That simulates a medium-traffic payment service processing about 86 million transactions per day.

• Run the load test in headless mode against your deployed ALB endpoint:

locust -f tests/locustfile.py --headless -u 100 -r 10 -t 5m -H http://[[ALB_DNS="your-alb-dns-name.region.elb.amazonaws.com"]] --html load_test_report.html

What does this command do?

• -f tests/locustfile.py specifies your load test file.
• --headless runs the test without the web UI, printing results to the terminal.
• -u 100 simulates 100 concurrent users.
• -r 10 spawns 10 new users per second until reaching 100.
• -t 5m runs the test for 5 minutes.
• -H sets the target host (your ALB endpoint).
• --html load_test_report.html saves a full HTML report you can review in your browser.

After the test completes (about 5 minutes), Locust prints a summary table to your terminal showing request counts, failure rates, and response time percentiles.

• Look at the Aggregated row in the output. Check the Avg, Min, Max, and percentile columns against your SLOs.
• Open load_test_report.html in your browser for detailed charts showing response time distribution over the test duration.

Seeing connection errors?

If most requests fail with connection errors, verify your ALB is still accepting traffic. Run curl against your ALB endpoint's /health path first. If the health check succeeds but load tests fail, your ECS tasks may be scaling down to zero. Check the ECS service desired count.

Help me debug network connectivity to my ECS Fargate service

Tune Fargate configuration if SLOs are not met

If your response times exceed the SLO targets, the most common fix is increasing CPU and memory allocated to each Fargate task. More CPU means the model inference completes faster. More memory prevents garbage collection pressure from spiking latency.

✔️ My SLOs are met

Excellent. Your current Fargate configuration handles the load within acceptable latency bounds. No changes needed. Skip ahead to the launchpad at the end of this step.

ⓧ My p95 or p99 exceeds the targets

• Open terraform/main.tf and find your Fargate task definition resource.
• Increase the CPU from 256 to 512 (or from 512 to 1024).
• Increase the memory from 512 to 1024 (or from 1024 to 2048).
• Save terraform/main.tf.

• Apply the Terraform change by running:

cd terraform && terraform apply -auto-approve

Wait for the new task definition to register and ECS to replace the running task with the updated configuration.

• Re-run the load test with the same command:

cd .. && locust -f tests/locustfile.py --headless -u 100 -r 10 -t 5m -H http://[[ALB_DNS="your-alb-dns-name.region.elb.amazonaws.com"]] --html load_test_report.html

Compare the new results against your SLOs. If p95 and p99 now fall within targets, you are done. If not, increase CPU/memory one more step and repeat.

Diminishing returns on scaling up

If doubling CPU does not help, the bottleneck is likely network latency between the ALB and the task, or model loading time. Consider adding a model-warming step in your container startup so the first few requests do not pay a cold-start penalty.

ⓧ Seeing high error rates instead

A high failure percentage (above 5%) usually means your ECS service cannot handle the connection volume. Check that your ECS service has multiple tasks running (desired count of at least 2) and that your ALB health check is not killing tasks under load.

• Increase the ECS service desired count to 2 or 3 in your Terraform configuration.
• Re-apply and re-test.

Help me diagnose my ECS service error rates

Why not jump straight to maximum resources?

In production, cost optimization matters. ECS Fargate charges per vCPU-second and per GB-second. Running 1024 CPU units costs twice as much as 512. The goal is to find the minimum configuration that meets your SLOs, not the maximum that can be provisioned.

With your SLOs validated under load, you have proven that your fraud detection system can serve real-time predictions at production scale. The combination of monitoring, automated rollback, and validated latency targets gives you the confidence to serve live traffic.

Secret mission

Build a Drift Simulation Harness

Your production system can detect drift and self-heal. But can you prove it works end-to-end? Build a drift simulator that injects realistic concept drift into your production data stream, then sit back and watch the full autonomous lifecycle respond: detect, retrain, evaluate, promote or rollback.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project creates AWS resources that incur charges while running.

Cost warning

Your ECS Fargate tasks cost approximately $0.04/hour per task, and the ALB costs approximately $0.02/hour. Left running 24/7, this adds up to $1-2/day.

If you are not actively experimenting, choose Pause or Delete below to avoid unnecessary charges.

Resources you used:

• AWS ECS Fargate cluster and service (fraud-detection-cluster, fraud-detection-service).
• AWS ECR repository (Docker images).
• AWS S3 bucket (MLflow artifacts).
• AWS Application Load Balancer.
• AWS IAM roles (task execution role, task role).
• Local Docker Compose stack (PostgreSQL, MinIO, MLflow, Prometheus, Grafana).
• Local project directory on your Desktop.
• GitHub repository with CI/CD workflows.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the system.

• Your AWS infrastructure continues running and incurring charges (~$1-2/day).
• Your local Docker Compose stack stays up (no cost, but uses memory).
• The GitHub Actions workflows continue to run on schedule.

Pause - I'll come back to this later

Shut down running processes to free up memory and reduce AWS costs, while keeping your project files intact.

• Scale down your ECS service to stop Fargate charges by running:
• Stop the local Docker Compose stack by running:

Scale the ECS service to zero tasks:

aws ecs update-service --cluster fraud-detection-cluster \
  --service fraud-detection-service \
  --desired-count 0

What does this do?

Setting the desired count to 0 stops all running Fargate tasks, which eliminates the per-task compute charges. The ALB remains active (~$0.02/hour) so your endpoint stays configured. When you return, set the desired count back to 1 to resume.

Stop the local Docker Compose stack (preserves volumes):

docker compose down

What does this preserve?

Running docker compose down without the -v flag stops and removes containers but keeps your named volumes (postgres_data, minio_data, grafana_data). Your MLflow experiments, model artifacts, and Grafana dashboards are all preserved.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

• Destroy all AWS infrastructure by navigating to your Terraform directory and running:

cd terraform
terraform destroy

What does this remove?

This removes your ECS cluster, ECS service, ECR repository, S3 bucket, ALB, IAM roles, and all associated networking resources. Terraform will show you a plan of what it will destroy and ask for confirmation before proceeding.

• Stop and remove all local Docker containers and volumes by running:

docker compose down -v

What does the -v flag do?

The -v flag removes named volumes declared in your Compose file. This deletes your local PostgreSQL database, MinIO storage, and Grafana data permanently.

• Delete the project directory from your Desktop by running:

macOS

cd ~/Desktop
rm -rf fraud-detection

Windows

cd ~\Desktop
rmdir /s /q fraud-detection

• Delete your GitHub repository if you no longer need it. Go to your repository on GitHub, click Settings, scroll to the bottom of the General page, and click Delete this repository in the Danger Zone section.

Nice Work!

Nice Work!

You just built a production-grade, self-healing ML system that most ML engineers never get to build at their day jobs. This is the kind of end-to-end system that top ML platform teams at companies like Stripe, Uber, and Google operate in production.

You've learned how to:

• Train and optimize a fraud detection model with Optuna hyperparameter tuning, full experiment tracking in MLflow, and a feature store pattern that prevents training-serving skew.
• Deploy a FastAPI inference service on AWS ECS Fargate via Terraform with champion/challenger shadow deployment, automated promotion, drift detection with Evidently, and self-healing rollback.
• Build a complete CI/CD pipeline with GitHub Actions for the full ML lifecycle, validated under load with Locust to meet latency SLOs (p95 < 50ms at 1000 RPS).
• Secret Mission: Demonstrated the full autonomous ML lifecycle with a drift simulation harness that triggers detect, retrain, evaluate, and promote/rollback end-to-end.

Ready to quiz yourself?