Model Deployment

ML Model Deployment

Deploy trained models to production with proper serving and monitoring.

Deployment Options

FastAPI Model Server

from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()
model = joblib.load('model.pkl')

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: float
    probability: float

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

@app.post('/predict', response_model=PredictionResponse)
def predict(request: PredictionRequest):
    features = np.array(request.features).reshape(1, -1)
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features)[0].max()
    return PredictionResponse(prediction=prediction, probability=probability)

Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY model.pkl .
COPY app.py .

EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Model Monitoring

class ModelMonitor:
    def __init__(self):
        self.predictions = []
        self.latencies = []

    def log_prediction(self, input_data, prediction, latency):
        self.predictions.append({
            'input': input_data,
            'prediction': prediction,
            'latency': latency,
            'timestamp': datetime.now()
        })

    def detect_drift(self, reference_distribution):
        # Compare current predictions to reference
        pass

Deployment Checklist

• [ ] Model validated on test set
• [ ] API endpoints documented
• [ ] Health check endpoint
• [ ] Authentication configured
• [ ] Logging and monitoring setup
• [ ] Model versioning in place
• [ ] Rollback procedure documented

Quick Start: Deploy Model in 6 Steps

# 1. Save trained model
import joblib
joblib.dump(model, 'model.pkl')

# 2. Create FastAPI app (see references/fastapi-production-server.md)
# app.py with /predict and /health endpoints

# 3. Create Dockerfile
cat > Dockerfile << 'EOF'
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py model.pkl ./
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
EOF

# 4. Build and test locally
docker build -t model-api:v1.0.0 .
docker run -p 8000:8000 model-api:v1.0.0

# 5. Push to registry
docker tag model-api:v1.0.0 registry.example.com/model-api:v1.0.0
docker push registry.example.com/model-api:v1.0.0

# 6. Deploy to Kubernetes
kubectl apply -f deployment.yaml
kubectl rollout status deployment/model-api

Known Issues Prevention

1. No Health Checks = Downtime

Problem: Load balancer sends traffic to unhealthy pods, causing 503 errors.

Solution: Implement both liveness and readiness probes:

# app.py
@app.get("/health")  # Liveness: Is service alive?
async def health():
    return {"status": "healthy"}

@app.get("/ready")  # Readiness: Can handle traffic?
async def ready():
    try:
        _ = model_store.model  # Verify model loaded
        return {"status": "ready"}
    except:
        raise HTTPException(503, "Not ready")

# deployment.yaml
livenessProbe:
  httpGet:
    path: /health
    port: 8000
  initialDelaySeconds: 30
readinessProbe:
  httpGet:
    path: /ready
    port: 8000
  initialDelaySeconds: 5

2. Model Not Found Errors in Container

Problem: FileNotFoundError: model.pkl when container starts.

Solution: Verify model file is copied in Dockerfile and path matches:

# ❌ Wrong: Model in wrong directory
COPY model.pkl /app/models/  # But code expects /app/model.pkl

# ✅ Correct: Consistent paths
COPY model.pkl /models/model.pkl
ENV MODEL_PATH=/models/model.pkl

# In Python:
model_path = os.getenv("MODEL_PATH", "/models/model.pkl")

3. Unhandled Input Validation = 500 Errors

Problem: Invalid inputs crash API with unhandled exceptions.

Solution: Use Pydantic for automatic validation:

from pydantic import BaseModel, Field, validator

class PredictionRequest(BaseModel):
    features: List[float] = Field(..., min_items=1, max_items=100)

    @validator('features')
    def validate_finite(cls, v):
        if not all(np.isfinite(val) for val in v):
            raise ValueError("All features must be finite")
        return v

# FastAPI auto-validates and returns 422 for invalid requests
@app.post("/predict")
async def predict(request: PredictionRequest):
    # Request is guaranteed valid here
    pass

4. No Drift Monitoring = Silent Degradation

Problem: Model performance degrades over time, no one notices until users complain.

Solution: Implement drift detection (see references/model-monitoring-drift.md):

monitor = ModelMonitor(reference_data=training_data, drift_threshold=0.1)

@app.post("/predict")
async def predict(request: PredictionRequest):
    prediction = model.predict(features)
    monitor.log_prediction(features, prediction, latency)

    # Alert if drift detected
    if monitor.should_retrain():
        alert_manager.send_alert("Model drift detected - retrain recommended")

    return prediction

5. Missing Resource Limits = OOM Kills

Problem: Pod killed by Kubernetes OOMKiller, service goes down.

Solution: Set memory/CPU limits and requests:

resources:
  requests:
    memory: "512Mi"  # Guaranteed
    cpu: "500m"
  limits:
    memory: "1Gi"    # Max allowed
    cpu: "1000m"

# Monitor actual usage:
kubectl top pods

6. No Rollback Plan = Stuck on Bad Deploy

Problem: New model version has bugs, no way to revert quickly.

Solution: Tag images with versions, keep previous deployment:

# Deploy with version tag
kubectl set image deployment/model-api model-api=registry/model-api:v1.2.0

# If issues, rollback to previous
kubectl rollout undo deployment/model-api

# Or specify version
kubectl set image deployment/model-api model-api=registry/model-api:v1.1.0

7. Synchronous Prediction = Slow Batch Processing

Problem: Processing 10,000 predictions one-by-one takes hours.

Solution: Implement batch endpoint:

@app.post("/predict/batch")
async def predict_batch(request: BatchPredictionRequest):
    # Process all at once (vectorized)
    features = np.array(request.instances)
    predictions = model.predict(features)  # Much faster!
    return {"predictions": predictions.tolist()}

8. No CI/CD Validation = Deploy Bad Models

Problem: Deploying model that fails basic tests, breaking production.

Solution: Validate in CI pipeline (see references/cicd-ml-models.md):

# .github/workflows/deploy.yml
- name: Validate model performance
  run: |
    python scripts/validate_model.py \
      --model model.pkl \
      --test-data test.csv \
      --min-accuracy 0.85  # Fail if below threshold

Best Practices

• Version everything: Models (semantic versioning), Docker images, deployments
• Monitor continuously: Latency, error rate, drift, resource usage
• Test before deploy: Unit tests, integration tests, performance benchmarks
• Deploy gradually: Canary (10%), then full rollout
• Plan for rollback: Keep previous version, document procedure
• Log predictions: Enable debugging and drift detection
• Set resource limits: Prevent OOM kills and resource contention
• Use health checks: Enable proper load balancing

