「また新しい手口か…」
実務の現場では、セキュリティ担当者や監査部門がログを眺めながら頭を抱えるケースが少なくありません。従来のルールベース検知は既知のパターンには強い反面、少し条件を変えた「未知の手口」には無力です。検知の基準（閾値）を厳しくすれば誤検知（False Positive）が増え、緩めれば不正を見逃す（False Negative）というイタチごっこが続いてしまいます。
内部不正や不正取引の検知において、「過去の不正データが少なすぎてAIに学習させられない」という課題は頻出します。一般的なAI（教師あり学習）は、正解ラベルが付いた大量のデータを必要としますが、不正自体が稀にしか起きないため、十分なデータを集めることが困難だからです。
そこで今回は、「教師なし学習（Unsupervised Learning）」というアプローチを採用し、ディープラーニングの一種であるAutoencoder（自己符号化器）を用いた異常検知システムの実装に挑戦します。実証に基づいた論理的なステップで、未知の異常をあぶり出す仕組みを構築していきましょう。
なぜルールベースでは内部不正を防げないのか
ルールベースの限界は、人間の想像力の限界に直結しています。たとえば「深夜2時に100万円以上の送金」というルールを設定したとしましょう。この場合、「深夜1時に99万円」の送金が実行されると、システムは検知できません。ルールを増やせばシステムは複雑化し、メンテナンスコストが指数関数的に増大するという構造的な課題も抱えています。
「教師なし学習（Autoencoder）」が不正検知に効く理由
Autoencoderのアプローチは、非常にシンプルかつ強力です。
- 正常なデータだけをAIに大量に学習させる。
- AIは「正常なデータのパターン（特徴）」を圧縮して記憶する。
- 新しいデータを入力した際、「正常」とかけ離れていればうまく再現できず、エラー（誤差）が生じる。
- この「再現できなかった誤差の大きさ」を異常スコアとして検知する。
つまり、「不正とは何か」を教えるのではなく、「正常とは何か」を徹底的に教え込み、そこから逸脱するものを「異常」とみなす戦略です。これにより、過去に例のない未知の手口であっても、「いつもと違う」という論理的な理由で検知が可能になります。
本記事で作る成果物のイメージ
本チュートリアルでは、PythonとKerasを用いて、以下のステップで検知モデルを構築します。
- クレジットカードの取引データ（Kaggleの公開データセット）を使用。
- 正常な取引のみを使ってAutoencoderを学習。
- テストデータに対する「再構成誤差（Reconstruction Error）」を計算。
- 誤差の分布を可視化し、実証データに基づいて適切な閾値を設定し、不正をあぶり出す。
このコードはクレジットカードの不正検知だけでなく、社内のアクセスログ、経費精算データ、製造ラインのセンサーデータなど、あらゆる「異常検知」の課題に応用可能です。それでは、Google Colabを開いて実装を進めていきましょう。
2. 環境構築とデータセットの準備
まずは環境とデータを準備します。手軽に実行できるGoogle Colab環境を想定していますが、ローカルのJupyter Notebookでも同様に動作します。
Google Colabのセットアップとライブラリ導入
必要なライブラリをインポートします。ディープラーニングの構築にTensorFlow/Keras、データ操作にPandas、数値計算にNumPy、可視化にMatplotlib/Seaborn、前処理にScikit-learnを使用します。
# 必要なライブラリのインポート
import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.callbacks import ModelCheckpoint, TensorBoard
from tensorflow.keras import regularizers
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
import seaborn as sns
# ランダムシードの固定（再現性のため）
LABELS = ["Normal", "Fraud"]
np.random.seed(42)
tf.random.set_seed(42)
データセットの読み込みと構造理解
Kaggleで公開されているデータセット「Credit Card Fraud Detection」を使用します。2日間のクレジットカード取引データが含まれており、その中にわずかな不正取引（Fraud）が混ざっています。
※実業務に適用する際は、自社の「経費申請ログ」や「サーバーアクセスログ」を読み込ませるイメージを持ってください。
# データの読み込み（パスは適宜変更してください）
# データセットURL: https://www.kaggle.com/mlg-ulb/creditcardfraud
df = pd.read_csv('creditcard.csv')
# データの形状確認
print(f"データ形状: {df.shape}")
print(df.head())
本データセットは、個人情報保護の観点からPCA（主成分分析）によって変換されたV1〜V28の匿名化特徴量を持っています。さらに、Time（最初の取引からの経過秒数）とAmount（取引金額）が含まれています。
ビジネスの文脈では、以下のように置き換えて考えると分かりやすいでしょう。
V1: 申請部門コードのベクトル表現。V2: 承認者のIDベクトル。Amount: 経費申請金額。Time: 申請時刻。
前処理：データの正規化と分割戦略
ディープラーニングにおいて、データのスケール（単位）を統一することは非常に重要です。数万秒という大きな単位を持つTimeと、0を中心に分布するV1をそのまま学習させると、モデルが値の大きな特徴量に引きずられてしまいます。
ここではモデルをシンプルにするためTimeを削除し、Amountを標準化（平均0、分散1）します。
# Timeカラムの削除（モデルの単純化）
df = df.drop(['Time'], axis=1)
# Amountの標準化
scaler = StandardScaler()
df['Amount'] = scaler.fit_transform(df['Amount'].values.reshape(-1, 1))
次にデータを分割しますが、ここが「教師なし学習」特有の重要なポイントです。
「正常なデータのパターン」を学習させるため、学習データには「正常（Class=0）」のみを使用します。一方、テストデータには正常と不正の両方を含め、不正を正しく「異常」と判定できるかを検証します。
# 正常データと不正データに分離
# Class = 0 が正常、1 が不正
frauds = df[df.Class == 1]
normal = df[df.Class == 0]
print(f"正常取引数: {normal.shape[0]}, 不正取引数: {frauds.shape[0]}")
# 正常データのみを学習用(80%)とテスト用(20%)に分割
X_train, X_test = train_test_split(normal, test_size=0.2, random_state=42)
# テストデータには不正データも全て混ぜる（評価用）
# 注: 実際の運用では不正データは「未知」なので学習には使えません
X_test = pd.concat([X_test, frauds])
# ラベル（Class）を削除して入力データとする
# Autoencoderは入力と同じものを出力するため、y_trainは不要（X_train自体が正解）
X_train = X_train.drop(['Class'], axis=1)
y_test = X_test['Class'] # 評価用にラベルを保存
X_test = X_test.drop(['Class'], axis=1)
# numpy配列に変換
X_train = X_train.values
X_test = X_test.values
print(f"学習データ形状: {X_train.shape}")
print(f"テストデータ形状: {X_test.shape}")
これで、「正常な振る舞い」のみを持った学習データと、評価用のテストデータの準備が整いました。
3. 実装Part 1：Autoencoderモデルの設計と学習
ここからが実装の中核です。Keras（TensorFlowのハイレベルAPI）を使用し、異常検知用のAutoencoderを構築していきます。
エンコーダとデコーダの構造設計
Autoencoderは、情報の「圧縮」と「復元」を行う砂時計型のニューラルネットワークであり、大きく2つのパートに分かれます。
Encoder（入力層 → 中間層）:
入力データを圧縮し、潜在空間（Latent Space）と呼ばれる低次元に情報を凝縮します。ノイズを削ぎ落とし、「本質的な特徴」だけを抽出する役割を担います。Decoder（中間層 → 出力層）:
圧縮された潜在変数から、元の入力データを忠実に復元します。
設計のポイントは、中間層の次元数を入力層よりも小さくする「ボトルネック（Bottleneck）」を設けることです。
正常データのみを学習させることで、モデルは「正常なデータの圧縮・復元方法」を習得します。そこに未知の不正データが入力されると、うまく復元できずに大きな「再構成誤差（Reconstruction Error）」が発生します。この誤差の大きさが、そのまま異常検知スコアとなります。
モデルの定義（Keras Functional API）
以下の層構造でモデルを定義します。KerasのFunctional APIを使うことで、柔軟なネットワーク設計が可能です。
- 入力層: 29次元（前処理済みの特徴量）。
- エンコーダ: [29] -> [14] -> [7]。
- デコーダ: [7] -> [14] -> [29]。
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras import regularizers
# 入力次元数（ここでは29列）
input_dim = X_train.shape[1]
# --- エンコーダ部分 ---
input_layer = Input(shape=(input_dim, ))
# 第1層: 情報を圧縮しつつ、L1正則化でスパース性を持たせる
# activity_regularizerを加えることで、重要な特徴だけが発火するように促します
encoder = Dense(14, activation="tanh",
activity_regularizer=regularizers.l1(10e-5))(input_layer)
# 第2層: ボトルネック層（最も圧縮された表現）
encoder = Dense(7, activation="relu")(encoder)
# --- デコーダ部分 ---
# 圧縮された情報から元の次元へ復元を開始
decoder = Dense(14, activation='tanh')(encoder)
# 出力層: 元のデータと同じ次元に戻す
decoder = Dense(input_dim, activation='linear')(decoder)
# Autoencoderモデルの結合
autoencoder = Model(inputs=input_layer, outputs=decoder)
# モデル構造の確認
autoencoder.summary()
正常データのみを用いたモデル学習
モデルをコンパイルし、学習を開始します。過学習（Overfitting）を防ぎ、最適なモデルを得るために、「EarlyStopping（早期終了）」の導入を推奨します。
学習データX_trainを、入力と正解ラベルの両方に指定する点が、教師なし学習（自己教師あり学習）の大きな特徴です。
注意: 学習には計算リソースを消費します。GPUを利用する場合、最新のTensorFlow環境ではWindowsネイティブではなく、WSL2（Windows Subsystem for Linux 2）上での実行が推奨されています。
from tensorflow.keras.callbacks import EarlyStopping
# モデルのコンパイル
# 最適化アルゴリズム: Adam
# 損失関数: 平均二乗誤差(MSE) - 入力と出力のズレを評価
autoencoder.compile(optimizer='adam', loss='mean_squared_error')
# EarlyStoppingの設定
# val_loss（検証データの損失）が改善しなくなったら学習を自動停止
# patience=5: 5エポック様子を見て改善しなければ停止
# restore_best_weights=True: 最も良かった状態の重みを復元
early_stopping = EarlyStopping(
monitor='val_loss',
patience=5,
mode='min',
restore_best_weights=True
)
# 学習の実行
history = autoencoder.fit(
X_train, X_train, # 入力と正解に同じデータを使用
epochs=100, # 最大エポック数（EarlyStoppingがあるため多めに設定）
batch_size=32, # バッチサイズ
shuffle=True, # データの順序をシャッフル
validation_data=(X_test, X_test), # 検証用データ
callbacks=[early_stopping], # コールバックの指定
verbose=1
).history
学習が完了したら、損失（Loss）の推移をグラフで確認します。学習データ（loss）と検証データ（val_loss）が共に低下していれば、「正常データのパターン」を正しく学習できている証拠です。
import matplotlib.pyplot as plt
plt.figure(figsize=(8, 4))
plt.plot(history['loss'], label='Train Loss')
plt.plot(history['val_loss'], label='Validation Loss')
plt.title('Model Loss Progress')
plt.ylabel('Loss (MSE)')
plt.xlabel('Epoch')
plt.legend()
plt.show()
Lossが十分に低い値に収束していれば、正常データを復元する能力をしっかりと獲得したと言えます。
4. 実装Part 2：異常スコアの算出と閾値設定
学習済みモデルを用いて「どれくらい異常か」を数値化し、不正判定のライン（閾値）を論理的に決定していきます。
再構成誤差（Reconstruction Error）の計算
学習済みモデルにテストデータを入力して予測値（復元データ）を取得し、元のデータとの差（二乗誤差）を計算します。これが「異常スコア」となります。
# テストデータの予測（復元）
predictions = autoencoder.predict(X_test)
# 再構成誤差（MSE）の計算
# axis=1で行ごとの平均をとる
mse = np.mean(np.power(X_test - predictions, 2), axis=1)
# 結果をDataFrameにまとめる
error_df = pd.DataFrame({'reconstruction_error': mse,
'true_class': y_test})
print(error_df.describe())
誤差分布の可視化と「異常」の境界線
正常データと不正データにおける異常スコア（MSE）の分布を確認します。理想的な状態では、正常データは低スコアに、不正データは高スコアに集中します。
# 正常データと不正データの誤差分布をプロット
plt.figure(figsize=(12, 6))
# 正常データのヒストグラム（青）
sns.histplot(error_df[error_df['true_class'] == 0]['reconstruction_error'],
bins=50, kde=True, color='blue', label='Normal', alpha=0.6)
# 不正データのヒストグラム（赤）
# ※不正データは見やすいように数を強調する場合もありますが、ここではそのまま表示
sns.histplot(error_df[error_df['true_class'] == 1]['reconstruction_error'],
bins=50, kde=True, color='red', label='Fraud', alpha=0.6)
plt.title('Reconstruction Error Distribution')
plt.xlabel('Reconstruction error (MSE)')
plt.ylabel('Count')
plt.legend()
plt.yscale('log') # 対数スケールにすると分布の違いが見やすい
plt.show()
グラフを見ると、青い山（正常）は左側（誤差が小さい領域）に集中し、赤い分布（不正）は右側（誤差が大きい領域）に広がっていることが確認できます。「青と赤が交わるあたり」が、閾値の有力な候補となります。
ビジネス要件に基づいた閾値（Threshold）の決定
閾値の設定は、最終的にはビジネス上の判断になります。
- 閾値を低くする: 不正の見逃し（False Negative）は減りますが、誤検知（False Positive）が増加し、監査部門の確認工数を圧迫します。
- 閾値を高くする: 確実な異常のみを検知するため運用は楽になりますが、巧妙な不正を見逃すリスクが高まります。
ここでは仮に「正常データの誤差の上位数%」をラインとするか、グラフの分布を見て手動で設定します。
# 閾値の設定（例: 2.9）
# 実際の運用では、Precision-Recall曲線を見て決定します
threshold = 2.9
# 閾値を超えたものを「不正(1)」と判定
y_pred = [1 if e > threshold else 0 for e in error_df.reconstruction_error.values]
5. 実装Part 3：モデル評価とビジネス適用シミュレーション
最後に、設定した閾値での精度を評価し、実務のシステムへどのように組み込むかを検討します。
混同行列（Confusion Matrix）による精度評価
from sklearn.metrics import confusion_matrix, classification_report
# 混同行列の表示
conf_matrix = confusion_matrix(error_df.true_class, y_pred)
plt.figure(figsize=(8, 6))
sns.heatmap(conf_matrix, annot=True, fmt="d", cmap='Blues')
plt.title("Confusion Matrix")
plt.ylabel('True Class')
plt.xlabel('Predicted Class')
plt.show()
# 詳細レポート
print(classification_report(error_df.true_class, y_pred))
結果を確認しましょう。「不正（1）」を正しく「予測（1）」できた数だけでなく、「正常（0）」を誤って「予測（1）」してしまった数にも注目してください。これが、現場で発生する「無駄なアラート対応」の数に直結します。
誤検知（False Positive）時の運用フロー検討
AIによる異常検知は、システムを導入して終わりではありません。「AIが怪しいと判定した取引を、人間がどのように処理するか」というプロセス設計こそが成功の鍵を握ります。
たとえば、以下のようなフローが考えられます。
- AI検知: Autoencoderが再構成誤差の高い取引を抽出する。
- 一次スクリーニング: スコアが極端に高いものは即時ブロックし、ボーダーライン上のものは「要確認リスト」へ送る。
- 人間による監査: 担当者がリストを確認し、AIが異常と判断した論理的な理由（寄与した特徴量）を分析する。
- フィードバック: 「月末特有の正常な処理」などの知見を蓄積し、次回のモデル学習時に除外リストへ入れる等の調整を実施する。
実運用に向けたパイプライン化のヒント
PoC（概念実証）で効果が確認できたら、次は本番適用です。Pythonスクリプトを用いた夜間バッチ処理や、API化によるリアルタイム検知などが考えられます。
また、データの傾向は時間の経過とともに変化します（Concept Drift）。半年に一度など、定期的に最新の正常データを用いてモデルを再学習させるパイプラインの構築も忘れないようにしてください。
6. まとめと次のステップ：プロトタイプから本番運用へ
今回は、教師なし学習のAutoencoderを用い、ラベルのないデータから未知の不正を検知する手法を解説しました。
本チュートリアルの重要なポイントは以下の通りです。
- 内部不正対策には教師なし学習が有効: 過去の不正パターンに依存せず、「正常からの逸脱」を論理的に検知できる。
- データの正規化は必須: Amountなどの数値スケールを揃えることで、モデルの精度が劇的に向上する。
- 閾値設定はビジネス判断: 誤検知の対応コストと、見逃しによるリスクのバランスを実証データに基づいて考慮し決定する。
このコードはあくまで出発点です。実務においてより複雑なデータ構造に対応するためには、LSTM（時系列データ向け）を用いたAutoencoderや、VAE（変分オートエンコーダ）による確率的なアプローチが有効な場合もあります。
さらなる学習のために
異常検知の世界は非常に奥深く、日々新しいアルゴリズムが登場しています。本チュートリアルを通じて「さらに深く知りたい」「自社の特殊なデータに適用してみたい」と感じた方は、ぜひ継続的に最新の技術動向をキャッチアップし、実践的な検証を続けてみてください。
コメント