01 - MLOps: Da Esperimento a Produzione

    +----------+     +---------+     +----------+
    |   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  |
    +----------+     +---------+     +----------+

# Cella 1: Tutto nello stesso 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

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

# Feature engineering inline
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 - parametri hardcoded
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X_train, y_train)

# Valutazione - print a schermo
y_pred = model.predict(X_test)
print(f"Accuracy: {accuracy_score(y_test, y_pred)}")
print(f"F1: {f1_score(y_test, y_pred)}")

# Salvataggio - pickle senza versioning
with open("model.pkl", "wb") as f:
    pickle.dump(model, f)
print("Modello salvato!")

"""Modulo per la preparazione e trasformazione dei dati."""
import pandas as pd
from pathlib import Path
from typing import Tuple


def load_data(path: str) -> pd.DataFrame:
    """Carica il dataset dal path specificato."""
    filepath = Path(path)
    if not filepath.exists():
        raise FileNotFoundError(f"Dataset non trovato: {path}")
    return pd.read_csv(filepath)


def create_features(df: pd.DataFrame) -> pd.DataFrame:
    """Crea le feature derivate per il modello."""
    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 dei dati in train/test con seed fisso per riproducibilità."""
    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)

"""Modulo per il training e la valutazione del modello."""
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:
    """Addestra un RandomForestClassifier con parametri configurabili."""
    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]:
    """Valuta il modello e restituisce un dizionario di metriche."""
    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),
    }

"""Pipeline principale che orchestra tutte le fasi."""
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:
    """Esegue l'intera pipeline ML con configurazione esterna."""
    # 1. Carica configurazione
    with open(config_path) as f:
        config = yaml.safe_load(f)

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

    # 3. Split
    print("[2/4] Split 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 modello...")
    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] Valutazione...")
    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 - Tutti i parametri in un unico 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

# Installazione
pip install mlflow

# Avvio del tracking server locale
mlflow server --host 127.0.0.1 --port 5000

"""Pipeline ML con 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:
    """Esegue la pipeline tracciando tutto con MLflow."""

    # Imposta il tracking URI (server locale o remoto)
    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 dei parametri
        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 dimensioni dataset
        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 delle metriche
        for name, value in metrics.items():
            mlflow.log_metric(name, value)

        # Log del modello come artefatto
        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            registered_model_name="churn-classifier"
        )

        # Log della configurazione come artefatto
        mlflow.log_artifact("config.yaml")

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

"""Gestione del ciclo di vita del modello con MLflow Model Registry."""
from mlflow.tracking import MlflowClient

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

# Recupera l'ultima versione del modello in staging
latest_versions = client.get_latest_versions(
    name="churn-classifier",
    stages=["Staging"]
)

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

    # Promuovi a Production dopo validazione
    client.transition_model_version_stage(
        name="churn-classifier",
        version=version,
        stage="Production",
        archive_existing_versions=True  # Archivia versione precedente
    )
    print(f"Modello v{version} promosso a Production")

# Carica il modello in produzione per inference
import mlflow.pyfunc

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

"""API REST per servire predizioni del modello ML."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import mlflow.pyfunc
import pandas as pd
import logging
from typing import List

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

app = FastAPI(
    title="Churn Prediction API",
    description="API per predizioni di churn basata su ML",
    version="1.0.0"
)


class PredictionRequest(BaseModel):
    """Schema della richiesta di predizione."""
    age: int = Field(..., ge=0, le=120, description="Eta del cliente")
    total_spend: float = Field(..., ge=0, description="Spesa totale")
    visits: int = Field(..., ge=0, description="Numero visite")
    days_since_last: int = Field(..., ge=0, description="Giorni dall'ultima visita")


class PredictionResponse(BaseModel):
    """Schema della risposta di predizione."""
    prediction: int
    probability: float
    model_version: str


# Caricamento modello all'avvio
MODEL_NAME = "churn-classifier"
MODEL_STAGE = "Production"
model = None
model_version = "unknown"


@app.on_event("startup")
async def load_model():
    """Carica il modello MLflow all'avvio del server."""
    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"Modello caricato: {MODEL_NAME} ({model_version})")
    except Exception as e:
        logger.error(f"Errore caricamento modello: {e}")
        raise


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


@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """Genera una predizione di churn per un cliente."""
    if model is None:
        raise HTTPException(status_code=503, detail="Modello non caricato")

    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"Errore predizione: {e}")
        raise HTTPException(status_code=500, detail="Errore nella predizione")


@app.post("/predict/batch", response_model=List[PredictionResponse])
async def predict_batch(requests: List[PredictionRequest]):
    """Genera predizioni batch per più clienti."""
    if model is None:
        raise HTTPException(status_code=503, detail="Modello non caricato")

    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 per il serving del modello ML
FROM python:3.11-slim

WORKDIR /app

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

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

# Codice applicazione
COPY src/serving/ ./serving/
COPY config.yaml .

# Porta del servizio
EXPOSE 8000

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

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

# Build dell'immagine
docker build -t churn-api:v1.0.0 .

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

# Test dell'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}'

"""Rilevamento data drift con test statistici."""
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]:
    """
    Rileva data drift confrontando distribuzioni con il test KS.

    Args:
        reference_data: dati di training (riferimento)
        production_data: dati di produzione (attuali)
        feature_names: nomi delle feature
        threshold: soglia p-value per il drift (default 0.05)

    Returns:
        Report di drift per ogni feature
    """
    drift_report = {}

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

        # Test Kolmogorov-Smirnov
        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 RILEVATO su '{feature}': "
                  f"KS={ks_stat:.4f}, p={p_value:.4f}")

    return drift_report

churn-prediction/
  data/
    raw/                  # Dati grezzi (versionati con DVC)
    processed/            # Dati trasformati
    data.dvc              # File di tracking DVC
  src/
    data/
      preprocessing.py    # Pulizia e feature engineering
      validation.py       # Validazione qualità dati
    models/
      trainer.py          # Logica di training
      evaluator.py        # Valutazione e metriche
    serving/
      app.py              # FastAPI application
      schemas.py          # Pydantic schemas
    monitoring/
      drift_detector.py   # Rilevamento drift
      metrics.py          # Metriche custom
    pipeline.py           # Orchestrazione pipeline
  tests/
    test_preprocessing.py
    test_trainer.py
    test_api.py
  config.yaml             # Configurazione pipeline
  Dockerfile              # Container per serving
  docker-compose.yaml     # Stack locale completo
  requirements.txt        # Dipendenze Python
  .dvc/                   # Configurazione DVC
  .github/
    workflows/
      train.yaml          # CI/CD per training
      deploy.yaml         # CI/CD per deployment
  mlflow/                 # Artefatti MLflow (locale)
  README.md