Autoencoder異常検知

Autoencoderの原理

基本構造

入力 → [エンコーダ] → 潜在表現(z) → [デコーダ] → 復元
  x        圧縮          低次元         復元        x'

再構成誤差 = ||x - x'||^2

正常データ: 再構成誤差が小さい（学習済みパターンなので復元できる）
異常データ: 再構成誤差が大きい（未知のパターンなので復元できない）

不正検知への応用

学習フェーズ:
  正常取引データのみでAutoencoderを学習
  → 正常パターンの圧縮・復元を学習

検知フェーズ:
  新しい取引データを入力
  → 再構成誤差を計算
  → 誤差が閾値を超えたら「異常（不正の疑い）」と判定

PyTorchでの実装

Autoencoderモデル

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

class FraudAutoencoder(nn.Module):
    def __init__(self, input_dim, encoding_dim=14):
        super().__init__()

        # エンコーダ
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.BatchNorm1d(64),
            nn.Dropout(0.2),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.BatchNorm1d(32),
            nn.Linear(32, encoding_dim),
            nn.ReLU()
        )

        # デコーダ
        self.decoder = nn.Sequential(
            nn.Linear(encoding_dim, 32),
            nn.ReLU(),
            nn.BatchNorm1d(32),
            nn.Dropout(0.2),
            nn.Linear(32, 64),
            nn.ReLU(),
            nn.BatchNorm1d(64),
            nn.Linear(64, input_dim),
        )

    def forward(self, x):
        z = self.encoder(x)
        x_reconstructed = self.decoder(z)
        return x_reconstructed

    def get_reconstruction_error(self, x):
        x_reconstructed = self.forward(x)
        error = torch.mean((x - x_reconstructed) ** 2, dim=1)
        return error

学習

# データ準備（正常データのみで学習）
X_train_normal = X_train_scaled[y_train == 0]
X_val = X_test_scaled  # テスト用（正常+不正）

train_tensor = torch.FloatTensor(X_train_normal)
train_dataset = TensorDataset(train_tensor, train_tensor)
train_loader = DataLoader(train_dataset, batch_size=256, shuffle=True)

# モデル初期化
input_dim = X_train_scaled.shape[1]
model = FraudAutoencoder(input_dim, encoding_dim=14)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5)

# 学習ループ
n_epochs = 100
for epoch in range(n_epochs):
    model.train()
    total_loss = 0
    for batch_x, _ in train_loader:
        optimizer.zero_grad()
        x_reconstructed = model(batch_x)
        loss = criterion(x_reconstructed, batch_x)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()

    avg_loss = total_loss / len(train_loader)
    scheduler.step(avg_loss)

    if (epoch + 1) % 10 == 0:
        print(f"Epoch {epoch+1}/{n_epochs}, Loss: {avg_loss:.6f}")

異常スコアの計算と評価

from sklearn.metrics import average_precision_score

model.eval()
with torch.no_grad():
    X_test_tensor = torch.FloatTensor(X_test_scaled)
    reconstruction_errors = model.get_reconstruction_error(X_test_tensor).numpy()

# 再構成誤差 = 異常スコア
pr_auc = average_precision_score(y_test, reconstruction_errors)
print(f"Autoencoder PR-AUC: {pr_auc:.4f}")

# スコア分布の確認
import matplotlib.pyplot as plt

errors_normal = reconstruction_errors[y_test == 0]
errors_fraud = reconstruction_errors[y_test == 1]

plt.figure(figsize=(10, 5))
plt.hist(errors_normal, bins=100, alpha=0.7, label='正常', density=True)
plt.hist(errors_fraud, bins=50, alpha=0.7, label='不正', density=True, color='red')
plt.xlabel('Reconstruction Error')
plt.ylabel('Density')
plt.title('Autoencoder: 再構成誤差の分布')
plt.legend()
plt.show()

変分Autoencoder（VAE）

VAEは潜在空間に確率分布の制約を加えたAutoencoderである。

VAEの構造

通常のAE:  入力 → z（確定的な値） → 復元
VAE:       入力 → μ, σ → z 〜 N(μ, σ²) → 復元

VAEの損失 = 再構成誤差 + KLダイバージェンス
  再構成誤差: 復元の正確さ
  KLダイバージェンス: 潜在分布を標準正規分布に近づける正則化

VAEの実装

class FraudVAE(nn.Module):
    def __init__(self, input_dim, latent_dim=14):
        super().__init__()

        # エンコーダ
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU()
        )
        self.fc_mu = nn.Linear(32, latent_dim)
        self.fc_logvar = nn.Linear(32, latent_dim)

        # デコーダ
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 32),
            nn.ReLU(),
            nn.Linear(32, 64),
            nn.ReLU(),
            nn.Linear(64, input_dim)
        )

    def encode(self, x):
        h = self.encoder(x)
        return self.fc_mu(h), self.fc_logvar(h)

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std

    def decode(self, z):
        return self.decoder(z)

    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        x_reconstructed = self.decode(z)
        return x_reconstructed, mu, logvar

def vae_loss(x, x_recon, mu, logvar):
    recon_loss = nn.functional.mse_loss(x_recon, x, reduction='sum')
    kl_loss = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    return recon_loss + kl_loss

VAEの異常スコア

model_vae.eval()
with torch.no_grad():
    x_tensor = torch.FloatTensor(X_test_scaled)
    x_recon, mu, logvar = model_vae(x_tensor)

    # 方法1: 再構成誤差のみ
    recon_error = torch.mean((x_tensor - x_recon) ** 2, dim=1).numpy()

    # 方法2: 再構成誤差 + KLダイバージェンス
    kl = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp(), dim=1).numpy()
    combined_score = recon_error + 0.1 * kl  # 重み付き

    # 方法3: 複数回サンプリングの平均（より安定）
    n_samples = 10
    errors = []
    for _ in range(n_samples):
        z = model_vae.reparameterize(mu, logvar)
        x_recon = model_vae.decode(z)
        error = torch.mean((x_tensor - x_recon) ** 2, dim=1)
        errors.append(error)
    avg_error = torch.stack(errors).mean(dim=0).numpy()

AE vs VAE vs Isolation Forest

まとめ

チェックリスト

• Autoencoderによる異常検知の原理を説明できる
• 正常データのみで学習する理由を説明できる
• VAEとAEの違いを理解した
• 再構成誤差を異常スコアとして利用する方法を実装できる

次のステップへ

異常検知モデルを理解したところで、次は閾値最適化の手法を学び、モデルの性能を最大化しよう。

推定読了時間: 30分