【Python実践】LOF（Local Outlier Factor）で異常検知を行う（コピペで使えるサンプルコード付き）

import numpy as np
from sklearn.neighbors import LocalOutlierFactor

# 学習データ（100個の値を持つ2次元の正常データ）の作成
np.random.seed(42)
X_train = 0.3 * np.random.randn(100, 2)
X_train = np.r_[X_train + 2, X_train - 2, X_train]

# LOFモデルの作成と学習＋判定
# n_neighborsは近傍点の数、contaminationは異常の割合を推定
lof = LocalOutlierFactor(n_neighbors=20, contamination=0.05)
y_pred_train = lof.fit_predict(X_train)

print("正常データの判定結果:", y_pred_train.tolist())

# テストデータを生成（異常と判定されるよう、振幅を-4～4で生成）
X_test = np.random.uniform(low=-4, high=4, size=(20, 2))

# LOFは新しいデータに対してpredictを直接使えないため、再度fit_predictを実行
y_pred_test = lof.fit_predict(np.r_[X_train, X_test])[-20:]

print("異常データの判定結果:", y_pred_test.tolist())

import matplotlib.pyplot as plt
import numpy as np
from matplotlib import rcParams
rcParams['font.family'] = 'Meiryo'


def plot_lof_results(X_train, y_train_pred, X_test, y_test_pred):
    """
    LOF（Local Outlier Factor）の判定結果を2次元散布図で可視化する
    """

    plt.figure(figsize=(8, 8))

    # 背景色（薄いブルー）
    ax = plt.gca()
    ax.set_facecolor("#eef7ff")

    # -------- 学習データ（薄い色・丸） --------
    plt.scatter(
        X_train[y_train_pred == 1, 0],
        X_train[y_train_pred == 1, 1],
        c="#b6f2b6",          # 薄い緑
        marker="o",
        s=60,
        alpha=0.8,
        edgecolor="black",
        label="学習データ（正常）"
    )

    plt.scatter(
        X_train[y_train_pred == -1, 0],
        X_train[y_train_pred == -1, 1],
        c="#ffb6b6",          # 薄い赤
        marker="o",
        s=60,
        alpha=0.8,
        edgecolor="black",
        label="学習データ（異常）"
    )

    # -------- テストデータ（濃い色・★） --------
    plt.scatter(
        X_test[y_test_pred == 1, 0],
        X_test[y_test_pred == 1, 1],
        c="darkgreen",
        marker="o",
        s=80,
        alpha=0.95,
        edgecolor="black",
        linewidth=0.8,
        label="テストデータ（正常）"
    )

    plt.scatter(
        X_test[y_test_pred == -1, 0],
        X_test[y_test_pred == -1, 1],
        c="red",
        marker="o",
        s=80,
        alpha=0.95,
        edgecolor="black",
        linewidth=0.8,
        label="テストデータ（異常）"
    )

    plt.title("LOF 異常検知結果")
    plt.xlabel("特徴量 1")
    plt.ylabel("特徴量 2")
    plt.grid(True, alpha=0.3)
    plt.legend()
    plt.tight_layout()
    plt.show()

# プロットを実行
plot_lof_results(
    X_train=X_train,
    y_train_pred=y_pred_train,
    X_test=X_test,
    y_test_pred=y_pred_test
)

from sklearn.neighbors import LocalOutlierFactor
from sklearn.model_selection import GridSearchCV
import numpy as np

# LOFのハイパーパラメータの探索範囲を設定
param_lof_grid = {
    # ★ 必須パラメータ 1: n_neighbors (近傍点の数)
    # n_neighbors の値は、データセットのサイズや密度に応じて調整が必要です
    'n_neighbors': [5, 10, 20, 30, 50],
    
    # ★ 必須パラメータ 2: contamination (外れ値の割合)
    # これはデータセット固有の値であり、モデル性能評価には直接関わりませんが、
    # 決定境界を計算する際に閾値として使用されます。
    'contamination': [0.01, 0.05, 0.1],
    
    # metric: 距離計算に使用するメトリック
    'metric': ['euclidean', 'manhattan'],
    
    # p: Minkowski距離のパラメータ (metric='minkowski'の場合に有効)
    # 例としてp=2 (Euclidean) と p=1 (Manhattan) を含めることもできます。
    'p': [1, 2],
}

