Detección de valores atípicos utilizando algoritmos de Scikit-Learn

Introducción

Esta práctica demuestra cómo utilizar Scikit-Learn para realizar la detección de valores atípicos en conjuntos de datos de detección de anomalías clásicos utilizando los algoritmos de factor de atipicidad local (LOF, por sus siglas en inglés) y bosque de aislamiento (IForest). El rendimiento de los algoritmos se evalúa en un contexto de detección de valores atípicos, y se utilizan curvas ROC para graficar los resultados.

Consejos sobre la VM

Una vez finalizada la inicialización de la VM, haga clic en la esquina superior izquierda para cambiar a la pestaña Cuaderno y acceder a Jupyter Notebook para practicar.

A veces, es posible que tenga que esperar unos segundos a que Jupyter Notebook termine de cargarse. La validación de las operaciones no se puede automatizar debido a las limitaciones de Jupyter Notebook.

Si tiene problemas durante el aprendizaje, no dude en preguntar a Labby. Deje sus comentarios después de la sesión y lo resolveremos rápidamente para usted.

Skills Graph

Preprocesamiento de datos

El primer paso es preprocesar el conjunto de datos. En este ejemplo, utilizamos conjuntos de datos del mundo real disponibles en el módulo datasets de Scikit-Learn. El tamaño de la muestra de algunos conjuntos de datos se reduce para acelerar el cálculo. Después del preprocesamiento de los datos, las etiquetas de los conjuntos de datos tendrán dos clases, 0 que representa valores normales y 1 que representa valores atípicos. La función preprocess_dataset devuelve los datos y la etiqueta.

import numpy as np
from sklearn.datasets import fetch_kddcup99, fetch_covtype, fetch_openml
from sklearn.preprocessing import LabelBinarizer
import pandas as pd

rng = np.random.RandomState(42)

def preprocess_dataset(dataset_name):
    ## carga y vectorización
    print(f"Cargando datos de {dataset_name}")
    if dataset_name in ["http", "smtp", "SA", "SF"]:
        dataset = fetch_kddcup99(subset=dataset_name, percent10=True, random_state=rng)
        X = dataset.data
        y = dataset.target
        lb = LabelBinarizer()

        if dataset_name == "SF":
            idx = rng.choice(X.shape[0], int(X.shape[0] * 0.1), replace=False)
            X = X[idx]  ## reduce el tamaño de la muestra
            y = y[idx]
            x1 = lb.fit_transform(X[:, 1].astype(str))
            X = np.c_[X[:, :1], x1, X[:, 2:]]
        elif dataset_name == "SA":
            idx = rng.choice(X.shape[0], int(X.shape[0] * 0.1), replace=False)
            X = X[idx]  ## reduce el tamaño de la muestra
            y = y[idx]
            x1 = lb.fit_transform(X[:, 1].astype(str))
            x2 = lb.fit_transform(X[:, 2].astype(str))
            x3 = lb.fit_transform(X[:, 3].astype(str))
            X = np.c_[X[:, :1], x1, x2, x3, X[:, 4:]]
        y = (y!= b"normal.").astype(int)
    if dataset_name == "forestcover":
        dataset = fetch_covtype()
        X = dataset.data
        y = dataset.target
        idx = rng.choice(X.shape[0], int(X.shape[0] * 0.1), replace=False)
        X = X[idx]  ## reduce el tamaño de la muestra
        y = y[idx]

        ## los valores normales son aquellos con atributo 2
        ## los valores atípicos son aquellos con atributo 4
        s = (y == 2) + (y == 4)
        X = X[s, :]
        y = y[s]
        y = (y!= 2).astype(int)
    if dataset_name in ["glass", "wdbc", "cardiotocography"]:
        dataset = fetch_openml(
            name=dataset_name, version=1, as_frame=False, parser="pandas"
        )
        X = dataset.data
        y = dataset.target

        if dataset_name == "glass":
            s = y == "tableware"
            y = s.astype(int)
        if dataset_name == "wdbc":
            s = y == "2"
            y = s.astype(int)
            X_mal, y_mal = X[s], y[s]
            X_ben, y_ben = X[~s], y[~s]

            ## submuestreo a 39 puntos (9.8% de valores atípicos)
            idx = rng.choice(y_mal.shape[0], 39, replace=False)
            X_mal2 = X_mal[idx]
            y_mal2 = y_mal[idx]
            X = np.concatenate((X_ben, X_mal2), axis=0)
            y = np.concatenate((y_ben, y_mal2), axis=0)
        if dataset_name == "cardiotocography":
            s = y == "3"
            y = s.astype(int)
    ## 0 representa valores normales, y 1 representa valores atípicos
    y = pd.Series(y, dtype="category")
    return (X, y)

