01 - MLOps 101: From Experiment to Production

    +----------+     +---------+     +----------+
    |   DATA   |---->|  TRAIN  |---->| EVALUATE |
    | Collect  |     | Feature |     | Validate |
    | Clean    |     | Train   |     | Compare  |
    | Version  |     | Tune    |     | Approve  |
    +----------+     +---------+     +----------+
         ^                                |
         |                                v
    +----------+     +---------+     +----------+
    | RETRAIN  |<----| MONITOR |<----| DEPLOY   |
    | Trigger  |     | Drift   |     | Stage    |
    | Schedule |     | Metrics |     | Canary   |
    | Auto     |     | Alert   |     | Release  |
    +----------+     +---------+     +----------+

# Cell 1: Everything in one notebook
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import pickle

# Load data
df = pd.read_csv("data/customers.csv")

# Inline feature engineering
df["age_group"] = pd.cut(df["age"], bins=[0, 25, 45, 65, 100],
                          labels=["young", "adult", "senior", "elderly"])
df["total_spend"] = df["orders"] * df["avg_order_value"]

# Split
X = df[["age", "total_spend", "visits", "days_since_last"]]
y = df["churned"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Training - hardcoded parameters
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X_train, y_train)

# Evaluation - print to screen
y_pred = model.predict(X_test)
print(f"Accuracy: {accuracy_score(y_test, y_pred)}")
print(f"F1: {f1_score(y_test, y_pred)}")

# Save - pickle with no versioning
with open("model.pkl", "wb") as f:
    pickle.dump(model, f)
print("Model saved!")

"""Module for data preparation and transformation."""
import pandas as pd
from pathlib import Path
from typing import Tuple


def load_data(path: str) -> pd.DataFrame:
    """Load the dataset from the specified path."""
    filepath = Path(path)
    if not filepath.exists():
        raise FileNotFoundError(f"Dataset not found: {path}")
    return pd.read_csv(filepath)


def create_features(df: pd.DataFrame) -> pd.DataFrame:
    """Create derived features for the model."""
    result = df.copy()
    result["age_group"] = pd.cut(
        result["age"],
        bins=[0, 25, 45, 65, 100],
        labels=["young", "adult", "senior", "elderly"]
    )
    result["total_spend"] = result["orders"] * result["avg_order_value"]
    return result


def split_data(
    df: pd.DataFrame,
    target_col: str = "churned",
    feature_cols: list = None,
    test_size: float = 0.2,
    random_state: int = 42
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
    """Split data into train/test with a fixed seed for reproducibility."""
    from sklearn.model_selection import train_test_split

    if feature_cols is None:
        feature_cols = ["age", "total_spend", "visits", "days_since_last"]

    X = df[feature_cols]
    y = df[target_col]
    return train_test_split(X, y, test_size=test_size, random_state=random_state)

"""Module for model training and evaluation."""
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from typing import Dict, Any
import pandas as pd


def train_model(
    X_train: pd.DataFrame,
    y_train: pd.Series,
    n_estimators: int = 100,
    max_depth: int = 10,
    random_state: int = 42
) -> RandomForestClassifier:
    """Train a RandomForestClassifier with configurable parameters."""
    model = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        random_state=random_state
    )
    model.fit(X_train, y_train)
    return model


def evaluate_model(
    model: RandomForestClassifier,
    X_test: pd.DataFrame,
    y_test: pd.Series
) -> Dict[str, float]:
    """Evaluate the model and return a metrics dictionary."""
    y_pred = model.predict(X_test)
    return {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1_score": f1_score(y_test, y_pred),
        "precision": precision_score(y_test, y_pred),
        "recall": recall_score(y_test, y_pred),
    }

"""Main pipeline orchestrating all stages."""
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model
import yaml
from pathlib import Path


def run_pipeline(config_path: str = "config.yaml") -> None:
    """Run the full ML pipeline using an external config."""
    # 1. Load configuration
    with open(config_path) as f:
        config = yaml.safe_load(f)

    # 2. Data preparation
    print("[1/4] Loading data...")
    df = load_data(config["data"]["path"])
    df = create_features(df)

    # 3. Split
    print("[2/4] Splitting train/test...")
    X_train, X_test, y_train, y_test = split_data(
        df,
        test_size=config["data"]["test_size"],
        random_state=config["data"]["random_state"]
    )

    # 4. Training
    print("[3/4] Training model...")
    model = train_model(
        X_train, y_train,
        n_estimators=config["model"]["n_estimators"],
        max_depth=config["model"]["max_depth"],
        random_state=config["model"]["random_state"]
    )

    # 5. Evaluation
    print("[4/4] Evaluating...")
    metrics = evaluate_model(model, X_test, y_test)
    for name, value in metrics.items():
        print(f"  {name}: {value:.4f}")


