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

import numpy as np
import pandas as pd
from sklearn.svm import OneClassSVM

# 学習データ（100個の値を持つ２次元の正常データ）を生成
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]

# OneClassSVM モデルの定義と学習（X_train は２次元以上のデータが必要）
model = OneClassSVM(nu=0.01, gamma='auto').fit(X_train)

# 出来上がったモデルに学習データを投入し、判定させてみる
y_pred_train = model.predict(X_train)
print("正常データの判定結果:",y_pred_train.tolist())

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

# モデルに異常データを投入し、判定させてみる
y_pred_outliers = model.predict(X_outliers)
print("異常データの判定結果:",y_pred_outliers.tolist())

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

def plot_ocsvm_results(X_train, y_pred_train, X_test, y_pred_test, model):
    """
    OneClassSVMモデルの異常検知結果を可視化します。

    Args:
        X_train (ndarray): 学習データの特徴量。
        y_pred_train (ndarray): 学習データに対する予測ラベル (1: 正常, -1: 異常)。
        X_test (ndarray): テストデータの特徴量。
        y_pred_test (ndarray): テストデータに対する予測ラベル。
        model (OneClassSVM): 学習済みのOneClassSVMモデル。
    Returns:
        None: グラフを表示します。
    """
    # ======== グラフによる可視化 ========
    plt.figure(figsize=(8, 6))

    # 正常データの表示（分類結果ごとに色分け）
    plt.scatter(X_train[y_pred_train == 1][:, 0], X_train[y_pred_train == 1][:, 1], 
                c='lightblue', edgecolors='k', s=70, label="正常データ")
    plt.scatter(X_train[y_pred_train == -1][:, 0], X_train[y_pred_train == -1][:, 1], 
                c='orange', edgecolors='k', s=70, label="異常と検出されたデータ（学習データ）")

    # 異常データの表示（分類結果ごとに色分け）
    plt.scatter(X_test[y_pred_test == 1][:, 0], X_test[y_pred_test == 1][:, 1], 
                c='green', edgecolors='k', s=70, label="異常データ（正常と検出）")
    plt.scatter(X_test[y_pred_test == -1][:, 0], X_test[y_pred_test == -1][:, 1], 
                c='red', edgecolors='k', s=70, label="異常データ（異常と検出）")

    # 境界の等高線を描画するための範囲を設定
    xx, yy = np.meshgrid(np.linspace(-5, 5, 500), np.linspace(-5, 5, 500))

    # OneClassSVM モデルの決定関数の結果を計算
    Z = model.decision_function(np.c_[xx.ravel(), yy.ravel()])
    Z = Z.reshape(xx.shape)

    # 境界の等高線を描画
    plt.contour(xx, yy, Z, levels=[0], linewidths=2, colors='black')
    plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), 0, 7), cmap=plt.cm.Blues_r, alpha=0.3)

    # グラフの設定
    plt.legend()
    plt.title("OneClassSVM 異常検知と決定境界", fontsize=16)
    plt.xlabel("特徴量 1", fontsize=12)
    plt.ylabel("特徴量 2", fontsize=12)
    plt.grid(True, linestyle='--', alpha=0.7)

    # 結果を表示
    plt.show()

# ======== 可視化 ========
plot_ocsvm_results(X_train, y_pred_train, X_outliers, y_pred_outliers, model)

from sklearn.model_selection import GridSearchCV

param_grid = {'nu': [0.1, 0.5, 0.9], 'gamma': ['scale', 'auto']}
grid_search = GridSearchCV(OneClassSVM(), param_grid, cv=5)
grid_search.fit(X_train)
print("Best parameters:", grid_search.best_params_)

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

# 学習データの生成
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'])

# インスタンスを生成
oc_svm = OneClassSvmUtil(df_train)

# モデルを学習
oc_svm.fit(['feature1', 'feature2'], nu=0.01)

