금융 분야의 AI: 사기 탐지, 신용 평가 및 위험 관리
2024년에는 금융 사기로 인한 전 세계 손실액이 초과되었습니다. 125억 달러, 전년 대비 25% 증가했다. 같은 해 사기 탐지 시스템 인공지능을 기반으로 약 255억 달러의 손실을 예방했습니다. 전 세계적으로. 결론은 분명합니다. 금융 부문의 AI는 더 이상 경쟁 우위가 아닙니다. 그리고 운영상의 필요성.
오늘은 금융기관의 99% 일종의 기계 학습이나 AI를 사용합니다. 사기를 퇴치하기 위해 93%는 AI가 사기 탐지 기능에 혁명을 일으킬 것이라고 믿고 있습니다. 내년. 하지만 이러한 시스템은 실제로 어떻게 작동합니까? 신용 모델을 구축하는 방법 유럽 규제 요구 사항을 초과하는 점수를 매겼습니까? 탐지가 가능한 AML 시스템을 구현하는 방법 수천 건의 거래 네트워크에서 돈세탁 계획을 세우나요?
이 기사에서는 실제 코드, 구체적인 아키텍처 및 사례 연구를 통해 이러한 질문에 답합니다. 이탈리아 은행 환경에서 영감을 얻었습니다. 우리는 실시간 사기 감지 카프카(Kafka)와 플링크(Flink)와 함께 해석 가능한 신용 점수 XGBoost 및 SHAP를 사용하면 AML 감지 Graph Neural Networks를 사용하고AI법 EU e PSD3.
이 기사에서 배울 내용
- ML에서 사기 탐지가 작동하는 방식: 격리 포레스트, 그래디언트 부스팅, 신경망
- XGBoost 및 LightGBM을 사용한 AI 신용 점수: 기능 엔지니어링 및 실제 성능(>95% 정확도)
- 규정을 준수하는 재무 결정을 위한 설명 가능한 AI(SHAP 및 LIME)
- Kafka 및 Flink를 사용한 트랜잭션 모니터링을 위한 실시간 아키텍처
- 그래프 신경망을 사용한 자금세탁 방지: 규칙 기반 시스템이 놓친 패턴을 탐지합니다.
- RegTech: AI Act EU, PSD3, DORA 및 이탈리아 은행에 미치는 영향
- 사례 연구: 이탈리아 소매 은행의 사기 탐지 구현
데이터 웨어하우스, AI 및 디지털 혁신 시리즈
| # | Articolo | 집중하다 |
|---|---|---|
| 1 | 데이터 웨어하우스의 진화 | SQL Server에서 데이터 레이크하우스로 |
| 2 | 데이터 메시 및 분산형 아키텍처 | 데이터의 도메인 소유권 |
| 3 | ETL과 최신 ELT | dbt, 에어바이트, Fivetran |
| 4 | 파이프라인 오케스트레이션 | Airflow, Dagster 및 Prefect |
| 5 | 제조 분야의 AI | 예측 유지 관리 및 디지털 트윈 |
| 6 | 현재 위치 - 금융 분야의 AI | 사기 탐지 및 신용 점수 |
| 7 | 소매업의 AI | 수요예측 및 추천 |
| 8 | 헬스케어 분야의 AI | 진단 및 약물 발견 |
| 9 | 물류 분야의 AI | 경로 최적화 및 창고 자동화 |
| 10 | 비즈니스 LLM | RAG Enterprise 및 가드레일 |
금융 분야 AI 환경: 2025년 데이터 및 동향
금융 부문은 대규모 머신러닝 기술을 최초로 도입한 분야 중 하나였으며, 오늘날에도 여전히 가장 발전된 기술 중 하나입니다. 그 이유는 구조적입니다. 풍부한 역사적 데이터, 인센티브 막대한 경제적 비용(10,000유로 규모의 사기를 예방하면 즉각적인 ROI를 얻을 수 있음) 및 규제 체계 복잡하기는 하지만 명시적으로 자동화된 모니터링 시스템이 필요합니다.
시장 및 주요 데이터 2025
| 지시자 | 가치 2025 | 동향 |
|---|---|---|
| 글로벌 금융 사기 손실 | 125억 달러(2024년) | 전년 대비 +25% |
| AI로 예방하는 사기 | 255억 달러 | 2025년 추정 |
| AI를 활용하는 금융기관 | 99% | 사기 탐지를 위해 |
| 생성 AI와 관련된 사기 | >50% | 딥페이크, 합성 신원 |
| 정확성 LightGBM 신용 점수 | 98.13% | SHAP 설명 기능 사용 |
| RegTech AI 시장 | 217억 달러(CAGR) | 연간 성장 예상 |
2024~2025년의 근본적인 변화와 등장도구로서의 생성적 AI 공격적인: 오늘날 사기의 50%는 범죄자의 AI 사용과 관련되어 있습니다. 그들은 대규모로 비디오 딥페이크, 합성 신원 및 맞춤형 피싱 캠페인을 만듭니다. 이 현상은 더욱 정교한 방어 시스템의 채택을 가속화하여 기준을 이동시켰습니다. 단순한 이상 탐지부터 복잡한 행동 패턴을 탐지할 수 있는 모델까지.
세 가지 주요 응용 분야
금융 분야의 AI는 각각 요구 사항이 있는 서로 다르지만 상호 연결된 세 가지 도메인으로 구성됩니다. 특정 기술 및 규제:
금융 분야의 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) XGBoost e 라이트GBM, 최근 벤치마크에서 최대 98.13%의 정확도로 상당히 뛰어난 성능을 보여주었습니다. 고전적인 로지스틱 회귀의 70-75%와 비교됩니다.
GBM 채택에 대한 장벽은 기술적인 것이 아니라 규제적인 것이었습니다. 이를 고객(및 다른 사람)에게 어떻게 설명합니까? 규제 기관) 모델이 "블랙박스"인 경우 대출이 거부된 이유는 무엇입니까? 대답은설명 가능한 AI(XAI) ~와 함께 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.5설명 가능한 AI(XAI): 재무 결정을 위한 SHAP 및 LIME
L'AI법 EU 신용 점수 시스템을 다음과 같이 분류합니다. 고위험 AI (부속서 III, 5항) 이는 상세한 기술 문서, 적합성 평가, 사람의 감독, 그리고 무엇보다도 의사결정의 투명성 자동화된. GDPR(제22조) 및 CRD IV 은행 규정에서는 이미 다음 사항을 요구하고 있습니다. 자동화된 신용 결정은 고객에게 설명 가능합니다.
SHAP(SHapley 첨가제 설명) 그리고 이를 준수하기 위해 가장 많이 사용되는 도구는 요구 사항. 협동 게임 이론을 기반으로 합니다. 각 예측에 대해 기여도를 계산합니다. 수학적으로 엄격하고 일관된 방식으로 각 기능을 최종 결과에 미미하게 적용합니다.
# 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: 언제 어느 것을 사용해야 할까요?
| 나는 기다린다 | SHAP | 라임 |
|---|---|---|
| 이론적 기초 | 샤플리 가치(게임이론) | 선형 국소 근사 |
| 일관성 | 높음(수학적 보장) | 평균(확률적) |
| 속도 | 느림(그러나 GBM의 경우 TreeSHAP 및 빠름) | 더 빠르게 |
| 설명 유형 | 글로벌 + 로컬 | 로컬 전용 |
| 은행 생산에 사용 | 사실상의 표준(감사 준비) | 디버깅을 위한 보완 |
| AI법 준수 | 적절함(문서 포함) | 그 자체로는 부족함 |
Kafka 및 Flink를 통한 실시간 트랜잭션 모니터링
결제 거래에 대한 사기 탐지는 기다릴 수 없습니다: 사기 거래 신용 카드로 차단해야합니다 전에 완료 후가 아니라 완료 후입니다. 이를 위해서는 초당 수백만 개의 이벤트를 처리할 수 있는 스트리밍 아키텍처가 필요합니다. 지연 시간은 100밀리초 미만입니다.
2025년의 표준 아키텍처는 다음을 결합합니다. 아파치 카프카 메시지 브로커로서 이벤트 수집을 위해 아파치 플링크 상태 저장 처리를 위해 ML 모델을 적용하여 실시간으로 이벤트를 풍부하게 하는 기능 저장소 미리 계산된 행동 특징을 가지고 있습니다.
실시간 사기 탐지 아키텍처
| 요소 | 기술 | 역할 | 일반적인 지연 시간 |
|---|---|---|---|
| 이벤트 수집 | 아파치 카프카 | 들어오는 트랜잭션 버퍼링 | <5ms |
| 기능 강화 | Redis + 피처 스토어 | 행동 기능 추가 | <10ms |
| ML 추론 | 아파치 플링크 + ONNX | GBM 모델을 사용한 채점 | <20ms |
| 의사결정 엔진 | 맞춤 규칙 + ML 점수 | 승인/검토/차단 | <5ms |
| 경고 및 사례 관리 | 엘라스틱서치 + 키바나 | 대시보드 및 조사자 경고 | 거의 실시간 |
| 피드백 루프 | 카프카 + 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 Act, PSD3 및 자동화된 규정 준수
유럽 금융 부문은 전례 없는 규제 변화를 겪고 있습니다. 세 가지 규정은 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, 실소유자, 필수 거래 모니터링 |
AI Act EU: 은행에 대한 실질적인 영향
L'AI법 EU AI 뱅킹 시스템을 세 가지 주요 범주로 분류합니다. 점점 커지는 의무. 자동 신용 점수는 다음과 같이 명시적으로 나열됩니다. 고위험 AI Annex III에서 이는 은행이 다음을 수행해야 함을 의미합니다.
은행 AI법 준수 체크리스트(고위험 AI)
- AI 시스템 로그: EU 고위험 시스템 등록부에 신용 결정에 사용되는 각 AI 시스템을 문서화합니다(2026년 8월부터 필수).
- 적합성 평가: 위험 분석, 편향 테스트, 보호된 인구통계 데이터에 대한 성능을 통한 배포 전 평가
- 인간 감독: 영향력이 큰 결정을 위한 필수 인적 검토 프로세스(정의된 기준을 초과하는 대출 거부)
- 투명도: AI 시스템이 결정을 내렸거나 지원했다는 사실을 고객에게 알려야 합니다.
- 정확성과 견고성: 지속적인 성능 모니터링, 적대적 테스트, 데이터 드리프트 감지
- 기술 문서: 모델 체계, 교육 데이터, 성과 지표, 편향 분석(제11조)
- 로그 및 감사 추적: 각 AI 결정에 대한 로그를 최소 5년 동안 보관합니다(제12조).
- 설명의 권리: 고객은 AI 결정에 대해 의미 있는 설명을 들을 권리가 있습니다(GDPR 제22조 + AI법).
PSD3 및 오픈 뱅킹: 더 나은 모델을 위한 새로운 데이터
PSD3, 결제 서비스 지침 개정판에서는 프레임워크를 도입합니다. 의 오픈뱅킹 PSD2보다 훨씬 더 강력합니다. 시스템용 신용 점수를 통해 이는 구체적인 가능성을 열어줍니다. 고객의 동의가 있으면 은행 다른 기관의 계좌 데이터에 접근하여 재무 개요를 얻을 수 있습니다. 내부 데이터만 있는 것보다 훨씬 더 완전합니다.
항상 다른 은행에서 담보대출을 제때에 납부하셨으나 신규고객이신 고객님 대출을 요청하는 은행의 고객은 이제 긍정적인 이야기를 들려줄 수 있습니다. 당신과 함께. 이것은오픈파이낸스: 동의 하에 공유되는 금융 데이터 사용자는 위험 모델의 정확성을 향상시킵니다.
사례 연구: 이탈리아 소매 은행의 사기 탐지
우리는 이탈리아 은행 업무 환경의 실제 구현에서 영감을 얻은 사례 연구를 제시합니다. 하나의 시스템에서 마이그레이션한 중간 규모 은행(약 500,000명의 소매 고객) 하이브리드 ML 시스템에 대한 규칙 기반 레거시 + 거래 사기 탐지 규칙 직불카드와 신용카드로.
시나리오 시작
| 미터법 | 이전(규칙 기반) | 이후(하이브리드 ML) | 개선 |
|---|---|---|---|
| 사기 감지(%) | 72% | 91% | +19포인트 |
| 거짓 긍정(불량 블록) | 5.2% | 0.8% | -84% |
| 평균 결정 대기 시간 | 450ms | 85ms | -81% |
| 잘못된 차단에 대한 고객 불만 | 1,200/월 | 190/월 | -84% |
| 연간 순 사기 비용 | €420만 | €110만 | -74% |
| 사기팀 운영자 | FTE 22명 | FTE 14명 | -36% |
구현된 아키텍처
이 시스템은 14개월에 걸쳐 3단계로 구축되었습니다.
구현 로드맵(14개월)
- 1단계 - 기초(1~4개월): 과거 사기 데이터 통합 (5년, 1억 2천만 건의 트랜잭션), 기능 엔지니어링, 라벨링이 포함된 교육 데이터 세트 수사관. 트랜잭션 스트리밍을 위한 Kafka 인프라. 피처 스토어 Redis에서 실시간 기능을 확인하세요. 기준선 로지스틱 회귀 모델(AUC 0.78).
- 2단계 - ML Core(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 지원 조사자"로 재배치 운영자가 승인/차단을 결정하지 않고 REVIEW 사례를 검토하도록 유도합니다. 자동. 팀은 복잡한 사례와 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은 복잡하고 새로운 패턴에 탁월합니다. 최적의 아키텍처는 두 가지를 모두 사용합니다. 별도의 레이어.
- 중앙 집중식 특성 저장소: 행동 특징 계산 (속도, 이동 평균, 집계)을 한 번만 특성에 저장합니다. 저장소(Redis, Feast, Tecton). 이를 통해 훈련과 추론 간의 일관성이 보장됩니다. 생산 대기 시간을 줄입니다.
- 챔피언-도전자 테스트: 빅뱅 배포를 수행하지 마십시오. 사용 항상 트래픽 분할을 통한 A/B 테스트. "도전자" 모델은 10~20%를 받습니다. 트래픽이 증가하며 측정항목이 현재 '챔피언'을 초과하는 경우에만 승격됩니다.
- 포괄적인 의사결정 로깅: 모든 예측을 다음과 같이 유지하세요. 완전한 특징 벡터, SHAP 값, 최종 결정 및 결과 확인되었습니다. 이는 감사, 디버깅 및 재교육에 필수적입니다.
피해야 할 안티패턴
- 유출 날짜: 기능에 사용 가능한 정보를 포함하지 마세요. 라벨링 시에만(예: "이 계정은 차단되었습니다." 기능으로 훈련의). 모델은 실제 현상 대신 라벨을 예측하는 방법을 학습합니다.
- 알려진 사기에 대한 과대적합: 패턴에만 학습된 모델 새로운 패턴에서는 역사적 사기가 실패합니다. 항상 강력한 네거티브 샘플링 포함 미래의 기간에 대해 테스트합니다(정방향 검증).
- 피드백 루프를 무시합니다. 모델이 거래를 차단하는 경우에는 거래를 차단하지 않습니다. 그것이 정말로 사기인지는 결코 알 수 없습니다. 이러한 선택 편향은 점진적으로 왜곡됩니다. 훈련 세트. 차단된 거래에 대한 샘플 검토 시스템을 구현합니다.
- 교정 없이 점수: XGBoost 및 LightGBM은 확률을 생성하지 않습니다. 기본적으로 보정됩니다. Fraud_score 0.8은 "80% 사기 확률"을 의미하지 않습니다. Platt 스케일링 또는 등장성 회귀를 사용하여 확률 출력을 보정합니다.
- 모니터링 없이 배포: 재무 모델은 빠르게 저하됩니다. 6월의 우수한 모델이 변화로 인해 12월에는 평범할 수도 있습니다. 계절적 요인과 사기 패턴의 진화.
결론 및 다음 단계
금융분야 AI는 단순 이상징후 탐지에서 시스템으로 급속히 성숙됐다. 그래디언트 부스팅, 그래프 신경망, 실시간 스트리밍 및 점점 더 엄격해지는 규제 요구 사항을 충족할 수 있는 설명 가능성. 2025~2026년은 결정적인 시기:AI법 EU 신용 점수 시스템을 범위에 포함시킵니다. 공식적인 규제를 하는 반면, PSD3 e 도라 그들은 재정의한다 보안 및 탄력성 요구 사항.
이탈리아 은행과 핀테크 기업의 메시지는 분명합니다. 아직 현대화되지 않은 기업입니다. 사기 탐지 및 신용 점수 시스템에는 기회의 창이 있습니다. 2025~2026년에는 규제 의무를 예상하여 그렇게 하여 규정 준수를 통해 경쟁 우위를 확보할 수 있습니다.
기억해야 할 핵심 사항
- Il 금융기관의 99% 이미 사기 탐지를 위해 ML을 사용하고 있지만 구현 품질은 크게 다릅니다.
- XGBoost 및 LightGBM SMOTE를 통해 신용 점수를 지배합니다. 규정 준수에 대한 SHAP 설명 기능을 갖춘 성능 >95% AUC
- 실시간 아키텍처 카프카 + 플링크 + 레디스 100ms 미만의 트랜잭션 모니터링 표준
- Le 그래프 신경망 AML을 혁신하고 있습니다. 규칙으로 포착할 수 없는 복잡한 네트워크에서 자금 세탁 패턴을 탐지합니다.
- L'AI법 EU 신용 점수를 고위험 AI로 분류: 2026년부터 공식적인 의무이지만 지금 준비하는 것이 좋습니다
- Il 하이브리드 규칙 + ML 모델 프로덕션에서의 최선의 선택: 결정론적 규칙의 견고성과 감사 가능성을 갖춘 ML 성능
시리즈의 다음 기사에서는소매업의 AI, 우리가 직면하게 될 곳 시계열을 통한 수요 예측, 협업 필터링을 통한 추천 엔진 강화 학습을 통한 행렬 분해, 동적 가격 책정 등이 있습니다. 다양한 기술 금융과 비교하지만 공통점이 많은 아키텍처 패턴(피처 스토어, A/B 테스트) 프로덕션 모델에 대한 MLOps.
자세히 알아볼 수 있는 리소스
- 시리즈와의 연결 MLOps: 재무 모델의 수명주기 관리를 위한 "비즈니스용 MLOps: MLflow를 사용한 프로덕션의 AI 모델"
- 시리즈와의 연결 AI 엔지니어링: 가드레일과 규정 준수 기능을 갖춘 은행 고객 관리 챗봇을 위한 "회사 내 LLM: RAG Enterprise"
- 공식 문서: 내부 거버넌스에 관한 EBA 지침(2021) e 기후 관련 및 환경 위험에 대한 ECB 가이드 EU 은행 규제 상황에 대해
- 실험할 공개 데이터세트: IEEE-CIS 사기 탐지 Kaggle(590,000개 거래, 1% 사기) e 주택 신용 부실 위험 신용 점수를 위해