When to Load References

Load reference files for detailed implementations:

• FastAPI Production Server: Load references/fastapi-production-server.md for complete production-ready FastAPI implementation with error handling, validation (Pydantic models), logging, health/readiness probes, batch predictions, model versioning, middleware, exception handlers, and performance optimizations (caching, async)

• Model Monitoring & Drift: Load references/model-monitoring-drift.md for ModelMonitor implementation with KS-test drift detection, Jensen-Shannon divergence, Prometheus metrics integration, alert configuration (Slack, email), continuous monitoring service, and dashboard endpoints

• Containerization & Deployment: Load references/containerization-deployment.md for multi-stage Dockerfiles, model versioning in containers, Docker Compose setup, A/B testing with Nginx, Kubernetes deployments (rolling update, blue-green, canary), GitHub Actions CI/CD, and deployment checklists

• CI/CD for ML Models: Load references/cicd-ml-models.md for complete GitHub Actions pipeline with model validation, data validation, automated testing, security scanning, performance benchmarks, automated rollback, and deployment strategies

CI/CD for ML Models

Complete CI/CD pipeline for machine learning models including automated testing, validation, and deployment strategies.

Overview

ML CI/CD differs from traditional software:

• Model validation beyond unit tests
• Data validation for training/inference
• Performance benchmarks (accuracy, latency)
• Drift detection in production
• Model versioning and artifact management

Complete GitHub Actions Workflow

# .github/workflows/ml-pipeline.yml
name: ML Model CI/CD Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  workflow_dispatch:  # Manual trigger

env:
  PYTHON_VERSION: '3.11'
  MODEL_REGISTRY: 's3://my-model-bucket'