# 学習データで異常検知を実施
y_pred_train = oc_svm.predict(['feature1', 'feature2'])
print("Train predictions:", y_pred_train)

# モデルの保存
oc_svm.save_model('one_class_svm_model.pkl')

# グラフ描画
plot_ocsvm_results(oc_svm.model,X_train=df_train[['feature1', 'feature2']].values, y_pred_train=y_pred_train)

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

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

# インスタンスの生成とモデルの読み込み
oc_svm_new = OneClassSvmUtil(model_path='one_class_svm_model.pkl')
print("読み込んだモデル:", oc_svm_new.model)

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

# グラフ描画
plot_ocsvm_results(oc_svm_new.model,X_test=df_test[['feature1', 'feature2']].values, y_pred_test=y_pred_test)

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

# 学習データの生成
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'])

# インスタンスを生成
oc_svm = OneClassSvmUtil(df_train)

# モデルを学習
oc_svm.fit(['feature1', 'feature2'], nu=0.01)

# 学習データで異常検知を実施
y_pred_train = oc_svm.predict(['feature1', 'feature2'])
print("Train predictions:", y_pred_train)

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

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

# 学習データとテストデータの検知結果を合わせてグラフ描画
plot_ocsvm_results(oc_svm.model,df_train[['feature1', 'feature2']].values, y_pred_train,df_test[['feature1', 'feature2']].values, y_pred_test)

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

# 学習データの生成
np.random.seed(42)
X_train_data = 0.5 * np.random.randn(100).reshape(-1, 1)
# 1次元データを2次元配列に変換
df_train = pd.DataFrame(X_train_data, columns=['feature1'])

# インスタンスを生成
oc_svm = OneClassSvmUtil(df_train)

# モデルを学習
oc_svm.fit(['feature1'], nu=0.01)

# 学習データで異常検知を実施
y_pred_train = oc_svm.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 = oc_svm.predict(['feature1'], df=df_test)
print("Test predictions:", y_pred_test)

# １次元用の関数を使ってグラフ描画
plot_ocsvm_results_1d(
    oc_svm.model,  # 学習したモデルを直接使用
    df_train[['feature1']].values,
    y_pred_train,
    df_test[['feature1']].values,
    y_pred_test
)

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


def plot_ocsvm_results_1d(model, X_train=None, y_pred_train=None, X_test=None, y_pred_test=None):
    """
    1次元データのOne-Class SVMの結果をプロットする。

    学習データとテストデータを表示し、正常データと異常と検出されたデータを色分けしてプロットする。
    また、決定境界を描画する。

    Parameters:
    model : One-Class SVMモデル
        学習済みのOne-Class SVMモデル。
    X_train : array-like, shape (n_samples,), optional
        学習データの特徴量。1次元データを想定。
    y_pred_train : array-like, shape (n_samples,), optional
        学習データに対する予測結果。1（正常）または-1（異常）。
    X_test : array-like, shape (n_samples,), optional
        テストデータの特徴量。1次元データを想定。
    y_pred_test : array-like, shape (n_samples,), optional
        テストデータに対する予測結果。1（正常）または-1（異常）。
    """
    plt.figure(figsize=(8, 6))

    # 学習データのプロット
    if X_train is not None and y_pred_train is not None:
        plt.scatter(X_train[y_pred_train == 1], np.zeros_like(X_train[y_pred_train == 1]),
                    c='lightblue', edgecolors='k', s=70, label="正常データ")
        plt.scatter(X_train[y_pred_train == -1], np.zeros_like(X_train[y_pred_train == -1]),
                    c='orange', edgecolors='k', s=70, label="異常と検出されたデータ（学習データ）")

    # テストデータのプロット
    if X_test is not None and y_pred_test is not None:
        plt.scatter(X_test[y_pred_test == 1], np.zeros_like(X_test[y_pred_test == 1]),
                    c='green', edgecolors='k', s=70, label="異常データ（正常として検出）")
        plt.scatter(X_test[y_pred_test == -1], np.zeros_like(X_test[y_pred_test == -1]),
                    c='red', edgecolors='k', s=70, label="異常データ（異常として検出）")

    # 決定境界の描画
    if X_train is not None:
        x_range = np.linspace(X_train.min() - 1, X_train.max() + 1, 500).reshape(-1, 1)
    elif X_test is not None:
        x_range = np.linspace(X_test.min() - 1, X_test.max() + 1, 500).reshape(-1, 1)
    else:
        # 両方のデータがない場合は固定範囲を使用（適宜調整）
        x_range = np.linspace(-5, 5, 500).reshape(-1, 1)

    Z = model.decision_function(x_range)
    plt.plot(x_range, Z, color='black', linewidth=2, label='決定境界')

    # グラフの設定
    plt.axhline(0, color='black', linewidth=2)
    plt.legend()
    plt.title("One-Class SVM 異常検知（1次元）", fontsize=16)
    plt.xlabel("特徴量", fontsize=12)
    plt.ylabel("値", fontsize=12)
    plt.grid(True, linestyle='--', alpha=0.7)

    # 結果を表示
    plt.show()