# LOFのGridSearchCVのインスタンスを作成
lof_grid_search = GridSearchCV(
    LocalOutlierFactor(novelty=True),
    param_lof_grid,
    cv=5, 
    n_jobs=-1,  # 並列処理で高速化
    # 評価指標 (スコアリング): 異常検知では 'roc_auc' や 'neg_mean_squared_error' などを使用
    # ただし、LOFは教師なしのため、ここでは単純なスコアリングを指定（実際の異常検知では工夫が必要）
    scoring='neg_mean_squared_error' 
)

# グリッドサーチを実行
# lof_grid_search.fit(X_train, y_train) # 評価指標を使う場合は y_train (異常/正常のラベル) が必要

# 最適なパラメータを表示
# print("Best parameters:", lof_grid_search.best_params_)

# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～
# ここにLocalOutlierFactorUtilとグラフ描画関数のコードを張り付けるか、インポートしてください
# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～

# 学習データの生成(3つのクラスターを生成)
np.random.seed(42)
X_train_data = 0.5 * np.random.randn(100, 2)
X_train_data = np.r_[X_train_data + 1, X_train_data - 1, X_train_data] 
# 異常値（外れ値）をいくつか追加
outliers = 3 * np.random.randn(10, 2) + 10
X_train_data = np.r_[X_train_data, outliers]
df_train = pd.DataFrame(X_train_data, columns=['feature1', 'feature2'])

# インスタンスを生成
lof_util = LocalOutlierFactorUtil(df=df_train)

# モデルを学習し、学習データで異常検知を実施
# LOF (novelty=False) は fit_predict を実行し、結果を返します
model, y_pred_train = lof_util.fit(
    column=['feature1', 'feature2'], 
    n_neighbors=20, 
    contamination=0.05, 
    novelty=False # ★ novelty=False: 学習データ内の異常検知
)

print("Train predictions (1:正常, -1:異常):", y_pred_train)

# --- グラフ描画 ---
plot_lof_results(
    X_train=df_train[['feature1', 'feature2']].values, 
    y_train_pred=y_pred_train
)

# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～
# ここにLocalOutlierFactorUtilとグラフ描画関数のコードを張り付けるか、インポートしてください
# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～

# --- データの生成 ---
np.random.seed(42)
# 正常データ群 (3つのクラスター)
X_train_data = 0.5 * np.random.randn(100, 2)
X_train_data = np.r_[X_train_data + 1, X_train_data - 1, X_train_data] 
df_train = pd.DataFrame(X_train_data, columns=['feature1', 'feature2'])

# テストデータ生成（一部を異常値とする）
np.random.seed(20)
X_test_normal = 0.5 * np.random.randn(50, 2) + 0.5
X_test_outliers = 3 * np.random.randn(5, 2) + 10 # 異常値
X_test_data = np.r_[X_test_normal, X_test_outliers]
df_test = pd.DataFrame(X_test_data, columns=['feature1', 'feature2'])

# インスタンスを生成
lof_util = LocalOutlierFactorUtil(df=df_train)

# novelty=True で学習（fit のみ実行）
model_trained, _ = lof_util.fit(column=['feature1', 'feature2'], n_neighbors=20, contamination=0.05, novelty=True)

# モデルを保存
MODEL_PATH = 'lof_novelty_model.pkl'
lof_util.save_model(MODEL_PATH)

# 異常検知の実行
y_pred_test = lof_util.predict(column=['feature1', 'feature2'])
print(f"テストデータ中の異常値 (-1) の数: {np.sum(y_pred_test == -1)}")
print(f"テストデータ中の正常値 (1) の数: {np.sum(y_pred_test == 1)}")

