Narcis Miclaus

9 min di lettura

Python, dalle fondamenta Lezione 54 / 60

ML project: un problema di classificazione, end to end

Da CSV grezzo a modello in produzione: le lezioni del Modulo 9 rese tangibili.

Un modello che ottiene 0.92 di AUC in un Jupyter notebook e non lascia mai il notebook vale zero. Un modello che ottiene 0.85 di AUC ed è dietro un endpoint HTTP che i sistemi reali possono chiamare vale soldi veri. La distanza tra i due è dove si bloccano la maggior parte dei data scientist, ed è dove si vede davvero l’engineering nel machine learning engineering.

Questa lezione è la conclusione pratica del Modulo 9. Prendiamo un vero problema di classificazione binaria, customer churn, da CSV grezzo a servizio di predizione in produzione. Il dataset è l’IBM Telco Customer Churn dataset, disponibile gratuitamente su Kaggle e in molti repo replicati. Circa 7.000 clienti, una ventina di feature, una colonna binaria Churn. La stessa forma di problema che vedresti in fraud detection, default sul credito, conversion prediction, o qualsiasi domanda del tipo “questo utente farà X?”.

Ogni step mantiene deliberatamente il codice diretto. Il punto non è massimizzare lo score di classifica; è dimostrare il workflow end-to-end con codice che saresti davvero disposto a mandare in produzione.

Step 1: load and explore

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

df = pd.read_csv("telco_churn.csv")
print(df.shape)
print(df.dtypes)
print(df["Churn"].value_counts(normalize=True))

Prime domande: qual è la distribuzione del target? Quali colonne sono numeriche e quali categoriche? Ci sono valori mancanti?

print(df.isna().sum())

# TotalCharges is read as object — common gotcha, has empty strings for new customers
df["TotalCharges"] = pd.to_numeric(df["TotalCharges"], errors="coerce")
print(df["TotalCharges"].isna().sum())  # ~11 rows
df = df.dropna(subset=["TotalCharges"])

df["Churn"] = (df["Churn"] == "Yes").astype(int)
df = df.drop(columns=["customerID"])

Il churn rate è circa 27%: lo squilibrio di classi è lieve ma reale. Vale la pena saperlo per la stratificazione più avanti. Un rapido controllo visivo:

fig, axes = plt.subplots(1, 2, figsize=(10, 4))
df["tenure"].hist(by=df["Churn"], bins=30, ax=axes)
axes[0].set_title("tenure | Churn=0")
axes[1].set_title("tenure | Churn=1")
plt.tight_layout()
plt.show()

I clienti che fanno churn hanno una tenure molto più breve. Prior utile: ci dice che un albero probabilmente farà split su tenure presto.

Step 2: feature engineering

La maggior parte delle colonne è categorica con due o tre livelli. Qualche feature derivata che la conoscenza del dominio suggerisce:

df["AvgChargePerMonth"] = df["TotalCharges"] / df["tenure"].replace(0, 1)
df["IsLongTermContract"] = (df["Contract"] != "Month-to-month").astype(int)
df["NumServices"] = (
    (df["PhoneService"] == "Yes").astype(int)
    + (df["MultipleLines"] == "Yes").astype(int)
    + (df["InternetService"] != "No").astype(int)
    + (df["OnlineSecurity"] == "Yes").astype(int)
    + (df["OnlineBackup"] == "Yes").astype(int)
    + (df["DeviceProtection"] == "Yes").astype(int)
    + (df["TechSupport"] == "Yes").astype(int)
    + (df["StreamingTV"] == "Yes").astype(int)
    + (df["StreamingMovies"] == "Yes").astype(int)
)

Queste sono ipotesi di dominio: i clienti che pagano molto al mese rispetto al loro storico potrebbero fare churn, il tipo di contratto conta, la profondità totale dei servizi conta. Lasceremo decidere al modello se sono utili.

Per lo step di modellazione vogliamo una netta separazione tra categoriche e numeriche, così possiamo costruire un column transformer:

y = df["Churn"]
X = df.drop(columns=["Churn"])

categorical = X.select_dtypes(include="object").columns.tolist()
numeric = X.select_dtypes(exclude="object").columns.tolist()
print(f"{len(categorical)} categorical, {len(numeric)} numeric")

Step 3: stratified split

Split a tre vie: train, validation (per early stopping durante il tuning), test (toccato una sola volta, alla fine).

from sklearn.model_selection import train_test_split

X_temp, X_test, y_temp, y_test = train_test_split(
    X, y, test_size=0.15, stratify=y, random_state=42
)
X_train, X_val, y_train, y_val = train_test_split(
    X_temp, y_temp, test_size=0.15 / 0.85, stratify=y_temp, random_state=42
)
print(X_train.shape, X_val.shape, X_test.shape)

