偽陽性管理

偽陽性の業務インパクト

現状の問題:
  日次アラート数: 500件
  偽陽性率: 85%（425件が正常取引）
  調査チーム: 5名
  1人あたり調査可能件数: 60件/日
  → 調査カバー率: 60%（200件が未調査のまま）

偽陽性削減のアプローチ

1. 閾値の多段階化

class MultiTierAlertSystem:
    """多段階アラートシステム"""

    def __init__(self):
        self.tiers = {
            'critical': {
                'threshold': 0.95,
                'action': 'immediate_block',
                'review_required': True,
            },
            'high': {
                'threshold': 0.80,
                'action': 'hold_and_review',
                'review_required': True,
            },
            'medium': {
                'threshold': 0.60,
                'action': 'flag_for_batch_review',
                'review_required': False,
            },
            'low': {
                'threshold': 0.40,
                'action': 'monitor_only',
                'review_required': False,
            },
        }

    def classify_alert(self, fraud_score):
        """スコアに応じてアラートを分類"""
        for tier_name, tier_config in self.tiers.items():
            if fraud_score >= tier_config['threshold']:
                return {
                    'tier': tier_name,
                    'action': tier_config['action'],
                    'review_required': tier_config['review_required'],
                    'score': fraud_score,
                }
        return {'tier': 'safe', 'action': 'allow', 'review_required': False}

2. ルールベースフィルタの併用

class FalsePositiveFilter:
    """ルールベースの偽陽性フィルタ"""

    def __init__(self):
        self.whitelist_rules = []
        self.context_rules = []

    def add_whitelist_rule(self, rule_name, condition_fn):
        """ホワイトリストルールを追加"""
        self.whitelist_rules.append({
            'name': rule_name,
            'condition': condition_fn,
        })

    def apply_filters(self, transaction, fraud_score):
        """フィルタを適用して偽陽性を削減"""
        adjustments = []

        # ホワイトリストチェック
        for rule in self.whitelist_rules:
            if rule['condition'](transaction):
                adjustments.append({
                    'rule': rule['name'],
                    'action': 'score_reduction',
                    'factor': 0.5,
                })

        # コンテキストルールチェック
        if self._is_recurring_pattern(transaction):
            adjustments.append({
                'rule': 'recurring_pattern',
                'action': 'score_reduction',
                'factor': 0.7,
            })

        # スコア調整
        adjusted_score = fraud_score
        for adj in adjustments:
            adjusted_score *= adj['factor']

        return {
            'original_score': fraud_score,
            'adjusted_score': adjusted_score,
            'adjustments': adjustments,
        }

    def _is_recurring_pattern(self, transaction):
        """過去に同様のパターンで正常と判定されたか"""
        # 同一顧客の同一加盟店への定期的な支払いなど
        return (
            transaction.get('is_recurring', False)
            or transaction.get('same_merchant_count_30d', 0) >= 3
        )

3. 顧客リスクプロファイル

class CustomerRiskProfile:
    """顧客ごとのリスクプロファイル"""

    def __init__(self):
        self.profiles = {}

    def update_profile(self, customer_id, transaction_history):
        """取引履歴からリスクプロファイルを更新"""
        history = transaction_history

        profile = {
            'customer_id': customer_id,
            'avg_transaction_amount': np.mean([t['amount'] for t in history]),
            'std_transaction_amount': np.std([t['amount'] for t in history]),
            'frequent_merchants': self._top_merchants(history),
            'typical_hours': self._typical_hours(history),
            'typical_locations': self._typical_locations(history),
            'account_age_days': history[-1].get('account_age_days', 0),
            'false_positive_count': sum(
                1 for t in history if t.get('was_false_positive', False)
            ),
        }

        self.profiles[customer_id] = profile
        return profile

    def is_typical_behavior(self, customer_id, transaction):
        """取引が顧客の典型的な行動パターンに合致するか"""
        profile = self.profiles.get(customer_id)
        if not profile:
            return False

        checks = {
            'amount_normal': self._check_amount(profile, transaction),
            'merchant_known': transaction.get('merchant_id') in profile['frequent_merchants'],
            'time_normal': transaction.get('hour') in profile['typical_hours'],
            'location_normal': transaction.get('location') in profile['typical_locations'],
        }

        # 4つ中3つ以上合致すれば典型的な行動
        return sum(checks.values()) >= 3

    def _check_amount(self, profile, transaction):
        """金額が正常範囲か"""
        mean = profile['avg_transaction_amount']
        std = profile['std_transaction_amount']
        amount = transaction['amount']
        return abs(amount - mean) < 3 * std

    def _top_merchants(self, history, top_n=10):
        from collections import Counter
        merchants = [t.get('merchant_id') for t in history]
        return [m for m, _ in Counter(merchants).most_common(top_n)]

    def _typical_hours(self, history):
        hours = [t.get('hour', 12) for t in history]
        return list(set(hours))

    def _typical_locations(self, history):
        locations = [t.get('location') for t in history if t.get('location')]
        return list(set(locations))

偽陽性削減の効果測定

def measure_fp_reduction(before_metrics, after_metrics):
    """偽陽性削減の効果を測定"""
    result = {
        'fp_rate_before': before_metrics['fp_rate'],
        'fp_rate_after': after_metrics['fp_rate'],
        'fp_reduction': round(
            (1 - after_metrics['fp_rate'] / before_metrics['fp_rate']) * 100, 1
        ),
        'recall_before': before_metrics['recall'],
        'recall_after': after_metrics['recall'],
        'recall_change': round(
            (after_metrics['recall'] - before_metrics['recall']) * 100, 2
        ),
        'daily_alerts_before': before_metrics['daily_alerts'],
        'daily_alerts_after': after_metrics['daily_alerts'],
        'investigation_cost_saving': (
            (before_metrics['daily_alerts'] - after_metrics['daily_alerts'])
            * 2500 * 365
        ),
    }
    return result

# 期待される結果
# 偽陽性率: 85% → 45%（47%削減）
# 日次アラート: 500件 → 200件
# Recall維持: 95% → 93%（許容範囲）
# 年間調査コスト削減: 約2.7億円

まとめ

チェックリスト

• 偽陽性の業務インパクトを定量的に説明できる
• 多段階アラートシステムを設計できる
• ルールベースフィルタの併用方法を理解した
• 顧客リスクプロファイルの活用方法を説明できる
• 偽陽性削減とRecallのトレードオフを管理できる

次のステップへ

偽陽性管理を理解した。次はフィードバックループによる継続的改善を学ぼう。

推定読了時間: 30分