Détecter des lignes inhabituelles dans un CSV avec Python

Vous voulez repérer les lignes inhabituelles d’un CSV et obtenir un fichier de résultats exploitable ? Ce projet Python lit des mesures numériques, calcule un score avec Isolation Forest et ajoute une colonne a_examiner. Il conserve l’ordre des observations, leurs identifiants et les cellules d’origine.

Le résultat est une liste de pistes pour votre analyse. Une valeur inhabituelle peut être correcte : le script ne supprime aucune ligne et ne transforme pas une alerte en diagnostic. Ici, nous auditons le fichier chargé ; nous ne mesurons pas une performance sur de futures observations.

Télécharger le projet Python complet — ZIP, 12,6 Ko
Fichier : csv-isolation-forest-thepingouin-v1.zip. Gratuit, sans inscription : script, CSV synthétique, dépendances, résultat de référence et guide de lancement. Vous pouvez aussi lire le script complet plus bas.

Un CSV de mesures, avec des identifiants stables

Le fichier mesures_synthetiques.csv contient 210 observations entièrement fabriquées pour ce tutoriel. Les durées et tailles évoquent des traitements informatiques, mais ne proviennent d’aucune application réelle. Deux cellules numériques sont volontairement vides pour vérifier leur traitement.

Les premières lignes ressemblent à ceci :

id,duree_ms,taille_ko,lot
M0001,125.77,43.54,demo
M0002,115.82,45.78,demo
M0003,108.42,44.24,demo
M0004,119.51,49.22,demo

Le séparateur est une virgule et les décimales utilisent un point. Le script s’appuie sur csv.DictReader de la bibliothèque standard Python ; pandas n’est pas nécessaire. L’ouverture utilise newline="" et l’encodage utf-8-sig, qui accepte aussi un marqueur BOM. Voir la documentation Python du module csv.

Installer les dépendances et créer le fichier de résultats

Environnement testé le 14 septembre 2026 : Python 3.12.14, NumPy 2.5.3 et scikit-learn 1.9.1. Décompressez l’archive dans un nouveau dossier, puis ouvrez un terminal dans ce dossier. Le fichier requirements.txt contient les versions utilisées.

Créez un environnement Python dédié :

python -m venv .venv

Avec Windows PowerShell

Ces commandes appellent directement le Python de l’environnement ; aucune activation n’est nécessaire :

.\.venv\Scripts\python.exe -m pip install -r requirements.txt
.\.venv\Scripts\python.exe analyser_csv.py mesures_synthetiques.csv resultats.csv --colonnes duree_ms taille_ko

Avec macOS ou Linux

.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python analyser_csv.py mesures_synthetiques.csv resultats.csv --colonnes duree_ms taille_ko

La première installation télécharge les dépendances depuis votre dépôt Python habituel. Ensuite, l’analyse du CSV est locale : le script n’envoie pas vos lignes à une API. Voici la sortie obtenue avec le fichier fourni :

Observations lues : 210
Observations analysées : 208
Observations incomplètes conservées : 2
Alertes à examiner : 11
Offset : -0.606394
Résultat écrit : resultats.csv

resultats.csv est créé à côté des fichiers du projet. resultats_exemple.csv, inclus dans l’archive, sert de référence. Pour recommencer une analyse, choisissez un autre nom de sortie : le programme refuse d’écraser un fichier existant, y compris son entrée.

Lire le score, la décision et les lignes non analysées

La sortie reprend les colonnes d’origine et ajoute quatre champs :

• score_isolation : les valeurs les plus basses correspondent aux observations les plus atypiques pour ce modèle.
• decision_isolation : le score moins le seuil interne offset_.
• a_examiner : oui si la décision est strictement négative, sinon non pour une ligne analysée.
• statut_analyse : analyse ou donnees_manquantes.

Une ligne incomplète n’est pas une ligne jugée normale. Si une colonne numérique sélectionnée est vide, l’observation reste dans le résultat, mais ses scores et son indicateur a_examiner sont vides. Il faut traiter séparément ces lignes avant de relancer l’analyse. Aucun zéro ni aucune moyenne n’est inséré à leur place.

Le tableau ci-dessous présente les cinq scores les plus bas de notre exécution, arrondis pour la lecture. Le CSV de sortie conserve davantage de chiffres. Il garde aussi l’ordre initial ; seul ce tableau de présentation est trié.

Le graphique place les 12 scores les plus bas face au seuil du fichier. Onze observations sont signalées dans cette exécution. Le score n’est pas une probabilité d’erreur : -0.79 ne veut pas dire « 79 % de risque ». Pour les conventions détaillées, consultez notre guide Isolation Forest : scores, exemples et limites et la référence officielle scikit-learn.

Le script complet pour analyser votre CSV