stratify=y mantiene il churn rate consistente fra gli split. Con classi sbilanciate non è opzionale: senza, puoi finire con un test set con un positive rate notevolmente diverso e la tua valutazione diventa rumorosa.

Step 4: baseline, logistic regression

La baseline rituale. Costruisci una Pipeline così che preprocessing e modello siano accoppiati:

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score, classification_report

preprocess = ColumnTransformer([
    ("num", StandardScaler(), numeric),
    ("cat", OneHotEncoder(handle_unknown="ignore", drop="if_binary"), categorical),
])

baseline = Pipeline([
    ("pre", preprocess),
    ("clf", LogisticRegression(max_iter=1000, class_weight="balanced")),
])

baseline.fit(X_train, y_train)
val_proba = baseline.predict_proba(X_val)[:, 1]
print(f"Baseline AUC = {roc_auc_score(y_val, val_proba):.3f}")

Nota class_weight="balanced": utile per il lieve squilibrio, anche se per un positive rate del 27% puoi anche lasciarlo a default. Il drop="if_binary" sull’OneHotEncoder elimina la colonna ridondante sulle feature yes/no.

AUC tipica su questo dataset: intorno a 0.84. Quello è il numero che ogni altro modello deve battere.

Step 5: modelli migliori con default sensati

from xgboost import XGBClassifier
from lightgbm import LGBMClassifier

xgb = Pipeline([
    ("pre", preprocess),
    ("clf", XGBClassifier(
        n_estimators=500,
        max_depth=6,
        learning_rate=0.05,
        subsample=0.8,
        colsample_bytree=0.8,
        random_state=42,
        eval_metric="logloss",
    )),
])
xgb.fit(X_train, y_train)
print(f"XGB AUC = {roc_auc_score(y_val, xgb.predict_proba(X_val)[:, 1]):.3f}")

lgbm = Pipeline([
    ("pre", preprocess),
    ("clf", LGBMClassifier(
        n_estimators=500,
        learning_rate=0.05,
        num_leaves=31,
        random_state=42,
    )),
])
lgbm.fit(X_train, y_train)
print(f"LGBM AUC = {roc_auc_score(y_val, lgbm.predict_proba(X_val)[:, 1]):.3f}")

Su Telco, entrambi gli ensemble di solito fanno score di 0.85-0.86: solo un pelo sopra la baseline lineare. Questo è, tra l’altro, uno dei pattern di cui parlava la Lezione 52: su un problema dominato da segnali additivi, i modelli lineari regolarizzati reggono.

Per scopo pedagogico, fingiamo che il gap conti e facciamo il tuning.

Step 6: tuning con Optuna

import optuna
from sklearn.model_selection import StratifiedKFold, cross_val_score

def objective(trial):
    params = {
        "n_estimators": trial.suggest_int("n_estimators", 200, 1000),
        "max_depth": trial.suggest_int("max_depth", 3, 10),
        "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.2, log=True),
        "subsample": trial.suggest_float("subsample", 0.6, 1.0),
        "colsample_bytree": trial.suggest_float("colsample_bytree", 0.6, 1.0),
        "reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10.0, log=True),
        "reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10.0, log=True),
    }
    pipe = Pipeline([
        ("pre", preprocess),
        ("clf", XGBClassifier(**params, random_state=42, eval_metric="logloss")),
    ])
    cv = StratifiedKFold(n_splits=3, shuffle=True, random_state=42)
    scores = cross_val_score(pipe, X_train, y_train, scoring="roc_auc", cv=cv, n_jobs=1)
    return scores.mean()

study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=40, show_progress_bar=True)

print("Best AUC (CV):", study.best_value)
print("Best params:", study.best_params)

best_xgb = Pipeline([
    ("pre", preprocess),
    ("clf", XGBClassifier(**study.best_params, random_state=42, eval_metric="logloss")),
])
best_xgb.fit(X_train, y_train)
val_auc = roc_auc_score(y_val, best_xgb.predict_proba(X_val)[:, 1])
print(f"Tuned XGB val AUC = {val_auc:.3f}")

Quaranta trial sono sufficienti per vedere un miglioramento significativo. Su questo dataset potresti guadagnare altri 0.005-0.01 di AUC. Vale la pena? Dipende dal business. Il punto è il workflow.

Step 7: valuta in modo onesto

Ora il test set, toccato per la prima volta:

from sklearn.metrics import (
    confusion_matrix, classification_report,
    precision_recall_curve, roc_curve, auc,
)

test_proba = best_xgb.predict_proba(X_test)[:, 1]
test_pred = (test_proba > 0.5).astype(int)

print(f"Test AUC = {roc_auc_score(y_test, test_proba):.3f}")
print(classification_report(y_test, test_pred))
print(confusion_matrix(y_test, test_pred))

