Model Deployment & MLOps

import joblib
import os

# Save without compression
joblib.dump(pipeline, 'model_uncompressed.joblib')

# Save with compression
joblib.dump(pipeline, 'model_compressed.joblib', compress=3)  # 1-9 compression level

# Compare sizes
size_uncompressed = os.path.getsize('model_uncompressed.joblib') / 1024
size_compressed = os.path.getsize('model_compressed.joblib') / 1024

print(f"Uncompressed: {size_uncompressed:.2f} KB")
print(f"Compressed: {size_compressed:.2f} KB")
print(f"Reduction: {(1 - size_compressed/size_uncompressed)*100:.1f}%")

# Load compressed model
loaded = joblib.load('model_compressed.joblib')
print(f"Model works: {loaded.score(X_test, y_test):.4f}")

import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

# Train model
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

print(f"Original model accuracy: {model.score(X_test, y_test):.4f}")

# Save
joblib.dump(model, 'my_model.joblib')
print("Model saved!")

# Load
loaded_model = joblib.load('my_model.joblib')
print(f"Loaded model accuracy: {loaded_model.score(X_test, y_test):.4f}")

# Verify they're the same!
assert model.score(X_test, y_test) == loaded_model.score(X_test, y_test)
print("✓ Model works identically after loading!")

import joblib
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

# Create pipeline
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression(random_state=42))
])

# Train
pipeline.fit(X_train, y_train)

# Save
joblib.dump(pipeline, 'full_pipeline.joblib')

# Load and test with RAW (unscaled) data
loaded_pipeline = joblib.load('full_pipeline.joblib')

# This should work - pipeline handles scaling internally!
raw_sample = [[5.1, 3.5, 1.4, 0.2]]  # Raw values, not scaled
prediction = loaded_pipeline.predict(raw_sample)
print(f"Prediction from raw input: {prediction}")

# Verify scaler has correct statistics
scaler = loaded_pipeline.named_steps['scaler']
print(f"Scaler mean (from training): {scaler.mean_.round(2)}")
print("✓ Pipeline remembers how to scale!")

import joblib
import json
from datetime import datetime
import sklearn

# Train and save model
pipeline.fit(X_train, y_train)
joblib.dump(pipeline, 'models/model_v1.joblib')

# Create metadata
metadata = {
    'model_name': 'iris_classifier',
    'version': '1.0.0',
    'created_at': datetime.now().isoformat(),
    'sklearn_version': sklearn.__version__,
    'python_version': '3.10',
    'metrics': {
        'train_accuracy': float(pipeline.score(X_train, y_train)),
        'test_accuracy': float(pipeline.score(X_test, y_test))
    },
    'features': ['sepal_length', 'sepal_width', 'petal_length', 'petal_width'],
    'classes': ['setosa', 'versicolor', 'virginica'],
    'notes': 'First production model'
}

# Save metadata
with open('models/model_v1_metadata.json', 'w') as f:
    json.dump(metadata, f, indent=2)

print("Model and metadata saved!")
print(json.dumps(metadata, indent=2))

import joblib
import json
import sklearn
from packaging import version

def safe_load_model(model_path, metadata_path):
    """Load model with version compatibility check."""
    
    # Load metadata first
    with open(metadata_path, 'r') as f:
        metadata = json.load(f)
    
    saved_version = metadata.get('sklearn_version', 'unknown')
    current_version = sklearn.__version__
    
    print(f"Model was saved with sklearn {saved_version}")
    print(f"Current sklearn version: {current_version}")
    
    # Check major.minor version match
    saved_major_minor = '.'.join(saved_version.split('.')[:2])
    current_major_minor = '.'.join(current_version.split('.')[:2])
    
    if saved_major_minor != current_major_minor:
        print(f"WARNING: Version mismatch!")
        print(f"   Model: {saved_major_minor}.x, Current: {current_major_minor}.x")
        print("   This may cause errors. Consider retraining.")
        
        # Ask for confirmation
        proceed = input("Load anyway? (y/n): ")
        if proceed.lower() != 'y':
            raise ValueError("Loading cancelled due to version mismatch")
    
    # Load the model
    try:
        model = joblib.load(model_path)
        print("✓ Model loaded successfully!")
        return model, metadata
    except Exception as e:
        print(f"✗ Failed to load: {e}")
        raise

