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

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.optimizers import Adam

# 1. 正常データの生成（2次元の3クラスタ）
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]

# 2. オートエンコーダの構築
input_dim = X_train.shape[1]
encoding_dim = 2

input_layer = Input(shape=(input_dim,))
encoded = Dense(encoding_dim, activation='relu')(input_layer)
decoded = Dense(input_dim, activation='linear')(encoded)

autoencoder = Model(inputs=input_layer, outputs=decoded)
autoencoder.compile(optimizer=Adam(learning_rate=0.01), loss='mse')

# 3. 学習
autoencoder.fit(X_train, X_train, epochs=100, batch_size=16, verbose=0)

# 4. テストデータの生成（異常値を含む）
X_test = np.random.uniform(low=-4, high=4, size=(20, 2))
X_all = np.vstack([X_train, X_test])

# 5. 再構成誤差の計算
X_pred = autoencoder.predict(X_all)
mse = np.mean(np.square(X_all - X_pred), axis=1)

# 6. 異常判定（閾値は経験的に設定）
threshold = np.percentile(mse[:len(X_train)], 95)
y_pred = (mse > threshold).astype(int)  # 1: 異常, 0: 正常

print("異常判定結果（1:異常, 0:正常）:", y_pred[-20:].tolist())

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

def plot_ae_results(X_train, y_train_pred, X_test, y_test_pred):
    """
    オートエンコーダによる異常検知結果を2次元散布図で可視化する
    """

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

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

    # -------- 学習データ（薄い色・丸） --------
    plt.scatter(
        X_train[y_train_pred == 0, 0],
        X_train[y_train_pred == 0, 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 == 0, 0],
        X_test[y_test_pred == 0, 1],
        c="darkgreen",
        marker="*",
        s=120,
        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="*",
        s=120,
        alpha=0.95,
        edgecolor="black",
        linewidth=0.8,
        label="テストデータ（異常）"
    )

    plt.title("オートエンコーダ 異常検知結果")
    plt.xlabel("特徴量 1")
    plt.ylabel("特徴量 2")
    plt.grid(True, alpha=0.3)
    plt.legend()
    plt.tight_layout()
    plt.show()

# プロットを実行
y_pred_train = y_pred[:len(X_train)] 
y_pred_test = y_pred[len(X_train):]

plot_ae_results(
    X_train=X_train,
    y_train_pred=y_pred_train,
    X_test=X_test,
    y_test_pred=y_pred_test
)

from sklearn.model_selection import GridSearchCV
from tensorflow.keras.wrappers.scikit_learn import KerasRegressor
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.optimizers import Adam

# モデル構築関数（KerasRegressor用）
def build_autoencoder(encoding_dim=2, learning_rate=0.01):
    input_dim = X_train.shape[1]
    input_layer = Input(shape=(input_dim,))
    encoded = Dense(encoding_dim, activation='relu')(input_layer)
    decoded = Dense(input_dim, activation='linear')(encoded)
    autoencoder = Model(inputs=input_layer, outputs=decoded)
    autoencoder.compile(optimizer=Adam(learning_rate=learning_rate), loss='mse')
    return autoencoder

# ラッパーでKerasモデルをscikit-learn互換に
ae_regressor = KerasRegressor(build_fn=build_autoencoder, epochs=100, batch_size=16, verbose=0)

# 探索するパラメータの範囲
param_grid = {
    'encoding_dim': [1, 2, 4],
    'learning_rate': [0.001, 0.005, 0.01],
    'epochs': [50, 100],
    'batch_size': [8, 16, 32]
}

# グリッドサーチの実行
grid = GridSearchCV(estimator=ae_regressor, param_grid=param_grid, cv=3, scoring='neg_mean_squared_error', n_jobs=-1)
grid_result = grid.fit(X_train, X_train)

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

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

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

# --- モデルの構築と学習 ---
detector = AutoEncoderUtil(df=df_train)
detector.fit(column=["feature1", "feature2"], epochs=50)

# --- 学習データに対する異常検知 ---
y_pred_train, loss, threshold = detector.predict(column=["feature1", "feature2"])
print("Train predictions (-1:異常, 1:正常):", y_pred_train)

# --- グラフの描画 ---
plot_ae_results(X_train_data, y_pred_train)

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

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

# --- テストデータの生成（正常 + 異常） ---
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'])

# --- モデルの構築と学習 ---
detector = AutoEncoderUtil(df=df_train)
detector.fit(column=["feature1", "feature2"], epochs=100, batch_size=16, lr=1e-3)

# --- 異常検知の実行 ---
y_pred_test, _, threshold = detector.predict(["feature1", "feature2"], df=df_test)
y_pred_train, _, _ = detector.predict(["feature1", "feature2"], df=df_train)