L’AUC da sola nasconde dettagli operativamente importanti. La confusion matrix e precision/recall per classe ti dicono che tipo di errori fa il modello. Per il churn, i falsi negativi (abbiamo detto “non farà churn”, il cliente ha fatto churn) sono di solito più costosi dei falsi positivi (abbiamo detto “farà churn”, il cliente no, abbiamo sprecato un coupon di retention).

Quell’asimmetria suggerisce di tunare la soglia di classificazione lontano da 0.5:

prec, rec, thresh = precision_recall_curve(y_test, test_proba)
fig, ax = plt.subplots(figsize=(8, 4))
ax.plot(thresh, prec[:-1], label="precision")
ax.plot(thresh, rec[:-1], label="recall")
ax.set_xlabel("threshold")
ax.legend()
ax.grid(True)
plt.show()

Leggi questa curva e scegli una soglia che corrisponde al tuo rapporto di costo business. Magari 0.35: soglia più bassa, più predizioni di churn, recall più alto, precision più bassa. Accettabile se una chiamata di retention costa 5 dollari e un cliente salvato ne vale 200.

Step 8: interpreta con SHAP

Gli stakeholder chiederanno “perché il modello ha segnalato questo cliente?”. I valori SHAP ti danno una decomposizione per riga.

import shap

# Pull the trained classifier out of the pipeline
clf = best_xgb.named_steps["clf"]
X_train_transformed = best_xgb.named_steps["pre"].transform(X_train)
X_test_transformed = best_xgb.named_steps["pre"].transform(X_test)

# Recover feature names from the ColumnTransformer
feat_names = best_xgb.named_steps["pre"].get_feature_names_out()

explainer = shap.TreeExplainer(clf)
shap_values = explainer.shap_values(X_test_transformed)

# Global view
shap.summary_plot(shap_values, X_test_transformed, feature_names=feat_names)

# Single-customer explanation
i = 0
shap.force_plot(
    explainer.expected_value,
    shap_values[i],
    X_test_transformed[i],
    feature_names=feat_names,
    matplotlib=True,
)

Il summary plot ordina le feature per quanto muovono in media le predizioni. Il force plot per un singolo cliente ti dice “la probabilità di churn di questo cliente è 0.74 perché contract=month-to-month l’ha spinta di +0.18, tenure=3 l’ha spinta di +0.12, …”. Quella seconda vista è ciò che metti davanti a un team di prevenzione churn: gli dice quale leva tirare.

Step 9: salva la pipeline

L’intera pipeline, preprocessing più modello, salvata come un singolo oggetto. Quello è il punto di Pipeline.

import joblib
import json
from datetime import date

# Refit on train+val for the final production model
X_full = pd.concat([X_train, X_val])
y_full = pd.concat([y_train, y_val])
final_model = Pipeline([
    ("pre", preprocess),
    ("clf", XGBClassifier(**study.best_params, random_state=42, eval_metric="logloss")),
])
final_model.fit(X_full, y_full)

joblib.dump(final_model, "churn_model_v1.joblib")

# Schema documentation — the contract for callers
schema = {
    "model_version": "v1",
    "trained_on": str(date.today()),
    "test_auc": round(float(roc_auc_score(y_test, test_proba)), 3),
    "input_columns": {
        col: str(X[col].dtype) for col in X.columns
    },
    "categorical_levels": {
        col: sorted(X[col].dropna().unique().tolist()) for col in categorical
    },
}
with open("churn_model_v1.schema.json", "w") as f:
    json.dump(schema, f, indent=2, default=str)

Il file di schema è critico. Senza di esso, fra sei mesi nessuno si ricorderà quali colonne questo modello si aspetta, in che ordine, o quali valori sono validi per le categoriche. Salva il contratto.

Step 10: endpoint di serving minimo

Un servizio FastAPI che carica la pipeline e predice su un record JSON inviato:

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Literal, Optional
import pandas as pd
import joblib

app = FastAPI(title="Churn Predictor v1")
model = joblib.load("churn_model_v1.joblib")

class CustomerFeatures(BaseModel):
    gender: Literal["Male", "Female"]
    SeniorCitizen: int
    Partner: Literal["Yes", "No"]
    Dependents: Literal["Yes", "No"]
    tenure: int
    PhoneService: Literal["Yes", "No"]
    MultipleLines: Literal["Yes", "No", "No phone service"]
    InternetService: Literal["DSL", "Fiber optic", "No"]
    OnlineSecurity: Literal["Yes", "No", "No internet service"]
    OnlineBackup: Literal["Yes", "No", "No internet service"]
    DeviceProtection: Literal["Yes", "No", "No internet service"]
    TechSupport: Literal["Yes", "No", "No internet service"]
    StreamingTV: Literal["Yes", "No", "No internet service"]
    StreamingMovies: Literal["Yes", "No", "No internet service"]
    Contract: Literal["Month-to-month", "One year", "Two year"]
    PaperlessBilling: Literal["Yes", "No"]
    PaymentMethod: str
    MonthlyCharges: float
    TotalCharges: float

