金融における AI: 不正検出、信用スコアリング、リスク管理
2024 年、金融詐欺による世界の損失額は 125億ドル、 前年と比較して25％増加しました。同年、不正検知システム 人工知能に基づいて、 推定255億ドルの損失を回避 世界的に。結論は明らかです。金融分野における AI はもはや競争上の優位性ではありません。 そして運用上の必要性。
今日は 金融機関の99％ 何らかの形式の機械学習または AI を使用する 93% が、AI が不正検出機能に革命を起こすと信じています。 来年。しかし、これらのシステムは実際にどのように機能するのでしょうか?信用モデルを構築する方法 欧州の規制要件を超えるスコア?検出可能な AML システムを実装する方法 何千もの取引のネットワークにおけるマネーロンダリング計画?
この記事では、実際のコード、具体的なアーキテクチャ、およびケーススタディを使用して、これらの質問に答えます。 イタリアの銀行業界の背景からインスピレーションを受けています。を探索していきます。 リアルタイムで不正行為を検知 カフカとフリンクと一緒に、 解釈可能な信用スコアリング XGBoost と SHAP を使用すると、 AMLの検出 グラフ ニューラル ネットワークを使用し、AI法EU e PSD3.
この記事で学べること
- 不正検出が ML でどのように機能するか: 分離フォレスト、勾配ブースティング、ニューラル ネットワーク
- XGBoost と LightGBM による AI 信用スコアリング: 特徴エンジニアリングと実際のパフォーマンス (95% 以上の精度)
- 規制に準拠した財務上の意思決定のための説明可能な AI (SHAP および LIME)
- Kafka と Flink を使用したトランザクション監視のためのリアルタイム アーキテクチャ
- グラフニューラルネットワークによるマネーロンダリング対策: ルールベースのシステムが見逃すパターンを検出
- RegTech: AI 法 EU、PSD3、DORA、およびイタリアの銀行への影響
- ケーススタディ: イタリアの小売銀行向けの不正検出の導入
データ ウェアハウス、AI、デジタル トランスフォーメーション シリーズ
| # | アイテム | 集中 |
|---|---|---|
| 1 | データウェアハウスの進化 | SQL Server からデータ レイクハウスへ |
| 2 | データメッシュと分散型アーキテクチャ | データのドメイン所有権 |
| 3 | ETL と最新の ELT の比較 | dbt、Airbyte、Fivetran |
| 4 | パイプライン オーケストレーション | エアフロー、ダグスター、プリフェクト |
| 5 | 製造業における AI | 予知保全とデジタルツイン |
| 6 | あなたはここにいます - 金融における AI | 不正行為の検出と信用スコアリング |
| 7 | 小売における AI | 需要予測と推奨 |
| 8 | ヘルスケアにおける AI | 診断と創薬 |
| 9 | 物流におけるAI | ルートの最適化と倉庫の自動化 |
| 10 | ビジネスにおけるLLM | RAG Enterprise とガードレール |
金融における AI の展望: 2025 年のデータとトレンド
金融セクターは、大規模な機械学習技術を最初に採用したセクターの 1 つです。 そして今日でもそれは最も先進的なものの一つです。理由は構造的なものです: 豊富な過去のデータ、インセンティブ 莫大な経済コスト（10,000ユーロの詐欺を防ぐとすぐにROIが得られます）、および規制の枠組み これは複雑ではありますが、自動監視システムを明示的に必要とします。
2025 年の市場と主要データ
| インジケータ | バリュー2025 | トレンド |
|---|---|---|
| 世界的な金融詐欺被害 | 125億ドル（2024年） | 前年比 +25% |
| AIで詐欺を防ぐ | 255億ドル | 推定 2025 年 |
| AIを活用した金融機関 | 99% | 不正行為検知のため |
| 生成AIを利用した詐欺 | >50% | ディープフェイク、合成アイデンティティ |
| LightGBM クレジット スコアリングの精度 | 98.13% | SHAPの説明可能性を備えた |
| レグテックAI市場 | 217 億ドル (CAGR) | 予想される年間成長率 |
2024 年から 2025 年の根本的な変化とツールとしての生成 AI 攻撃的な: 今日の詐欺行為の 50% には、犯罪者による AI の使用が含まれています。 彼らはビデオディープフェイク、合成アイデンティティ、カスタマイズされたフィッシングキャンペーンを大規模に作成します。これ この現象により、より洗練された防御システムの導入が加速され、基準が変わりました 単純な異常検出から複雑な動作パターンを検出できるモデルまで。
3 つの主な応用分野
金融における AI は 3 つの異なるドメインを中心に構造化されていますが、それぞれに要件があります。 特定の技術的および規制:
金融における Domeni AI
| ドメイン | 主なテクニック | 必要なレイテンシ | 関連規格 |
|---|---|---|---|
| 不正行為の検出 | アイソレーションフォレスト、GBM、ディープラーニング | <100ms (リアルタイム) | PSD2/PSD3、AI法 |
| 信用スコアリング | XGBoost、LightGBM、SHAP | 秒-分 | CRR/CRD IV、AI 法高リスク |
| AML (マネーロンダリング対策) | グラフ ニューラル ネットワーク、NLP | 時間（バッチ）+リアルタイムアラート | AMLD6、FATF ガイドライン |
機械学習による不正行為の検出
不正検出は、金融分野で最も成熟した ML ユースケースです。主な課題はそうではありません 技術的だが統計的: 詐欺データセットは 非常にアンバランスな。 10,000 件中 トランザクション、通常 1 ～ 3 のみが不正 (ポジティブ クラス)、残りは正当です (ネガティブクラス)。 「すべて正当なもの」を予測するモデルはすでに 99.9% の精度に達しています。 しかしそれはまったく役に立たないでしょう。
階級の不均衡の問題
不均衡に対処するには、さまざまなテクニックが使用されます。 リモート (合成マイノリティ オーバーサンプリング技術）を使用して、少数派クラスの合成サンプルを生成し、クラスの重み付けを行います。 まれなクラスのエラーにペナルティを課し、次のような適切な評価指標を追加します。 AUC-ROC, 精度リコール AUC e F1 スコア 正確さの代わりに。
# fraud_detection_pipeline.py
# Pipeline completa di fraud detection con gestione sbilanciamento
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, StratifiedKFold
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import IsolationForest, RandomForestClassifier
from sklearn.metrics import (
classification_report, roc_auc_score,
precision_recall_curve, average_precision_score,
confusion_matrix
)
from sklearn.pipeline import Pipeline
import xgboost as xgb
from imblearn.over_sampling import SMOTE
from imblearn.pipeline import Pipeline as ImbPipeline
import warnings
warnings.filterwarnings('ignore')
# ============================================================
# 1. FEATURE ENGINEERING PER TRANSAZIONI FINANZIARIE
# ============================================================
def engineer_transaction_features(df: pd.DataFrame) -> pd.DataFrame:
"""
Costruisce feature comportamentali e di contesto
dalle transazioni grezze.
"""
df = df.copy()
# Feature temporali
df['ora_del_giorno'] = pd.to_datetime(df['timestamp']).dt.hour
df['giorno_settimana'] = pd.to_datetime(df['timestamp']).dt.dayofweek
df['e_weekend'] = df['giorno_settimana'].isin([5, 6]).astype(int)
df['e_notte'] = df['ora_del_giorno'].between(0, 6).astype(int)
# Feature di velocità (velocity) - window scorrevole
df = df.sort_values(['account_id', 'timestamp'])
df['importo_medio_7d'] = (
df.groupby('account_id')['importo']
.transform(lambda x: x.rolling('7D', min_periods=1).mean())
)
df['num_transazioni_1h'] = (
df.groupby('account_id')['importo']
.transform(lambda x: x.rolling('1H', min_periods=1).count())
)
df['num_transazioni_24h'] = (
df.groupby('account_id')['importo']
.transform(lambda x: x.rolling('24H', min_periods=1).count())
)
# Deviazione dall'importo tipico del cliente
df['z_score_importo'] = (
df.groupby('account_id')['importo']
.transform(lambda x: (x - x.mean()) / (x.std() + 1e-8))
)
# Feature geografiche
df['distanza_km'] = calcola_distanza_ultima_transazione(df)
# Cambio di paese rispetto alla transazione precedente
df['cambio_paese'] = (
df.groupby('account_id')['paese']
.transform(lambda x: x != x.shift(1))
.astype(int)
)
return df
def calcola_distanza_ultima_transazione(df: pd.DataFrame) -> pd.Series:
"""Calcola distanza in km dalla transazione precedente dello stesso account."""
from math import radians, sin, cos, sqrt, atan2
def haversine(lat1, lon1, lat2, lon2):
R = 6371 # raggio Terra in km
dlat = radians(lat2 - lat1)
dlon = radians(lon2 - lon1)
a = sin(dlat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2
return R * 2 * atan2(sqrt(a), sqrt(1-a))
distanze = []
for account_id, group in df.groupby('account_id'):
group = group.sort_values('timestamp')
d = [0.0] # prima transazione: distanza 0
for i in range(1, len(group)):
d.append(haversine(
group.iloc[i-1]['lat'], group.iloc[i-1]['lon'],
group.iloc[i]['lat'], group.iloc[i]['lon']
))
distanze.extend(d)
return pd.Series(distanze, index=df.index)
# ============================================================
# 2. MODELLO XGBOOST CON SMOTE E CROSS-VALIDATION
# ============================================================
def train_fraud_detector(df: pd.DataFrame):
"""
Allena un rilevatore di frodi con XGBoost + SMOTE.
Restituisce il modello, lo scaler e i risultati.
"""
feature_cols = [
'importo', 'ora_del_giorno', 'giorno_settimana', 'e_weekend',
'e_notte', 'importo_medio_7d', 'num_transazioni_1h',
'num_transazioni_24h', 'z_score_importo', 'distanza_km',
'cambio_paese', 'categoria_merchant', 'canale'
]
X = df[feature_cols].fillna(0)
y = df['is_frode'].astype(int)
print(f"Distribuzione classi: {y.value_counts().to_dict()}")
print(f"Rapporto frodi: {y.mean():.4%}")
# Split stratificato per mantenere proporzione frodi
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# Pipeline: SMOTE -> StandardScaler -> XGBoost
pipeline = ImbPipeline([
('smote', SMOTE(
sampling_strategy=0.1, # porta le frodi al 10% del totale
random_state=42,
k_neighbors=5
)),
('scaler', StandardScaler()),
('classifier', xgb.XGBClassifier(
n_estimators=500,
max_depth=6,
learning_rate=0.05,
subsample=0.8,
colsample_bytree=0.8,
# Peso classi: penalizza errori su frodi
scale_pos_weight=len(y_train[y_train==0]) / len(y_train[y_train==1]),
eval_metric='aucpr', # Precision-Recall AUC per classi sbilanciate
tree_method='hist',
device='cpu',
random_state=42
))
])
# Cross-validation stratificata (5 fold)
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
cv_scores = []
for fold, (train_idx, val_idx) in enumerate(cv.split(X_train, y_train)):
X_cv_train, X_cv_val = X_train.iloc[train_idx], X_train.iloc[val_idx]
y_cv_train, y_cv_val = y_train.iloc[train_idx], y_train.iloc[val_idx]
pipeline.fit(X_cv_train, y_cv_train)
y_pred_proba = pipeline.predict_proba(X_cv_val)[:, 1]
score = roc_auc_score(y_cv_val, y_pred_proba)
cv_scores.append(score)
print(f"Fold {fold+1} AUC-ROC: {score:.4f}")
print(f"\nAUC-ROC medio CV: {np.mean(cv_scores):.4f} +/- {np.std(cv_scores):.4f}")
# Training finale su tutto il training set
pipeline.fit(X_train, y_train)
# Valutazione sul test set
y_pred = pipeline.predict(X_test)
y_pred_proba = pipeline.predict_proba(X_test)[:, 1]
print("\n--- Report di Classificazione ---")
print(classification_report(y_test, y_pred, target_names=['Legittima', 'Frode']))
print(f"AUC-ROC: {roc_auc_score(y_test, y_pred_proba):.4f}")
print(f"Average Precision (PR-AUC): {average_precision_score(y_test, y_pred_proba):.4f}")
return pipeline, X_test, y_test, y_pred_proba監視されていない異常検出のための隔離フォレスト
不正ラベルが利用できない場合 (新しいシステムまたはドメインの一般的なシナリオ) 歴史的)、私たちは孤立の森: を識別する教師なしアルゴリズム デシジョンツリーによる通常の観察よりも迅速に異常を特定します。 ランダム。異常なトランザクションは、通常のトランザクションよりもカットが少なく「隔離」されます。
# anomaly_detection.py
# Isolation Forest per fraud detection unsupervised
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
def detect_anomalies_isolation_forest(
df: pd.DataFrame,
feature_cols: list,
contamination: float = 0.02 # stima del 2% di frodi nel dataset
) -> pd.DataFrame:
"""
Rileva anomalie con Isolation Forest.
contamination: percentuale attesa di anomalie (prior)
"""
X = df[feature_cols].fillna(0)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
iso_forest = IsolationForest(
n_estimators=200,
max_samples='auto',
contamination=contamination,
max_features=1.0,
bootstrap=False,
n_jobs=-1,
random_state=42
)
# -1 = anomalia, 1 = normale
df = df.copy()
df['anomaly_flag'] = iso_forest.fit_predict(X_scaled)
df['anomaly_score'] = iso_forest.score_samples(X_scaled)
# Normalizza score in [0, 1] dove 1 = alta probabilità frode
min_score = df['anomaly_score'].min()
max_score = df['anomaly_score'].max()
df['fraud_probability'] = 1 - (df['anomaly_score'] - min_score) / (max_score - min_score)
anomalie = df[df['anomaly_flag'] == -1]
print(f"Transazioni anomale rilevate: {len(anomalie)} ({len(anomalie)/len(df):.2%})")
print(f"Importo medio anomalie: €{anomalie['importo'].mean():.2f}")
print(f"Importo medio normali: €{df[df['anomaly_flag']==1]['importo'].mean():.2f}")
return df
# Uso combinato: Isolation Forest per pre-screening + XGBoost per scoring
def dual_model_fraud_pipeline(transaction: dict, iso_model, xgb_pipeline) -> dict:
"""
Pipeline a due stadi:
1. Isolation Forest: filtro rapido (latenza <1ms)
2. XGBoost: scoring preciso solo per sospetti (latenza <10ms)
"""
features = extract_features(transaction)
# Stadio 1: fast pre-screening
anomaly_score = iso_model.score_samples([features])[0]
if anomaly_score > -0.1: # threshold per "normale"
return {
'fraud_probability': 0.01,
'decision': 'APPROVE',
'model': 'isolation_forest_fast_path'
}
# Stadio 2: scoring dettagliato solo per sospetti
fraud_prob = xgb_pipeline.predict_proba([features])[0][1]
decision = 'BLOCK' if fraud_prob > 0.7 else 'REVIEW' if fraud_prob > 0.3 else 'APPROVE'
return {
'fraud_probability': float(fraud_prob),
'decision': decision,
'model': 'xgboost_detailed',
'anomaly_score': float(anomaly_score)
}信用スコアリング AI: ロジスティック回帰から LightGBM まで
従来の信用スコアリングは、次のような線形統計モデルに基づいています。 回帰 物流、歴史的にその解釈のしやすさから銀行に好まれてきました。しかし、グラデーションは 特に最新のブースティング マシン (GBM) XGブースト e ライトGBM、 最近のベンチマークでは最大 98.13% の精度を誇り、非常に優れたパフォーマンスを実証しています。 従来のロジスティック回帰の 70 ～ 75% と比較します。
GBM の導入に対する障壁は技術的なものではなく、顧客 (および他の人) にどのように説明するかという規制にありました。 規制当局）モデルが「ブラックボックス」である場合、なぜ融資が拒否されたのですか？ 答えは、説明可能な AI (XAI) con SHAP値.
信用スコアリングのための特徴エンジニアリング
信用スコアリングの品質は、使用される機能に大きく依存します。銀行が結合する 内部データ (取引履歴、口座の行動) と外部データ (銀行リスク センター) イタリアの CRIF) を活用して、多次元のリスク プロファイルを構築します。
# credit_scoring_features.py
# Feature engineering per credit scoring bancario
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def build_credit_features(
applicant_id: str,
transactions: pd.DataFrame,
credit_history: pd.DataFrame,
application: dict
) -> dict:
"""
Costruisce il vettore di feature per credit scoring.
Combina dati comportamentali, storici e di application.
"""
today = datetime.now()
lookback_12m = today - timedelta(days=365)
# Filtra transazioni degli ultimi 12 mesi
txn_12m = transactions[
(transactions['account_id'] == applicant_id) &
(pd.to_datetime(transactions['date']) >= lookback_12m)
]
# ---------- FEATURE COMPORTAMENTALI ----------
features = {}
# Stabilità del saldo
features['saldo_medio_12m'] = txn_12m['saldo'].mean()
features['saldo_min_12m'] = txn_12m['saldo'].min()
features['volatilita_saldo'] = txn_12m['saldo'].std()
features['giorni_saldo_negativo'] = (txn_12m['saldo'] < 0).sum()
# Pattern di entrate (reddito stimato)
entrate = txn_12m[txn_12m['tipo'] == 'ACCREDITO']
features['reddito_medio_mensile'] = entrate['importo'].sum() / 12
features['stabilita_reddito'] = 1 - (entrate['importo'].std() / (entrate['importo'].mean() + 1e-8))
features['num_fonti_reddito'] = entrate['controparte'].nunique()
# Pattern di spesa
uscite = txn_12m[txn_12m['tipo'] == 'ADDEBITO']
features['spesa_media_mensile'] = uscite['importo'].sum() / 12
features['ratio_risparmio'] = 1 - (features['spesa_media_mensile'] / (features['reddito_medio_mensile'] + 1e-8))
# Utilizzo del fido (se disponibile)
if 'fido_disponibile' in txn_12m.columns:
utilizzo = txn_12m[txn_12m['saldo'] < 0]
features['utilizzo_fido_medio'] = abs(utilizzo['saldo'].mean()) / application.get('fido_richiesto', 1)
features['max_utilizzo_fido'] = abs(utilizzo['saldo'].min()) / application.get('fido_richiesto', 1)
# ---------- FEATURE STORICO CREDITIZIO ----------
credit_hist = credit_history[credit_history['applicant_id'] == applicant_id]
features['num_prestiti_attivi'] = credit_hist[credit_hist['status'] == 'attivo'].shape[0]
features['num_prestiti_chiusi'] = credit_hist[credit_hist['status'] == 'chiuso'].shape[0]
features['num_rate_scadute'] = credit_hist['rate_scadute'].sum()
features['max_giorni_ritardo'] = credit_hist['max_giorni_ritardo'].max()
features['ratio_rimborso'] = (
credit_hist['rate_pagate'].sum() /
(credit_hist['rate_totali'].sum() + 1e-8)
)
features['anzianita_relazione_banca'] = (
today - pd.to_datetime(credit_hist['data_prima_apertura'].min())
).days / 365.25
# ---------- FEATURE DEMOGRAFICHE E APPLICAZIONE ----------
features['eta'] = application.get('eta', 0)
features['anni_impiego_attuale'] = application.get('anni_impiego', 0)
features['importo_richiesto'] = application['importo']
features['durata_mesi'] = application['durata_mesi']
features['ratio_debt_income'] = (
features['importo_richiesto'] / 12 /
(features['reddito_medio_mensile'] + 1e-8)
)
# Tipo impiego (encoded)
tipo_impiego_map = {
'dipendente_tempo_indeterminato': 3,
'dipendente_tempo_determinato': 2,
'autonomo': 2,
'libero_professionista': 2,
'pensionato': 3,
'disoccupato': 0
}
features['tipo_impiego_score'] = tipo_impiego_map.get(
application.get('tipo_impiego', 'altro'), 1
)
return featuresLightGBM トレーニングとしきい値の最適化
# credit_scoring_model.py
# Training LightGBM per credit scoring con threshold optimization
import lightgbm as lgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, precision_recall_curve
import shap
import matplotlib.pyplot as plt
import numpy as np
def train_credit_scoring_model(df: pd.DataFrame):
"""
Allena un modello LightGBM per credit scoring.
Target: 1 = default (non rimborsa), 0 = bono pagatore
"""
feature_cols = [
'saldo_medio_12m', 'saldo_min_12m', 'volatilita_saldo',
'giorni_saldo_negativo', 'reddito_medio_mensile',
'stabilita_reddito', 'num_fonti_reddito', 'spesa_media_mensile',
'ratio_risparmio', 'utilizzo_fido_medio', 'num_prestiti_attivi',
'num_rate_scadute', 'max_giorni_ritardo', 'ratio_rimborso',
'anzianita_relazione_banca', 'eta', 'anni_impiego_attuale',
'ratio_debt_income', 'tipo_impiego_score'
]
X = df[feature_cols].fillna(df[feature_cols].median())
y = df['default_flag'].astype(int)
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# Dataset LightGBM nativo (più efficiente)
dtrain = lgb.Dataset(X_train, label=y_train)
dval = lgb.Dataset(X_test, label=y_test, reference=dtrain)
params = {
'objective': 'binary',
'metric': ['binary_logloss', 'auc'],
'boosting_type': 'gbdt',
'num_leaves': 63,
'max_depth': 8,
'learning_rate': 0.03,
'n_estimators': 1000,
'feature_fraction': 0.8,
'bagging_fraction': 0.8,
'bagging_freq': 5,
'min_child_samples': 30, # min campioni per foglia (regularizzazione)
'lambda_l1': 0.1,
'lambda_l2': 0.1,
'is_unbalance': True, # gestisce classi sbilanciate
'verbose': -1,
'random_state': 42
}
callbacks = [
lgb.early_stopping(stopping_rounds=50, verbose=False),
lgb.log_evaluation(period=100)
]
model = lgb.train(
params,
dtrain,
valid_sets=[dtrain, dval],
valid_names=['train', 'val'],
callbacks=callbacks
)
# Ottimizzazione soglia di decisione
y_pred_proba = model.predict(X_test)
threshold = optimize_threshold(y_test, y_pred_proba)
y_pred = (y_pred_proba >= threshold).astype(int)
auc = roc_auc_score(y_test, y_pred_proba)
print(f"AUC-ROC: {auc:.4f}")
print(f"Soglia ottimale: {threshold:.3f}")
return model, threshold, X_test, y_test
def optimize_threshold(y_true, y_prob) -> float:
"""
Trova la soglia che massimizza l'F1-score.
In credit scoring si può calibrare per bilanciare
tra costo di un default (FN) e costo di un rifiuto errato (FP).
"""
precisions, recalls, thresholds = precision_recall_curve(y_true, y_prob)
f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-8)
optimal_idx = np.argmax(f1_scores)
return thresholds[optimal_idx] if optimal_idx < len(thresholds) else 0.5Explainable AI (XAI): 財務上の意思決定のための SHAP と LIME
L'AI法EU 信用スコアリング システムを次のように分類します ハイリスクAI (付録 III、ポイント 5)。これは、詳細な技術文書、 適合性評価、人間による監督、そして何よりも、 意思決定の透明性 自動化された。 GDPR (第 22 条) および CRD IV 銀行規制では、現在すでに次のことを要求しています。 自動化された与信判断は顧客に説明可能です。
SHAP (シャープリー添加剤の説明) これらに準拠するために最もよく使用されるツール 要件。これは協力ゲーム理論に基づいており、予測ごとに寄与度を計算します。 数学的に厳密かつ一貫した方法で、各特徴の限界を最終結果に導きます。
# explainable_credit_scoring.py
# SHAP per spiegare decisioni di credit scoring
import shap
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import lightgbm as lgb
def explain_credit_decision(
model: lgb.Booster,
application_features: pd.DataFrame,
feature_names: list,
threshold: float = 0.4
) -> dict:
"""
Genera una spiegazione SHAP per una singola decisione di credito.
Ritorna sia la spiegazione tecnica che un testo human-readable.
"""
# Calcola SHAP values per l'istanza
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(application_features)
# Per classificazione binaria LightGBM, shap_values ha shape (n_samples, n_features)
if isinstance(shap_values, list):
shap_vals = shap_values[1] # classe positiva (default)
else:
shap_vals = shap_values
# Contributi ordinati per importanza assoluta
contributions = pd.DataFrame({
'feature': feature_names,
'valore': application_features.values[0],
'shap_value': shap_vals[0],
'impatto': ['AUMENTA_RISCHIO' if v > 0 else 'RIDUCE_RISCHIO' for v in shap_vals[0]]
}).sort_values('shap_value', key=abs, ascending=False)
base_value = explainer.expected_value
if isinstance(base_value, list):
base_value = base_value[1]
# Probabilità di default
prob_default = model.predict(application_features)[0]
decisione = 'RIFIUTO' if prob_default >= threshold else 'APPROVAZIONE'
# Genera spiegazione human-readable (per il cliente)
motivi_principali = contributions.head(5)
spiegazione = genera_spiegazione_testuale(motivi_principali, decisione, prob_default)
return {
'decisione': decisione,
'probabilita_default': float(prob_default),
'score_credito': int((1 - prob_default) * 1000), # scala 0-1000
'spiegazione_testuale': spiegazione,
'contributi_feature': contributions.to_dict('records'),
'base_value': float(base_value),
'shap_values': shap_vals[0].tolist()
}
def genera_spiegazione_testuale(contributi: pd.DataFrame, decisione: str, prob: float) -> str:
"""
Genera testo in linguaggio naturale per spiegare la decisione.
Richiesto da GDPR Art. 22 e AI Act per sistemi ad alto rischio.
"""
linee = []
linee.append(f"Decisione: {decisione} (score: {int((1-prob)*1000)}/1000)")
linee.append("\nFattori principali che hanno influenzato la decisione:")
for _, row in contributi.iterrows():
feature_label = FEATURE_LABELS.get(row['feature'], row['feature'])
valore = format_valore(row['feature'], row['valore'])
if row['shap_value'] > 0:
linee.append(f" [-] {feature_label}: {valore} (aumenta il rischio)")
else:
linee.append(f" [+] {feature_label}: {valore} (riduce il rischio)")
if decisione == 'RIFIUTO':
linee.append("\nPer richiedere una revisione o maggiori informazioni, contatta il tuo referente bancario.")
return "\n".join(linee)
# Dizionario di etichette human-readable per le feature
FEATURE_LABELS = {
'ratio_debt_income': 'Rapporto rata/reddito',
'giorni_saldo_negativo': 'Giorni con saldo negativo negli ultimi 12 mesi',
'max_giorni_ritardo': 'Massimo ritardo nei pagamenti (giorni)',
'ratio_risparmio': 'capacità di risparmio',
'anzianita_relazione_banca': 'Anzianita relazione bancaria (anni)',
'reddito_medio_mensile': 'Reddito medio mensile stimato',
'stabilita_reddito': 'Stabilità del reddito',
'num_rate_scadute': 'Rate scadute storiche',
'volatilita_saldo': 'Variabilità del saldo',
'tipo_impiego_score': 'Tipo di impiego'
}
def visualizza_shap_waterfall(model, application_df, feature_names, filepath: str):
"""Genera grafici SHAP per documentazione e audit trail."""
explainer = shap.TreeExplainer(model)
shap_values = explainer(application_df)
fig, axes = plt.subplots(1, 2, figsize=(16, 6))
# Waterfall plot: contributi per singola predizione
plt.sca(axes[0])
shap.plots.waterfall(shap_values[0], max_display=10, show=False)
axes[0].set_title("Contributi SHAP - Singola Predizione")
# Summary plot: importanza globale
plt.sca(axes[1])
shap.summary_plot(shap_values.values, application_df,
feature_names=feature_names, show=False)
axes[1].set_title("Importanza Feature - Globale")
plt.tight_layout()
plt.savefig(filepath, dpi=150, bbox_inches='tight')
plt.close()
print(f"Report SHAP salvato: {filepath}")SHAP と LIME: いつどちらを使用するか
| 待ってます | シャープ | ライム |
|---|---|---|
| 理論的基礎 | シャプリー価値観（ゲーム理論） | 線形局所近似 |
| 一貫性 | 高 (数学的保証) | 平均（確率的） |
| スピード | 遅い (ただし TreeSHAP と GBM の場合は速い) | もっと早く |
| 説明タイプ | グローバル + ローカル | ローカルのみ |
| 銀行の生産での使用 | 事実上の標準 (監査対応) | デバッグ用に補完 |
| AI法の遵守 | 適切（文書付き） | それだけでは不十分 |
Kafka と Flink によるリアルタイムのトランザクション監視
支払いトランザクションの不正検出は待ってはくれません: 不正なトランザクション クレジットカードによる支払いはブロックする必要があります 前に 後ではなく、完了のとき。 これには、1 秒あたり数百万のイベントを処理できるストリーミング アーキテクチャが必要です。 遅延は 100 ミリ秒未満です。
2025 年の標準アーキテクチャの組み合わせ アパッチ カフカ メッセージブローカーとして イベントの取り込みのため、 アパッチフリンク ステートフル処理用 ML モデルのアプリケーションとイベントを強化する機能ストアによるリアルタイム 事前に計算された行動特徴を備えています。
リアルタイム不正検出アーキテクチャ
| 成分 | テクノロジー | 役割 | 一般的なレイテンシー |
|---|---|---|---|
| イベントの取り込み | アパッチ カフカ | 受信トランザクションをバッファリングする | <5ms |
| 機能の強化 | Redis + 機能ストア | 行動的な特徴を追加する | <10ms |
| ML推論 | Apache Flink + ONNX | GBMモデルによるスコアリング | <20ms |
| 意思決定エンジン | カスタムルール + ML スコア | 承認/レビュー/ブロック | <5ms |
| アラートとケースの管理 | Elasticsearch + キバナ | ダッシュボードと調査員のアラート | ほぼリアルタイム |
| フィードバックループ | カフカ + MLflow | 増分再トレーニング | 毎日/毎週 |
# real_time_fraud_flink.py
# Pseudo-codice architetturale per Flink fraud detection
# (PyFlink API - Apache Flink 1.18+)
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors.kafka import (
FlinkKafkaConsumer, FlinkKafkaProducer
)
import json
import redis
import joblib
import numpy as np
# Modello ML pre-caricato (ONNX o joblib)
MODEL = joblib.load('/models/fraud_detector_v2.pkl')
REDIS_CLIENT = redis.Redis(host='redis', port=6379, db=0)
def enrich_with_behavioral_features(transaction: dict) -> dict:
"""
Arricchisce la transazione con feature comportamentali
dal feature store (Redis).
"""
account_id = transaction['account_id']
# Feature pre-calcolate aggiornate ogni 5 minuti
cached = REDIS_CLIENT.hgetall(f"features:{account_id}")
if cached:
transaction['importo_medio_7d'] = float(cached.get(b'importo_medio_7d', 0))
transaction['num_txn_24h'] = int(cached.get(b'num_txn_24h', 0))
transaction['importo_max_30d'] = float(cached.get(b'importo_max_30d', 0))
transaction['paesi_visitati_30d'] = int(cached.get(b'paesi_visitati_30d', 1))
else:
# Default per account senza storico (nuovi clienti)
transaction['importo_medio_7d'] = 0.0
transaction['num_txn_24h'] = 0
transaction['importo_max_30d'] = 0.0
transaction['paesi_visitati_30d'] = 1
return transaction
def score_transaction(transaction: dict) -> dict:
"""
Applica il modello ML per calcolare il fraud score.
Questa funzione viene eseguita in Flink per ogni evento.
"""
features = [
transaction.get('importo', 0),
transaction.get('ora_del_giorno', 12),
transaction.get('e_weekend', 0),
transaction.get('e_notte', 0),
transaction.get('importo_medio_7d', 0),
transaction.get('num_txn_24h', 0),
transaction.get('z_score_importo', 0),
transaction.get('distanza_km', 0),
transaction.get('cambio_paese', 0),
transaction.get('canale_encoded', 0),
]
fraud_prob = MODEL.predict_proba([features])[0][1]
# Regole ibride: combina ML score con regole business hard-coded
# Le regole business catturano pattern noti sempre validi
hard_block = apply_hard_rules(transaction)
if hard_block:
final_decision = 'BLOCK'
final_score = 1.0
elif fraud_prob >= 0.70:
final_decision = 'BLOCK'
final_score = fraud_prob
elif fraud_prob >= 0.30:
final_decision = 'REVIEW'
final_score = fraud_prob
else:
final_decision = 'APPROVE'
final_score = fraud_prob
return {
**transaction,
'fraud_score': float(final_score),
'decision': final_decision,
'scored_at': datetime.utcnow().isoformat()
}
def apply_hard_rules(txn: dict) -> bool:
"""
Regole deterministiche ad alta precision.
Eseguono PRIMA del modello ML per blocchi immediati.
"""
# Regola 1: importo anomalo (10x superiore alla media del cliente)
if txn.get('importo', 0) > txn.get('importo_medio_7d', 0) * 10:
return True
# Regola 2: transazione da paese blacklist
BLACKLIST_PAESI = {'COUNTRY_A', 'COUNTRY_B'} # configurabile
if txn.get('paese', '') in BLACKLIST_PAESI:
return True
# Regola 3: velocity check - più di 5 transazioni in 1 ora
if txn.get('num_txn_1h', 0) > 5:
return True
# Regola 4: impossibile travel (2 paesi diversi in meno di 2 ore)
if txn.get('distanza_km', 0) > 1000 and txn.get('ore_dalla_precedente', 999) < 2:
return True
return Falseグラフニューラルネットワークによるマネーロンダリング対策
マネーロンダリングはクレジット カード詐欺とは構造的に異なります。 エンティティネットワーク (人、企業、銀行口座) 取引によってつながっています コンプレックス。従来のルールベースのシステムは、既知のパターン (構造化、階層化、 統合）が、新たなパターンと間接的な接続が欠落しています。
Le グラフ ニューラル ネットワーク (GNN) AML の自然な進化を表す： ノードがエンティティ (アカウント、顧客、企業) であるグラフとして金融システムをモデル化します。 そしてアークはトランザクションです。 GNN はノードを疑わしいノードまたは正常なノードとして分類することを学習します グラフ内の近隣からの情報を集約し、パターンのタイプを正確に捕捉する マネーロンダリング業者はネットワークの複雑さの中に隠れようとしているのです。
# aml_graph_neural_network.py
# Graph Neural Network per Anti-Money Laundering con PyTorch Geometric
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import GATConv, SAGEConv, global_mean_pool
from torch_geometric.data import Data, DataLoader
import pandas as pd
import numpy as np
class AMLGraphAttentionNetwork(nn.Module):
"""
Graph Attention Network per rilevamento AML.
Usa meccanismo di attenzione per pesare i vicini più rilevanti.
Architettura: GAT + SAGEConv + Classificatore
"""
def __init__(
self,
in_channels: int, # dimensione feature nodo
edge_channels: int, # dimensione feature arco (transazione)
hidden_channels: int = 128,
num_heads: int = 8,
dropout: float = 0.3
):
super().__init__()
# Layer 1: Graph Attention (aggrega vicini con attenzione)
self.conv1 = GATConv(
in_channels,
hidden_channels // num_heads,
heads=num_heads,
dropout=dropout,
edge_dim=edge_channels
)
# Layer 2: GraphSAGE (aggregazione robusta per grafi sparsi)
self.conv2 = SAGEConv(hidden_channels, hidden_channels)
# Layer 3: Classificatore finale
self.classifier = nn.Sequential(
nn.Linear(hidden_channels, 64),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(64, 2) # classe 0: legittimo, classe 1: AML
)
self.dropout = nn.Dropout(dropout)
self.bn1 = nn.BatchNorm1d(hidden_channels)
def forward(self, x, edge_index, edge_attr):
# Propagazione del messaggio tra nodi connessi
x = self.conv1(x, edge_index, edge_attr)
x = F.elu(self.bn1(x))
x = self.dropout(x)
x = self.conv2(x, edge_index)
x = F.relu(x)
x = self.dropout(x)
# Classificazione per nodo
return self.classifier(x)
def build_transaction_graph(transactions: pd.DataFrame) -> Data:
"""
Costruisce il grafo finanziario da un DataFrame di transazioni.
Nodi: conti bancari / entità
Archi: transazioni (con feature: importo, tipo, data)
"""
# Crea mapping entità -> indice nodo
entità = pd.concat([
transactions['conto_mittente'],
transactions['conto_destinatario']
]).unique()
node_map = {e: i for i, e in enumerate(entità)}
# Feature dei nodi (aggregate dalle transazioni)
num_nodes = len(entità)
node_features = np.zeros((num_nodes, 8)) # 8 feature per nodo
for _, txn in transactions.iterrows():
i = node_map[txn['conto_mittente']]
j = node_map[txn['conto_destinatario']]
# Feature mittente
node_features[i][0] += txn['importo'] # volume uscite
node_features[i][1] += 1 # num transazioni uscita
node_features[i][2] = max(node_features[i][2], txn['importo']) # max uscita
# Feature destinatario
node_features[j][3] += txn['importo'] # volume entrate
node_features[j][4] += 1 # num transazioni entrata
# Normalizza feature
node_features = (node_features - node_features.mean(0)) / (node_features.std(0) + 1e-8)
# Archi: lista di (mittente, destinatario)
edge_index = torch.tensor([
[node_map[r] for r in transactions['conto_mittente']],
[node_map[r] for r in transactions['conto_destinatario']]
], dtype=torch.long)
# Feature degli archi (transazioni)
edge_attr = torch.tensor(transactions[[
'importo_normalizzato',
'tipo_transazione_encoded',
'ora_del_giorno_sin', # encoding ciclico dell'ora
'ora_del_giorno_cos',
'num_round_amounts' # importi tondi: segnale AML tipico
]].values, dtype=torch.float)
# Label dei nodi (da sistema di indagine / SAR storico)
y = torch.tensor(
[1 if e in KNOWN_AML_ENTITIES else 0 for e in entità],
dtype=torch.long
)
return Data(
x=torch.tensor(node_features, dtype=torch.float),
edge_index=edge_index,
edge_attr=edge_attr,
y=y,
num_nodes=num_nodes
)
def train_aml_gnn(graph_data: Data, epochs: int = 200):
"""Allena la GNN per AML detection."""
model = AMLGraphAttentionNetwork(
in_channels=graph_data.x.shape[1],
edge_channels=graph_data.edge_attr.shape[1],
hidden_channels=128,
num_heads=8
)
optimizer = torch.optim.AdamW(model.parameters(), lr=0.001, weight_decay=1e-4)
# Pesi per classi sbilanciate (pochissimi AML nel totale)
class_weights = torch.tensor([1.0, 20.0]) # penalizza false negative su AML
criterion = nn.CrossEntropyLoss(weight=class_weights)
model.train()
for epoch in range(epochs):
optimizer.zero_grad()
out = model(graph_data.x, graph_data.edge_index, graph_data.edge_attr)
loss = criterion(out[graph_data.train_mask], graph_data.y[graph_data.train_mask])
loss.backward()
optimizer.step()
if epoch % 20 == 0:
with torch.no_grad():
pred = out.argmax(dim=1)
accuracy = (pred[graph_data.val_mask] == graph_data.y[graph_data.val_mask]).float().mean()
print(f"Epoch {epoch} | Loss: {loss:.4f} | Val Acc: {accuracy:.4f}")
return modelAML 向けの GNN とルールベースのシステム: 比較
| メトリック | ルールベース | GNN |
|---|---|---|
| 誤検知率 | 高 (40-60%) | 低い (10-20%) |
| 新しいパターン | 検出されない (ゼロデイ) | 検出可能 (一般化) |
| 間接的なつながり | 最大1～2ホップ | マルチホップ (最大 5 度以上) |
| メンテナンス | 高（手動ルール） | 低 (再トレーニングによる自動更新) |
| 説明可能性 | 高 (可読ルール) | 平均（注目の重み） |
| 導入コスト | ベース | 高 (データ、インフラストラクチャ) |
RegTech: AI 法、PSD3、および自動化されたコンプライアンス
欧州の金融セクターは前例のない規制変革を迎えています。 3 つの規制は重複しており、AI システムに直接的な影響を与えます。 銀行業:
金融における AI に関する EU 規制枠組み 2025 ～ 2027 年
| 規則 | 発効 | AI 金融への影響 |
|---|---|---|
| AI法EU | 2025 年 2 月 (2027 年 8 月まで段階的に) | 信用スコアリング = 高リスク。義務: 登録、監査、人間による監督、透明性 |
| PSD3 | 採用は2025～2026年、転置は2027年 | SCAの強化、オープンバンキングAPI、不正行為責任のシフト、データ共有の拡大 |
| ドーラ (デジタル・オペレーショナル・レジリエンス法) | 2025年1月（発効） | 必須の ICT レジリエンス、4 時間のインシデント報告、AI システム侵入テスト |
| AMLD6 （マネーロンダリング防止指令） | プログレッシブ適用 2025 | デジタルKYC、実質的所有権、強制的な取引監視 |
EU における AI 法: 銀行への実際的な影響
L'AI法EU AI バンキング システムを 3 つの主要なカテゴリに分類します。 増大する義務。自動信用スコアリングは次のように明示的にリストされています。 ハイリスクAI これは、銀行が次のことを行う必要があることを意味します。
銀行向けAI法遵守チェックリスト（ハイリスクAI）
- AI システムログ: 高リスク システムの EU 登録簿に信用判断に使用される各 AI システムを文書化する (2026 年 8 月から義務化)
- 適合性評価: リスク分析、バイアステスト、保護された人口統計データのパフォーマンスを含む導入前評価
- 人間による監督: 大きな影響を与える意思決定（定義されたしきい値を超えるローンの拒否）に対する人間によるレビュープロセスの必須化
- 透明度: 顧客には、その決定が AI システムによって行われた (または支援された) ことを通知する必要があります。
- 精度と堅牢性: 継続的なパフォーマンス監視、敵対的テスト、データドリフト検出
- 技術文書: モデルスキーム、トレーニングデータ、パフォーマンスメトリクス、バイアス分析（第11条）
- ログと監査証跡: 各 AI 決定のログを少なくとも 5 年間保存する (第 12 条)
- 説明権: 顧客は、AI の決定について意味のある説明を受ける権利を有します (GDPR 第 22 条 + AI 法)
PSD3 とオープン バンキング: より良いモデルのための新しいデータ
PSD3決済サービス指令の改訂版では、フレームワークが導入されています。 の オープンバンキング PSD2 よりも大幅に堅牢です。システム向け 信用スコアリングにより、具体的な可能性が広がります。顧客の同意を得て、銀行が 他の機関の口座データにアクセスして財務概要を取得できるようになります 内部データだけよりもはるかに完全です。
別の銀行で常に住宅ローンを期日通りに支払ってきたが、新規の顧客 ローンを要求している銀行の顧客は、その前向きな話を持ち込むことができるようになります。 あなたと一緒に。これは、オープンファイナンス: 財務データは同意を得て共有されます ユーザーはリスクモデルの精度を向上させることができます。
ケーススタディ: イタリアの小売銀行における不正行為の検出
イタリアの銀行業界における実際の実装にインスピレーションを得たケーススタディを紹介します。 1 つのシステムから移行した中規模銀行 (約 500,000 の個人顧客) ハイブリッド ML システムへのルールベースのレガシー + トランザクション不正検出ルール デビットカードとクレジットカードで。
開始シナリオ
| メトリック | 以前 (ルールベース) | 後（ハイブリッドML） | 改善 |
|---|---|---|---|
| 検出された不正行為 (%) | 72% | 91% | +19ポイント |
| 誤検知 (不良ブロック) | 5.2% | 0.8% | -84% |
| 平均意思決定待ち時間 | 450ミリ秒 | 85ミリ秒 | -81% |
| 誤ったブロックに対する顧客からの苦情 | 1,200/月 | 190/月 | -84% |
| 年間純詐欺コスト | 420万ユーロ | 110万ユーロ | -74% |
| 詐欺チームのオペレーター | 22人のFTE | 14 名の FTE | -36% |
実装されたアーキテクチャ
このシステムは、14 か月にわたって 3 つのフェーズで構築されました。
導入ロードマップ (14 か月)
- フェーズ 1 - 基礎 (1 ～ 4 か月目): 過去の不正行為データの統合 (5 年間、1 億 2,000 万トランザクション)、特徴エンジニアリング、ラベル付けされたトレーニング データセット 調査員。トランザクション ストリーミング用の Kafka インフラストラクチャ。特集ストア リアルタイム機能については Redis 上で使用します。ベースライン ロジスティック回帰モデル (AUC 0.78)。
- フェーズ 2 - ML コア (5 ～ 10 か月目): XGBoost + SMOTE のトレーニング (AUC 0.94)。 リアルタイム スコアリングのための Flink パイプラインの実装。基幹システムとの統合 銀行業。 A/B テスト: 新しいシステムでは 20% のトラフィック、古いシステムでは 80% のトラフィック。モニタリング グラファナのダッシュボード。 FP/FN のバランスをとるためのしきい値の調整。
- フェーズ 3 - 生産とコンプライアンス (11 ～ 14 か月目): 完全展開 トラフィックの 100%。説明ログ用の SHAP 実装。ドキュメント AI 法準拠のための技術 (2026 年の義務の予測)。毎月のパイプライン再トレーニング MLflowを使って。捜査官の事件管理システムとの統合。
直面する課題と解決策
この実装では、イタリアの銀行業界に特有のいくつかの課題に直面しました。
課題とその解決方法
- 履歴データのデータ品質が低い: 5 年間のデータセットには次のものが含まれていました 不正確な詐欺ラベル (すべての詐欺が正しくラベル付けされているわけではありません)。 解決策: 層別サンプル + ルールに対する手動再ラベル付けキャンペーン 明らかなケース（承認されたチャージバック = 特定の不正行為など）の自動クリーニング。
- コアバンキングシステムのレイテンシ: 従来の Temenos T24 システムは、 100ms 未満のリアルタイム統合向けに設計されています。解決策: 事前認証パターン Kafka と非同期で、スコアリング システムを最終承認から切り離します。
- 詐欺チームのオペレーターに対する抵抗: チームはAIに取って代わられるのではないかと懸念している 彼らの仕事。解決策: ワークフローを使用して「AI 支援調査員」として再配置する そのため、オペレーターは承認/ブロックを決定するのではなく、ケースを検討することになります。 自動。チームは複雑な事件とAMLの調査に焦点を当てました。
- 人口統計上の偏り: 初期モデルは誤ったレートを示しました 南イタリアの特定の地域での取引のプラスが増加（歴史的には低い） トランザクション)。解決策: トレーニング中の公平性の制約、差別化されたしきい値 履歴が限られているセグメントの場合、月次モニタリングのバイアス。
財務モデルの MLOps: モニタリングと再トレーニング
不正行為の検出と信用スコアリングのモデルは静的ではありません: 詐欺師の行動 継続的に進化し、経済パターンが変化し、新しい金融商品が導入される 新たなリスク。の データドリフト そして コンセプトドリフト それらは現象です ML 金融システムでは一般的です。
# mlops_financial_monitoring.py
# Monitoring e alerting per modelli finanziari in produzione
import mlflow
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from scipy import stats
from sklearn.metrics import roc_auc_score
def monitor_model_performance(
model_name: str,
production_window_days: int = 7
) -> dict:
"""
Monitora le performance del modello in produzione.
Rileva data drift e performance degradation.
"""
client = mlflow.tracking.MlflowClient()
# Recupera le predizioni degli ultimi N giorni dal data warehouse
end_date = datetime.now()
start_date = end_date - timedelta(days=production_window_days)
predictions_df = load_predictions_from_dw(model_name, start_date, end_date)
actuals_df = load_actuals_from_dw(model_name, start_date, end_date)
# Unisci predizioni e outcome reali
merged = predictions_df.merge(actuals_df, on='transaction_id', how='inner')
monitoring_results = {}
# 1. Performance metrics (solo per transazioni con label confermato)
labeled = merged[merged['label_confirmed'] == True]
if len(labeled) > 0:
current_auc = roc_auc_score(labeled['is_fraud'], labeled['fraud_score'])
monitoring_results['current_auc'] = current_auc
# Confronto con baseline di riferimento
baseline_auc = get_baseline_auc(model_name)
auc_degradation = baseline_auc - current_auc
monitoring_results['auc_degradation'] = auc_degradation
if auc_degradation > 0.05:
monitoring_results['alert_performance'] = True
send_alert(f"ALERT: AUC degraded by {auc_degradation:.3f} for {model_name}")
# 2. Score distribution drift (PSI - Population Stability Index)
reference_scores = load_reference_score_distribution(model_name)
current_scores = merged['fraud_score'].values
psi = calculate_psi(reference_scores, current_scores, buckets=10)
monitoring_results['psi'] = psi
# PSI < 0.1: nessun drift; 0.1-0.25: drift moderato; > 0.25: drift severo
if psi > 0.25:
monitoring_results['alert_drift'] = True
send_alert(f"ALERT: Severe score drift detected (PSI={psi:.3f}) for {model_name}")
elif psi > 0.1:
monitoring_results['warning_drift'] = True
# 3. Feature drift (KS test su feature principali)
reference_features = load_reference_feature_distribution(model_name)
feature_drift = {}
for feature in reference_features.columns:
ks_stat, ks_pvalue = stats.ks_2samp(
reference_features[feature].dropna(),
merged[feature].dropna()
)
feature_drift[feature] = {'ks_stat': ks_stat, 'pvalue': ks_pvalue}
if ks_pvalue < 0.01:
monitoring_results[f'drift_{feature}'] = True
monitoring_results['feature_drift'] = feature_drift
# 4. Log su MLflow
with mlflow.start_run(run_name=f"monitoring_{model_name}_{datetime.now().date()}"):
mlflow.log_metrics({
'monitoring_auc': monitoring_results.get('current_auc', 0),
'monitoring_psi': psi,
'monitoring_window_days': production_window_days
})
mlflow.log_dict(monitoring_results, 'monitoring_results.json')
return monitoring_results
def calculate_psi(expected: np.ndarray, actual: np.ndarray, buckets: int = 10) -> float:
"""
Population Stability Index: misura il drift nella distribuzione degli score.
Formula: PSI = sum((A_i - E_i) * ln(A_i/E_i))
"""
# Crea bucket basati sulla distribuzione di riferimento
breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1))
breakpoints[0] = -np.inf
breakpoints[-1] = np.inf
expected_pct = np.histogram(expected, bins=breakpoints)[0] / len(expected)
actual_pct = np.histogram(actual, bins=breakpoints)[0] / len(actual)
# Evita divisione per zero
expected_pct = np.where(expected_pct == 0, 1e-8, expected_pct)
actual_pct = np.where(actual_pct == 0, 1e-8, actual_pct)
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return float(psi)金融における AI のベスト プラクティスとアンチパターン
統合されたベストプラクティス
- データ品質から始めます。 ML モデルを構築する前に、 履歴データの理解とクリーニングに 2 ～ 3 か月を投資します。データ上のモデル 汚くて、ルールベースのシステムよりも悪い。ガベージインガベージアウトルールは二重です 金融業界では、誤った決定が法的影響を及ぼします。
- 不均衡なクラスの正しいメトリクス: 決して精度を使用しないでください 不正行為検出の主な指標として使用されます。 AUC-ROC、プレシジョン-リコール AUC、 F1 スコア。 FP の相対コストを明示的に定義します (顧客が不必要にブロックされる) および FN (不正が検出されない) を使用して、最適なしきい値を調整します。
- ハイブリッド ルール + ML モデル: ルールを完全に排除しないでください 決定論的。ルールは既知のパターンを高精度でキャプチャし、監査可能です。 ML は、複雑で斬新なパターンに優れています。最適なアーキテクチャでは両方を使用します。 別々のレイヤー。
- 一元化された機能ストア: 行動特徴を計算する (速度、移動平均、集計) を 1 回だけ実行し、特徴に保存します ストア (Redis、Feast、Tecton)。これにより、トレーニングと推論の間の一貫性が確保されます。 そして本番のレイテンシを短縮します。
- チャンピオンとチャレンジャーのテスト: ビッグバン展開はしないでください。使用する 常にトラフィック分割を使用した A/B テストを実行します。 「チャレンジャー」モデルは 10 ～ 20% を受け取ります メトリクスが現在の「チャンピオン」を超えた場合にのみ昇格されます。
- 包括的な意思決定ログ: すべての予測を 完全な特徴ベクトル、SHAP 値、最終決定と結果 確認されました。これは、監査、デバッグ、再トレーニングに不可欠です。
避けるべきアンチパターン
- 漏洩日: 利用可能な情報を機能に含めないでください ラベル付け時のみ（機能として「このアカウントはブロックされています」など） トレーニングの）。モデルは、実際の現象ではなくラベルを予測することを学習します。
- 既知の不正行為へのオーバーフィッティング: パターンのみでトレーニングされたモデル 歴史的な詐欺は新しいパターンでは失敗します。常に堅牢なネガティブ サンプリングを含める 将来の時間枠でテストします (前方検証)。
- フィードバック ループを無視します。 モデルがトランザクションをブロックしても、ブロックされません。 それが本当に詐欺だったのかどうかは決してわかりません。この選択バイアスは徐々に歪んでいきます トレーニングセット。ブロックされたトランザクションに対するサンプル レビュー システムを実装します。
- キャリブレーションなしのスコア: XGBoost と LightGBM は確率を生成しません デフォルトで調整されています。 0.8のfraudture_scoreは、「80%の不正行為の可能性」を意味するものではありません。 Platt スケーリングまたは等張回帰を使用して、確率出力を調整します。
- 監視なしで導入: 財務モデルはすぐに劣化します。 6月には優れたモデルでも、12月には変更により凡庸になる可能性がある 季節性と不正パターンの進化。
結論と次のステップ
金融分野における AI は、単純な異常検出からシステムまで急速に成熟しています 勾配ブースティング、グラフ ニューラル ネットワーク、リアルタイム ストリーミング、および ますます厳しくなる規制要件を満たすための説明可能性。 2025年から2026年は 重要な時期：AI法EU 信用スコアリング システムを範囲に組み込む 正式な規制の一方、 PSD3 e ドーラ 彼らは再定義する セキュリティと復元力の要件。
イタリアの銀行とフィンテックにとって、メッセージは明確です。 同社の不正行為検出および信用スコアリング システムにはチャンスがある 2025年から2026年には規制上の義務を見据えてそうする予定であり、 コンプライアンスを競争上の優位性へと導きます。
覚えておくべき重要なポイント
- Il 99％の金融機関が すでに不正検出に ML を使用していますが、実装の品質は大きく異なります
- XGBoost と LightGBM SMOTE を使用すると、信用スコアリングを支配します: パフォーマンス >95% AUC、SHAP によるコンプライアンスの説明可能性
- リアルタイム アーキテクチャ カフカ + フリンク + Redis および 100ms 未満のトランザクション監視の標準
- Le グラフニューラルネットワーク AML に革命をもたらしています: ルールでは捉えることが不可能な複雑なネットワーク内のマネーロンダリング パターンを検出します
- L'AI法EU 信用スコアリングを高リスク AI として分類: 2026 年から正式に義務付けられるが、今から準備をしたほうがよい
- Il ハイブリッド ルール + ML モデル 本番環境での最良の選択: 決定論的ルールの堅牢性と監査可能性を備えた ML パフォーマンス
シリーズの次の記事では、小売における AI、どこに直面するか 時系列による需要予測、協調フィルタリングによるレコメンデーションエンジン 行列因数分解と強化学習による動的価格設定。さまざまなテクニック 金融と比較されますが、多くのアーキテクチャ パターンが共通しています: フィーチャー ストア、A/B テスト 実稼働モデルの MLOps。
詳細を学ぶためのリソース
- シリーズとのつながり MLOps: 財務モデルのライフサイクルを管理するための「MLOps for Business: MLflow を使用した本番環境の AI モデル」
- シリーズとのつながり AIエンジニアリング: ガードレールとコンプライアンスを備えた銀行カスタマー ケア チャットボット用の「LLM in the Company: RAG Enterprise」
- 公式ドキュメント: 内部ガバナンスに関する EBA ガイドライン (2021 年) e 気候関連および環境リスクに関する ECB ガイド EU の銀行規制の文脈について
- 実験用の公開データセット: IEEE-CIS 不正検出 Kaggle 上 (590,000 トランザクション、1% 詐欺) e 住宅信用デフォルトリスク 信用スコアリング用