print(f"テストデータ中の異常値 (-1) の数: {np.sum(y_pred_test == -1)}")
print(f"テストデータ中の正常値 (1) の数: {np.sum(y_pred_test == 1)}")

# --- モデルの保存 ---
MODEL_PATH = 'ae_model.keras'
detector.save_model(MODEL_PATH)

# --- NumPy配列に変換（可視化用） ---
X_train = df_train[["feature1", "feature2"]].values
X_test = df_test[["feature1", "feature2"]].values

# --- 結果の可視化 ---
plot_ae_results(
    X_train=X_train,
    y_train_pred=y_pred_train,
    X_test=X_test,
    y_test_pred=y_pred_test,
    model_util=detector,
    threshold=threshold
)

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

# --- 学習データの生成 ---
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'])
X_train = df_train[['feature1', 'feature2']].values

# --- モデルの構築と学習 ---
detector = AutoEncoderUtil(df=df_train)
detector.fit(column=["feature1", "feature2"], epochs=100, batch_size=16, lr=1e-3)

# --- 学習データに対する異常検知 ---
y_pred_train, _, _ = detector.predict(["feature1", "feature2"], df=df_train)

# --- 学習データの結果を可視化 ---
plot_ae_results(
    X_train=X_train,
    y_train_pred=y_pred_train,
    X_test=None,
    y_test_pred=None,
    model_util=detector
)

# --- モデルの保存 ---
MODEL_PATH = 'ae_model.keras'
detector.save_model(MODEL_PATH)

# --- モデルの再ロード ---
detector_loaded = AutoEncoderUtil()
detector_loaded.input_dim = 2  # 必要！モデル構築に使う
detector_loaded.load_model(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'])
X_test = df_test[['feature1', 'feature2']].values

# --- テストデータに対する異常検知 ---
y_pred_test, _, _ = detector_loaded.predict(["feature1", "feature2"], df=df_test)
print("Test predictions:", y_pred_test)

# --- テストデータの結果を可視化 ---
plot_ae_results(
    X_train=None,
    y_train_pred=None,
    X_test=X_test,
    y_test_pred=y_pred_test,
    model_util=detector_loaded
)

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

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

# --- モデルの構築と学習 ---
detector = AutoEncoderUtil(df=df_train)
detector.fit(column=["feature1"], epochs=100, batch_size=16, lr=1e-3)

# --- 学習データでの異常検知 ---
y_pred_train, errors_train, threshold = detector.predict(["feature1"], df=df_train)
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, errors_test, _ = detector.predict(["feature1"], df=df_test)
print("Test predictions:", y_pred_test)

# --- NumPy配列に変換（プロット用） ---
X_train = df_train[["feature1"]].values
X_test = df_test[["feature1"]].values

# --- 1次元プロットで可視化 ---
plot_ae_results_1d(
    X_train=X_train,
    y_train_pred=y_pred_train,
    X_test=X_test,
    y_test_pred=y_pred_test,
    model_util=detector,
    threshold=threshold
)

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import os

class AutoEncoderUtil:
    def __init__(self, df=None, model_path=None):
        self.df = df.copy() if df is not None else None
        self.model = None
        self.input_dim = None
        if model_path:
            self.load_model(model_path)

    def build_model(self, input_dim):
        input_layer = keras.Input(shape=(input_dim,))
        encoded = layers.Dense(8, activation='relu')(input_layer)
        encoded = layers.Dense(4, activation='relu')(encoded)
        decoded = layers.Dense(8, activation='relu')(encoded)
        output_layer = layers.Dense(input_dim)(decoded)

        model = keras.Model(inputs=input_layer, outputs=output_layer)
        model.compile(optimizer='adam', loss='mse')
        return model

    def fit(self, column, epochs=50, batch_size=32, lr=1e-3):
        if self.df is None:
            raise ValueError("DataFrameがセットされていません。")

        X = self.df[column].values.astype(np.float32)
        if X.ndim == 1:
            X = X.reshape(-1, 1)
        self.input_dim = X.shape[1]

        self.model = self.build_model(self.input_dim)

        # 学習率を指定してオプティマイザを作成
        optimizer = keras.optimizers.Adam(learning_rate=lr)
        self.model.compile(optimizer=optimizer, loss='mse')

        self.model.fit(X, X, epochs=epochs, batch_size=batch_size, verbose=1)
        print(f"学習完了！（lr={lr}, epochs={epochs}, batch_size={batch_size}）")


    def predict(self, column, df=None, threshold=None):
        if self.model is None:
            raise RuntimeError("モデルが学習またはロードされていません。")

        if df is not None:
            self.df = df

        X = self.df[column].values.astype(np.float32)
        if X.ndim == 1:
            X = X.reshape(-1, 1)

        reconstructed = self.model.predict(X, verbose=0)
        loss = np.mean(np.square(X - reconstructed), axis=1)

        if threshold is None:
            threshold = np.percentile(loss, 95)

        y_pred = np.where(loss > threshold, -1, 1)
        return y_pred, loss, threshold

    def save_model(self, path):
        if self.model is None:
            raise RuntimeError("モデルが学習されていません。")
        self.model.save(path)
        print(f"モデルを {path} に保存しました。")

    def load_model(self, path):
        if not os.path.exists(path):
            raise FileNotFoundError(f"ファイルが見つかりません: {path}")
        self.model = keras.models.load_model(path)
        self.input_dim = self.model.input_shape[1]
        print(f"モデルを {path} からロードしました。")


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

def plot_ae_results_1d(X_train=None, y_train_pred=None,
                       X_test=None, y_test_pred=None,
                       model_util=None, threshold=None):
    """
    AutoEncoder の 1次元異常検知結果をプロット。
    model_util: AutoEncoderUtil インスタンス
    threshold: 再構成誤差のしきい値（Noneなら自動計算）
    """
    plt.figure(figsize=(10, 6))
    ax = plt.gca()
    ax.set_facecolor("#eef7ff")  # 背景色を設定！

    # --- データ点の描画 ---
    if X_train is not None and y_train_pred is not None:
        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)

    # --- スコアカーブの描画 ---
    if model_util is not None and model_util.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)
        df_range = pd.DataFrame(x_range, columns=["feature"])

        _, loss_scores, _ = model_util.predict(["feature"], df=df_range)

        if threshold is None:
            threshold = np.percentile(loss_scores, 95)

        plt.plot(x_range, loss_scores, color='blue', linestyle='--', linewidth=2,
                 label="再構成誤差スコア", zorder=2)

        plt.axhline(threshold, color='red', linestyle='-', linewidth=2,
                    label="判定閾値", zorder=1)

        min_z, max_z = loss_scores.min(), loss_scores.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)

    plt.axhline(0, color='gray', linestyle=':', alpha=0.5)

    plt.title("AutoEncoder 異常検知（1次元）とスコアカーブ")
    plt.xlabel("特徴量 X")
    plt.ylabel("再構成誤差")
    plt.grid(True, alpha=0.6)
    plt.legend()
    plt.tight_layout()
    plt.show()