def plot_ocsvm_results(model, X_train=None, y_pred_train=None, X_test=None, y_pred_test=None):
    """
    2次元データのOne-Class SVMの結果をプロットする。

    学習データとテストデータを表示し、正常データと異常と検出されたデータを色分けしてプロットする。
    また、決定境界を描画する。

    Parameters:
    model : One-Class SVMモデル
        学習済みのOne-Class SVMモデル。
    X_train : array-like, shape (n_samples, 2), optional
        学習データの特徴量。2次元データを想定。
    y_pred_train : array-like, shape (n_samples,), optional
        学習データに対する予測結果。1（正常）または-1（異常）。
    X_test : array-like, shape (n_samples, 2), optional
        テストデータの特徴量。2次元データを想定。
    y_pred_test : array-like, shape (n_samples,), optional
        テストデータに対する予測結果。1（正常）または-1（異常）。
    """
    
    # 描画サイズの設定
    plt.figure(figsize=(8, 6))

    # 学習データのプロット（2次元データのみ）
    if X_train is not None and y_pred_train is not None and X_train.ndim > 1:
        plt.scatter(X_train[y_pred_train == 1][:, 0], X_train[y_pred_train == 1][:, 1],
                    c='lightblue', edgecolors='k', s=70, label="正常データ")
        plt.scatter(X_train[y_pred_train == -1][:, 0], X_train[y_pred_train == -1][:, 1],
                    c='orange', edgecolors='k', s=70, label="異常と検出されたデータ（学習データ）")

    # テストデータのプロット（2次元データのみ）
    if X_test is not None and y_pred_test is not None and X_test.ndim > 1:
        plt.scatter(X_test[y_pred_test == 1][:, 0], X_test[y_pred_test == 1][:, 1],
                    c='green', edgecolors='k', s=70, label="異常データ（正常として検出）")
        plt.scatter(X_test[y_pred_test == -1][:, 0], X_test[y_pred_test == -1][:, 1],
                    c='red', edgecolors='k', s=70, label="異常データ（異常として検出）")

    # 境界の等高線を描画するための範囲を設定
    if X_train is not None and X_train.ndim > 1:
        xx, yy = np.meshgrid(np.linspace(X_train[:, 0].min() - 1, X_train[:, 0].max() + 1, 500),
                             np.linspace(X_train[:, 1].min() - 1, X_train[:, 1].max() + 1, 500))
    else:
        # X_trainがない場合、テストデータの範囲を使用
        if X_test is not None and X_test.ndim > 1:
            xx, yy = np.meshgrid(np.linspace(X_test[:, 0].min() - 1, X_test[:, 0].max() + 1, 500),
                                 np.linspace(X_test[:, 1].min() - 1, X_test[:, 1].max() + 1, 500))
        else:
            # X_trainとX_testがない場合は範囲を固定（適宜調整）
            xx, yy = np.meshgrid(np.linspace(-5, 5, 500), np.linspace(-5, 5, 500))

    x_range = np.c_[xx.ravel(), yy.ravel()]

    # OneClassSVM モデルの決定関数の結果を計算
    Z = model.decision_function(x_range)
    Z = Z.reshape(xx.shape)

    # 境界の等高線を描画
    plt.contour(xx, yy, Z, levels=[0], linewidths=2, colors='black')
    plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), 0, 7), cmap=plt.cm.Blues_r, alpha=0.3)

    # グラフの設定
    plt.legend()
    plt.title("One-Class SVM 異常検知と決定境界", fontsize=16)
    plt.xlabel("特徴量 1", fontsize=12)
    plt.ylabel("特徴量 2", fontsize=12)
    plt.grid(True, linestyle='--', alpha=0.7)

    # 結果を表示
    plt.show()

