Atelier : logistique en pipeline
CV, seuil validé et test final unique.
Énoncé
Sur data/colis_propres.csv, prédisez retard avec distance, poids, urgence et transporteur. delai_jours définit la cible après livraison : il ne doit pas figurer dans les entrées. Comparez le taux de retards du train (référence constante) à une régression logistique. Choisissez C et l'usage éventuel de class_weight="balanced" par validation croisée par client ; choisissez ensuite un seuil de décision sur les prédictions hors pli du train. Réservez les clients de test pour un verdict final : matrice de confusion, précision, rappel, F1 et score de Brier.
Contraintes
Même découpage par client_id que dans l'atelier de régression. Toutes les transformations sont dans le pipeline. Notez la proportion de retards dans chaque côté : GroupShuffleSplit respecte les groupes, mais ne garantit pas la stratification exacte. Le seuil est choisi avant de toucher au test ; indiquez pourquoi vous privilégiez le rappel ou la précision selon le coût d'un retard manqué ou d'une fausse alerte.
Indices
GroupKFold évite qu'un client apparaisse dans deux plis. cross_val_score(..., scoring="average_precision") compare les modèles sans fixer le seuil. cross_val_predict(..., method="predict_proba") donne une probabilité hors pli pour chaque ligne du train ; balayez les seuils uniquement sur ces probabilités. Un score de Brier plus petit indique de meilleures probabilités, mais examinez aussi leur calibration par groupes de probabilité si ce score doit piloter une décision réelle.
Correction disponibleCherchez d’abord par vous-même.Voir la correction
Correction
Ce script se lance après clean.py ; enregistrez-le sous retard.py :
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
average_precision_score, brier_score_loss, confusion_matrix,
f1_score, precision_score, recall_score,
)
from sklearn.model_selection import (
GroupKFold, GroupShuffleSplit, cross_val_predict, cross_val_score,
)
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
df = pd.read_csv("data/colis_propres.csv")
colonnes = ["distance_km", "poids_kg", "urgence", "transporteur"]
X, y, groupes = df[colonnes], df["retard"], df["client_id"]
split = GroupShuffleSplit(n_splits=1, test_size=0.2, random_state=42)
train_idx, test_idx = next(split.split(X, y, groupes))
X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
g_train, g_test = groupes.iloc[train_idx], groupes.iloc[test_idx]
assert set(g_train).isdisjoint(g_test)
print("retards train/test :", round(y_train.mean(), 3), round(y_test.mean(), 3))
def construire(c, poids):
nombres = make_pipeline(SimpleImputer(strategy="median"), StandardScaler())
categories = make_pipeline(
SimpleImputer(strategy="most_frequent"),
OneHotEncoder(handle_unknown="ignore"),
)
preparation = ColumnTransformer([
("nombres", nombres, ["distance_km", "poids_kg"]),
("categories", categories, ["urgence", "transporteur"]),
], sparse_threshold=0)
return make_pipeline(
preparation,
LogisticRegression(C=c, class_weight=poids, max_iter=1000),
)
plis = list(GroupKFold(n_splits=5).split(X_train, y_train, g_train))
essais = {}
for c in (0.1, 1.0, 10.0):
for poids in (None, "balanced"):
modele = construire(c, poids)
scores = cross_val_score(
modele, X_train, y_train, cv=plis, scoring="average_precision"
)
essais[(c, poids)] = scores.mean()
print(c, poids, "AP CV :", round(scores.mean(), 3))
choix = max(essais, key=essais.get)
modele = construire(*choix)
hors_pli = cross_val_predict(
modele, X_train, y_train, cv=plis, method="predict_proba"
)[:, 1]
seuils = np.arange(0.1, 0.91, 0.05)
seuil = max(seuils, key=lambda s: f1_score(y_train, hors_pli >= s))
print("AP hors pli :", round(average_precision_score(y_train, hors_pli), 3))
print("référence AP ≈ prévalence train :", round(y_train.mean(), 3))
print("choix et seuil :", choix, round(seuil, 2))
modele.fit(X_train, y_train)
proba_test = modele.predict_proba(X_test)[:, 1]
prediction = proba_test >= seuil
print("matrice test :\n", confusion_matrix(y_test, prediction))
print("précision :", round(precision_score(y_test, prediction), 3))
print("rappel :", round(recall_score(y_test, prediction), 3))
print("F1 :", round(f1_score(y_test, prediction), 3))
print("Brier :", round(brier_score_loss(y_test, proba_test), 3))
La meilleure combinaison, le seuil et les scores exacts sont des résultats de l'expérience, pas des valeurs à deviner. class_weight="balanced" peut aider au rappel tout en dégradant la calibration ; regardez les deux. Si le modèle ne dépasse pas la référence, documentez l'échec et inspectez les caractéristiques disponibles avant d'essayer un modèle plus complexe.