# Usage
model, meta = safe_load_model('models/model_v1.joblib', 'models/model_v1_metadata.json')

import joblib
import json
from pathlib import Path
from datetime import datetime

class ModelRegistry:
    """Simple local model registry."""
    
    def __init__(self, registry_dir='model_registry'):
        self.registry_dir = Path(registry_dir)
        self.registry_dir.mkdir(exist_ok=True)
        self.registry_file = self.registry_dir / 'registry.json'
        self.registry = self._load_registry()
    
    def _load_registry(self):
        if self.registry_file.exists():
            with open(self.registry_file, 'r') as f:
                return json.load(f)
        return {'models': {}}
    
    def _save_registry(self):
        with open(self.registry_file, 'w') as f:
            json.dump(self.registry, f, indent=2)
    
    def register(self, model, name, version, metrics=None):
        """Save and register a model version."""
        model_path = self.registry_dir / f"{name}_{version}.joblib"
        joblib.dump(model, model_path)
        
        if name not in self.registry['models']:
            self.registry['models'][name] = {}
        
        self.registry['models'][name][version] = {
            'path': str(model_path),
            'created_at': datetime.now().isoformat(),
            'metrics': metrics or {}
        }
        self._save_registry()
        print(f"Registered {name} v{version}")
    
    def load(self, name, version='latest'):
        """Load a specific model version."""
        if name not in self.registry['models']:
            raise ValueError(f"Model '{name}' not found")
        
        versions = self.registry['models'][name]
        if version == 'latest':
            version = max(versions.keys())
        
        if version not in versions:
            raise ValueError(f"Version '{version}' not found")
        
        path = versions[version]['path']
        return joblib.load(path)
    
    def list_versions(self, name):
        """List all versions of a model."""
        return list(self.registry['models'].get(name, {}).keys())

# Usage
registry = ModelRegistry()
registry.register(pipeline, 'iris_classifier', 'v1.0', {'accuracy': 0.95})
registry.register(pipeline, 'iris_classifier', 'v1.1', {'accuracy': 0.97})

print(registry.list_versions('iris_classifier'))
model = registry.load('iris_classifier', 'latest')

import logging
from datetime import datetime
import time

# Configure logging
logging.basicConfig(
    filename='predictions.log',
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)

@app.route('/predict', methods=['POST'])
def predict_with_logging():
    start_time = time.time()
    data = request.get_json()
    
    try:
        features = np.array(data['features'])
        predictions = model.predict(features)
        
        # Calculate latency
        latency_ms = (time.time() - start_time) * 1000
        
        # Log request
        logging.info(
            f"REQUEST - samples: {len(features)}, "
            f"predictions: {predictions.tolist()}, "
            f"latency_ms: {latency_ms:.2f}"
        )
        
        return jsonify({
            'predictions': predictions.tolist(),
            'latency_ms': latency_ms
        })
    
    except Exception as e:
        logging.error(f"ERROR - {str(e)}")
        return jsonify({'error': str(e)}), 400

# hello_api.py
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/')
def home():
    return "Hello! The API is running."

@app.route('/health')
def health():
    return jsonify({
        'status': 'healthy',
        'message': 'API is up and running!'
    })

if __name__ == '__main__':
    app.run(debug=True, port=5000)

# Run: python hello_api.py
# Test: 
#   curl http://localhost:5000/
#   curl http://localhost:5000/health

# test_my_api.py
import requests
import json

# API URL
BASE_URL = 'http://localhost:5000'

# Test 1: Health check
print("Testing health endpoint...")
response = requests.get(f'{BASE_URL}/health')
print(f"Status: {response.status_code}")
print(f"Response: {response.json()}")

# Test 2: Single prediction
print("\nTesting single prediction...")
payload = {
    'features': [[5.1, 3.5, 1.4, 0.2]]  # One iris sample
}
response = requests.post(
    f'{BASE_URL}/predict',
    json=payload
)
print(f"Status: {response.status_code}")
print(f"Prediction: {json.dumps(response.json(), indent=2)}")

