MLOps and Model Deployment

# app.py -- FastAPI ML prediction service
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import joblib
import numpy as np
from datetime import datetime

app = FastAPI(
    title="Sentiment Classifier API",
    description="Classifies text sentiment as positive or negative",
    version="1.0.0"
)

# Load model and vectorizer at startup
model = joblib.load("sentiment_model.joblib")
vectorizer = joblib.load("tfidf_vectorizer.joblib")

# Request/Response schemas
class TextRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000, 
                      description="Text to classify")

class PredictionResponse(BaseModel):
    text: str
    sentiment: str
    confidence: float
    model_version: str
    timestamp: str

class HealthResponse(BaseModel):
    status: str
    model_loaded: bool
    timestamp: str

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: TextRequest):
    try:
        # Vectorize text
        features = vectorizer.transform([request.text])
        
        # Predict
        prediction = model.predict(features)[0]
        proba = model.predict_proba(features)[0]
        confidence = float(max(proba))
        sentiment = "positive" if prediction == 1 else "negative"
        
        return PredictionResponse(
            text=request.text,
            sentiment=sentiment,
            confidence=round(confidence, 4),
            model_version="1.0.0",
            timestamp=datetime.now().isoformat()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health", response_model=HealthResponse)
async def health_check():
    return HealthResponse(
        status="healthy",
        model_loaded=model is not None,
        timestamp=datetime.now().isoformat()
    )

# Run with: uvicorn app:app --host 0.0.0.0 --port 8000
# Docs at: http://localhost:8000/docs
print("FastAPI app defined with /predict and /health endpoints")
print("Run: uvicorn app:app --host 0.0.0.0 --port 8000")

import pickle
import joblib
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
import os
import time

# Create a model
X, y = make_classification(n_samples=1000, n_features=20, random_state=42)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
print(f"Model accuracy: {model.score(X, y):.4f}")

# Method 1: pickle
start = time.time()
with open("model.pkl", "wb") as f:
    pickle.dump(model, f)
pickle_time = time.time() - start
pickle_size = os.path.getsize("model.pkl") / 1024

with open("model.pkl", "rb") as f:
    model_pkl = pickle.load(f)
print(f"\npickle: {pickle_size:.1f} KB, save time: {pickle_time:.4f}s")
print(f"  Predictions match: {np.array_equal(model.predict(X[:5]), model_pkl.predict(X[:5]))}")

# Method 2: joblib
start = time.time()
joblib.dump(model, "model.joblib")
joblib_time = time.time() - start
joblib_size = os.path.getsize("model.joblib") / 1024

model_jl = joblib.load("model.joblib")
print(f"\njoblib: {joblib_size:.1f} KB, save time: {joblib_time:.4f}s")
print(f"  Predictions match: {np.array_equal(model.predict(X[:5]), model_jl.predict(X[:5]))}")

# Method 3: joblib with compression
start = time.time()
joblib.dump(model, "model_compressed.joblib", compress=3)
comp_time = time.time() - start
comp_size = os.path.getsize("model_compressed.joblib") / 1024
print(f"\njoblib (compressed): {comp_size:.1f} KB, save time: {comp_time:.4f}s")
print(f"  Compression ratio: {joblib_size/comp_size:.1f}x")

# Cleanup
for f in ["model.pkl", "model.joblib", "model_compressed.joblib"]:
    os.remove(f)

# This example shows the Docker configuration files needed

# === Dockerfile ===
dockerfile_content = """FROM python:3.11-slim

WORKDIR /app

# Install system dependencies (if needed for numpy/scipy)
RUN apt-get update && apt-get install -y --no-install-recommends \\
    build-essential && rm -rf /var/lib/apt/lists/*

# Copy and install Python dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code and model
COPY app.py .
COPY model.joblib .
COPY vectorizer.joblib .

# Create non-root user for security
RUN useradd -m appuser && chown -R appuser /app
USER appuser

# Expose port and set healthcheck
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s \\
    CMD curl -f http://localhost:8000/health || exit 1

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

# === requirements.txt ===
requirements_content = """fastapi==0.109.0
uvicorn==0.27.0
joblib==1.3.2
scikit-learn==1.4.0
numpy==1.26.3
pydantic==2.5.3
"""

# === docker-compose.yml ===
compose_content = """version: '3.8'
services:
  ml-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/app/model.joblib
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 5s
      retries: 3
    restart: unless-stopped
"""

print("=== Dockerfile ===")
print(dockerfile_content)
print("=== requirements.txt ===")
print(requirements_content)
print("=== Docker Commands ===")
print("docker build -t ml-api .")
print("docker run -p 8000:8000 ml-api")
print("docker-compose up --build")
print("\nKey Docker concepts:")
print("  Image: Blueprint (read-only template)")
print("  Container: Running instance of an image")
print("  Layer caching: Dependencies installed before code for faster rebuilds")
print("  Non-root user: Security best practice")

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score

# Create dataset
X, y = make_classification(n_samples=2000, n_features=20, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Set experiment
mlflow.set_experiment("model-comparison")

# Define experiments
experiments = [
    ("LogisticRegression", LogisticRegression(max_iter=200)),
    ("RandomForest", RandomForestClassifier(n_estimators=100, random_state=42)),
    ("GradientBoosting", GradientBoostingClassifier(n_estimators=100, random_state=42))
]

results = []
for name, model in experiments:
    with mlflow.start_run(run_name=name):
        # Log model type
        mlflow.log_param("model_type", name)
        
        # Train
        model.fit(X_train, y_train)
        predictions = model.predict(X_test)
        
        # Calculate metrics
        accuracy = accuracy_score(y_test, predictions)
        f1 = f1_score(y_test, predictions)
        
        # Log metrics
        mlflow.log_metric("accuracy", accuracy)
        mlflow.log_metric("f1_score", f1)
        
        # Log model artifact
        mlflow.sklearn.log_model(model, "model")
        
        results.append((name, accuracy, f1))
        print(f"{name:25s} | Accuracy: {accuracy:.4f} | F1: {f1:.4f}")

# Summary
print("\n=== Experiment Summary ===")
best = max(results, key=lambda x: x[1])
print(f"Best model: {best[0]} (Accuracy: {best[1]:.4f})")
print("\nAll runs logged in MLflow. View with: mlflow ui")

import numpy as np
from scipy import stats

def detect_data_drift(reference_data, production_data, feature_names, alpha=0.05):
    """
    Detect data drift using Kolmogorov-Smirnov test.
    Compares the distribution of each feature between reference and production data.
    """
    results = []
    
    for i, name in enumerate(feature_names):
        ref_col = reference_data[:, i]
        prod_col = production_data[:, i]
        
        # KS test: null hypothesis = same distribution
        ks_stat, p_value = stats.ks_2samp(ref_col, prod_col)
        drift = p_value < alpha
        
        results.append({
            "feature": name,
            "ks_statistic": round(ks_stat, 4),
            "p_value": round(p_value, 4),
            "drift_detected": drift
        })
    
    return results

# Simulate training data (reference) and production data
np.random.seed(42)
n_samples = 1000

# Reference data (training distribution)
ref_data = np.column_stack([
    np.random.normal(0, 1, n_samples),      # Feature A: mean=0, std=1
    np.random.normal(5, 2, n_samples),      # Feature B: mean=5, std=2
    np.random.uniform(0, 10, n_samples),    # Feature C: uniform 0-10
])

# Production data (with drift on Feature B)
prod_data = np.column_stack([
    np.random.normal(0, 1, n_samples),      # Feature A: NO drift
    np.random.normal(7, 2.5, n_samples),    # Feature B: DRIFTED (mean 5->7, std 2->2.5)
    np.random.uniform(0, 10, n_samples),    # Feature C: NO drift
])

feature_names = ["Feature A", "Feature B", "Feature C"]
results = detect_data_drift(ref_data, prod_data, feature_names)

print("=== Data Drift Report ===")
print(f"{'Feature':12s} | {'KS Stat':8s} | {'P-Value':8s} | {'Drift?':8s}")
print("-" * 46)
for r in results:
    drift_str = "YES" if r["drift_detected"] else "No"
    print(f"{r['feature']:12s} | {r['ks_statistic']:8.4f} | {r['p_value']:8.4f} | {drift_str:8s}")

drifted = [r for r in results if r["drift_detected"]]
print(f"\nDrifted features: {len(drifted)}/{len(results)}")
if drifted:
    print("Action: Investigate drifted features and consider retraining the model.")

from flask import Flask, request, jsonify
import joblib
import numpy as np
import json
from datetime import datetime
import logging

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

app = Flask(__name__)

# Simulated model for demonstration
class SimpleModel:
    def predict(self, X):
        return (np.sum(X, axis=1) > 0).astype(int)
    def predict_proba(self, X):
        scores = 1 / (1 + np.exp(-np.sum(X, axis=1)))
        return np.column_stack([1 - scores, scores])

model = SimpleModel()

# Request logging
prediction_log = []

@app.route("/predict", methods=["POST"])
def predict():
    start_time = datetime.now()
    data = request.get_json()
    
    if "features" not in data:
        return jsonify({"error": "Missing 'features' in request"}), 400
    
    features = np.array(data["features"]).reshape(1, -1)
    prediction = int(model.predict(features)[0])
    proba = model.predict_proba(features)[0]
    confidence = float(max(proba))
    
    # Log the prediction
    latency_ms = (datetime.now() - start_time).total_seconds() * 1000
    log_entry = {
        "timestamp": start_time.isoformat(),
        "features": data["features"],
        "prediction": prediction,
        "confidence": confidence,
        "latency_ms": round(latency_ms, 2)
    }
    prediction_log.append(log_entry)
    logger.info(f"Prediction: {prediction} | Confidence: {confidence:.3f} | Latency: {latency_ms:.2f}ms")
    
    return jsonify({
        "prediction": prediction,
        "confidence": round(confidence, 4),
        "latency_ms": round(latency_ms, 2)
    })

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "healthy", "total_predictions": len(prediction_log)})

@app.route("/metrics", methods=["GET"])
def metrics():
    if not prediction_log:
        return jsonify({"message": "No predictions yet"})
    
    latencies = [log["latency_ms"] for log in prediction_log]
    return jsonify({
        "total_predictions": len(prediction_log),
        "avg_latency_ms": round(np.mean(latencies), 2),
        "p99_latency_ms": round(np.percentile(latencies, 99), 2),
        "avg_confidence": round(np.mean([log["confidence"] for log in prediction_log]), 4)
    })

print("Flask ML API with logging and metrics")
print("Endpoints: /predict (POST), /health (GET), /metrics (GET)")
print("Run: flask run --port 5000")