jobs:
  # Job 1: Code Quality & Unit Tests
  code-quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: 'pip'

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

      - name: Lint with flake8
        run: |
          flake8 src/ --count --select=E9,F63,F7,F82 --show-source --statistics
          flake8 src/ --count --max-complexity=10 --max-line-length=127 --statistics

      - name: Type check with mypy
        run: mypy src/

      - name: Run unit tests
        run: |
          pytest tests/unit/ \
            --cov=src \
            --cov-report=xml \
            --cov-report=html \
            --junit-xml=pytest.xml

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

  # Job 2: Data Validation
  data-validation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Validate training data
        run: |
          python scripts/validate_data.py \
            --data-path data/train.csv \
            --schema schema.json

      - name: Check data quality
        run: |
          python scripts/data_quality_checks.py

  # Job 3: Model Training & Validation
  train-model:
    needs: [code-quality, data-validation]
    runs-on: ubuntu-latest
    outputs:
      model-version: ${{ steps.version.outputs.version }}
      metrics: ${{ steps.train.outputs.metrics }}

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}

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

      - name: Generate model version
        id: version
        run: |
          VERSION=$(date +%Y%m%d)-${GITHUB_SHA::7}
          echo "version=$VERSION" >> $GITHUB_OUTPUT
          echo "Model version: $VERSION"

      - name: Train model
        id: train
        run: |
          python src/train.py \
            --output models/model_${{ steps.version.outputs.version }}.pkl \
            --config config/train_config.yaml

          # Extract metrics
          METRICS=$(cat metrics.json)
          echo "metrics=$METRICS" >> $GITHUB_OUTPUT

      - name: Validate model performance
        run: |
          python scripts/validate_model.py \
            --model models/model_${{ steps.version.outputs.version }}.pkl \
            --test-data data/test.csv \
            --min-accuracy 0.85

      - name: Upload model artifact
        uses: actions/upload-artifact@v3
        with:
          name: model-${{ steps.version.outputs.version }}
          path: models/model_${{ steps.version.outputs.version }}.pkl

  # Job 4: Integration Tests
  integration-tests:
    needs: train-model
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:alpine
        ports:
          - 6379:6379

    steps:
      - uses: actions/checkout@v3

      - name: Download model artifact
        uses: actions/download-artifact@v3
        with:
          name: model-${{ needs.train-model.outputs.model-version }}
          path: models/

      - name: Start API server
        run: |
          pip install -r requirements.txt
          uvicorn app:app --host 0.0.0.0 --port 8000 &
          sleep 10  # Wait for server to start

      - name: Run integration tests
        run: |
          pytest tests/integration/ \
            --base-url http://localhost:8000

      - name: Performance benchmarks
        run: |
          python scripts/benchmark_api.py \
            --url http://localhost:8000/predict \
            --requests 1000 \
            --max-latency 200  # ms

  # Job 5: Security Scanning
  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          format: 'sarif'
          output: 'trivy-results.sarif'

      - name: Upload Trivy results to GitHub Security
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: 'trivy-results.sarif'

      - name: Check Python dependencies
        run: |
          pip install safety
          safety check --json

  # Job 6: Build Docker Image
  build-image:
    needs: [train-model, integration-tests, security]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    outputs:
      image-tag: ${{ steps.meta.outputs.tags }}

    steps:
      - uses: actions/checkout@v3

      - name: Download model artifact
        uses: actions/download-artifact@v3
        with:
          name: model-${{ needs.train-model.outputs.model-version }}
          path: models/

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2

      - name: Log in to Registry
        uses: docker/login-action@v2
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Extract metadata
        id: meta
        uses: docker/metadata-action@v4
        with:
          images: ghcr.io/${{ github.repository }}
          tags: |
            type=sha
            type=semver,pattern={{version}}
            latest

      - name: Build and push
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          build-args: |
            MODEL_VERSION=${{ needs.train-model.outputs.model-version }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

  # Job 7: Deploy to Staging
  deploy-staging:
    needs: build-image
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment:
      name: staging
      url: https://staging.api.example.com

    steps:
      - uses: actions/checkout@v3

      - name: Configure kubectl
        uses: azure/k8s-set-context@v3
        with:
          kubeconfig: ${{ secrets.KUBECONFIG_STAGING }}

      - name: Deploy to staging
        run: |
          kubectl set image deployment/model-api \
            model-api=${{ needs.build-image.outputs.image-tag }} \
            --namespace=staging

          kubectl rollout status deployment/model-api --namespace=staging

      - name: Run smoke tests
        run: |
          python scripts/smoke_tests.py \
            --url https://staging.api.example.com

  # Job 8: Deploy to Production
  deploy-production:
    needs: deploy-staging
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment:
      name: production
      url: https://api.example.com

    steps:
      - uses: actions/checkout@v3

      - name: Configure kubectl
        uses: azure/k8s-set-context@v3
        with:
          kubeconfig: ${{ secrets.KUBECONFIG_PROD }}

      - name: Deploy to production (canary)
        run: |
          # Deploy to 10% of traffic first
          kubectl apply -f k8s/canary-deployment.yaml

          # Wait and monitor
          sleep 300

          # Check canary metrics
          python scripts/check_canary_metrics.py

      - name: Full rollout
        run: |
          kubectl set image deployment/model-api \
            model-api=${{ needs.build-image.outputs.image-tag }} \
            --namespace=production

          kubectl rollout status deployment/model-api --namespace=production

Model Validation Script

# scripts/validate_model.py
import joblib
import pandas as pd
import numpy as np
from sklearn.metrics import accuracy_score, precision_score, recall_score
import argparse
import sys

def validate_model(model_path, test_data_path, min_accuracy=0.85):
    """Validate model meets performance thresholds."""

    # Load model
    model = joblib.load(model_path)

    # Load test data
    test_df = pd.read_csv(test_data_path)
    X_test = test_df.drop('target', axis=1)
    y_test = test_df['target']

    # Make predictions
    y_pred = model.predict(X_test)

    # Compute metrics
    accuracy = accuracy_score(y_test, y_pred)
    precision = precision_score(y_test, y_pred, average='weighted')
    recall = recall_score(y_test, y_pred, average='weighted')

    print(f"Model Performance:")
    print(f"  Accuracy:  {accuracy:.4f}")
    print(f"  Precision: {precision:.4f}")
    print(f"  Recall:    {recall:.4f}")

    # Validate thresholds
    if accuracy < min_accuracy:
        print(f"❌ Model accuracy {accuracy:.4f} below threshold {min_accuracy}")
        sys.exit(1)

    print("✅ Model validation passed")
    return True

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--model', required=True)
    parser.add_argument('--test-data', required=True)
    parser.add_argument('--min-accuracy', type=float, default=0.85)
    args = parser.parse_args()

    validate_model(args.model, args.test_data, args.min_accuracy)

Data Validation Script

# scripts/validate_data.py
import pandas as pd
import json
import argparse
import sys

def validate_data(data_path, schema_path):
    """Validate data against schema."""

    # Load data and schema
    df = pd.read_csv(data_path)
    with open(schema_path) as f:
        schema = json.load(f)

    errors = []

    # Check columns
    expected_cols = set(schema['columns'])
    actual_cols = set(df.columns)

    if expected_cols != actual_cols:
        missing = expected_cols - actual_cols
        extra = actual_cols - expected_cols
        if missing:
            errors.append(f"Missing columns: {missing}")
        if extra:
            errors.append(f"Extra columns: {extra}")

    # Check data types
    for col, dtype in schema.get('dtypes', {}).items():
        if col in df.columns and df[col].dtype != dtype:
            errors.append(f"Column {col} has dtype {df[col].dtype}, expected {dtype}")

    # Check null values
    max_nulls = schema.get('max_null_percentage', 0.05)
    for col in df.columns:
        null_pct = df[col].isnull().sum() / len(df)
        if null_pct > max_nulls:
            errors.append(f"Column {col} has {null_pct:.1%} nulls (max: {max_nulls:.1%})")

    # Report results
    if errors:
        print("❌ Data validation failed:")
        for error in errors:
            print(f"  - {error}")
        sys.exit(1)

    print("✅ Data validation passed")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--data-path', required=True)
    parser.add_argument('--schema', required=True)
    args = parser.parse_args()

    validate_data(args.data_path, args.schema)

Performance Benchmark Script

# scripts/benchmark_api.py
import requests
import time
import numpy as np
import argparse

def benchmark_api(url, n_requests=1000, max_latency_ms=200):
    """Benchmark API performance."""

    latencies = []

    for i in range(n_requests):
        start = time.time()

        response = requests.post(
            url,
            json={"features": [1.2, 3.4, 5.6, 7.8]}
        )

        latency = (time.time() - start) * 1000
        latencies.append(latency)

        if response.status_code != 200:
            print(f"❌ Request {i} failed: {response.status_code}")
            return False

    # Compute statistics
    p50 = np.percentile(latencies, 50)
    p95 = np.percentile(latencies, 95)
    p99 = np.percentile(latencies, 99)

    print(f"Benchmark Results ({n_requests} requests):")
    print(f"  P50: {p50:.2f}ms")
    print(f"  P95: {p95:.2f}ms")
    print(f"  P99: {p99:.2f}ms")

    if p95 > max_latency_ms:
        print(f"❌ P95 latency {p95:.2f}ms exceeds {max_latency_ms}ms")
        return False

    print("✅ Performance benchmarks passed")
    return True

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--url', required=True)
    parser.add_argument('--requests', type=int, default=1000)
    parser.add_argument('--max-latency', type=float, default=200)
    args = parser.parse_args()

    benchmark_api(args.url, args.requests, args.max_latency)

Automated Rollback on Failure

# In CI/CD pipeline
- name: Deploy with auto-rollback
  run: |
    # Record previous deployment
    PREVIOUS_IMAGE=$(kubectl get deployment model-api -o jsonpath='{.spec.template.spec.containers[0].image}')

    # Deploy new version
    kubectl set image deployment/model-api model-api=${{ needs.build-image.outputs.image-tag }}

    # Wait for rollout
    if ! kubectl rollout status deployment/model-api --timeout=5m; then
      echo "Deployment failed, rolling back"
      kubectl set image deployment/model-api model-api=$PREVIOUS_IMAGE
      exit 1
    fi

    # Run health checks
    sleep 30
    if ! python scripts/health_check.py; then
      echo "Health checks failed, rolling back"
      kubectl set image deployment/model-api model-api=$PREVIOUS_IMAGE
      exit 1
    fi

    echo "Deployment successful"

Best Practices

1. Pin all dependency versions (requirements.txt with ==) 2. Cache build artifacts (Docker layers, pip packages) 3. Run tests in parallel when possible 4. Use separate staging environment 5. Implement gradual rollout (canary, blue-green) 6. Monitor deployment health automatically 7. Keep rollback procedure simple (one command) 8. Log all deployment events for audit trail

Containerization and Deployment

Production Docker configuration, multi-stage builds, model versioning, A/B testing, and deployment strategies for ML models.

Multi-Stage Dockerfile for ML Models

# Stage 1: Builder - Install dependencies
FROM python:3.11-slim as builder

WORKDIR /build

# Install build dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements
COPY requirements.txt .

# Install Python dependencies to /install
RUN pip install --no-cache-dir --prefix=/install \
    -r requirements.txt

# Stage 2: Runtime - Minimal image
FROM python:3.11-slim

# Create non-root user
RUN useradd -m -u 1000 mluser && \
    mkdir -p /app /models && \
    chown -R mluser:mluser /app /models

WORKDIR /app

# Copy installed dependencies from builder
COPY --from=builder /install /usr/local

# Copy application code
COPY --chown=mluser:mluser app.py .
COPY --chown=mluser:mluser model.pkl /models/

# Switch to non-root user
USER mluser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')" || exit 1

# Run application
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Model Versioning Strategy

# Dockerfile with versioned model
ARG MODEL_VERSION=1.0.0
FROM python:3.11-slim

WORKDIR /app

# Copy model with version
COPY models/model_v${MODEL_VERSION}.pkl /models/model.pkl

ENV MODEL_VERSION=${MODEL_VERSION}

# Rest of Dockerfile...

Build with specific version:

docker build --build-arg MODEL_VERSION=1.2.3 -t model-api:v1.2.3 .

Docker Compose for Local Development

# docker-compose.yml
version: '3.8'

services:
  model-api:
    build:
      context: .
      args:
        MODEL_VERSION: "1.0.0"
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/models/model.pkl
      - LOG_LEVEL=INFO
      - WORKERS=2
    volumes:
      - ./models:/models:ro  # Read-only models
      - ./logs:/app/logs     # Persistent logs
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus-data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana-data:/var/lib/grafana
    depends_on:
      - prometheus

volumes:
  prometheus-data:
  grafana-data:

A/B Testing Deployment

Nginx Configuration for A/B Testing

# nginx.conf
upstream model_a {
    server model-api-a:8000;
}

upstream model_b {
    server model-api-b:8000;
}

split_clients "${remote_addr}${http_user_agent}" $model_upstream {
    70%    model_a;  # 70% traffic to model A
    30%    model_b;  # 30% traffic to model B
}

server {
    listen 80;

    location /predict {
        proxy_pass http://$model_upstream;
        proxy_set_header X-Model-Version $upstream_addr;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }

    location /health {
        proxy_pass http://model_a;  # Use primary for health
    }
}

Docker Compose for A/B Testing

# docker-compose-ab.yml
version: '3.8'

services:
  model-api-a:
    build:
      context: .
      args:
        MODEL_VERSION: "1.0.0"
    environment:
      - MODEL_VERSION=1.0.0
    deploy:
      replicas: 3

  model-api-b:
    build:
      context: .
      args:
        MODEL_VERSION: "1.1.0"
    environment:
      - MODEL_VERSION=1.1.0
    deploy:
      replicas: 1  # Less replicas for testing

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - model-api-a
      - model-api-b

Kubernetes Deployment

Deployment YAML

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-api
  labels:
    app: model-api
    version: v1.0.0
spec:
  replicas: 3
  selector:
    matchLabels:
      app: model-api
  template:
    metadata:
      labels:
        app: model-api
        version: v1.0.0
    spec:
      containers:
      - name: model-api
        image: your-registry/model-api:v1.0.0
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_PATH
          value: "/models/model.pkl"
        - name: WORKERS
          value: "2"
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
        volumeMounts:
        - name: model-storage
          mountPath: /models
          readOnly: true
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: model-api-service
spec:
  selector:
    app: model-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

Rolling Update Strategy

# deployment-rolling.yaml
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1        # Max 1 pod above desired count
      maxUnavailable: 0  # No downtime
  template:
    # ... container spec

Blue-Green Deployment

# Service for active (blue) deployment
apiVersion: v1
kind: Service
metadata:
  name: model-api
spec:
  selector:
    app: model-api
    version: blue  # Switch to 'green' to cut over
  ports:
  - port: 80
    targetPort: 8000

CI/CD Pipeline (GitHub Actions)

# .github/workflows/deploy.yml
name: Build and Deploy Model API

on:
  push:
    branches: [main]
    tags: ['v*']

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}/model-api

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

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

      - name: Run tests
        run: pytest tests/ --cov=app --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v3

  build-and-push:
    needs: test
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write

    steps:
      - uses: actions/checkout@v3

      - name: Log in to Container Registry
        uses: docker/login-action@v2
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Extract metadata
        id: meta
        uses: docker/metadata-action@v4
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
            type=ref,event=branch
            type=semver,pattern={{version}}
            type=semver,pattern={{major}}.{{minor}}
            type=sha

      - name: Build and push Docker image
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

  deploy:
    needs: build-and-push
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v3

      - name: Configure kubectl
        uses: azure/k8s-set-context@v3
        with:
          kubeconfig: ${{ secrets.KUBECONFIG }}

      - name: Deploy to Kubernetes
        run: |
          kubectl set image deployment/model-api \
            model-api=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:sha-${{ github.sha }}
          kubectl rollout status deployment/model-api