# 学習データに対する予測を実行
y_pred_train_for_plot = lof_util.predict(column=['feature1', 'feature2']) 

plot_lof_results(
    X_train=df_train[['feature1', 'feature2']].values, 
    y_train_pred=y_pred_train_for_plot,
    model=lof_util.model
)

# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～
# ここにLocalOutlierFactorUtilとグラフ描画関数のコードを張り付けるか、インポートしてください
# ～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～～

# 学習データの生成
np.random.seed(42)
X_train_data = 0.5 * np.random.randn(100, 2)
X_train_data = np.r_[X_train_data + 1, X_train_data - 1, X_train_data]
df_train = pd.DataFrame(X_train_data, columns=['feature1', 'feature2'])

# インスタンスを生成
lof_util = LocalOutlierFactorUtil(df_train)

# novelty=True で学習（fit のみ実行）
model_trained, _ = lof_util.fit(column=['feature1', 'feature2'], n_neighbors=20, contamination=0.05, novelty=True)

# 異常検知の実行
y_pred_test = lof_util.predict(column=['feature1', 'feature2'])

# 学習データの結果
plot_lof_results(X_train=X_train_data, y_train_pred=y_pred_test,model=lof_util.model)

# モデルを保存
MODEL_PATH = 'lof_novelty_model.pkl'
lof_util.save_model(MODEL_PATH)

# LOF モデルのロード 
lof_util = LocalOutlierFactorUtil(model_path=MODEL_PATH)

# テストデータの生成 
np.random.seed(20)
X_test_data = np.random.uniform(low=-4, high=4, size=(100, 2))
df_test = pd.DataFrame(X_test_data, columns=['feature1', 'feature2'])

# 新しいデータで異常検知
y_pred_test = lof_util.predict(['feature1', 'feature2'], df=df_test)
print("Test predictions:", y_pred_test)

## テストデータの結果
plot_lof_results(X_train=X_test_data, y_train_pred=y_pred_test,model=lof_util.model)

# 学習データの生成
np.random.seed(42)
X_train_data = 0.5 * np.random.randn(100).reshape(-1, 1)
df_train = pd.DataFrame(X_train_data, columns=['feature1'])

# インスタンスを生成
lof_util = LocalOutlierFactorUtil(df_train)

# モデルを学習 (fit を使用し、novelty=True に設定)
model_trained, _ = lof_util.fit(column=['feature1'], n_neighbors=20, contamination=0.05, novelty=True)

# 学習データでの異常検知を実施 (可視化のために明示的に predict を実行)
y_pred_train = lof_util.predict(['feature1'])
print("Train predictions:", y_pred_train)

# テストデータを生成
np.random.seed(20)
# 異常値を含む広範囲のデータを作成
X_test_data = np.random.uniform(low=-4, high=4, size=(20,)).reshape(-1, 1)
df_test = pd.DataFrame(X_test_data, columns=['feature1'])

# テストデータで異常検知を実施
y_pred_test = lof_util.predict(['feature1'], df=df_test)
print("Test predictions:", y_pred_test)

# １次元用の関数を使ってグラフ描画 ---
plot_lof_results_1d(X_train=df_train[['feature1']].values,
    y_train_pred=y_pred_train,
    X_test=df_test[['feature1']].values,
    y_test_pred=y_pred_test,
    model=lof_util.model
)

from sklearn.neighbors import LocalOutlierFactor
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib import rcParams
import joblib # ★ joblib をインポート
import os
rcParams['font.family'] = 'Meiryo'


