適応学習

不正手口の進化パターン

時期1: カード番号の総当たり攻撃
  → 検知済み: 短時間の大量試行を検知
時期2: 少額分散型攻撃（1件あたり500円以下）
  → 金額閾値をすり抜ける
時期3: アカウント乗っ取り+正常パターン模倣
  → 顧客プロファイルに近い行動をとる
時期4: 合成ID詐欺（架空の信用情報を構築）
  → 過去の取引履歴が存在しない

適応学習の3つの戦略

オンライン学習の実装

class OnlineFraudLearner:
    """オンライン学習による適応的な不正検知"""

    def __init__(self, base_model, learning_rate=0.01):
        self.model = base_model
        self.lr = learning_rate
        self.update_buffer = []
        self.performance_window = []

    def partial_fit(self, X_new, y_new, sample_weight=None):
        """新しいデータで部分的に更新"""
        self.update_buffer.append({
            'X': X_new,
            'y': y_new,
            'weight': sample_weight,
            'timestamp': datetime.now(),
        })

        # バッファが一定量溜まったら更新
        if len(self.update_buffer) >= 100:
            self._batch_update()

    def _batch_update(self):
        """バッファのデータでバッチ更新"""
        X_batch = np.vstack([b['X'] for b in self.update_buffer])
        y_batch = np.concatenate([b['y'] for b in self.update_buffer])

        # 重み: 新しいサンプルほど高重み
        n = len(self.update_buffer)
        time_weights = np.linspace(0.5, 1.0, n)
        sample_weights = np.repeat(time_weights, [len(b['y']) for b in self.update_buffer])

        # 増分学習
        self.model.fit(
            X_batch, y_batch,
            sample_weight=sample_weights,
            init_model=self.model,
        )
        self.update_buffer = []

アンサンブル更新戦略

class AdaptiveEnsemble:
    """時間変化に適応するアンサンブル"""

    def __init__(self, max_models=5):
        self.models = []
        self.max_models = max_models

    def add_model(self, model, trained_on_period, validation_score):
        """新しいモデルをアンサンブルに追加"""
        self.models.append({
            'model': model,
            'period': trained_on_period,
            'score': validation_score,
            'weight': validation_score,
            'added_at': datetime.now(),
        })

        # 最大数を超えたら最も古い/性能の低いモデルを除外
        if len(self.models) > self.max_models:
            self.models.sort(key=lambda m: m['score'])
            removed = self.models.pop(0)

        # 重みを正規化
        total = sum(m['weight'] for m in self.models)
        for m in self.models:
            m['weight'] /= total

    def predict(self, X):
        """加重平均で予測"""
        predictions = np.zeros(len(X))
        for m in self.models:
            pred = m['model'].predict_proba(X)[:, 1]
            predictions += pred * m['weight']
        return predictions

新パターン検知

class NoveltyDetector:
    """未知の不正パターンを検知"""

    def __init__(self, contamination=0.01):
        from sklearn.ensemble import IsolationForest
        self.detector = IsolationForest(
            contamination=contamination,
            random_state=42,
        )
        self.known_patterns = []

    def fit(self, X_normal):
        """正常データで学習"""
        self.detector.fit(X_normal)

    def detect_novel(self, X_new):
        """未知のパターンを検知"""
        anomaly_scores = self.detector.decision_function(X_new)
        is_novel = self.detector.predict(X_new) == -1

        return {
            'novel_indices': np.where(is_novel)[0],
            'novel_count': int(is_novel.sum()),
            'anomaly_scores': anomaly_scores,
        }

まとめ

チェックリスト

• 不正手口の進化パターンを理解した
• オンライン学習と全面再学習の使い分けを判断できる
• アンサンブル更新戦略の仕組みを説明できる
• 未知の不正パターン検知のアプローチを理解した

次のステップへ

適応学習を理解した。次は演習で不正検知システムの運用計画を策定しよう。

推定読了時間: 15分