Model Registry Integration

# model_registry.py
import mlflow
from typing import Optional

class ModelRegistry:
    """
    Manage model versions with MLflow.
    """

    def __init__(self, tracking_uri: str):
        mlflow.set_tracking_uri(tracking_uri)

    def register_model(
        self,
        model_path: str,
        model_name: str,
        version: str,
        metrics: dict
    ) -> str:
        """Register model in MLflow."""
        with mlflow.start_run():
            # Log metrics
            mlflow.log_metrics(metrics)

            # Log model
            mlflow.sklearn.log_model(
                model_path,
                "model",
                registered_model_name=model_name
            )

            # Tag with version
            mlflow.set_tag("version", version)

        return f"{model_name}:{version}"

    def promote_to_production(
        self,
        model_name: str,
        version: str
    ):
        """Promote model version to production."""
        client = mlflow.tracking.MlflowClient()

        # Transition to production
        client.transition_model_version_stage(
            name=model_name,
            version=version,
            stage="Production"
        )

Deployment Checklist

Pre-Deployment

• [ ] Model validated on test set (metrics documented)
• [ ] Docker image built and scanned for vulnerabilities
• [ ] Resource limits configured (CPU, memory)
• [ ] Environment variables configured
• [ ] Secrets managed securely (K8s secrets, Vault)
• [ ] Health check endpoints tested
• [ ] Model file size optimized (<500MB)