if __name__ == "__main__":
    run_pipeline()

# config.yaml - All parameters in one file
data:
  path: "data/customers.csv"
  test_size: 0.2
  random_state: 42
  feature_cols:
    - age
    - total_spend
    - visits
    - days_since_last

model:
  algorithm: "random_forest"
  n_estimators: 100
  max_depth: 10
  random_state: 42

evaluation:
  metrics:
    - accuracy
    - f1_score
    - precision
    - recall
  min_accuracy: 0.85

# Installation
pip install mlflow

# Start a local tracking server
mlflow server --host 127.0.0.1 --port 5000

"""ML pipeline with experiment tracking via MLflow."""
import mlflow
import mlflow.sklearn
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model


def run_tracked_pipeline(config: dict) -> None:
    """Run the pipeline while tracking everything with MLflow."""

    # Set the tracking URI (local or remote server)
    mlflow.set_tracking_uri("http://127.0.0.1:5000")
    mlflow.set_experiment("churn-prediction")

    with mlflow.start_run(run_name="rf-baseline") as run:
        # Log parameters
        mlflow.log_param("algorithm", "RandomForest")
        mlflow.log_param("n_estimators", config["model"]["n_estimators"])
        mlflow.log_param("max_depth", config["model"]["max_depth"])
        mlflow.log_param("test_size", config["data"]["test_size"])
        mlflow.log_param("random_state", config["data"]["random_state"])

        # Data preparation
        df = load_data(config["data"]["path"])
        df = create_features(df)
        X_train, X_test, y_train, y_test = split_data(
            df,
            test_size=config["data"]["test_size"],
            random_state=config["data"]["random_state"]
        )

        # Log dataset dimensions
        mlflow.log_param("train_samples", len(X_train))
        mlflow.log_param("test_samples", len(X_test))
        mlflow.log_param("n_features", X_train.shape[1])

        # Training
        model = train_model(
            X_train, y_train,
            n_estimators=config["model"]["n_estimators"],
            max_depth=config["model"]["max_depth"]
        )

        # Evaluation
        metrics = evaluate_model(model, X_test, y_test)

        # Log metrics
        for name, value in metrics.items():
            mlflow.log_metric(name, value)

        # Log the model as an artifact
        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            registered_model_name="churn-classifier"
        )

        # Log the config as an artifact
        mlflow.log_artifact("config.yaml")

        print(f"Run ID: {run.info.run_id}")
        print(f"Metrics: {metrics}")

"""Model lifecycle management with MLflow Model Registry."""
from mlflow.tracking import MlflowClient

client = MlflowClient("http://127.0.0.1:5000")

# Retrieve the latest staging version
latest_versions = client.get_latest_versions(
    name="churn-classifier",
    stages=["Staging"]
)

if latest_versions:
    version = latest_versions[0].version
    print(f"Model in staging: v{version}")

    # Promote to Production after validation
    client.transition_model_version_stage(
        name="churn-classifier",
        version=version,
        stage="Production",
        archive_existing_versions=True  # Archive previous version
    )
    print(f"Model v{version} promoted to Production")

# Load the production model for inference
import mlflow.pyfunc

model = mlflow.pyfunc.load_model("models:/churn-classifier/Production")
prediction = model.predict(new_data)

