MLOps & Model Deployment

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Set experiment name
mlflow.set_experiment("fraud-detection-model")

# Start run tracking
with mlflow.start_run():
    # Define hyperparameters
    n_estimators = 100
    max_depth = 5
    
    # Log parameters
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_param("max_depth", max_depth)
    
    # Train model
    # (Assuming X_train, y_train, X_test, y_test are loaded)
    clf = RandomForestClassifier(n_estimators=n_estimators, max_depth=max_depth)
    clf.fit(X_train, y_train)
    
    # Log metrics
    y_pred = clf.predict(X_test)
    acc = accuracy_score(y_test, y_pred)
    mlflow.log_metric("accuracy", acc)
    
    # Log and register the model in the registry
    mlflow.sklearn.log_model(
        sk_model=clf,
        artifact_path="model",
        registered_model_name="RandomForestFraudModel"
    )

# app.py (Serving API)
from fastapi import FastAPI
import joblib
import numpy as np

app = FastAPI()
# Load model from registry / disk
model = joblib.load("model.joblib")

@app.post("/predict")
def predict(features: list[float]):
    # Expects features as list of float values
    prediction = model.predict([np.array(features)])
    return {"prediction": int(prediction[0])}

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy model artifact and serving script
COPY model.joblib .
COPY app.py .

# Expose API port
EXPOSE 8000

# Run FastAPI app with Uvicorn
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

name: ML CI/CD Pipeline

on: [push, pull_request]

jobs:
  test_model:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout repository
        uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.10"

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

      - name: Validate dataset schemas
        run: |
          pytest tests/test_data_validation.py

      - name: Evaluate model performance threshold
        run: |
          # Fails the build if accuracy falls below 85%
          python scripts/evaluate_regression.py --threshold 0.85

# Model Card: Fraud Detection Random Forest (v1.2.0)

## Overview
- **Developer**: Risk Platform Team
- **Model Type**: Random Forest Classifier
- **Release Date**: June 2026

## Intended Use
- **Primary Use**: Predicting credit card transaction fraud in real-time.
- **Out of Scope**: Credit score evaluations or long-term lending risks.

## Training Data & Methodology
- Trained on 1.2M anonymized transactions (Q1 2026 data).
- Features include: transaction amount, category, user location frequency, distance from home.

## Metrics & Performance
- **Accuracy**: 94.2%
- **Precision (Fraud class)**: 91.5% (minimizes false positive blocks)
- **Recall (Fraud class)**: 88.0% (captures true fraud events)

## Ethical Considerations & Limitations
- **Limitations**: The model lacks context on global traveling events, leading to higher false-positives during holidays.
- **Biases**: Tends to flag transactions from IP addresses in newly registered server pools; regular checks are run to ensure geographic fairness.

import math
from typing import List, Tuple

def calculate_psi(expected: List[float], actual: List[float], num_buckets: int = 10) -> Tuple[float, str]:
    """
    Calculate the Population Stability Index (PSI) between two 1D datasets.
    
    Args:
        expected: Baseline/training feature values.
        actual: Production/serving feature values.
        num_buckets: Number of quantiles to split the data.
        
    Returns:
        A tuple of (psi_value, drift_level).
    """
    # 1. Sort the baseline data to determine bucket boundaries (quantiles)
    sorted_expected = sorted(expected)
    n_expected = len(expected)
    n_actual = len(actual)
    
    if n_expected == 0 or n_actual == 0:
        raise ValueError("Inputs cannot be empty.")
        
    # Get split boundaries based on equal-frequency quantiles from expected distribution
    boundaries = []
    for i in range(1, num_buckets):
        idx = int(i * n_expected / num_buckets)
        boundaries.append(sorted_expected[idx])
        
    # Helper to count frequencies in each bucket
    def get_bucket_counts(data: List[float]) -> List[int]:
        counts = [0] * num_buckets
        for val in data:
            placed = False
            for bucket_idx, boundary in enumerate(boundaries):
                if val <= boundary:
                    counts[bucket_idx] += 1
                    placed = True
                    break
            if not placed:
                counts[-1] += 1
        return counts

    expected_counts = get_bucket_counts(expected)
    actual_counts = get_bucket_counts(actual)
    
    # 2. Calculate PSI
    psi_val = 0.0
    for e_count, a_count in zip(expected_counts, actual_counts):
        # Convert to percentages (probabilities)
        e_pct = e_count / n_expected
        a_pct = a_count / n_actual
        
        # Handle zero counts using a tiny epsilon to prevent division-by-zero/log-zero errors
        if e_pct == 0:
            e_pct = 0.0001
        if a_pct == 0:
            a_pct = 0.0001
            
        psi_val += (a_pct - e_pct) * math.log(a_pct / e_pct)
        
    # 3. Classify drift level
    # PSI < 0.1: No change / Stable
    # 0.1 <= PSI < 0.25: Slight change / Warning
    # PSI >= 0.25: Significant change / Action required
    if psi_val < 0.1:
        status = "Healthy (No significant distribution shift)"
    elif psi_val < 0.25:
        status = "Warning (Moderate distribution shift detected)"
    else:
        status = "Alert (Significant distribution shift detected - Action Required)"
        
    return psi_val, status

# Example usage with mock data
if __name__ == "__main__":
    # expected: centered around 0
    # actual: shifted slightly to the right (simulating drift)
    expected_data = [-2.0, -1.5, -1.0, -0.5, 0.0, 0.5, 1.0, 1.5, 2.0] * 10
    actual_data = [-1.0, -0.5, 0.0, 0.5, 1.0, 1.5, 2.0, 2.5, 3.0] * 10

    psi, level = calculate_psi(expected_data, actual_data, num_buckets=5)
    print(f"Calculated PSI: {psi:.4f}")
    print(f"Drift Level: {level}")