Deployment

• [ ] Rolling update configured (no downtime)
• [ ] Monitoring configured (Prometheus, Grafana)
• [ ] Logging aggregation setup (ELK, Loki)
• [ ] Alerts configured (PagerDuty, Slack)
• [ ] Load balancer configured
• [ ] Auto-scaling rules set

Post-Deployment

• [ ] Smoke tests passed
• [ ] Performance benchmarks met (latency <200ms)
• [ ] Drift monitoring enabled
• [ ] A/B test configured (if applicable)
• [ ] Rollback procedure documented
• [ ] On-call rotation updated

Best Practices

1. Use multi-stage builds to minimize image size 2. Run as non-root user for security 3. Pin dependency versions in requirements.txt 4. Implement health checks for orchestration 5. Use secrets management (never commit credentials) 6. Enable horizontal pod autoscaling in Kubernetes 7. Monitor resource usage and adjust limits 8. Test rollback procedure before deploying

FastAPI Production Server for ML Models

Complete production-ready FastAPI implementation with error handling, validation, logging, health checks, and performance optimizations.

Complete Production FastAPI Server

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, validator
from typing import List, Dict, Optional, Any
import joblib
import numpy as np
import logging
import time
from datetime import datetime
from pathlib import Path
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Initialize FastAPI app
app = FastAPI(
    title="ML Model API",
    description="Production ML model serving with FastAPI",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure for your domain in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global model storage
class ModelStore:
    """Singleton for model management."""
    _instance = None
    _model = None
    _model_version = None
    _model_loaded_at = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(ModelStore, cls).__new__(cls)
        return cls._instance

    def load_model(self, model_path: str):
        """Load model from disk with error handling."""
        try:
            logger.info(f"Loading model from {model_path}")
            self._model = joblib.load(model_path)
            self._model_version = self._extract_version(model_path)
            self._model_loaded_at = datetime.now()
            logger.info(f"Model loaded successfully. Version: {self._model_version}")
        except Exception as e:
            logger.error(f"Failed to load model: {e}")
            raise RuntimeError(f"Model loading failed: {e}")

    def _extract_version(self, path: str) -> str:
        """Extract version from model filename."""
        # Example: model_v1.2.3.pkl -> 1.2.3
        filename = Path(path).stem
        if '_v' in filename:
            return filename.split('_v')[1]
        return "unknown"

    @property
    def model(self):
        """Get loaded model."""
        if self._model is None:
            raise RuntimeError("Model not loaded")
        return self._model

    @property
    def version(self) -> str:
        """Get model version."""
        return self._model_version or "unknown"

    @property
    def loaded_at(self) -> Optional[datetime]:
        """Get model load timestamp."""
        return self._model_loaded_at


# Initialize model store
model_store = ModelStore()


# Request/Response models with validation
class Feature(BaseModel):
    """Single feature with validation."""
    name: str = Field(..., description="Feature name")
    value: float = Field(..., description="Feature value")

    @validator('value')
    def validate_value(cls, v):
        """Validate feature value is finite."""
        if not np.isfinite(v):
            raise ValueError(f"Feature value must be finite, got {v}")
        return v


class PredictionRequest(BaseModel):
    """Request model for predictions."""
    features: List[float] = Field(
        ...,
        description="Feature vector for prediction",
        min_items=1,
        max_items=100
    )
    model_version: Optional[str] = Field(
        None,
        description="Specific model version to use"
    )

    @validator('features')
    def validate_features(cls, v):
        """Validate all features are finite numbers."""
        if not all(np.isfinite(val) for val in v):
            raise ValueError("All features must be finite numbers")
        return v

    class Config:
        schema_extra = {
            "example": {
                "features": [1.2, 3.4, 5.6, 7.8],
                "model_version": "1.0.0"
            }
        }


class PredictionResponse(BaseModel):
    """Response model for predictions."""
    prediction: float = Field(..., description="Model prediction")
    probability: Optional[float] = Field(
        None,
        description="Prediction probability (if classification)"
    )
    model_version: str = Field(..., description="Model version used")
    latency_ms: float = Field(..., description="Prediction latency in milliseconds")
    timestamp: datetime = Field(default_factory=datetime.now)

    class Config:
        schema_extra = {
            "example": {
                "prediction": 1.0,
                "probability": 0.87,
                "model_version": "1.0.0",
                "latency_ms": 12.5,
                "timestamp": "2024-01-15T10:30:00"
            }
        }


class BatchPredictionRequest(BaseModel):
    """Batch prediction request."""
    instances: List[List[float]] = Field(
        ...,
        description="List of feature vectors",
        min_items=1,
        max_items=1000  # Limit batch size
    )

    @validator('instances')
    def validate_instances(cls, v):
        """Validate all instances have same length."""
        if not v:
            raise ValueError("instances cannot be empty")

        first_len = len(v[0])
        if not all(len(inst) == first_len for inst in v):
            raise ValueError("All instances must have same number of features")

        return v


class HealthResponse(BaseModel):
    """Health check response."""
    status: str
    model_loaded: bool
    model_version: Optional[str]
    uptime_seconds: float
    timestamp: datetime


# Startup event
@app.on_event("startup")
async def startup_event():
    """Load model on startup."""
    model_path = os.getenv("MODEL_PATH", "model.pkl")

    if not Path(model_path).exists():
        logger.error(f"Model file not found: {model_path}")
        raise FileNotFoundError(f"Model file not found: {model_path}")

    model_store.load_model(model_path)
    logger.info("API server started successfully")


# Middleware for request logging
@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log all requests with timing."""
    start_time = time.time()

    # Process request
    response = await call_next(request)

    # Log request details
    process_time = (time.time() - start_time) * 1000
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} "
        f"duration={process_time:.2f}ms"
    )

    # Add custom headers
    response.headers["X-Process-Time"] = str(process_time)
    response.headers["X-Model-Version"] = model_store.version

    return response


# Exception handlers
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Handle HTTP exceptions with consistent format."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "status_code": exc.status_code,
            "timestamp": datetime.now().isoformat()
        }
    )


@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Handle unexpected exceptions."""
    logger.error(f"Unexpected error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "Internal server error",
            "detail": str(exc) if os.getenv("DEBUG") else "An error occurred",
            "timestamp": datetime.now().isoformat()
        }
    )