# Test 3: Batch prediction
print("\nTesting batch prediction...")
payload = {
    'features': [
        [5.1, 3.5, 1.4, 0.2],
        [6.2, 2.9, 4.3, 1.3],
        [7.2, 3.0, 5.8, 1.6]
    ]
}
response = requests.post(f'{BASE_URL}/predict', json=payload)
print(f"Predictions: {response.json()}")

from flask import Flask, request, jsonify
import numpy as np
import joblib

app = Flask(__name__)
model = joblib.load('model.joblib')

EXPECTED_FEATURES = 4
FEATURE_NAMES = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width']

@app.route('/predict', methods=['POST'])
def predict():
    # Check content type
    if not request.is_json:
        return jsonify({'error': 'Content-Type must be application/json'}), 400
    
    data = request.get_json()
    
    # Check for 'features' key
    if 'features' not in data:
        return jsonify({
            'error': 'Missing "features" key',
            'expected_format': {'features': [[1.0, 2.0, 3.0, 4.0]]}
        }), 400
    
    try:
        features = np.array(data['features'])
    except Exception:
        return jsonify({'error': 'Features must be a list of lists of numbers'}), 400
    
    # Check dimensions
    if features.ndim == 1:
        features = features.reshape(1, -1)
    
    if features.shape[1] != EXPECTED_FEATURES:
        return jsonify({
            'error': f'Expected {EXPECTED_FEATURES} features, got {features.shape[1]}',
            'feature_names': FEATURE_NAMES
        }), 400
    
    # Check for NaN/Inf
    if np.any(np.isnan(features)) or np.any(np.isinf(features)):
        return jsonify({'error': 'Features cannot contain NaN or Inf values'}), 400
    
    # Make prediction
    predictions = model.predict(features)
    return jsonify({'predictions': predictions.tolist()})

from flask import Flask, request, jsonify
from collections import defaultdict
import time

app = Flask(__name__)

# Simple in-memory rate limiter
request_counts = defaultdict(list)
RATE_LIMIT = 10  # requests per minute

def check_rate_limit(ip):
    """Check if IP has exceeded rate limit."""
    now = time.time()
    minute_ago = now - 60
    
    # Clean old requests
    request_counts[ip] = [t for t in request_counts[ip] if t > minute_ago]
    
    if len(request_counts[ip]) >= RATE_LIMIT:
        return False, len(request_counts[ip])
    
    # Record this request
    request_counts[ip].append(now)
    return True, len(request_counts[ip])

@app.before_request
def rate_limit_check():
    """Check rate limit before every request."""
    ip = request.remote_addr
    allowed, count = check_rate_limit(ip)
    
    if not allowed:
        return jsonify({
            'error': 'Rate limit exceeded',
            'limit': f'{RATE_LIMIT} requests per minute',
            'retry_after': '60 seconds'
        }), 429

@app.route('/predict', methods=['POST'])
def predict():
    # Your normal prediction logic
    data = request.get_json()
    features = np.array(data['features'])
    predictions = model.predict(features)
    return jsonify({'predictions': predictions.tolist()})

# Note: For production, use Flask-Limiter or Redis-based solutions

from fastapi import FastAPI, HTTPException, Depends, Security
from fastapi.security import APIKeyHeader
import os

app = FastAPI()

# API key configuration
API_KEY = os.getenv("API_KEY", "your-secret-key")
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def verify_api_key(api_key: str = Security(api_key_header)):
    """Verify the API key."""
    if api_key is None:
        raise HTTPException(status_code=401, detail="API key missing")
    if api_key != API_KEY:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.post("/predict", dependencies=[Depends(verify_api_key)])
async def predict(request: PredictionRequest):
    """Protected endpoint - requires valid API key."""
    # ... prediction logic
    pass

# Test with:
# curl -X POST http://localhost:8000/predict \
#   -H "X-API-Key: your-secret-key" \
#   -H "Content-Type: application/json" \
#   -d '{"samples": [{"sepal_length": 5.1, ...}]}'

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="My First FastAPI",
    description="A simple API to learn FastAPI",
    version="1.0.0"
)

@app.get("/")
async def home():
    return {"message": "Hello from FastAPI!"}

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

@app.get("/greet/{name}")
async def greet(name: str):
    return {"message": f"Hello, {name}!"}

