Umělá inteligence ve financích: detekce podvodů, hodnocení úvěrů a řízení rizik
V roce 2024 přesáhly celosvětové ztráty z finančních podvodů 12,5 miliardy dolarů, s nárůstem o 25 % oproti předchozímu roku. Ve stejném roce systémy odhalování podvodů založené na umělé inteligenci mají zabránil odhadovaným ztrátám ve výši 25,5 miliardy dolarů globálně. Závěr je jasný: AI ve finančním sektoru již není konkurenční výhodou, a provozní nutnost.
Dnes 99 % finančních organizací používá nějakou formu strojového učení nebo AI v boji proti podvodům a 93 % věří, že umělá inteligence způsobí revoluci ve schopnostech odhalování podvodů další roky. Jak ale tyto systémy vlastně fungují? Jak vytvořit úvěrový model bodování, které překračuje evropské regulační požadavky? Jak implementovat AML systém schopný detekce schémata praní špinavých peněz v sítích tisíců transakcí?
Tento článek odpovídá na tyto otázky skutečným kódem, konkrétními architekturami a případovou studií inspirované italským bankovním kontextem. Prozkoumáme, odhalování podvodů v reálném čase s Kafkou a Flinkem interpretovatelné kreditní skóre s XGBoost a SHAP, Detekce AML s Graph Neuron Networks a soulad sAI zákon EU e PSD3.
Co se dozvíte v tomto článku
- Jak funguje detekce podvodů s ML: les izolace, zesílení gradientu, neuronové sítě
- Kreditní bodování AI s XGBoost a LightGBM: funkce a skutečný výkon (>95% přesnost)
- Vysvětlitelná umělá inteligence (SHAP a LIME) pro finanční rozhodnutí v souladu s předpisy
- Architektura v reálném čase pro sledování transakcí s Kafkou a Flinkem
- Anti-Money Laundering with Graph Neuron Networks: Detekuje vzory, které systémy založené na pravidlech postrádají
- RegTech: AI Act EU, PSD3, DORA a důsledky pro italské banky
- Případová studie: implementace detekce podvodů pro italskou retailovou banku
Série Data Warehouse, AI a Digital Transformation
| # | Položka | Soustředit |
|---|---|---|
| 1 | Vývoj datového skladu | Od SQL Serveru po Data Lakehouse |
| 2 | Data Mesh a decentralizovaná architektura | Doménové vlastnictví dat |
| 3 | ETL vs moderní ELT | dbt, Airbyte a Fivetran |
| 4 | Pipeline Orchestrace | Airflow, Dagster a Prefect |
| 5 | AI ve výrobě | Prediktivní údržba a digitální dvojče |
| 6 | Jste zde – AI ve financích | Detekce podvodů a kreditní hodnocení |
| 7 | AI v maloobchodě | Prognóza a doporučení poptávky |
| 8 | AI ve zdravotnictví | Diagnostika a objevování léků |
| 9 | AI v logistice | Optimalizace trasy a automatizace skladu |
| 10 | LLM v podnikání | RAG Enterprise a Guardrails |
Krajina AI ve financích: Data a trendy 2025
Finanční sektor byl jedním z prvních, kdo přijal rozsáhlé techniky strojového učení, a dnes patří mezi nejvyspělejší. Důvody jsou strukturální: bohaté a historické údaje, pobídky obrovské ekonomické náklady (zabránění podvodu ve výši 10 000 EUR má okamžitou návratnost investic) a regulační rámec který, i když je složitý, výslovně vyžaduje automatizované monitorovací systémy.
Trh a klíčová data 2025
| Indikátor | Hodnota 2025 | Trendy |
|---|---|---|
| Globální ztráty z finančních podvodů | $12.5 miliardi (2024) | +25% anno su anno |
| Frodi prevenute da AI | $25.5 miliardi | Stima 2025 |
| Istituzioni finanziarie che usano AI | 99% | Per fraud detection |
| Frodi che coinvolgono AI generativa | >50% | Deepfake, identità sintetiche |
| Accuracy LightGBM credit scoring | 98.13% | Con SHAP explainability |
| Mercato RegTech AI | $21.7B (CAGR) | Crescita annua prevista |
Un cambiamento fondamentale del 2024-2025 e l'emergere dell'AI generativa come strumento offensivo: il 50% delle frodi odierne coinvolge l'uso di AI da parte dei criminali, che creano deepfake video, identità sintetiche e campagne di phishing personalizzate a scala. Questo fenomeno ha accelerato l'adozione di sistemi di difesa più sofisticati, spostando l'asticella dalla semplice anomaly detection a modelli capaci di rilevare pattern comportamentali complessi.
Le Tre Aree di Applicazione Principali
L'AI nel finance si articola attorno a tre domeni distinti ma interconnessi, ciascuno con requisiti tecnici e normativi specifici:
Domeni AI nel Finance
| Dominio | Tecniche Principali | Latenza Richiesta | Norma Rilevante |
|---|---|---|---|
| Fraud Detection | Isolation Forest, GBM, Deep Learning | <100ms (real-time) | PSD2/PSD3, AI Act |
| Credit Scoring | XGBoost, LightGBM, SHAP | Secondi-minuti | CRR/CRD IV, AI Act High-Risk |
| AML (Anti-Money Laundering) | Graph Neural Networks, NLP | Ore (batch) + real-time alert | AMLD6, FATF guidelines |
Fraud Detection con Machine Learning
Il rilevamento delle frodi e il caso d'uso di ML più maturo nel finance. La sfida principale non e tecnica ma statistica: i dataset di frodi sono altamente sbilanciati. Su 10.000 transazioni, tipicamente solo 1-3 sono fraudolente (classe positiva), il resto sono legittime (classe negativa). Un modello che predice "tutto legittimo" raggiungerebbe già il 99.9% di accuracy, ma sarebbe completamente inutile.
Il Problema dello Sbilanciamento delle Classi
Per affrontare lo sbilanciamento si usano diverse tecniche: SMOTE (Synthetic Minority Over-sampling Technique) per generare campioni sintetici della classe minoritaria, class weights per penalizzare errori sulla classe rara, e metriche di valutazione appropriate come AUC-ROC, Precision-Recall AUC e F1-score místo přesnosti.
# 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_probaIsolation Forest pro nekontrolovanou detekci anomálií
Když nejsou k dispozici štítky pro podvody (běžný scénář pro nové systémy nebo domény bez historické), používámeIzolační les: Algoritmus bez dozoru, který identifikuje izolování anomálií rychleji než normální pozorování prostřednictvím rozhodovacích stromů náhodný. Abnormální transakce jsou „izolované“ s menším počtem škrtů než normální transakce.
# 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)
}Credit Scoring AI: Od logistické regrese k LightGBM
Tradiční kreditní scoring je založen na lineárních statistických modelech jako např regrese logistiky, historicky preferované bankami pro svou interpretovatelnost. Ale ty přechody zejména moderní posilovací stroje (GBM). XGBoost e LightGBM, prokázaly výrazně lepší výkon s přesností až 98,13 % v nedávných testech oproti 70-75 % klasické logistické regrese.
Překážka pro přijetí GBM nebyla technická, ale regulační: jak to vysvětlit zákazníkovi (a dalším). regulátor), proč byla půjčka zamítnuta, pokud je model „černá skříňka“? Odpověď jeVysvětlitelná umělá inteligence (XAI) con hodnoty SHAP.
Feature Engineering pro kreditní hodnocení
Kvalita kreditního bodování kriticky závisí na použitých funkcích. Banky se kombinují interní data (transakční historie, chování účtu) s externími daty (Bank Risk Center Itálie, CRIF) k vybudování vícerozměrného rizikového profilu.
# 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 školení a optimalizace prahů
# 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.5Vysvětlitelná AI (XAI): SHAP a LIME pro finanční rozhodnutí
L'AI zákon EU klasifikuje systémy kreditního hodnocení jako Vysoce riziková AI (Příloha III, bod 5). Z toho vyplývají přísné povinnosti: podrobná technická dokumentace, posuzování shody, lidský dohled a především transparentnost v rozhodování automatizované. GDPR (článek 22) a bankovní nařízení CRD IV to již dnes vyžadují automatizovaná úvěrová rozhodnutí jsou zákazníkovi vysvětlitelná.
SHAP (vysvětlení aditiva SHapley) a nejpoužívanějším nástrojem k jejich splnění požadavky. Je založena na kooperativní teorii her: pro každou předpověď vypočítejte příspěvek na okraj každého prvku až ke konečnému výsledku, a to matematicky přesným a koherentním způsobem.
# 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 vs LIME: Kdy použít který
| čekám | SHAP | VÁPNO |
|---|---|---|
| Teoretický základ | Shapleyho hodnoty (teorie her) | Lineární lokální aproximace |
| Konzistence | Vysoká (matematické záruky) | Průměrný (stochastický) |
| Rychlost | Pomalejší (ale TreeSHAP a rychlý pro GBM) | Rychleji |
| Typ vysvětlení | Globální + místní | Pouze místní |
| Použití v bankovní výrobě | De facto standardní (připraveno k auditu) | Doplňkové pro ladění |
| Soulad se zákonem o AI | Adekvátní (s dokumentací) | Sám o sobě nedostatečný |
Sledování transakcí v reálném čase pomocí Kafka a Flink
Detekce podvodů u platebních transakcí nemůže počkat: podvodná transakce kreditní kartou musí být zablokována Před dokončení, ne po. To vyžaduje streamovací architekturu schopnou zpracovat miliony událostí za sekundu s latencí menší než 100 milisekund.
Standardní architektura v roce 2025 kombinuje Apache Kafka jako zprostředkovatel zpráv pro přijímání událostí, Apache Flink pro stavové zpracování v v reálném čase s aplikací modelů ML a úložiště funkcí pro obohacení událostí s předem vypočítanými behaviorálními rysy.
Architektura detekce podvodů v reálném čase
| Komponent | Technologie | Role | Typická latence |
|---|---|---|---|
| Požití události | Apache Kafka | Vyrovnávací paměť příchozích transakcí | <5 ms |
| Obohacení funkcí | Redis + Obchod s funkcemi | Přidejte behaviorální funkce | <10 ms |
| ML závěr | Apache Flink + ONNX | Bodování s modelem GBM | <20 ms |
| Rozhodovací motor | Vlastní pravidla + skóre ML | SCHVÁLIT / RECENZOVAT / BLOKOVAT | <5 ms |
| Alert & Case Mgmt | Elasticsearch + Kibana | Dashboard a upozornění vyšetřovatele | Téměř v reálném čase |
| Smyčka zpětné vazby | Kafka + MLflow | Postupná rekvalifikace | Denně/týdně |
# 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 FalseProti praní špinavých peněz pomocí grafových neuronových sítí
Praní špinavých peněz se strukturálně liší od podvodů s kreditními kartami: zahrnuje sítě entit (lidé, firmy, bankovní účty) propojené transakcemi komplexní. Tradiční systémy založené na pravidlech detekují známé vzory (strukturování, vrstvení, integrace), ale chybí nové vzorce a nepřímé souvislosti.
Le Graf neuronových sítí (GNN) představují přirozený vývoj pro AML: modelují finanční systém jako graf, kde uzly jsou entity (účty, zákazníci, společnosti) a oblouky jsou transakce. GNN se učí klasifikovat uzly jako podezřelé nebo normální agregovat informace od svých sousedů v grafu a zachycovat přesně ten typ vzoru že se osoby, které perou špinavé peníze, snaží ukrýt ve složitosti sítě.
# 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 modelGNN vs systémy založené na pravidlech pro AML: Srovnání
| Metrický | Na základě pravidel | GNN |
|---|---|---|
| Falešně pozitivní sazba | Vysoká (40–60 %) | Nízká (10–20 %) |
| Nové vzory | Nezjištěno (nultý den) | Zjistitelné (zobecnění) |
| Nepřímé souvislosti | Maximálně 1-2 poskoky | Multi-hop (až 5+ stupňů) |
| Údržba | Vysoká (ruční pravidla) | Nízká (automatické aktualizace s přeškolením) |
| Vysvětlitelnost | Vysoká (čitelná pravidla) | Průměr (váhy pozornosti) |
| Náklady na implementaci | Bas | Vysoká (data, infrastruktura) |
RegTech: AI Act, PSD3 a Automated Compliance
Evropský finanční sektor prochází bezprecedentní regulační transformací. Tři nařízení se překrývají a mají přímé důsledky pro systémy umělé inteligence bankovnictví:
Regulační rámec EU pro AI ve financích 2025–2027
| Předpisy | Vstup v platnost | Dopad AI Finance |
|---|---|---|
| AI zákon EU | února 2025 (postupně do srpna 2027) | Kreditní bodování = vysoké riziko. Povinnosti: evidence, audit, lidský dohled, transparentnost |
| PSD3 | Přijetí 2025–2026, transpozice 2027 | Posílené SCA, otevřené bankovnictví API, posun odpovědnosti za podvody, rozšířené sdílení dat |
| DORA (Digital Operational Resilience Act) | ledna 2025 (účinné) | Povinná ICT odolnost, 4h hlášení incidentů, penetrační test systémů AI |
| AMLD6 (Směrnice proti praní špinavých peněz) | Progresivní aplikace 2025 | Digitální KYC, skutečné vlastnictví, povinné sledování transakcí |
AI Act EU: Praktické důsledky pro banky
L'AI zákon EU klasifikuje AI bankovní systémy do tří hlavních kategorií s rostoucí povinnosti. Automatizované kreditní hodnocení je výslovně uvedeno jako Vysoce riziková AI v příloze III, což znamená, že banky musí:
Kontrolní seznam pro dodržování zákona o umělé inteligenci pro banky (umělá inteligence s vysokým rizikem)
- Protokol systému AI: Dokumentujte každý systém umělé inteligence používaný pro úvěrová rozhodnutí v registru vysoce rizikových systémů EU (povinné od srpna 2026)
- Posouzení shody: Hodnocení před nasazením s analýzou rizik, testování zkreslení, výkon na chráněných demografických datech
- Lidský dohled: Povinný proces kontroly člověkem u rozhodnutí s velkým dopadem (odmítnutí úvěru nad stanovenými prahy)
- Průhlednost: Zákazník musí být informován, že rozhodnutí bylo učiněno (nebo asistováno) systémem AI
- Přesnost a robustnost: Nepřetržité sledování výkonu, testování protivníků, detekce posunu dat
- Technická dokumentace: Modelové schéma, tréninková data, výkonnostní metriky, analýza zkreslení (článek 11)
- Protokoly a auditní záznamy: Uchovávání protokolů o každém rozhodnutí AI po dobu nejméně 5 let (článek 12)
- Právo na vysvětlení: Zákazník má právo na smysluplné vysvětlení rozhodnutí AI (GDPR čl. 22 + zákon o AI)
PSD3 a otevřené bankovnictví: Nová data pro lepší modely
PSD3, revize směrnice o platebních službách, zavádí rámec z otevřené bankovnictví výrazně robustnější než PSD2. Pro systémy credit scoring, to otevírá konkrétní možnosti: se souhlasem zákazníka banka bude mít přístup k údajům o účtu v jiných institucích a získá finanční přehled mnohem úplnější než samotná interní data.
Zákazník, který vždy platil hypotéku včas v jiné bance, ale je to nový zákazník pro banku žádající o úvěr, bude nyní moci přinést tento pozitivní příběh s vámi. Toto jeOtevřete Finance: Finanční údaje sdílené se souhlasem uživatele ke zlepšení přesnosti modelů rizik.
Případová studie: Odhalování podvodů pro italskou retailovou banku
Představujeme případovou studii inspirovanou skutečnými implementacemi v italském bankovním kontextu: středně velká banka (přibližně 500 000 retailových zákazníků), která migrovala z jednoho systému dědictví založené na pravidlech k hybridnímu systému ML + pravidla pro odhalování transakčních podvodů debetní a kreditní kartou.
Spuštění scénáře
| Metrický | Před (na základě pravidel) | Po (Hybrid ML) | Zlepšení |
|---|---|---|---|
| Zjištěn podvod (%) | 72 % | 91 % | +19 bodů |
| Falešně pozitivní (špatné bloky) | 5,2 % | 0,8 % | -84 % |
| Průměrná rozhodovací latence | 450 ms | 85 ms | -81 % |
| Stížnosti zákazníků na nesprávné blokování | 1200/měsíc | 190/měsíc | -84 % |
| Roční čisté náklady na podvod | 4,2 milionu EUR | 1,1 milionu EUR | -74 % |
| Operátoři podvodného týmu | 22 FTE | 14 FTE | -36 % |
Implementovaná architektura
Systém byl vybudován ve třech fázích po dobu 14 měsíců:
Plán implementace (14 měsíců)
- Fáze 1 – Nadace (1.–4. měsíce): Konsolidace historických dat o podvodech (5 let, 120 milionů transakcí), inženýrství funkcí, tréninková datová sada s označením od vyšetřovatelé. Kafka infrastruktura pro streamování transakcí. Featurní obchody na Redis pro funkce v reálném čase. Základní model logistické regrese (AUC 0,78).
- Fáze 2 – ML Core (5.–10. měsíce): Trénink XGBoost + SMOTE (AUC 0,94). Implementace Flink pipeline pro hodnocení v reálném čase. Integrace se základním systémem bankovnictví. A/B test: 20 % provozu na novém systému, 80 % na starém. Sledování Palubní deska Grafana. Ladění prahů pro vyvážení FP/FN.
- Fáze 3 – Výroba a dodržování předpisů (11.–14. měsíce): Úplné zavedení do 100 % provozu. Implementace SHAP pro protokolování vysvětlení. Dokumentace technika pro dodržování zákona o umělé inteligenci (předvídání povinností v roce 2026). Měsíční rekvalifikace potrubí s MLflow. Integrace se systémem řízení případů vyšetřovatelů.
Výzvy a řešení
Implementace narazila na několik problémů typických pro italský bankovní kontext:
Výzvy a jak byly vyřešeny
- Nízká kvalita dat v historických datech: 5letý datový soubor obsahoval nepřesné označení podvodu (ne všechny podvody byly správně označeny). Řešení: ruční přeznačení kampaně na stratifikovaném vzorku + pravidla automatické čištění pro zřejmé případy (např. schválená zpětná zúčtování = jistý podvod).
- Latence základního bankovního systému: Starší systém Temenos T24 nebyl navrženo pro integraci v reálném čase pod 100 ms. Řešení: Vzor předběžné autorizace asynchronní s Kafkou, aby se oddělil systém hodnocení od konečného schválení.
- Odolnost vůči operátorům týmu podvodníků: Tým se obává, že AI nahradí jejich zaměstnání. Řešení: Změna pozice na „vyšetřovatele s pomocí AI“ s pracovním postupem což vede operátory k přezkoumání případů REVIEW, nikoli k rozhodování o SCHVÁLENÍ/BLOKOVÁNÍ automatické. Tým se zaměřil na složité případy a vyšetřování AML.
- Demografické zkreslení: Původní model vykazoval falešnou míru vyšší kladné hodnoty u transakcí v určitých regionech jižní Itálie (historicky nižší transakční). Řešení: spravedlivá omezení během tréninku, diferencované prahy pro segmenty s omezenou historií, měsíční zkreslení monitorování.
MLOps pro finanční modely: Monitoring a rekvalifikace
Modely detekce podvodů a úvěrového hodnocení nejsou statické: chování podvodníků se neustále vyvíjí, ekonomické vzorce se mění, zavádějí se nové finanční produkty nová rizika. The datový drift e la koncept drift jsou to fenomény common in ML financial systems.
# 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)Osvědčené postupy a anti-vzory pro AI ve financích
Konsolidované osvědčené postupy
- Začněte s kvalitou dat: Před vytvořením jakéhokoli modelu ML investujte 2–3 měsíce do porozumění a čištění historických dat. Model na datech špinavý a horší než systém založený na pravidlech. Pravidlo garbage-in-garbage-out-out je dvojí ve financích, kde nesprávná rozhodnutí mají právní dopady.
- Správné metriky pro nevyvážené třídy: Nikdy nepoužívejte přesnost jako hlavní metriku pro odhalování podvodů. Použijte AUC-ROC, Precision-Recall AUC, Skóre F1. Explicitně definovat relativní náklady na FP (zákazník je zbytečně blokován) a FN (podvod nebyl zjištěn) pro kalibraci optimální prahové hodnoty.
- Hybridní pravidla + ML model: Nikdy neodstraňujte pravidla úplně deterministický. Pravidla zachycují známé vzory s vysokou přesností a jsou auditovatelná. ML vyniká ve složitých a neotřelých vzorech. Optimální architektura využívá obojí v oddělené vrstvy.
- Centralizovaný obchod s funkcemi: Vypočítejte rysy chování (rychlost, klouzavý průměr, agregace) pouze jednou a uložte je do prvku prodejnách (Redis, Feast, Tecton). To zajišťuje konzistenci mezi školením a dedukcí a snižuje produkční latenci.
- Testování šampiona a vyzývatele: Nedělejte nasazení velkého třesku. Použijte vždy A/B testování s rozdělením provozu. Model „vyzývatele“ obdrží 10–20 % návštěvnosti a povýší se pouze v případě, že metriky překročí aktuálního „šampiona“.
- Komplexní protokolování rozhodnutí: Udržujte každou předpověď s kompletní vektor vlastností, hodnoty SHAP, konečné rozhodnutí a výsledek ověřeno. To je nezbytné pro audit, ladění a přeškolení.
Anti-vzory, kterým je třeba se vyhnout
- Datum úniku: Do funkcí nezahrnujte dostupné informace pouze v době označení (např. „tento účet byl zablokován“ jako funkce školení). Model se učí předpovídat označení místo skutečného jevu.
- Přepastování na známé podvody: Modelka cvičená pouze na vzorech Historické podvody selhávají na nových vzorcích. Vždy zahrňte robustní negativní vzorkování a test na budoucích časových oknech (dopředné ověření).
- Ignorujte zpětnovazební smyčku: Pokud model zablokuje transakci, neblokuje ji se nikdy nedozví, zda to byl skutečně podvod. Toto zkreslení výběru postupně zkresluje tréninkovou sadu. Implementujte vzorový systém kontroly zablokovaných transakcí.
- Skóre bez kalibrace: XGBoost a LightGBM nevytvářejí pravděpodobnosti standardně zkalibrováno. Skóre podvodu 0,8 neznamená „80% pravděpodobnost podvodu“. Ke kalibraci výstupů pravděpodobnosti použijte Plattovo škálování nebo izotonickou regresi.
- Nasazení bez monitorování: Finanční modely rychle degradují. Vynikající model v červnu může být v prosinci kvůli změnám průměrný sezónní a vývoj vzorů podvodů.
Závěry a další kroky
Umělá inteligence ve finančním sektoru rychle dospěla od jednoduché detekce anomálií k systémům komplexy, které kombinují zesílení gradientu, grafové neuronové sítě, streamování v reálném čase a vysvětlitelnost pro splnění stále přísnějších regulačních požadavků. 2025-2026 bude klíčové období:AI zákon EU přináší systémy kreditního bodování do oblasti působnosti formální regulace, přitom PSD3 e DORA předefinují požadavky na bezpečnost a odolnost.
Pro italské banky a fintechy je zpráva jasná: ti, kteří se ještě nemodernizovali její systémy odhalování podvodů a kreditního hodnocení mají příležitost v letech 2025-2026, aby tak učinily v očekávání regulačních povinností, transformace souladu do konkurenční výhody.
Klíčové body k zapamatování
- Il 99 % finančních institucí již používá ML pro detekci podvodů, ale kvalita implementací se velmi liší
- XGBoost a LightGBM se SMOTE dominují kreditnímu hodnocení: výkon >95 % AUC s vysvětlitelností SHAP pro shodu
- Architektura v reálném čase Kafka + Flink + Redis a standard pro monitorování transakcí pod 100 ms
- Le Graf neuronových sítí přinášejí revoluci v AML: odhalují vzorce praní špinavých peněz ve složitých sítích, které nelze zachytit pomocí pravidel
- L'AI zákon EU klasifikuje kreditní skóring jako vysoce rizikovou AI: formální povinnosti od roku 2026, ale nyní se lépe připravte
- Il hybridní pravidla + ML model a nejlepší volba v produkci: výkon ML s robustností a auditovatelností deterministických pravidel
Další článek v sérii zkoumáAI v maloobchodě, kde budeme čelit prognózování poptávky s časovými řadami, motor doporučení s kolaborativním filtrováním a maticová faktorizace a dynamické oceňování s posílením učení. Různé techniky ve srovnání s financemi, ale mnoho architektonických vzorů společných: obchody s funkcemi, A/B testování a MLO pro modely ve výrobě.
Zdroje, kde se dozvíte více
- Spojení se seriálem MLOps: „MLOps for Business: AI Models in Production with MLflow“ pro správu životního cyklu finančních modelů
- Spojení se seriálem AI inženýrství: "LLM ve společnosti: RAG Enterprise" pro bankovního chatbota péče o zákazníky s ochrannými zábradlími a dodržováním předpisů
- Oficiální dokumentace: Pokyny EBA k internímu řízení (2021) e Příručka ECB o rizicích souvisejících s klimatem a životním prostředím pro bankovní regulační kontext EU
- Veřejné datové sady k experimentování: Detekce podvodů IEEE-CIS na Kaggle (590 tisíc transakcí, 1 % podvod) e Riziko selhání Home Credit pro kreditní bodování