# Health check endpoint
@app.get(
    "/health",
    response_model=HealthResponse,
    status_code=status.HTTP_200_OK,
    tags=["Health"]
)
async def health_check():
    """
    Health check endpoint for load balancers and monitoring.

    Returns service status and model information.
    """
    try:
        model_loaded = model_store._model is not None
        uptime = (datetime.now() - model_store.loaded_at).total_seconds() \
            if model_store.loaded_at else 0

        return HealthResponse(
            status="healthy" if model_loaded else "degraded",
            model_loaded=model_loaded,
            model_version=model_store.version,
            uptime_seconds=uptime,
            timestamp=datetime.now()
        )
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        return JSONResponse(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            content={
                "status": "unhealthy",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }
        )


# Readiness probe
@app.get("/ready", tags=["Health"])
async def readiness_check():
    """
    Readiness probe for Kubernetes.

    Returns 200 when model is loaded and ready to serve.
    """
    try:
        # Verify model is loaded and functional
        _ = model_store.model
        return {"status": "ready", "timestamp": datetime.now()}
    except Exception as e:
        logger.error(f"Readiness check failed: {e}")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Service not ready"
        )


# Prediction endpoint
@app.post(
    "/predict",
    response_model=PredictionResponse,
    status_code=status.HTTP_200_OK,
    tags=["Predictions"]
)
async def predict(request: PredictionRequest):
    """
    Make a single prediction.

    Args:
        request: Prediction request with features

    Returns:
        Prediction with probability and metadata

    Raises:
        HTTPException: If prediction fails
    """
    start_time = time.time()

    try:
        # Validate model is loaded
        model = model_store.model

        # Prepare features
        features = np.array(request.features).reshape(1, -1)

        # Make prediction
        prediction = float(model.predict(features)[0])

        # Get probability if classification model
        probability = None
        if hasattr(model, 'predict_proba'):
            proba = model.predict_proba(features)[0]
            probability = float(proba.max())

        # Calculate latency
        latency_ms = (time.time() - start_time) * 1000

        logger.info(f"Prediction made: {prediction} (latency: {latency_ms:.2f}ms)")

        return PredictionResponse(
            prediction=prediction,
            probability=probability,
            model_version=model_store.version,
            latency_ms=latency_ms,
            timestamp=datetime.now()
        )

    except ValueError as e:
        logger.error(f"Validation error: {e}")
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid input features: {e}"
        )
    except Exception as e:
        logger.error(f"Prediction failed: {e}", exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Prediction failed"
        )


# Batch prediction endpoint
@app.post(
    "/predict/batch",
    status_code=status.HTTP_200_OK,
    tags=["Predictions"]
)
async def predict_batch(request: BatchPredictionRequest):
    """
    Make batch predictions.

    Args:
        request: Batch prediction request

    Returns:
        List of predictions with metadata
    """
    start_time = time.time()

    try:
        model = model_store.model

        # Prepare features
        features = np.array(request.instances)

        # Make predictions
        predictions = model.predict(features).tolist()

        # Get probabilities if available
        probabilities = None
        if hasattr(model, 'predict_proba'):
            probabilities = model.predict_proba(features).max(axis=1).tolist()

        latency_ms = (time.time() - start_time) * 1000

        logger.info(
            f"Batch prediction: {len(predictions)} instances "
            f"(latency: {latency_ms:.2f}ms)"
        )

        return {
            "predictions": predictions,
            "probabilities": probabilities,
            "model_version": model_store.version,
            "latency_ms": latency_ms,
            "count": len(predictions),
            "timestamp": datetime.now()
        }

    except Exception as e:
        logger.error(f"Batch prediction failed: {e}", exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Batch prediction failed"
        )


# Model metadata endpoint
@app.get("/model/info", tags=["Model"])
async def model_info():
    """
    Get model metadata and information.

    Returns:
        Model version, loaded time, and other metadata
    """
    try:
        return {
            "version": model_store.version,
            "loaded_at": model_store.loaded_at,
            "model_type": type(model_store.model).__name__,
            "uptime_seconds": (datetime.now() - model_store.loaded_at).total_seconds()
                if model_store.loaded_at else 0
        }
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to get model info: {e}"
        )

Production Configuration

Environment Variables

# .env file
MODEL_PATH=/models/model_v1.0.0.pkl
LOG_LEVEL=INFO
DEBUG=false
MAX_BATCH_SIZE=1000
WORKERS=4
HOST=0.0.0.0
PORT=8000

Running with Uvicorn

# Development
uvicorn app:app --reload --host 0.0.0.0 --port 8000

# Production with multiple workers
uvicorn app:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4 \
  --log-level info \
  --access-log

# With Gunicorn (recommended for production)
gunicorn app:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --access-logfile - \
  --error-logfile -

Testing the API

import requests
import json

# Test health endpoint
response = requests.get("http://localhost:8000/health")
print(response.json())

# Test prediction
response = requests.post(
    "http://localhost:8000/predict",
    json={
        "features": [1.2, 3.4, 5.6, 7.8]
    }
)
print(response.json())

# Test batch prediction
response = requests.post(
    "http://localhost:8000/predict/batch",
    json={
        "instances": [
            [1.2, 3.4, 5.6, 7.8],
            [2.3, 4.5, 6.7, 8.9],
            [3.4, 5.6, 7.8, 9.0]
        ]
    }
)
print(response.json())

Performance Optimizations

1. Response Caching

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def cached_predict(features_hash: str):
    """Cache predictions for identical inputs."""
    # Decode features from hash
    # Make prediction
    pass

def hash_features(features: List[float]) -> str:
    """Create hash of features for caching."""
    return hashlib.md5(str(features).encode()).hexdigest()

2. Async Model Loading

from fastapi import BackgroundTasks