# Run with: uvicorn main:app --reload
# Visit: http://localhost:8000/docs  ← Free Swagger UI!
# Also try: http://localhost:8000/redoc

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

# Define input schema with validation
class IrisSample(BaseModel):
    sepal_length: float = Field(..., ge=0, le=10, description="Sepal length in cm")
    sepal_width: float = Field(..., ge=0, le=10, description="Sepal width in cm")
    petal_length: float = Field(..., ge=0, le=10, description="Petal length in cm")
    petal_width: float = Field(..., ge=0, le=10, description="Petal width in cm")
    
    class Config:
        json_schema_extra = {
            "example": {
                "sepal_length": 5.1,
                "sepal_width": 3.5,
                "petal_length": 1.4,
                "petal_width": 0.2
            }
        }

@app.post("/predict")
async def predict(sample: IrisSample):
    # Validation happens automatically!
    # If values are out of range, FastAPI returns 422 error
    return {
        "received": sample.dict(),
        "message": "Validation passed!"
    }

# Test with invalid data to see automatic error:
# curl -X POST localhost:8000/predict -H "Content-Type: application/json" \
#   -d '{"sepal_length": 100}'  ← Will fail validation!

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Dict

app = FastAPI()

# Response model
class PredictionResult(BaseModel):
    prediction: int
    class_name: str
    confidence: float
    probabilities: Dict[str, float]

class PredictionResponse(BaseModel):
    predictions: List[PredictionResult]
    latency_ms: float

# The endpoint
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    start = time.time()
    
    # ... make predictions ...
    
    results = [
        PredictionResult(
            prediction=0,
            class_name="setosa",
            confidence=0.95,
            probabilities={"setosa": 0.95, "versicolor": 0.03, "virginica": 0.02}
        )
    ]
    
    return PredictionResponse(
        predictions=results,
        latency_ms=(time.time() - start) * 1000
    )

# FastAPI validates output matches PredictionResponse
# Documentation shows exact response format

from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
import joblib