Función de predicción de valores atípicos

El siguiente paso es definir una función de predicción de valores atípicos. En este ejemplo, utilizamos los algoritmos LocalOutlierFactor e IsolationForest. La función compute_prediction devuelve la puntuación media de valores atípicos de X.

from sklearn.neighbors import LocalOutlierFactor
from sklearn.ensemble import IsolationForest

def compute_prediction(X, model_name):
    print(f"Calculando la predicción con {model_name}...")
    if model_name == "LOF":
        clf = LocalOutlierFactor(n_neighbors=20, contamination="auto")
        clf.fit(X)
        y_pred = clf.negative_outlier_factor_
    if model_name == "IForest":
        clf = IsolationForest(random_state=rng, contamination="auto")
        y_pred = clf.fit(X).decision_function(X)
    return y_pred

Graficar e interpretar los resultados

El último paso es graficar e interpretar los resultados. El rendimiento del algoritmo está relacionado con lo bien que se desempeña la tasa de verdaderos positivos (TPR) a un bajo valor de la tasa de falsos positivos (FPR). Los mejores algoritmos tienen la curva en la esquina superior izquierda de la gráfica y el área bajo la curva (AUC) cercana a 1. La línea discontinua diagonal representa una clasificación aleatoria de valores atípicos y valores normales.

import math
import matplotlib.pyplot as plt
from sklearn.metrics import RocCurveDisplay

datasets_name = [
    "http",
    "smtp",
    "SA",
    "SF",
    "forestcover",
    "glass",
    "wdbc",
    "cardiotocography",
]

models_name = [
    "LOF",
    "IForest",
]

## parámetros de la gráfica
cols = 2
linewidth = 1
pos_label = 0  ## significa que 0 pertenece a la clase positiva
rows = math.ceil(len(datasets_name) / cols)

fig, axs = plt.subplots(rows, cols, figsize=(10, rows * 3), sharex=True, sharey=True)

for i, dataset_name in enumerate(datasets_name):
    (X, y) = preprocess_dataset(dataset_name=dataset_name)

    for model_idx, model_name in enumerate(models_name):
        y_pred = compute_prediction(X, model_name=model_name)
        display = RocCurveDisplay.from_predictions(
            y,
            y_pred,
            pos_label=pos_label,
            name=model_name,
            linewidth=linewidth,
            ax=axs[i // cols, i % cols],
            plot_chance_level=(model_idx == len(models_name) - 1),
            chance_level_kw={
                "linewidth": linewidth,
                "linestyle": ":",
            },
        )
    axs[i // cols, i % cols].set_title(dataset_name)
plt.tight_layout(pad=2.0)  ## espaciado entre los subgráficos
plt.show()

Resumen

Esta práctica demostró cómo utilizar Scikit-Learn para realizar la detección de valores atípicos en conjuntos de datos de detección de anomalías clásicos utilizando los algoritmos de factor de atipicidad local (LOF, por sus siglas en inglés) y bosque de aislamiento (IForest). El rendimiento de los algoritmos se evaluó en un contexto de detección de valores atípicos, y se utilizaron curvas ROC para graficar los resultados.