class LocalOutlierFactorUtil:
    def __init__(self, df=None, model_path=None):
        self.df = None if df is None else df.copy()
        self.model = None
        self.y_train_pred = None
        # model_path が指定されていればロードを試みる
        if model_path:
            self.load_model(model_path)

    def fit(self, column, n_neighbors=20, contamination=0.05, novelty=False):
        """
        LOFモデルの作成と学習。novelty=False で学習データの fit_predict を行う。
        """
        if self.df is None:
            raise ValueError("DataFrameがセットされていません。")
            
        if isinstance(column, list) and len(column) > 1:
            X = self.df[column].values
        else:
            X = self.df[column].values.reshape(-1, 1)

        # LOF は novelty=True なら predict が可能になる（sklearn 0.22+）
        self.model = LocalOutlierFactor(
            n_neighbors=n_neighbors,
            contamination=contamination,
            novelty=novelty
        )
        
        # novelty=True の場合、学習データは fit のみ
        if novelty:
            self.model.fit(X)
            self.y_train_pred = None
        
        # novelty=False の場合 fit_predict が必要
        else:
            self.y_train_pred = self.model.fit_predict(X)

        return self.model, self.y_train_pred

    def predict(self, column, df = None):
        """
        LOF で新しいデータを判定する。novelty=True の場合のみ使用可能。
        """       
        if self.model is None:
            raise RuntimeError("モデルが学習またはロードされていません。")
            
        if not hasattr(self.model, "predict"):
            raise RuntimeError("この LOF モデルは novelty=False のため predict できません。novelty=True で学習してください。")

        # df がNoneでなければ、インスタンス変数に代入
        if df is not None:
            self.df = df 

        if isinstance(column, list) and len(column) > 1:
            X = self.df[column].values
        else:
            X = self.df[column].values.reshape(-1, 1)

        return self.model.predict(X)

    def save_model(self, path):
        """
        学習済みモデルを joblib でファイルに保存する。
        """
        if self.model is None:
            raise RuntimeError("モデルが学習されていません。")
        
        if not self.model.novelty:
            print("警告: novelty=False のモデルは、学習データを保持するため、通常は保存・再利用に適しません。")

        joblib.dump(self.model, path)
        print(f"モデルを {path} に保存しました。")
        
    def load_model(self, path):
        """
        joblib で保存されたモデルをロードする。
        """
        if not os.path.exists(path):
            raise FileNotFoundError(f"ファイルが見つかりません: {path}")

        self.model = joblib.load(path)
        print(f"モデルを {path} からロードしました。")
        
        # novelty=True のモデルがロードされたか確認
        if not hasattr(self.model, "predict"):
             print("警告: ロードされたモデルは novelty=False のため、predict メソッドが使えません。")
        
        return self.model


import numpy as np
import matplotlib.pyplot as plt