def plot_ae_results(X_train=None, y_train_pred=None,
                    X_test=None, y_test_pred=None,
                    model_util=None, threshold=None):
    """
    AutoEncoder の 2次元異常検知結果をプロット。
    model_util: AutoEncoderUtil インスタンス
    threshold: 再構成誤差のしきい値（Noneなら自動計算）
    """
    plt.figure(figsize=(10, 8))
    ax = plt.gca()
    ax.set_facecolor("#eef7ff")  # 背景色を設定！

    # --- データ結合と描画範囲の決定 ---
    all_data = [arr for arr in [X_train, X_test] if arr is not None]
    if not all_data:
        print("プロットするデータがありません。")
        return

    X_all = np.concatenate(all_data)
    x_min, x_max = X_all[:, 0].min() - 1, X_all[:, 0].max() + 1
    y_min, y_max = X_all[:, 1].min() - 1, X_all[:, 1].max() + 1

    # --- 背景スコアマップの描画 ---
    if model_util is not None and model_util.model is not None:
        xx, yy = np.meshgrid(np.linspace(x_min, x_max, 200),
                             np.linspace(y_min, y_max, 200))
        grid = np.c_[xx.ravel(), yy.ravel()]
        df_grid = pd.DataFrame(grid, columns=["feature1", "feature2"])
        _, loss_grid, _ = model_util.predict(["feature1", "feature2"], df=df_grid)
        Z = loss_grid.reshape(xx.shape)

        if threshold is None:
            threshold = np.percentile(loss_grid, 95)

        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], colors='black', linewidths=2, linestyles='--')

    # --- データ点の描画 ---
    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='lightgreen', edgecolors='k', s=60, label="学習データ（正常）", zorder=3)
        plt.scatter(X_train[y_train_pred == -1, 0], X_train[y_train_pred == -1, 1],
                    c='lightcoral', edgecolors='k', s=60, 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='lightgreen', edgecolors='k', s=80, marker='*', label="テストデータ（正常）", zorder=4)
        plt.scatter(X_test[y_test_pred == -1, 0], X_test[y_test_pred == -1, 1],
                    c='red', edgecolors='k', s=80, marker='*', label="テストデータ（異常）", zorder=4)

    # --- グラフの体裁 ---
    plt.title("AutoEncoder 異常検知結果（2次元）")
    plt.xlabel("feature1")
    plt.ylabel("feature2")
    plt.xlim(x_min, x_max)
    plt.ylim(y_min, y_max)
    plt.grid(True, alpha=0.3)
    plt.legend()
    plt.tight_layout()
    plt.show()




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