import numpy as np
import pandas as pd
from sklearn import svm
import pickle
from sklearn.model_selection import GridSearchCV

class OneClassSvmUtil:
    def __init__(self, df=None,model_path=None):
        """
        OneClassSvmクラスの初期化メソッド
        :param df: 初期データフレーム (pd.DataFrame)
        """
        self.df = None if df is None else df.copy()  # データフレームのコピーを保持
        self.model = None if model_path is None else self.load_model(model_path) # SVMモデルファイルのパス
        self.pred = []  # 予測結果のリスト

    def fit(self, column, nu=0.1, gamma="auto",kernel="rbf"):
        """
        One-Class SVMモデルの作成と学習
        :param column: 学習に使用するカラム名 (str または list)
        :param nu: SVMのnuパラメータ (float)
        :param gamma: SVMのgammaパラメータ (str または float)
        :param kernel: 使用するカーネルの種類 (str)
            - 'linear': 線形カーネル
            - 'poly': 多項式カーネル
            - 'rbf': RBFカーネル（ガウシアンカーネル⇒デフォルト）
            - 'sigmoid': シグモイドカーネル
        :return: 学習したモデル
        """

        # 指定されたカラムのデータを抽出
        if isinstance(column, list) and len(column) > 1:
            X = self.df[column].values
        else:
            X = self.df[column].values.reshape(-1, 1)  # 複数カラムのデータを2次元配列に変換

        # One-Class SVMモデルの作成
        self.model = svm.OneClassSVM(nu=nu, gamma=gamma,kernel=kernel).fit(X)
        return self.model

    def predict(self, column,df=None):
        """
        学習したモデルを使用して予測を行う
        :param column: 学習に使用するカラム名 (str または list)
        :return: 予測結果の配列
        """
        
        # 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)  # 複数カラムのデータを2次元配列に変換

        # 予測する
        self.pred = self.model.predict(X) 
        return self.pred

    def read_csv(self, file_name, encoding="shift-jis"):
        """
        CSVファイルからデータを読み込むメソッド
        :param file_name: 読み込むファイルのパス (str)
        :param encoding: ファイルのエンコーディング (str)
        """
        self.df = pd.read_csv(file_name, encoding=encoding)  # データフレームに読み込み

    def save_model(self, model_path):
        """
        学習したモデルをファイルに保存するメソッド
        :param model_path: 保存するファイルのパス (str)
        """
        self.model_path = model_path
        with open(model_path, 'wb') as f:
            pickle.dump(self.model, f)  # モデルをバイナリ形式で保存

    def load_model(self, model_path=None):
        """
        モデルをファイルから読み込むメソッド
        :param model_path: 読み込むファイルのパス (str)
        :return: 読み込んだモデル
        """
        if model_path is None:
            model_path = self.model_path

        with open(model_path, 'rb') as f:
            self.model = pickle.load(f)  # モデルをバイナリ形式で読み込み
        return self.model