La même version de analyser_csv.py se trouve dans l’archive. Elle valide les colonnes et les identifiants, sélectionne les observations complètes, ajuste le modèle puis replace chaque résultat à sa position d’origine.

"""Classer un CSV local avec Isolation Forest, sans modifier les entrées.

Exemple pédagogique The Pingouin, septembre 2026. Une alerte n'est pas une erreur prouvée.
"""
import argparse
import csv
import math
from pathlib import Path

import numpy as np
from sklearn.ensemble import IsolationForest

AJOUTS = ["score_isolation", "decision_isolation", "a_examiner", "statut_analyse"]


def analyser(entree, sortie, colonnes, id_colonne="id", contamination=0.05,
             separateur=","):
    entree, sortie = Path(entree), Path(sortie)
    if entree.resolve() == sortie.resolve():
        raise ValueError("La sortie doit être différente du fichier d'entrée.")
    if sortie.exists():
        raise ValueError("La sortie existe déjà : choisissez un nouveau nom.")
    if len(separateur) != 1 or separateur in "\r\n\0":
        raise ValueError("Le séparateur doit être un caractère, par exemple ',' ou ';'.")
    if not math.isfinite(contamination) or not 0 < contamination <= 0.5:
        raise ValueError("La contamination doit être comprise dans ]0, 0.5].")
    if not colonnes or len(set(colonnes)) != len(colonnes) or id_colonne in colonnes:
        raise ValueError("Choisissez des colonnes distinctes, sans la colonne identifiant.")

    # utf-8-sig accepte aussi un CSV UTF-8 avec marqueur BOM.
    with entree.open(encoding="utf-8-sig", newline="") as fichier:
        lecteur = csv.DictReader(fichier, delimiter=separateur, strict=True)
        noms = lecteur.fieldnames
        if not noms or any(not nom.strip() for nom in noms) or len(set(noms)) != len(noms):
            raise ValueError("En-tête absent, vide ou contenant des doublons.")
        if not set([id_colonne, *colonnes]).issubset(noms):
            raise ValueError("Colonne demandée absente : vérifiez en-tête et séparateur.")
        if set(AJOUTS).intersection(noms):
            raise ValueError("Le fichier contient déjà une colonne de résultat réservée.")
        lignes = list(lecteur)
    if not lignes:
        raise ValueError("Le CSV ne contient aucune observation.")

    identifiants, positions, valeurs = set(), [], []
    for position, ligne in enumerate(lignes):
        if None in ligne or any(valeur is None for valeur in ligne.values()):
            raise ValueError(f"Observation {position + 1} : nombre de champs incorrect.")
        identifiant = ligne[id_colonne].strip()
        if not identifiant or identifiant in identifiants:
            raise ValueError(f"Observation {position + 1} : identifiant vide ou répété.")
        identifiants.add(identifiant)
        vecteur, incomplet = [], False
        for colonne in colonnes:
            texte = ligne[colonne].strip()
            if texte == "":
                incomplet = True
                continue
            try:
                nombre = float(texte)
            except ValueError:
                raise ValueError(f"Observation {position + 1}, colonne {colonne} : nombre attendu.") from None
            if not math.isfinite(nombre) or abs(nombre) > float(np.finfo(np.float32).max):
                raise ValueError(f"Observation {position + 1}, colonne {colonne} : nombre non fini ou trop grand.")
            vecteur.append(nombre)
        if not incomplet:
            positions.append(position)
            valeurs.append(vecteur)

    # Garde pédagogique, pas une limite technique de l'algorithme.
    if len(valeurs) < 20:
        raise ValueError("Cet exemple exige au moins 20 observations complètes.")
    X = np.asarray(valeurs, dtype=float)
    modele = IsolationForest(n_estimators=200, contamination=contamination,
                             max_samples="auto", random_state=42, n_jobs=1)
    modele.fit(X)
    # Audit de ce fichier : ce calcul n'est pas une évaluation sur un test indépendant.
    scores = modele.score_samples(X)
    decisions = modele.decision_function(X)
    alertes = decisions < 0

    resultat = [dict(ligne, score_isolation="", decision_isolation="",
                     a_examiner="", statut_analyse="donnees_manquantes") for ligne in lignes]
    for j, position in enumerate(positions):
        resultat[position].update(
            score_isolation=repr(float(scores[j])),
            decision_isolation=repr(float(decisions[j])),
            a_examiner="oui" if alertes[j] else "non",
            statut_analyse="analyse",
        )
    # Le mode x refuse d'écraser un fichier créé entre les contrôles et l'écriture.
    with sortie.open("x", encoding="utf-8", newline="") as fichier:
        writer = csv.DictWriter(fichier, fieldnames=noms + AJOUTS, delimiter=separateur)
        writer.writeheader()
        writer.writerows(resultat)

    print(f"Observations lues : {len(lignes)}")
    print(f"Observations analysées : {len(positions)}")
    print(f"Observations incomplètes conservées : {len(lignes) - len(positions)}")
    print(f"Alertes à examiner : {int(alertes.sum())}")
    print(f"Offset : {float(modele.offset_):.6f}")
    print(f"Résultat écrit : {sortie.name}")
    return {"lues": len(lignes), "analysees": len(positions),
            "incompletes": len(lignes) - len(positions),
            "alertes": int(alertes.sum()), "offset": float(modele.offset_)}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("entree", type=Path)
    parser.add_argument("sortie", type=Path)
    parser.add_argument("--colonnes", nargs="+", required=True)
    parser.add_argument("--id-colonne", default="id")
    parser.add_argument("--contamination", type=float, default=0.05)
    parser.add_argument("--separateur", default=",")
    args = parser.parse_args()
    try:
        analyser(args.entree, args.sortie, args.colonnes, args.id_colonne,
                 args.contamination, args.separateur)
    except (ValueError, OSError, csv.Error, UnicodeError) as erreur:
        parser.exit(2, f"Erreur : {erreur}\n")