class Prediction(BaseModel):
    churn_probability: float
    will_churn: bool
    threshold: float = 0.35

@app.post("/predict", response_model=Prediction)
def predict(features: CustomerFeatures):
    df = pd.DataFrame([features.dict()])
    df["AvgChargePerMonth"] = df["TotalCharges"] / df["tenure"].replace(0, 1)
    df["IsLongTermContract"] = (df["Contract"] != "Month-to-month").astype(int)
    df["NumServices"] = (
        (df["PhoneService"] == "Yes").astype(int)
        + (df["MultipleLines"] == "Yes").astype(int)
        + (df["InternetService"] != "No").astype(int)
        + (df["OnlineSecurity"] == "Yes").astype(int)
        + (df["OnlineBackup"] == "Yes").astype(int)
        + (df["DeviceProtection"] == "Yes").astype(int)
        + (df["TechSupport"] == "Yes").astype(int)
        + (df["StreamingTV"] == "Yes").astype(int)
        + (df["StreamingMovies"] == "Yes").astype(int)
    )
    try:
        proba = float(model.predict_proba(df)[0, 1])
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    return Prediction(
        churn_probability=proba,
        will_churn=proba > 0.35,
    )

@app.get("/health")
def health():
    return {"status": "ok", "model_version": "v1"}

Eseguilo con:

uv add fastapi uvicorn pydantic joblib pandas xgboost scikit-learn
uvicorn server:app --reload --port 8000

Test:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"gender":"Female","SeniorCitizen":0,"Partner":"Yes","Dependents":"No","tenure":1,"PhoneService":"No","MultipleLines":"No phone service","InternetService":"DSL","OnlineSecurity":"No","OnlineBackup":"Yes","DeviceProtection":"No","TechSupport":"No","StreamingTV":"No","StreamingMovies":"No","Contract":"Month-to-month","PaperlessBilling":"Yes","PaymentMethod":"Electronic check","MonthlyCharges":29.85,"TotalCharges":29.85}'

Riceverai indietro un JSON con churn_probability e un booleano. Quello è un modello deployabile.

Alcune cose che aggiungeresti per produzione e che abbiamo lasciato fuori per brevità: validazione dell’input contro lo schema salvato, structured logging di ogni predizione (input + output) per monitorare il drift, un endpoint Prometheus /metrics, tracing a livello di richiesta, e la stessa funzione di feature engineering importata da un modulo condiviso invece che copiata fra training e serving. L’ultima è la più importante. Il train/serve skew, le tue feature di training e di serving sono calcolate da codice diverso che si discosta in modo sottile, è la causa singola più comune di “il modello andava benissimo in valutazione, perché in produzione fa schifo?”. Fattorizza sempre la feature engineering in una funzione che sia il notebook di training sia l’endpoint di serving importano.

Cosa hai messo insieme

Quello è un progetto completo. Hai caricato dati grezzi, li hai puliti, fatto feature engineering, splittato con stratificazione, fatto il fit di una baseline, fatto il fit di modelli migliori, tunato con Optuna, valutato su un test set held-out, interpretato con SHAP, salvato la pipeline e il suo schema, e tirato su un endpoint HTTP. Ogni step corrisponde a uno strumento del Modulo 9.

La cosa che nessuno dice ai data scientist junior: la modellazione vera e propria è forse il 20% del lavoro. L’altro 80% è il loading, cleaning, splitting, evaluating, saving e serving. Una volta che hai quell’impalcatura, puoi cambiare modelli quasi senza fatica. L’impalcatura è l’asset.

Cosa viene dopo: deep learning

Questo chiude il Modulo 9. Siamo rimasti saldamente nel mondo dei modelli tabulari: lineari, ad albero, ensemble. Il modello più profondo che abbiamo toccato ha ancora manopole interpretabili chiare. Il Modulo 10 si rivolge al deep learning: PyTorch, training di reti neurali, transformer, e le parti del ML moderno che non si infilano comodamente in una chiamata model.fit(X, y). Toolchain diversa, stile di debugging diverso, modalità di failure diverse. Ma molte delle stesse lezioni di workflow di questo modulo, disciplina train/test, prima la baseline, salva la pipeline, attento allo skew, valgono ancora direttamente.

Ci vediamo nel Modulo 10.

References: scikit-learn user guide, Optuna documentation (https://optuna.org/), SHAP documentation (https://shap.readthedocs.io/), Telco customer churn dataset (IBM, available on Kaggle). Retrieval 2026-05-01.

Cerca