"""REST API for serving ML model predictions."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import mlflow.pyfunc
import pandas as pd
import logging
from typing import List

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

app = FastAPI(
    title="Churn Prediction API",
    description="ML-based churn prediction API",
    version="1.0.0"
)


class PredictionRequest(BaseModel):
    """Prediction request schema."""
    age: int = Field(..., ge=0, le=120, description="Customer age")
    total_spend: float = Field(..., ge=0, description="Total spending")
    visits: int = Field(..., ge=0, description="Number of visits")
    days_since_last: int = Field(..., ge=0, description="Days since last visit")


class PredictionResponse(BaseModel):
    """Prediction response schema."""
    prediction: int
    probability: float
    model_version: str


# Load model at startup
MODEL_NAME = "churn-classifier"
MODEL_STAGE = "Production"
model = None
model_version = "unknown"


@app.on_event("startup")
async def load_model():
    """Load the MLflow model on server startup."""
    global model, model_version
    try:
        model_uri = f"models:/{MODEL_NAME}/{MODEL_STAGE}"
        model = mlflow.pyfunc.load_model(model_uri)
        model_version = model.metadata.run_id[:8]
        logger.info(f"Model loaded: {MODEL_NAME} ({model_version})")
    except Exception as e:
        logger.error(f"Model loading error: {e}")
        raise


@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "model_loaded": model is not None}


@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """Generate a churn prediction for a customer."""
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    try:
        input_data = pd.DataFrame([request.model_dump()])
        prediction = model.predict(input_data)
        probability = float(prediction[0]) if hasattr(prediction[0], '__float__') else 0.0

        return PredictionResponse(
            prediction=int(prediction[0]),
            probability=probability,
            model_version=model_version
        )
    except Exception as e:
        logger.error(f"Prediction error: {e}")
        raise HTTPException(status_code=500, detail="Prediction error")


@app.post("/predict/batch", response_model=List[PredictionResponse])
async def predict_batch(requests: List[PredictionRequest]):
    """Generate batch predictions for multiple customers."""
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    input_data = pd.DataFrame([r.model_dump() for r in requests])
    predictions = model.predict(input_data)

    return [
        PredictionResponse(
            prediction=int(p),
            probability=float(p),
            model_version=model_version
        )
        for p in predictions
    ]

# Dockerfile for ML model serving
FROM python:3.11-slim

WORKDIR /app

# System dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

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

# Application code
COPY src/serving/ ./serving/
COPY config.yaml .

# Service port
EXPOSE 8000

# Healthcheck
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Start with uvicorn
CMD ["uvicorn", "serving.app:app", "--host", "0.0.0.0", "--port", "8000"]

# Build the image
docker build -t churn-api:v1.0.0 .

# Start the container
docker run -d \
  --name churn-api \
  -p 8000:8000 \
  -e MLFLOW_TRACKING_URI=http://mlflow-server:5000 \
  churn-api:v1.0.0

# Test the API
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"age": 35, "total_spend": 1250.50, "visits": 12, "days_since_last": 45}'

"""Data drift detection using statistical tests."""
import numpy as np
from scipy import stats
from typing import Dict, Tuple


def detect_drift(
    reference_data: np.ndarray,
    production_data: np.ndarray,
    feature_names: list,
    threshold: float = 0.05
) -> Dict[str, Dict]:
    """
    Detect data drift by comparing distributions with the KS test.

    Args:
        reference_data: training data (reference)
        production_data: production data (current)
        feature_names: feature names
        threshold: p-value threshold for drift (default 0.05)

    Returns:
        Drift report for each feature
    """
    drift_report = {}

    for i, feature in enumerate(feature_names):
        ref_values = reference_data[:, i]
        prod_values = production_data[:, i]

        # Kolmogorov-Smirnov test
        ks_stat, p_value = stats.ks_2samp(ref_values, prod_values)

        drift_detected = p_value < threshold
        drift_report[feature] = {
            "ks_statistic": round(ks_stat, 4),
            "p_value": round(p_value, 4),
            "drift_detected": drift_detected,
            "ref_mean": round(float(np.mean(ref_values)), 4),
            "prod_mean": round(float(np.mean(prod_values)), 4),
        }

        if drift_detected:
            print(f"DRIFT DETECTED on '{feature}': "
                  f"KS={ks_stat:.4f}, p={p_value:.4f}")

    return drift_report

churn-prediction/
  data/
    raw/                  # Raw data (versioned with DVC)
    processed/            # Transformed data
    data.dvc              # DVC tracking file
  src/
    data/
      preprocessing.py    # Cleaning and feature engineering
      validation.py       # Data quality validation
    models/
      trainer.py          # Training logic
      evaluator.py        # Evaluation and metrics
    serving/
      app.py              # FastAPI application
      schemas.py          # Pydantic schemas
    monitoring/
      drift_detector.py   # Drift detection
      metrics.py          # Custom metrics
    pipeline.py           # Pipeline orchestration
  tests/
    test_preprocessing.py
    test_trainer.py
    test_api.py
  config.yaml             # Pipeline configuration
  Dockerfile              # Serving container
  docker-compose.yaml     # Full local stack
  requirements.txt        # Python dependencies
  .dvc/                   # DVC configuration
  .github/
    workflows/
      train.yaml          # CI/CD for training
      deploy.yaml         # CI/CD for deployment
  mlflow/                 # MLflow artifacts (local)
  README.md