# Global model variable
ml_model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Load model on startup, cleanup on shutdown."""
    global ml_model
    
    # STARTUP
    print("Loading model...")
    try:
        ml_model = joblib.load('models/iris_classifier.joblib')
        print("✓ Model loaded successfully!")
    except Exception as e:
        print(f"✗ Failed to load model: {e}")
        raise
    
    yield  # App runs here
    
    # SHUTDOWN
    print("Shutting down...")
    ml_model = None

app = FastAPI(lifespan=lifespan)

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "model_loaded": ml_model is not None
    }

@app.post("/predict")
async def predict(request: PredictionRequest):
    if ml_model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    # Use ml_model for predictions
    predictions = ml_model.predict(...)
    return {"predictions": predictions.tolist()}

# Multi-stage Dockerfile
# Stage 1: Build
FROM python:3.10 AS builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y gcc

# Install Python packages
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# Stage 2: Production
FROM python:3.10-slim

WORKDIR /app

# Copy installed packages from builder
COPY --from=builder /root/.local /root/.local

# Make sure scripts are in PATH
ENV PATH=/root/.local/bin:$PATH

# Copy only application code
COPY app/ ./app/

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

# This produces a smaller image since build tools aren't included

# Dockerfile

# Line 1: Start from a Python image (the "base")
FROM python:3.10-slim

# Line 2: Create a folder in the container
WORKDIR /app

# Line 3: Copy requirements file
COPY requirements.txt .

# Line 4: Install Python packages
RUN pip install --no-cache-dir -r requirements.txt

# Line 5: Copy your application code
COPY . .

# Line 6: Tell Docker what port your app uses
EXPOSE 8000

# Line 7: Command to run when container starts
CMD ["python", "app.py"]

# Build: docker build -t my-ml-api .
# Run:   docker run -p 8000:8000 my-ml-api
# Test:  curl http://localhost:8000/health

# BUILD: Create an image from Dockerfile
docker build -t ml-api:v1 .
#         -t = tag/name the image
#         .  = use Dockerfile in current directory

# LIST IMAGES: See what images exist
docker images

# RUN: Start a container from the image
docker run -d -p 8000:8000 --name my-api ml-api:v1
#         -d = run in background (detached)
#         -p = map port (host:container)
#         --name = name the container

# CHECK RUNNING CONTAINERS
docker ps

# VIEW LOGS
docker logs my-api
docker logs -f my-api  # -f = follow (live)

# STOP the container
docker stop my-api

# REMOVE the container
docker rm my-api

# REMOVE the image
docker rmi ml-api:v1

# PRO TIP: One-liner to stop, remove, rebuild
docker stop my-api; docker rm my-api; docker build -t ml-api:v1 . && docker run -d -p 8000:8000 --name my-api ml-api:v1

# Dockerfile with environment variables
FROM python:3.10-slim
WORKDIR /app

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

# Set default environment variables
ENV MODEL_PATH=/app/models/model.joblib
ENV LOG_LEVEL=INFO
ENV PORT=8000

EXPOSE ${PORT}
CMD ["sh", "-c", "uvicorn app:app --host 0.0.0.0 --port ${PORT}"]

# In your Python app, read environment variables
import os

MODEL_PATH = os.getenv('MODEL_PATH', 'models/model.joblib')
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
PORT = int(os.getenv('PORT', 8000))

# Override env vars when running
docker run -d \
  -p 8080:8080 \
  -e MODEL_PATH=/app/models/v2.joblib \
  -e LOG_LEVEL=DEBUG \
  -e PORT=8080 \
  ml-api:v1

# docker-compose.yml
version: '3.8'

services:
  ml-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/app/models/model.joblib
      - LOG_LEVEL=INFO
    volumes:
      - ./logs:/app/logs  # Persist logs outside container
    restart: unless-stopped  # Auto-restart unless manually stopped
    
    # Health check - restart if unhealthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s        # Check every 30 seconds
      timeout: 10s         # Fail if no response in 10s
      retries: 3           # Mark unhealthy after 3 failures
      start_period: 40s    # Wait 40s before first check

    # Resource limits
    deploy:
      resources:
        limits:
          memory: 512M
        reservations:
          memory: 256M

# Run
docker-compose up -d

# Check health status
docker-compose ps

# View logs
docker-compose logs -f ml-api

# Stop
docker-compose down

from fastapi import FastAPI
import joblib
from pathlib import Path
import threading

app = FastAPI()

class ModelManager:
    """Manage multiple model versions."""
    
    def __init__(self, models_dir: str):
        self.models_dir = Path(models_dir)
        self.current_model = None
        self.current_version = None
        self.lock = threading.Lock()
    
    def list_versions(self):
        """List available model versions."""
        return [f.stem for f in self.models_dir.glob("*.joblib")]
    
    def load_version(self, version: str):
        """Load a specific model version."""
        model_path = self.models_dir / f"{version}.joblib"
        if not model_path.exists():
            raise ValueError(f"Model version {version} not found")
        
        with self.lock:
            self.current_model = joblib.load(model_path)
            self.current_version = version
        return True
    
    def predict(self, features):
        """Make prediction with current model."""
        with self.lock:
            if self.current_model is None:
                raise ValueError("No model loaded")
            return self.current_model.predict(features)

model_manager = ModelManager("models/")
model_manager.load_version("v1.0")

@app.get("/models/versions")
async def list_versions():
    return {"versions": model_manager.list_versions(), "current": model_manager.current_version}

@app.post("/models/switch/{version}")
async def switch_model(version: str):
    try:
        model_manager.load_version(version)
        return {"message": f"Switched to {version}", "current": model_manager.current_version}
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

import logging
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('api.log'),  # Write to file
        logging.StreamHandler()           # Also print to console
    ]
)

logger = logging.getLogger(__name__)

# Usage in your API
@app.route('/predict', methods=['POST'])
def predict():
    logger.info(f"Received prediction request from {request.remote_addr}")
    
    try:
        features = request.get_json()['features']
        prediction = model.predict(features)
        
        logger.info(f"Prediction successful: {prediction}")
        return jsonify({'prediction': prediction.tolist()})
        
    except Exception as e:
        logger.error(f"Prediction failed: {str(e)}", exc_info=True)
        return jsonify({'error': str(e)}), 500

# api.log will contain all logs with timestamps

import time
from functools import wraps

# Decorator to measure execution time
def track_latency(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        latency_ms = (time.time() - start) * 1000
        print(f"{func.__name__} took {latency_ms:.2f}ms")
        return result
    return wrapper

# Usage with Flask
@app.route('/predict', methods=['POST'])
def predict():
    start_time = time.time()
    
    # Your prediction code
    features = request.get_json()['features']
    prediction = model.predict(features)
    
    # Calculate and log latency
    latency_ms = (time.time() - start_time) * 1000
    
    return jsonify({
        'prediction': prediction.tolist(),
        'latency_ms': round(latency_ms, 2)
    })

# Track latencies over time
latencies = []

@app.route('/predict', methods=['POST'])
def predict():
    start = time.time()
    # ... prediction logic ...
    latency = (time.time() - start) * 1000
    latencies.append(latency)
    
    # Alert if latency is too high
    if latency > 100:  # 100ms threshold
        logging.warning(f"High latency detected: {latency:.2f}ms")

from collections import Counter
from datetime import datetime

class PredictionMonitor:
    """Track prediction distribution over time."""
    
    def __init__(self):
        self.hourly_counts = {}  # {hour: Counter}
        self.total_counts = Counter()
    
    def record(self, prediction):
        """Record a prediction."""
        hour = datetime.now().strftime('%Y-%m-%d %H:00')
        
        if hour not in self.hourly_counts:
            self.hourly_counts[hour] = Counter()
        
        self.hourly_counts[hour][prediction] += 1
        self.total_counts[prediction] += 1
    
    def get_distribution(self):
        """Get overall prediction distribution."""
        total = sum(self.total_counts.values())
        return {k: v/total for k, v in self.total_counts.items()}
    
    def check_drift(self, expected_distribution, threshold=0.1):
        """Check if current distribution differs from expected."""
        current = self.get_distribution()
        alerts = []
        
        for class_name, expected_pct in expected_distribution.items():
            actual_pct = current.get(class_name, 0)
            diff = abs(actual_pct - expected_pct)
            
            if diff > threshold:
                alerts.append(f"{class_name}: expected {expected_pct:.1%}, got {actual_pct:.1%}")
        
        return alerts

# Usage
monitor = PredictionMonitor()

@app.post("/predict")
async def predict(request: PredictionRequest):
    predictions = model.predict(...)
    
    for pred in predictions:
        monitor.record(pred)
    
    return {"predictions": predictions.tolist()}

@app.get("/monitoring/distribution")
async def get_distribution():
    return monitor.get_distribution()

import numpy as np
from collections import deque
from scipy import stats

class DriftDetector:
    """Detect when input data distribution shifts."""
    
    def __init__(self, reference_data, window_size=1000, threshold=0.05):
        self.reference_data = reference_data
        self.window = deque(maxlen=window_size)
        self.threshold = threshold
        self.feature_names = ['feature_1', 'feature_2', 'feature_3', 'feature_4']
    
    def add_sample(self, sample):
        """Add a new sample to the monitoring window."""
        self.window.append(sample)
    
    def check_drift(self):
        """Check if current data distribution differs from reference."""
        if len(self.window) < 100:
            return {"status": "insufficient_data", "samples": len(self.window)}
        
        current_data = np.array(self.window)
        drift_detected = {}
        
        for i, name in enumerate(self.feature_names):
            # Kolmogorov-Smirnov test
            stat, p_value = stats.ks_2samp(
                self.reference_data[:, i],
                current_data[:, i]
            )
            
            is_drift = p_value < self.threshold
            drift_detected[name] = {
                "p_value": round(p_value, 4),
                "drift": is_drift
            }
        
        any_drift = any(d["drift"] for d in drift_detected.values())
        
        return {
            "status": "drift_detected" if any_drift else "ok",
            "features": drift_detected
        }

# Initialize with training data
drift_detector = DriftDetector(X_train)

@app.post("/predict")
async def predict(request: PredictionRequest):
    features = np.array([s.to_array() for s in request.samples])
    
    # Record samples for drift detection
    for sample in features:
        drift_detector.add_sample(sample)
    
    return {"predictions": model.predict(features).tolist()}

@app.get("/monitoring/drift")
async def check_drift():
    return drift_detector.check_drift()