def plot_lof_results_1d(X_train=None, y_train_pred=None,
                        X_test=None, y_test_pred=None, model=None): # modelを最後に移動し、デフォルト値をNoneに
    """
    LOF の 1次元結果をプロット。
    モデルが渡された場合、異常度スコア（decision_function）を反転させて可視化します。
    """
    plt.figure(figsize=(10, 6))
    ax = plt.gca()

    # --- データのプロット (X軸: 特徴量, Y軸: 0) ---
    
    # 学習データ
    if X_train is not None and y_train_pred is not None:
        # X_train は (N, 1) の形状を想定
        X_train_flat = X_train.flatten()
        
        plt.scatter(X_train_flat[y_train_pred == 1],
                    np.zeros_like(X_train_flat[y_train_pred == 1]),
                    c='lightgreen', edgecolors='k', s=70, label="正常（学習）", zorder=3)

        plt.scatter(X_train_flat[y_train_pred == -1],
                    np.zeros_like(X_train_flat[y_train_pred == -1]),
                    c='lightcoral', edgecolors='k', s=70, label="異常（学習）", zorder=3)

    # テストデータ
    if X_test is not None and y_test_pred is not None:
        X_test_flat = X_test.flatten()
        
        plt.scatter(X_test_flat[y_test_pred == 1],
                    np.zeros_like(X_test_flat[y_test_pred == 1]),
                    c='green', edgecolors='k', s=90, marker='*',
                    label="正常（テスト）", zorder=4)

        plt.scatter(X_test_flat[y_test_pred == -1],
                    np.zeros_like(X_test_flat[y_test_pred == -1]),
                    c='red', edgecolors='k', s=90, marker='*',
                    label="異常（テスト）", zorder=4)

    # --- 決定関数プロット (モデルが存在する場合のみ実行) ---
    
    # プロット対象のデータ（X）が存在し、かつモデルが渡された場合
    if (X_train is not None or X_test is not None) and model is not None:
        
        # 描画範囲を決定するために全てのデータを結合
        xx = np.concatenate([arr.flatten() for arr in [X_train, X_test] if arr is not None])
        x_range = np.linspace(xx.min() - 1, xx.max() + 1, 400).reshape(-1, 1)

        try:
            # LOFの決定関数 Z を計算し、反転させる (反転後: 値が大きいほど異常度が高い)
            Z = -model.decision_function(x_range)
            
            # 決定関数をプロット
            plt.plot(x_range, Z, color='blue', linestyle='--', linewidth=2, label="反転異常度スコア", zorder=2)
            
            # 異常の閾値（境界）を示す線
            if hasattr(model, 'offset_'):
                # Sklearn 0.22+の novelty=True で学習した場合に利用可能
                threshold = -model.offset_ 
            else:
                # 訓練データから閾値を計算（novelty=False の場合）
                if X_train is not None:
                     raw_scores = model.decision_function(X_train.reshape(-1, 1))
                     threshold = -np.percentile(raw_scores, 100. * (1. - model.contamination))
                else:
                     # 簡易的な可視化のため、Zの範囲とcontaminationから暫定閾値を計算
                     threshold = Z.min() + (Z.max() - Z.min()) * model.contamination

            plt.axhline(threshold, color='red', linestyle='-', linewidth=2, label="判定閾値", zorder=1)
            
            # Y軸の範囲をスコアに合わせて調整し、データ点のY軸 (0) も含むようにする
            min_z, max_z = Z.min(), Z.max()
            y_lim_min = min(min_z, -0.5)
            y_lim_max = max(max_z, 0.5)
            plt.ylim(y_lim_min, y_lim_max)

        except AttributeError:
            print("警告: モデルに decision_function がないため、スコアカーブはスキップされます。")

    # データ点のY軸位置
    plt.axhline(0, color='gray', linestyle=':', alpha=0.5) 
    
    # --- グラフの体裁 ---
    title_suffix = "とスコアカーブ" if model is not None else ""
    plt.title(f"LOF 異常検知（1次元）{title_suffix}")
    plt.xlabel("特徴量 X")
    plt.ylabel("異常度スコア (反転済み)")
    plt.grid(True, alpha=0.6)
    plt.legend()
    plt.tight_layout()
    plt.show()

