Ajuster le seuil pour repérer les transactions atypiques.
Le score Z mesure l'écart à la moyenne en écarts-types sur montant et délai, combinés en une distance ; l'IQR signale tout point hors de Q1 − k×IQR ou Q3 + k×IQR. Les deux méthodes sont recalculées entièrement côté client.
Statistiques appliquées au contrôle qualité, traduction d'une logique de détection en SQL et Python, arbitrage précision / rappel selon le seuil.
Filtrées selon les mêmes critères que le dashboard ci-dessus.
| ID | Date | Fournisseur | Montant | Délai (j) | Score |
|---|
Contrairement aux deux autres projets, il n'y a pas ici de schéma en étoile : les transactions forment une table plate, transformée par un pipeline statistique en 3 étapes.
Voici quelques extraits de code utilisés pour la conception de ce projet.
Détection directement en SQL, sans étape intermédiaire, via une fenêtre statistique sur l'ensemble de la table.
WITH stats AS (
SELECT
AVG(montant) AS moy_montant, STDDEV(montant) AS ect_montant,
AVG(delai) AS moy_delai, STDDEV(delai) AS ect_delai
FROM transactions
)
SELECT
t.id, t.date, t.fournisseur, t.montant, t.delai,
ROUND(SQRT(
POWER((t.montant - s.moy_montant) / NULLIF(s.ect_montant, 0), 2) +
POWER((t.delai - s.moy_delai) / NULLIF(s.ect_delai, 0), 2)
), 2) AS score_z
FROM transactions t
CROSS JOIN stats s
QUALIFY score_z > 2.5 -- ou sous-requete + WHERE sur les moteurs sans QUALIFY
ORDER BY score_z DESC;Les deux méthodes disponibles dans le sélecteur ci-dessus, implémentées avec pandas / numpy.
import numpy as np
def detecter_anomalies_zscore(df, seuil=2.5):
z_montant = (df["montant"] - df["montant"].mean()) / df["montant"].std()
z_delai = (df["delai"] - df["delai"].mean()) / df["delai"].std()
df["score_z"] = np.sqrt(z_montant**2 + z_delai**2)
return df[df["score_z"] > seuil].sort_values("score_z", ascending=False)
def detecter_anomalies_iqr(df, k=1.5):
bornes = {}
for col in ["montant", "delai"]:
q1, q3 = df[col].quantile([0.25, 0.75])
iqr = q3 - q1
bornes[col] = (q1 - k * iqr, q3 + k * iqr)
masque = (~df["montant"].between(*bornes["montant"])
| ~df["delai"].between(*bornes["delai"]))
return df[masque]
anomalies = detecter_anomalies_zscore(transactions, seuil=2.5)
print(f"{len(anomalies)} anomalies sur {len(transactions)} transactions "
f"({len(anomalies) / len(transactions):.1%})")