Navigation : Index | << ML-8-Clustering | Jumeau .NET (ML.NET)
Ce notebook est le jumeau Python (scikit-learn) de ML-9-Anomaly-Detection.ipynb (ML.NET). Il couvre la même pédagogie avec les outils du écosystème Python. La logique de détection est identique : on apprend le régime normal, puis on mesure l’écart d’un point à ce régime.
À la fin de ce notebook, vous saurez :
Le clustering regroupe des points sans étiquettes. La détection d’anomalies répond à une question différente : étant donné un régime de fonctionnement normal (apprentissage), quels points s’en écartent suffisamment pour être soupçonnés anormaux ? C’est encore de l’apprentissage non-supervisé au sens où le modèle n’apprend qu’à partir de données normales — mais la tâche est un test (chaque point est-il dans la distribution normale ?), pas une partition.
Le cas d’usage industriel canonique est la maintenance prédictive : une machine saine (capteurs : température, pression, vibration) opère dans une étroite ellipse de régime ; une usure, une fuite ou un déséquilibre la fait dériver hors de cette zone. On veut déclencher une alerte avant la casse.
L’intuition (Lakhina et al., 2004 ; Pfiefer et al., 2011) : la PCA projette les données normales sur un sous-espace de faible dimension (les k premières composantes). Un point normal se reconstruit bien dans ce sous-espace (faible résidu) ; un point anormal, lui, s’étale hors du sous-espace et laisse un fort résidu de reconstruction. Ce résidu est le score d’anomalie : plus il est élevé, plus le point s’écarte du régime appris. C’est exactement la sémantique du trainer RandomizedPca de ML.NET, reproduit ici avec sklearn.decomposition.PCA.
Subtilité importante. Le modèle est non-supervisé au
fit(il ignore toute étiquette), mais l’évaluation a besoin d’étiquettes (1= anomalie,0= normal) sur le jeu de test pour calculer l’AUC. On s’entraîne donc sur des données normales et l’on évalue sur un jeu de test labellisé mélangeant normaux et anomalies.
# Imports : numpy pour les données, scikit-learn pour la PCA et les métriques, matplotlib pour visualiser
import numpy as np
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA
from sklearn.metrics import roc_auc_score
plt.rcParams["figure.dpi"] = 110
print(f"numpy {np.__version__} | scikit-learn {roc_auc_score.__module__.split('.')[0]} charge")numpy 2.4.4 | scikit-learn sklearn chargeOn fixe la graine (42) pour rendre l’expérience déterministe : les tirages aléatoires (génération des données) et la PCA seront reproductibles d’une exécution à l’autre.
On simule trois capteurs d’une machine industrielle. Plutôt que les valeurs brutes, on travaille avec les écarts au point de fonctionnement nominal : c’est cette déviation qui est informative. Le régime sain est ainsi centré sur l’origine :
| Feature | Écart-type (régime sain) | Régime sain (normal) | Régime défaillant (anomalie) |
|---|---|---|---|
| Température | 2.0 | ~0 (nominal) | surchauffe (~+5) |
| Pression | 0.1 | ~0 (nominal) | surpression / fuite (~+0.2) |
| Vibration | 0.8 | ~0 (nominal) | fortes vibrations (~+2.5) |
On construit deux jeux distincts :
X_train (200 points normaux uniquement) — c’est le régime que le modèle apprend.X_test (120 points normaux + 40 anomalies, labellisés) — sert à évaluer l’AUC et à tuner le seuil.Les anomalies sont volontairement modérées (~2 à 3 écarts-types par capteur) afin que la détection reste non-triviale : un peu de chevauchement subsiste, et le choix du seuil aura un vrai coût. Le pic de pression est particulièrement discriminant car, la pression étant étroitement régulée (faible variance), ce signal est orthogonal au plan de variance normale — c’est précisément ce que la PCA exploite.
# Régime NORMAL (machine saine) : écarts au point nominal, centrés sur 0, bruit gaussien
# 200 points d'entraînement (tous normaux) + 120 points de test normaux
X_train = np.column_stack([
rng.normal(0.0, 2.0, size=200), # Température
rng.normal(0.0, 0.1, size=200), # Pression (étroitement régulée)
rng.normal(0.0, 0.8, size=200), # Vibration
])
X_test_normal = np.column_stack([
rng.normal(0.0, 2.0, size=120),
rng.normal(0.0, 0.1, size=120),
rng.normal(0.0, 0.8, size=120),
])
# ANOMALIES (machine défaillante, ~2 à 3 écarts-types par capteur) : décalage orthogonal au plan de variance
X_test_anom = np.column_stack([
rng.normal(5.0, 2.0, size=40), # surchauffe
rng.normal(0.20, 0.1, size=40), # pic de pression (fuite) -- signal orthogonal discriminant
rng.normal(2.5, 0.8, size=40), # fortes vibrations
])
y_test_anom = np.ones(40, dtype=int) # étiquette = anomalie
print(f"Train : {X_train.shape[0]} points (tous normaux)")
print(f"Test : {X_test_normal.shape[0]} normaux + {X_test_anom.shape[0]} anomalies")Train : 200 points (tous normaux)
Test : 120 normaux + 40 anomaliesX_test : 160 points | y_test : 40 anomalies / 120 normauxLe détecteur a deux étapes :
X_train (les données normales),||x - x_reconstruit||² est le score d’anomalie.Pourquoi PAS de normalisation ici, contrairement au clustering K-Means ? K-Means mesure des distances euclidiennes et exige que les features soient à la même échelle. La détection d’anomalies par PCA, elle, exploite justement la structure de variance : les composantes principales sont définies par les directions de plus grande variance. Normaliser écraserait cette structure.
Pourquoi
n_components = 2? Avec 3 features, la PCA de rang 2 projette les données sur un plan ; le score d’anomalie est la composante hors-plan (le résidu le long de la 3ᵉ direction). Si l’on choisissaitn_components = 3(le maximum), tout serait expliqué et le résidu serait nul : la détection deviendrait impossible. Le rang contrôle donc la sensibilité du détecteur (cf. Exercice 1).
Nos données sont déjà recentrées sur l’origine (écarts au point nominal) : la PCA passe donc par le centroïde, condition essentielle pour que le résidu reflète bien l’écart au régime normal.
On encapsule le score dans une fonction réutilisable (elle servira pour les exercices).
def pca_anomaly_scores(X_train, X, n_components=2):
"""Ajuste une PCA sur X_train et renvoie l'erreur de reconstruction de chaque ligne de X."""
pca = PCA(n_components=n_components, svd_solver="full")
pca.fit(X_train)
X_reconstructed = pca.inverse_transform(pca.transform(X))
residu = np.sum((X - X_reconstructed) ** 2, axis=1)
return residu, pca
print("Fonction pca_anomaly_scores prête (PCA rang-k -> erreur de reconstruction)")Fonction pca_anomaly_scores prête (PCA rang-k -> erreur de reconstruction)On entraîne le détecteur sur X_train (données normales uniquement). Le modèle apprend ainsi la géométrie du régime sain ; il n’a jamais vu d’anomalies.
Détecteur PCA entraîné sur les données normales (rang=2) | variance expliquée : [0.823 0.174]Contrairement au clustering, la détection d’anomalies se laisse évaluer dès lors que le jeu de test est labellisé. On dispose de deux métriques complémentaires :
def detection_rate_at_k_fp(scores, labels, k=10):
"""Fraction des vraies anomalies détectées avant d'atteindre k faux positifs.
On trie par score décroissant et on compte les anomalies trouvées avant la k-ième fausse alarme."""
order = np.argsort(-scores) # score décroissant
labels_sorted = labels[order]
fp = 0
detected = 0
total_anom = int(labels.sum())
for lab in labels_sorted:
if lab == 1:
detected += 1
else:
fp += 1
if fp == k: # on s'arrête au k-ième faux positif
break
return detected / total_anom if total_anom > 0 else 0.0
auc = roc_auc_score(y_test, scores) # scores élevés = anormaux => sens direct
dr10 = detection_rate_at_k_fp(scores, y_test, k=10)
print("=== Métriques de détection d'anomalies (rang=2) ===")
print(f" AUC (aire sous ROC) : {auc:.4f}")
print(f" Detection Rate @ 10 faux positifs: {dr10:.4f}")=== Métriques de détection d'anomalies (rang=2) ===
AUC (aire sous ROC) : 0.8683
Detection Rate @ 10 faux positifs: 0.6000Un AUC élevé (proche de 1) signale une bonne séparation : un point anormal tiré au hasard a de fortes chances d’être scoré plus haut qu’un point normal. Ce n’est pas 1.0 car le chevauchement subsiste — une anomalie dont le pic de pression est modéré ressemble, dans le sous-espace PCA, à un point normal bruité. Le Detection Rate @ 10 FP traduit cette limite : à très faible budget de fausses alarmes, on ne rattrape qu’une fraction des anomalies. Le sweep de seuil ci-dessous montre comment on améliore ce taux en acceptant plus de fausses alarmes.
On calcule le score de trois points nouveaux. La PCA étant déjà ajustée, il suffit de projeter/reconstruire chacun et de mesurer son résidu.
# Trois profils : sain, légèrement décalé, défaillant (fuite = pic de pression)
test_points = np.array([
[0.0, 0.0, 0.0], # profil sain (au nominal)
[3.0, 0.05, 1.0], # légèrement off (petit écart)
[5.0, 0.20, 2.5], # profil défaillant (fuite) -- anomalie représentative
])
pt_reconstructed = pca_model.inverse_transform(pca_model.transform(test_points))
pt_scores = np.sum((test_points - pt_reconstructed) ** 2, axis=1)
print("=== Score d'anomalie de nouveaux points ===")
for s, sc in zip(test_points, pt_scores):
print(f" T={s[0]:4.1f} P={s[1]:4.2f} V={s[2]:4.1f} -> Score={sc:8.4f}")=== Score d'anomalie de nouveaux points ===
T= 0.0 P=0.00 V= 0.0 -> Score= 0.0000
T= 3.0 P=0.05 V= 1.0 -> Score= 0.0031
T= 5.0 P=0.20 V= 2.5 -> Score= 0.0433Le score brut est une grandeur continue ; la décision opérationnelle (alerte ou non) exige un seuil. Ce seuil arbitre entre deux erreurs :
Abaisser le seuil augmente le TPR (on rate moins d’anomalies) mais augmente aussi le FPR (plus de fausses alarmes). Le bon seuil dépend du coût relatif d’une casse ratée vs d’une intervention inutile.
Plutôt que de tester des seuils absolus (dont l’échelle dépend des données), on balaie des seuils relatifs répartis entre le min et le max des scores observés. Pour chacun, on calcule TPR, FPR et précision.
smin, smax = scores.min(), scores.max()
total_anom = int(y_test.sum())
total_norm = int((y_test == 0).sum())
print("=== Trade-off seuil : TPR (détection) vs FPR (fausse alarme) ===")
print(f" ({total_anom} vraies anomalies, {total_norm} vrais normaux ; scores de {smin:.3f} à {smax:.3f})")
print(f" Seuil | TPR(detect) | FPR(fausse alarme) | Precision")
print(f" ------+-------------+--------------------+----------")
for t in range(1, 10):
thr = smin + (smax - smin) * t / 10.0
flagged = scores > thr
tp = int(np.sum(flagged & (y_test == 1)))
fp = int(np.sum(flagged & (y_test == 0)))
tpr = tp / total_anom if total_anom else 0
fpr = fp / total_norm if total_norm else 0
prec = tp / (tp + fp) if (tp + fp) else 0
print(f" {thr:5.2f} | {tpr:11.2f} | {fpr:18.2f} | {prec:10.2f}")=== Trade-off seuil : TPR (détection) vs FPR (fausse alarme) ===
(40 vraies anomalies, 120 vrais normaux ; scores de 0.000 à 0.124)
Seuil | TPR(detect) | FPR(fausse alarme) | Precision
------+-------------+--------------------+----------
0.01 | 0.85 | 0.33 | 0.47
0.02 | 0.75 | 0.14 | 0.64
0.04 | 0.60 | 0.04 | 0.83
0.05 | 0.45 | 0.02 | 0.90
0.06 | 0.33 | 0.01 | 0.93
0.07 | 0.20 | 0.00 | 1.00
0.09 | 0.10 | 0.00 | 1.00
0.10 | 0.05 | 0.00 | 1.00
0.11 | 0.05 | 0.00 | 1.00Le sweep dessine la courbe ROC de façon lisible : à seuil bas on rattrape presque toutes les anomalies mais au prix de nombreuses fausses alarmes ; à seuil haut on devient précis mais on rate la plupart des anomalies. Un point de fonctionnement raisonnable détecte la grande majorité des anomalies pour une fraction de fausses alarmes acceptable. Si le métier exige FPR <= 5 %, il faut monter le seuil — on y détecte encore environ 60 % des anomalies (seuil ≈ 0,04, précision 0,83), un compromis serré mais pas un ratage majoritaire. Ce compromis n’a pas de réponse universelle : il dépend du coût relatif d’une casse ratée (très élevé en maintenance prédictive) face au coût d’une intervention inutile. L’Exercice 2 formalise cet accord sur une contrainte TPR >= 0,95.
Le rang de la PCA contrôle la sensibilité du détecteur. Avec 3 features :
n_components = 1 -> sous-espace 1D, résidu sur 2 dimensions (detecteur très sensible : beaucoup de normaux flaggés),n_components = 2 -> sous-espace plan, résidu sur 1 dimension (compromis utilisé dans le notebook),n_components = 3 -> tout est expliqué, résidu nul (aucune détection possible).Objectifs : 1. Boucler sur n_components = 1, 2, 3, calculer les scores pour chaque valeur 2. Évaluer l’AUC sur X_test 3. Vérifier que n_components = 3 donne un AUC ~0.5 (aucun pouvoir de détection) et comparer 1 vs 2
Indice : réutilisez pca_anomaly_scores(X_train, X_test, n_components=...). Pour n_components = 3, les résidus sont tous quasi nuls : la PCA a tout expliqué, il ne reste plus de résidu à mesurer.
# Exercice 1 : Effet du rang PCA sur l'AUC
# TODO étudiant : boucle sur n_components = 1, 2, 3, calcule les scores, évalue l'AUC.
# Indice : réutilise pca_anomaly_scores(X_train, X_test, n_components=k). Pour k=3, résidus ~0 -> AUC ~0.5.
# Étape 1 : pour k dans {1, 2, 3}, appelle pca_anomaly_scores(X_train, X_test, n_components=k)
# Étape 2 : calcule roc_auc_score(y_test, scores_k) pour chaque k
# Étape 3 : affiche k | AUC et conclus sur le compromis sensibilité / pouvoir discriminant
result_ex1 = None # TODO étudiant
print("Exercice 1 à compléter")Exercice 1 à compléterOn veut un détecteur qui rate au plus 5 % des anomalies (TPR >= 0.95) tout en minimisant les fausses alarmes. Le sweep ci-dessus est trop grossier (9 seuils réguliers).
Objectifs : 1. Calculer une courbe ROC fine : 50 seuils répartis entre smin et smax, TPR et FPR pour chacun 2. Identifier le seuil qui réalise TPR >= 0.95 avec FPR minimal 3. Reporter ce seuil opérationnel et sa précision
Indice : même boucle que le sweep, mais avec 50 seuils ; parcourir les couples (TPR, FPR) et garder le premier seuil (du plus élevé au plus bas) qui atteint TPR >= 0.95.
# Exercice 2 : Accorder le seuil à un coût opérationnel
# TODO étudiant : courbe ROC fine (50 seuils), trouver le seuil à TPR >= 0.95 et FPR minimal.
# TPR = TP / total_anom ; FPR = FP / total_norm
# Indice : 50 seuils entre smin et smax ; garde le seuil (le plus élevé d'abord) qui atteint TPR>=0.95.
# Étape 1 : boucle 50 seuils, calcule TPR + FPR pour chacun
# Étape 2 : cherche le seuil réalisant TPR >= 0.95 avec FPR minimal
# Étape 3 : affiche le seuil opérationnel, son FPR et sa précision
result_ex2 = None # TODO étudiant
print("Exercice 2 à compléter")Exercice 2 à compléterLes anomalies actuelles sont décalées de ~2 à 3 écarts-types par capteur : détectables (AUC ≈ 0,87) mais non-triviales — on n’en rattrape que 60 % pour un budget de 10 fausses alarmes (cellule 13). Que se passe-t-il avec des anomalies subtiles (~1.5 sigma), qui se rapprochent du régime normal ?
Objectifs : 1. Régénérer X_test_anom avec un décalage réduit (ex. Température ~ N(2, 2), Pression ~ N(0.10, 0.1), Vibration ~ N(1.0, 0.8)) 2. Reconstituer X_test, ré-évaluer l’AUC du modèle (sans ré-entraîner) 3. Constater la chute de l’AUC et conclure sur la limite de la détection PCA
Indice : plus les anomalies se rapprochent du régime normal, plus elles entrent dans le nuage appris et plus leur résidu ressemble à celui d’un point normal. L’AUC chute vers 0.5. La PCA ne détecte bien que les anomalies orthogonales au sous-espace normal ; les anomalies « inline » (le long d’une direction normale) lui échappent.
# Exercice 3 : Anomalies subtiles (limite de l'approche PCA)
# TODO étudiant : régénère des anomalies subtiles (~1.5 sigma) et ré-évalue l'AUC.
# Indice : anomalies plus proches du régime normal -> résidu ressemblant à un point normal -> AUC chute.
# La PCA ne détecte bien que les anomalies ORTHOGONALES au sous-espace normal (ici : un pic de pression).
# Étape 1 : régénère X_test_anom avec décalage réduit (Température~2, Pression~0.10, Vibration~1.0)
# Étape 2 : reconstitue X_test, ré-évalue l'AUC du modèle existant (sans re-fit)
# Étape 3 : compare l'AUC à la valeur du notebook et conclus sur la limite de la détection PCA
result_ex3 = None # TODO étudiant
print("Exercice 3 à compléter")Exercice 3 à compléterCe notebook a introduit la détection d’anomalies non-supervisée par PCA en Python :
| Concept | Description |
|---|---|
| Détection d’anomalies | Apprentissage non-supervisé : on apprend le régime normal, on teste si un point s’en écarte |
sklearn.decomposition.PCA |
PCA ajustée sur les données normales ; score = erreur de reconstruction |
| Score d’anomalie | Résidu de reconstruction dans le sous-espace PCA (plus élevé = plus anormal) |
n_components |
Dimension du sous-espace ; contrôle la sensibilité (n_components = n_features -> aucune détection) |
| AUC | Séparation normaux/anormaux (1.0 = parfait, 0.5 = hasard) |
| DetectionRate@K FP | Fraction des anomalies détectée avant K fausses alarmes (métrique opérationnelle) |
| Seuil de décision | Arbitre TPR vs FPR ; s’accorde au coût opérationnel |
Points clés :
fit est non-supervisé : on s’entraîne sur des données normales uniquement. Mais contrairement au clustering, l’évaluation exploite des étiquettes (0/1) pour calculer l’AUC.n_components ne peut pas dépasser le nombre de features, et n_components = n_features annule toute détection (résidu nul).Équivalence ML.NET : ce notebook est le jumeau Python de ML-9-Anomaly-Detection.ipynb. La PCA + erreur de reconstruction reproduit fidèlement le trainer RandomizedPca de ML.NET (le « randomized » est une optimisation de calcul pour les grandes dimensions ; avec 3 features, la PCA exacte de scikit-learn est équivalente et déterministe).
Algorithme (PCA pour la détection d’anomalies)
Métriques d’évaluation (ROC et detection rate)
Cas d’usage (maintenance prédictive)