async def load_new_model(model_path: str):
    """Load model in background."""
    model_store.load_model(model_path)

@app.post("/model/reload")
async def reload_model(background_tasks: BackgroundTasks):
    """Reload model without downtime."""
    background_tasks.add_task(load_new_model, "new_model.pkl")
    return {"status": "Model reload initiated"}

3. Connection Pooling

# For models that need database/cache connections
from redis import ConnectionPool
import redis

redis_pool = ConnectionPool(host='localhost', port=6379, db=0)
redis_client = redis.Redis(connection_pool=redis_pool)

Security Best Practices

API Key Authentication

from fastapi.security import APIKeyHeader
from fastapi import Security

API_KEY = os.getenv("API_KEY")
api_key_header = APIKeyHeader(name="X-API-Key")

def verify_api_key(api_key: str = Security(api_key_header)):
    """Verify API key."""
    if api_key != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API key"
        )
    return api_key

@app.post("/predict", dependencies=[Depends(verify_api_key)])
async def predict(...):
    # Protected endpoint
    pass

Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/predict")
@limiter.limit("100/minute")
async def predict(request: Request, ...):
    # Rate-limited endpoint
    pass

Model Monitoring and Drift Detection

Complete implementation of model monitoring, drift detection, and alerting for production ML systems.

Overview

Model performance degrades over time due to:

• Data drift: Input distribution changes
• Concept drift: Relationship between features and target changes
• Model staleness: Training data becomes outdated

Monitoring detects these issues before they impact users.

Complete Monitoring System

from typing import Dict, List, Optional, Tuple
import numpy as np
from scipy import stats
from scipy.spatial.distance import jensenshannon
from datetime import datetime, timedelta
from collections import deque
import logging
import json

logger = logging.getLogger(__name__)


class ModelMonitor:
    """
    Production model monitoring with drift detection.

    Tracks predictions, inputs, latency, and detects distribution shifts.
    """

    def __init__(
        self,
        reference_data: np.ndarray,
        reference_predictions: Optional[np.ndarray] = None,
        window_size: int = 1000,
        drift_threshold: float = 0.1
    ):
        """
        Args:
            reference_data: Training/baseline feature matrix (n_samples, n_features)
            reference_predictions: Optional array of predictions on reference_data.
                                  Required for prediction drift detection.
                                  Should be 1D array of scalar predictions.
            window_size: Number of recent predictions to track
            drift_threshold: Threshold for drift detection (0-1)
        """
        self.reference_data = reference_data
        self.reference_predictions = reference_predictions
        self.window_size = window_size
        self.drift_threshold = drift_threshold

        # Rolling windows for monitoring
        self.predictions = deque(maxlen=window_size)
        self.inputs = deque(maxlen=window_size)
        self.latencies = deque(maxlen=window_size)
        self.timestamps = deque(maxlen=window_size)
        self.errors = deque(maxlen=window_size)

        # Statistics
        self.total_predictions = 0
        self.drift_detections = 0

    def log_prediction(
        self,
        input_features: np.ndarray,
        prediction: float,
        latency_ms: float,
        actual: Optional[float] = None
    ):
        """
        Log a prediction with monitoring data.

        Args:
            input_features: Input feature vector
            prediction: Model prediction
            latency_ms: Prediction latency
            actual: Actual label (if available for validation)
        """
        self.inputs.append(input_features)
        self.predictions.append(prediction)
        self.latencies.append(latency_ms)
        self.timestamps.append(datetime.now())

        if actual is not None:
            error = abs(prediction - actual)
            self.errors.append(error)

        self.total_predictions += 1

        # Check for drift periodically
        if self.total_predictions % 100 == 0:
            drift_detected, drift_score = self.detect_drift()
            if drift_detected:
                self.drift_detections += 1
                logger.warning(
                    f"Drift detected! Score: {drift_score:.4f} "
                    f"(threshold: {self.drift_threshold})"
                )

    def detect_drift(self) -> Tuple[bool, float]:
        """
        Detect input distribution drift using KS test.

        Returns:
            (drift_detected: bool, drift_score: float)
        """
        if len(self.inputs) < 100:
            return False, 0.0

        current_data = np.array(list(self.inputs))

        # Kolmogorov-Smirnov test per feature
        drift_scores = []
        for i in range(current_data.shape[1]):
            ref_feature = self.reference_data[:, i]
            curr_feature = current_data[:, i]

            # KS test
            statistic, p_value = stats.ks_2samp(ref_feature, curr_feature)
            drift_scores.append(statistic)

        # Max drift across all features
        max_drift = max(drift_scores)
        drift_detected = max_drift > self.drift_threshold

        return drift_detected, max_drift

    def detect_prediction_drift(self) -> Tuple[bool, float]:
        """
        Detect drift in prediction distribution.

        Returns:
            (drift_detected: bool, divergence: float)
        """
        if len(self.predictions) < 100:
            return False, 0.0

        # Check if reference predictions are available
        if self.reference_predictions is None:
            logger.warning(
                "No reference predictions provided. Cannot detect prediction drift. "
                "Pass reference_predictions to __init__ to enable this feature."
            )
            return False, 0.0

        # Create histograms from prediction distributions
        ref_hist, bins = np.histogram(
            self.reference_predictions,  # Fixed: Use reference predictions, not row means
            bins=20,
            density=True
        )
        curr_hist, _ = np.histogram(
            list(self.predictions),
            bins=bins,
            density=True
        )

        # Jensen-Shannon divergence
        divergence = jensenshannon(ref_hist + 1e-10, curr_hist + 1e-10)

        drift_detected = divergence > self.drift_threshold
        return drift_detected, float(divergence)

    def get_metrics(self) -> Dict:
        """
        Get current monitoring metrics.

        Returns:
            Dictionary of metrics
        """
        metrics = {
            "total_predictions": self.total_predictions,
            "drift_detections": self.drift_detections,
            "window_size": len(self.predictions),
        }

        if self.latencies:
            metrics["latency_p50_ms"] = np.percentile(list(self.latencies), 50)
            metrics["latency_p95_ms"] = np.percentile(list(self.latencies), 95)
            metrics["latency_p99_ms"] = np.percentile(list(self.latencies), 99)

        if self.predictions:
            metrics["prediction_mean"] = np.mean(list(self.predictions))
            metrics["prediction_std"] = np.std(list(self.predictions))

        if self.errors:
            metrics["mae"] = np.mean(list(self.errors))
            metrics["rmse"] = np.sqrt(np.mean([e**2 for e in self.errors]))

        # Drift scores
        drift_detected, drift_score = self.detect_drift()
        metrics["input_drift_detected"] = drift_detected
        metrics["input_drift_score"] = drift_score

        pred_drift, pred_divergence = self.detect_prediction_drift()
        metrics["prediction_drift_detected"] = pred_drift
        metrics["prediction_divergence"] = pred_divergence

        return metrics

    def should_retrain(self) -> bool:
        """
        Decide if model should be retrained based on metrics.

        Returns:
            True if retraining recommended
        """
        metrics = self.get_metrics()

        # Retrain if:
        # 1. Input drift detected
        # 2. Prediction drift detected
        # 3. Error rate above threshold (if available)

        if metrics.get("input_drift_detected", False):
            logger.info("Retraining recommended: Input drift detected")
            return True

        if metrics.get("prediction_drift_detected", False):
            logger.info("Retraining recommended: Prediction drift detected")
            return True

        if "mae" in metrics and metrics["mae"] > 0.5:  # Threshold depends on use case
            logger.info(f"Retraining recommended: High MAE ({metrics['mae']:.3f})")
            return True

        return False