def plot_lof_results(X_train=None, y_train_pred=None, X_test=None, y_test_pred=None, model=None):
    """
    LOF の 2次元結果をプロットします。
    モデルが渡された場合、決定境界を背景に描画します。

    Parameters
    ----------
    X_train : np.ndarray, optional
        学習データの特徴量 (N, 2)。
    y_train_pred : np.ndarray, optional
        学習データの LOF 判定結果 (1:正常, -1:異常)。
    X_test : np.ndarray, optional
        テストデータの特徴量 (N, 2)。
    y_test_pred : np.ndarray, optional
        テストデータの LOF 判定結果 (1:正常, -1:異常)。
    model : object, optional
        学習済みの LOF モデルオブジェクト。Noneの場合、決定境界は描画しません。
    """
    plt.figure(figsize=(10, 8))
    ax = plt.gca()
    
    # ★ 1. データを結合し、プロット範囲を決定（常に実行）
    all_data_arrays = [arr for arr in [X_train, X_test] if arr is not None]
    
    if all_data_arrays:
        X = np.concatenate(all_data_arrays)
    
    # --- 2. 決定境界の計算と描画 (モデルとデータが提供された場合のみ実行) ---
    if model is not None and all_data_arrays:
        
        # 描画範囲を決定
        x_min_plot, x_max_plot = X[:, 0].min() - 1, X[:, 0].max() + 1
        y_min_plot, y_max_plot = X[:, 1].min() - 1, X[:, 1].max() + 1
        
        # グリッドを作成
        xx, yy = np.meshgrid(np.linspace(x_min_plot, x_max_plot, 100),
                             np.linspace(y_min_plot, y_max_plot, 100))

        # モデルの決定関数を使って異常度スコアを計算 (Z)
        try:
            Z = model.decision_function(np.c_[xx.ravel(), yy.ravel()])
            Z = Z.reshape(xx.shape)
        except AttributeError as e:
            print(f"警告: モデルに decision_function がありません。決定境界はスキップされます。エラー: {e}")
            Z = None # スコア計算に失敗した場合、決定境界の描画をスキップ

        if Z is not None:
            # 閾値（決定境界）を計算
            if hasattr(model, 'offset_'):
                threshold = model.offset_
            elif X_train is not None and hasattr(model, 'contamination'):
                # 訓練データから生の閾値を計算
                raw_scores = model.decision_function(X_train)
                threshold = np.percentile(raw_scores, 100. * (1. - model.contamination))
            else:
                threshold = 0.0 # 代替値
                
            # 背景色（異常度スコアの等高線塗りつぶし）
            plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), threshold, 7),
                          cmap=plt.cm.coolwarm, alpha=0.3)
            
            # 決定境界線
            plt.contour(xx, yy, Z, levels=[threshold], linewidths=2, colors='black', linestyles='--')
            
            # 決定境界を描画する場合、軸の範囲も設定
            plt.xlim(x_min_plot, x_max_plot)
            plt.ylim(y_min_plot, y_max_plot)
    
    # --- 3. データ点の描画 (ロジック) ---
    
    # データが存在する場合のみ背景色を設定
    if all_data_arrays:
        ax.set_facecolor("#eef7ff")
    
    # 学習データが存在する場合
    if X_train is not None and y_train_pred is not None:
        plt.scatter(X_train[y_train_pred == 1, 0], X_train[y_train_pred == 1, 1],
                    c="#b6f2b6", marker="o", s=60, alpha=0.8, edgecolor="black", label="学習データ（正常）", zorder=3)
        plt.scatter(X_train[y_train_pred == -1, 0], X_train[y_train_pred == -1, 1],
                    c="#ffb6b6", marker="o", s=60, alpha=0.8, edgecolor="black", label="学習データ（異常）", zorder=3)

    # テストデータが存在する場合
    if X_test is not None and y_test_pred is not None:
        plt.scatter(X_test[y_test_pred == 1, 0], X_test[y_test_pred == 1, 1],
                    c="green", marker="o", s=80, alpha=0.95, edgecolor="black", linewidth=0.8, label="テストデータ（正常）", zorder=4)
        plt.scatter(X_test[y_test_pred == -1, 0], X_test[y_test_pred == -1, 1],
                    c="red", marker="o", s=80, alpha=0.95, edgecolor="black", linewidth=0.8, label="テストデータ（異常）", zorder=4)

    # --- 4. グラフの体裁 ---
    
    # ★ モデルが None の場合、データに基づいて軸の範囲を自動調整
    if model is None and all_data_arrays:
        # データの外側に少し余白を持たせる (描画がscatterに任せきりになるのを防ぐ)
        x_min, x_max = X[:, 0].min(), X[:, 0].max()
        y_min, y_max = X[:, 1].min(), X[:, 1].max()
        x_padding = (x_max - x_min) * 0.1 if (x_max - x_min) > 0 else 1
        y_padding = (y_max - y_min) * 0.1 if (y_max - y_min) > 0 else 1
        
        plt.xlim(x_min - x_padding, x_max + x_padding)
        plt.ylim(y_min - y_padding, y_max + y_padding)
    
    title_suffix = "と決定境界" if model is not None else ""
    plt.title(f"LOF 異常検知結果{title_suffix}")
    plt.xlabel("特徴量 1")
    plt.ylabel("特徴量 2")
    plt.grid(True, alpha=0.3)
    plt.legend()
    plt.tight_layout()
    plt.show()