Scikit-Learn アルゴリズムを用いたアウトライア検出

はじめに

この実験では、Scikit-Learn を使用して、局所的なアウトライアー係数（LOF）とアイソレーションフォレスト（IForest）アルゴリズムを使って、古典的な異常検出データセットに対してアウトライアー検出を行う方法を示します。アウトライアー検出のコンテキストでアルゴリズムの性能を評価し、結果をプロットするために ROC 曲線を使用します。

VM のヒント

VM の起動が完了したら、左上隅をクリックしてノートブックタブに切り替えて、Jupyter Notebook を使った練習にアクセスします。

時々、Jupyter Notebook が読み込み終わるまで数秒待つ必要があります。Jupyter Notebook の制限により、操作の検証を自動化することはできません。

学習中に問題に遭遇した場合は、Labby にお問い合わせください。セッション後にフィードバックを提供してください。そうすれば、迅速に問題を解決します。

データ前処理

最初のステップは、データセットを前処理することです。この例では、Scikit-Learn のdatasetsモジュールにある実際のデータセットを使用します。一部のデータセットのサンプルサイズを削減して、計算を高速化します。データ前処理後、データセットのターゲットは 2 つのクラスになり、0 はインライアを表し、1 はアウトライアを表します。preprocess_dataset関数はデータとターゲットを返します。

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):
    ## ロードとベクトル化
    print(f"{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]  ## サンプルサイズを削減
            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]  ## サンプルサイズを削減
            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]  ## サンプルサイズを削減
        y = y[idx]

        ## インライアは属性 2 を持つもの
        ## アウトライアは属性 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]

            ## 39 点にダウンサンプリング (9.8% アウトライア)
            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 はインライアを表し、1 はアウトライアを表す
    y = pd.Series(y, dtype="category")
    return (X, y)

アウトライア予測関数

次のステップは、アウトライア予測関数を定義することです。この例では、LocalOutlierFactor と IsolationForest アルゴリズムを使用します。compute_prediction 関数は、X の平均アウトライアスコアを返します。

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

def compute_prediction(X, model_name):
    print(f"{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

結果のプロットと解釈

最後のステップは、結果をプロットして解釈することです。アルゴリズムの性能は、偽陽性率（FPR）が低い値のときの真陽性率（TPR）がどれだけ良いかに関係します。最良のアルゴリズムは、プロットの左上に曲線があり、曲線下の面積（AUC）が 1 に近いものです。対角線の破線は、アウトライアとインライアのランダムな分類を表します。

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",
]

## プロットパラメータ
cols = 2
linewidth = 1
pos_label = 0  ## 平均的に 0 は陽性クラスに属する
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)  ## サブプロット間の間隔
plt.show()

まとめ

この実験では、Scikit-Learn を使って、局所的なアウトライア係数（LOF）とアイソレーションフォレスト（IForest）アルゴリズムを用いて、古典的な異常検出データセットに対するアウトライア検出を行う方法を示しました。アウトライア検出のコンテキストでアルゴリズムの性能を評価し、結果をプロットするために ROC 曲線を使用しました。