## Prometheus Metrics Integration

from prometheus_client import Counter, Histogram, Gauge, generate_latest

Define metrics

prediction_counter = Counter( 'model_predictions_total', 'Total number of predictions', ['model_version', 'status'] )

prediction_latency = Histogram( 'model_prediction_latency_seconds', 'Prediction latency in seconds', ['model_version'], buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0] )

drift_score = Gauge( 'model_drift_score', 'Current drift score', ['drift_type'] # 'input' or 'prediction' )

error_rate = Gauge( 'model_error_rate', 'Model error rate' )

def record_prediction( model_version: str, latency: float, success: bool ): """Record prediction in Prometheus.""" status = 'success' if success else 'failure' prediction_counter.labels( model_version=model_version, status=status ).inc()

prediction_latency.labels( model_version=model_version ).observe(latency)

@app.get("/metrics") async def metrics(): """Prometheus metrics endpoint.""" return Response(generate_latest(), media_type="text/plain")

## Alert Configuration

class AlertManager: """ Manage alerts for model degradation. """

def __init__( self, slack_webhook: Optional[str] = None, email_config: Optional[Dict] = None ): self.slack_webhook = slack_webhook self.email_config = email_config

def send_drift_alert( self, drift_type: str, drift_score: float, metrics: Dict ): """Send alert when drift detected.""" message = f""" 🚨 Model Drift Alert

Type: {drift_type} Score: {drift_score:.4f}

Current Metrics:

• Total Predictions: {metrics['total_predictions']}
• Latency P95: {metrics.get('latency_p95_ms', 'N/A')} ms
• MAE: {metrics.get('mae', 'N/A')}

Action: Consider retraining model """

if self.slack_webhook: self._send_slack(message)

logger.critical(message)

def _send_slack(self, message: str): """Send message to Slack.""" import requests requests.post( self.slack_webhook, json={"text": message} )

## Continuous Monitoring Pipeline

import asyncio

class ContinuousMonitor: """ Background monitoring service. """

def __init__( self, monitor: ModelMonitor, alert_manager: AlertManager, check_interval_seconds: int = 300 # 5 minutes ): self.monitor = monitor self.alert_manager = alert_manager self.check_interval = check_interval_seconds self.running = False

async def start(self): """Start continuous monitoring.""" self.running = True logger.info("Starting continuous monitoring")

while self.running: try:

Get current metrics

metrics = self.monitor.get_metrics()

Check for drift

if metrics.get("input_drift_detected"): self.alert_manager.send_drift_alert( "Input Distribution Drift", metrics["input_drift_score"], metrics )

if metrics.get("prediction_drift_detected"): self.alert_manager.send_drift_alert( "Prediction Distribution Drift", metrics["prediction_divergence"], metrics )

Update Prometheus gauges

drift_score.labels(drift_type='input').set( metrics["input_drift_score"] ) drift_score.labels(drift_type='prediction').set( metrics["prediction_divergence"] )

if "mae" in metrics: error_rate.set(metrics["mae"])

Log metrics

logger.info(f"Monitoring check: {json.dumps(metrics, indent=2)}")

except Exception as e: logger.error(f"Monitoring error: {e}", exc_info=True)

Wait before next check

await asyncio.sleep(self.check_interval)

def stop(self): """Stop monitoring.""" self.running = False logger.info("Stopping continuous monitoring")

Start monitoring on FastAPI startup

monitor = ModelMonitor(reference_data=training_data) alert_manager = AlertManager(slack_webhook=os.getenv("SLACK_WEBHOOK")) continuous_monitor = ContinuousMonitor(monitor, alert_manager)

@app.on_event("startup") async def start_monitoring(): asyncio.create_task(continuous_monitor.start())

@app.on_event("shutdown") async def stop_monitoring(): continuous_monitor.stop()

## Dashboard Integration

Endpoint for monitoring dashboard

@app.get("/monitoring/metrics") async def get_monitoring_metrics(): """Get current monitoring metrics for dashboard.""" metrics = monitor.get_metrics()

return { "metrics": metrics, "should_retrain": monitor.should_retrain(), "last_updated": datetime.now(), "window_info": { "size": len(monitor.predictions), "max_size": monitor.window_size, "oldest_timestamp": monitor.timestamps[0] if monitor.timestamps else None, "newest_timestamp": monitor.timestamps[-1] if monitor.timestamps else None } }

@app.get("/monitoring/drift-history") async def get_drift_history(): """Get historical drift scores."""

This would query from time-series database in production

return { "input_drift": list(monitor.inputs), "prediction_drift": list(monitor.predictions), "timestamps": [ts.isoformat() for ts in monitor.timestamps] }

## Best Practices

1. **Set appropriate thresholds** based on your use case (0.05-0.2 typically)
2. **Monitor multiple metrics** (inputs, predictions, errors, latency)
3. **Use time-series database** (Prometheus, InfluxDB) for long-term storage
4. **Implement gradual rollback** when drift detected
5. **Log all predictions** for future analysis
6. **Test drift detection** on historical data before deploying