if __name__ == "__main__":
    main()

Les colonnes sont sélectionnées explicitement avec --colonnes. Cela évite d’ajouter par accident un identifiant, un numéro de ligne ou une information dont le sens ne convient pas à l’analyse. La graine random_state=42 stabilise le tirage aléatoire du modèle ; elle ne garantit pas la pertinence des données choisies.

Adapter le projet à un autre fichier

Commencez avec une copie de votre CSV et choisissez deux ou trois mesures dont vous comprenez les unités. Les noms de colonnes doivent correspondre exactement à l’en-tête, avec la même casse. Chaque observation doit disposer d’un identifiant renseigné et unique. Si l’identifiant s’appelle mesure_id, utilisez --id-colonne mesure_id.

Pour un fichier séparé par des points-virgules, ajoutez --separateur ";" à la commande. Le point reste le séparateur décimal : une valeur telle que 12,5 n’est pas convertie automatiquement. Vérifiez le format à l’export plutôt que de remplacer toutes les virgules du fichier, ce qui pourrait endommager d’autres champs.

Les fichiers mal formés sont refusés avant l’écriture du résultat. Les cellules d’origine restent identiques dans la sortie, mais les guillemets et fins de ligne peuvent être normalisés par le lecteur CSV. Le fichier d’entrée, lui, n’est jamais réécrit. Pour revoir les bases, lisez notre tutoriel de lecture et d’écriture de fichiers Python.

Ce que cette analyse permet de conclure

Nous ajustons le modèle sur le fichier et classons ce même fichier. Ce choix convient à une exploration des observations présentes. Il ne constitue pas une évaluation indépendante et ne démontre pas que les prochains lots seront correctement analysés. Pour mesurer cela, il faudrait un protocole distinct, avec données réservées et critères d’évaluation adaptés.

La valeur contamination=0.05 règle le seuil à partir des scores du fichier. Elle ne démontre pas que 5 % des lignes sont réellement erronées. Un nombre entier d’observations et d’éventuels scores égaux peuvent faire varier la proportion exacte signalée. Augmenter ce paramètre peut surtout augmenter le travail de vérification.

Les colonnes et le contexte changent les résultats. Une durée inhabituelle dans un lot peut être habituelle dans un autre ; mélanger des activités très différentes peut produire des alertes difficiles à interpréter. Le script ne comprend ni les catégories textuelles ni l’ordre temporel. Il charge aussi le fichier en mémoire : ce projet n’est pas conçu pour des exports de plusieurs gigaoctets.

Trois exercices avec les fichiers fournis

1. Retrouver une observation. Cherchez M0006 dans le CSV d’entrée et dans la sortie. Vérifiez son identifiant, ses deux mesures et son indicateur d’alerte.
2. Repérer les données manquantes. Retrouvez les deux lignes de statut donnees_manquantes. Expliquez pourquoi leur indicateur vide ne doit pas être compté comme non.
3. Changer les variables. Dans une nouvelle sortie, analysez seulement duree_ms. Comparez les identifiants signalés. Une liste différente ne prouve pas que l’une des analyses est meilleure.

Pour poursuivre votre apprentissage, notre parcours de ressources Python rassemble des exercices gratuits et des livres selon le niveau. Ce projet reste intégralement utilisable avec les fichiers proposés ici.

Projet original exécuté le 14 septembre 2026. Trente contrôles couvrent notamment les erreurs d’entrée, la conservation des observations, la répétabilité et la commande documentée. Les chiffres présentés décrivent uniquement les données synthétiques et les versions indiquées.