AI nel Finance: Fraud Detection, Credit Scoring e Risk Management
Nel 2024, le perdite globali per frodi finanziarie hanno superato i 12.5 miliardi di dollari, con un aumento del 25% rispetto all'anno precedente. Nello stesso anno, i sistemi di fraud detection basati sull'intelligenza artificiale hanno prevenuto perdite stimate per 25.5 miliardi di dollari a livello globale. Il bilancio e chiaro: l'AI nel settore finanziario non e più un vantaggio competitivo, e una necessità operativa.
Oggi il 99% delle organizzazioni finanziarie utilizza qualche forma di machine learning o AI per contrastare le frodi, e il 93% ritiene che l'AI rivoluzionera le capacità di fraud detection nei prossimi anni. Ma come funzionano concretamente questi sistemi? Come si costruisce un modello di credit scoring che supera i requisiti normativi europei? Come si implementa un sistema AML capace di rilevare schemi di riciclaggio in reti di migliaia di transazioni?
Questo articolo risponde a queste domande con codice reale, architetture concrete e un caso studio ispirato al contesto bancario italiano. Esploreremo la fraud detection in tempo reale con Kafka e Flink, il credit scoring interpretabile con XGBoost e SHAP, il rilevamento AML con Graph Neural Networks, e la compliance con l'AI Act EU e PSD3.
Cosa Imparerai in Questo Articolo
- Come funziona la fraud detection con ML: isolation forest, gradient boosting, reti neurali
- Credit scoring AI con XGBoost e LightGBM: feature engineering e performance reale (>95% accuracy)
- Explainable AI (SHAP e LIME) per decisioni finanziarie conformi alla normativa
- Architettura real-time per transaction monitoring con Kafka e Flink
- Anti-Money Laundering con Graph Neural Networks: rileva pattern che i sistemi rule-based mancano
- RegTech: AI Act EU, PSD3, DORA e le implicazioni per le banche italiane
- Case study: implementazione fraud detection per una banca retail italiana
La Serie Data Warehouse, AI e Trasformazione Digitale
| # | Articolo | Focus |
|---|---|---|
| 1 | Evoluzione del Data Warehouse | Da SQL Server a Data Lakehouse |
| 2 | Data Mesh e Architettura Decentralizzata | Domain ownership dei dati |
| 3 | ETL vs ELT Moderno | dbt, Airbyte e Fivetran |
| 4 | Orchestrazione Pipeline | Airflow, Dagster e Prefect |
| 5 | AI nella Manifattura | Predictive Maintenance e Digital Twin |
| 6 | Sei qui - AI nel Finance | Fraud Detection e Credit Scoring |
| 7 | AI nel Retail | Demand Forecasting e Recommendation |
| 8 | AI in Healthcare | Diagnostica e Drug Discovery |
| 9 | AI nella Logistica | Route Optimization e Warehouse Automation |
| 10 | LLM in Azienda | RAG Enterprise e Guardrails |
Il Panorama dell'AI nel Finance: Dati e Tendenze 2025
Il settore finanziario e stato uno dei primi ad adottare tecniche di machine learning su larga scala, e oggi rimane tra i più avanzati. Le ragioni sono strutturali: dati abbondanti e storici, incentivi economici enormi (prevenire una froda da 10.000 euro ha un ROI immediato), e un framework normativo che, pur complesso, richiede esplicitamente sistemi di monitoraggio automatizzato.
Mercato e Dati Chiave 2025
| Indicatore | Valore 2025 | Trend |
|---|---|---|
| Perdite per frodi finanziarie globali | $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 invece dell'accuracy.
# 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 per Anomaly Detection Unsupervised
Quando i label di frode non sono disponibili (scenario comune per sistemi nuovi o domini senza storico), si usa l'Isolation Forest: un algoritmo non supervisionato che identifica le anomalie isolandole più rapidamente delle osservazioni normali attraverso alberi di decisione casuali. Le transazioni anomale vengono "isolate" con meno tagli rispetto a quelle normali.
# 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: Da Logistic Regression a LightGBM
Il credit scoring tradizionale si basa su modelli statistici lineari come la regressione logistica, storicamente preferita dalle banche per la sua interpretabilita. Ma i gradient boosting machines (GBM) moderni, in particolare XGBoost e LightGBM, hanno dimostrato performance nettamente superiori, con accuracy fino al 98.13% nei benchmark recenti contro il 70-75% della logistic regression classica.
La barriera all'adozione dei GBM non era tecnica ma normativa: come si spiega a un cliente (e al regolatore) perchè il prestito e stato rifiutato, se il modello e una "black box"? La risposta e l'Explainable AI (XAI) con SHAP values.
Feature Engineering per Credit Scoring
La qualità del credit scoring dipende in modo critico dalle feature utilizzate. Le banche combinano dati interni (storico transazionale, comportamento conto) con dati esterni (Centrale Rischi Banca d'Italia, CRIF) per costruire un profilo di rischio multidimensionale.
# 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 featuresTraining LightGBM e Ottimizzazione Soglia
# credit_scoring_model.py
# Training LightGBM per credit scoring con threshold optimization
import lightgbm as lgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, precision_recall_curve
import shap
import matplotlib.pyplot as plt
import numpy as np
def train_credit_scoring_model(df: pd.DataFrame):
"""
Allena un modello LightGBM per credit scoring.
Target: 1 = default (non rimborsa), 0 = bono pagatore
"""
feature_cols = [
'saldo_medio_12m', 'saldo_min_12m', 'volatilita_saldo',
'giorni_saldo_negativo', 'reddito_medio_mensile',
'stabilita_reddito', 'num_fonti_reddito', 'spesa_media_mensile',
'ratio_risparmio', 'utilizzo_fido_medio', 'num_prestiti_attivi',
'num_rate_scadute', 'max_giorni_ritardo', 'ratio_rimborso',
'anzianita_relazione_banca', 'eta', 'anni_impiego_attuale',
'ratio_debt_income', 'tipo_impiego_score'
]
X = df[feature_cols].fillna(df[feature_cols].median())
y = df['default_flag'].astype(int)
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# Dataset LightGBM nativo (più efficiente)
dtrain = lgb.Dataset(X_train, label=y_train)
dval = lgb.Dataset(X_test, label=y_test, reference=dtrain)
params = {
'objective': 'binary',
'metric': ['binary_logloss', 'auc'],
'boosting_type': 'gbdt',
'num_leaves': 63,
'max_depth': 8,
'learning_rate': 0.03,
'n_estimators': 1000,
'feature_fraction': 0.8,
'bagging_fraction': 0.8,
'bagging_freq': 5,
'min_child_samples': 30, # min campioni per foglia (regularizzazione)
'lambda_l1': 0.1,
'lambda_l2': 0.1,
'is_unbalance': True, # gestisce classi sbilanciate
'verbose': -1,
'random_state': 42
}
callbacks = [
lgb.early_stopping(stopping_rounds=50, verbose=False),
lgb.log_evaluation(period=100)
]
model = lgb.train(
params,
dtrain,
valid_sets=[dtrain, dval],
valid_names=['train', 'val'],
callbacks=callbacks
)
# Ottimizzazione soglia di decisione
y_pred_proba = model.predict(X_test)
threshold = optimize_threshold(y_test, y_pred_proba)
y_pred = (y_pred_proba >= threshold).astype(int)
auc = roc_auc_score(y_test, y_pred_proba)
print(f"AUC-ROC: {auc:.4f}")
print(f"Soglia ottimale: {threshold:.3f}")
return model, threshold, X_test, y_test
def optimize_threshold(y_true, y_prob) -> float:
"""
Trova la soglia che massimizza l'F1-score.
In credit scoring si può calibrare per bilanciare
tra costo di un default (FN) e costo di un rifiuto errato (FP).
"""
precisions, recalls, thresholds = precision_recall_curve(y_true, y_prob)
f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-8)
optimal_idx = np.argmax(f1_scores)
return thresholds[optimal_idx] if optimal_idx < len(thresholds) else 0.5Explainable AI (XAI): SHAP e LIME per Decisioni Finanziarie
L'AI Act EU classifica i sistemi di credit scoring come High-Risk AI (Allegato III, punto 5). Questo implica obblighi stringenti: documentazione tecnica dettagliata, valutazione di conformità, supervisione umana e, soprattutto, trasparenza nelle decisioni automatizzate. Il GDPR (Art. 22) e la normativa bancaria CRD IV richiedono già oggi che le decisioni automatizzate di credito siano spiegabili al cliente.
SHAP (SHapley Additive exPlanations) e lo strumento più usato per rispettare questi requisiti. Si basa sulla teoria dei giochi cooperativi: per ogni predizione, calcola il contributo marginale di ciascuna feature al risultato finale, in modo matematicamente rigoroso e coerente.
# 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: Quando Usare Quale
| Aspetto | SHAP | LIME |
|---|---|---|
| Fondamento teorico | Shapley values (teoria giochi) | Approssimazione locale lineare |
| Consistenza | Alta (garanzie matematiche) | Media (stocastico) |
| Velocita | Più lento (ma TreeSHAP e veloce per GBM) | Più veloce |
| Tipo spiegazione | Globale + locale | Solo locale |
| Uso in produzione bancaria | Standard de facto (audit-ready) | Complementare per debug |
| Conformità AI Act | Adeguato (con documentazione) | Insufficiente da solo |
Real-Time Transaction Monitoring con Kafka e Flink
La fraud detection su transazioni di pagamento non può aspettare: una transazione fraudolenta con carta di credito deve essere bloccata prima del completamento, non dopo. Questo richiede un'architettura di streaming capace di processare milioni di eventi al secondo con latenza inferiore ai 100 millisecondi.
L'architettura standard nel 2025 combina Apache Kafka come message broker per l'ingestione degli eventi, Apache Flink per l'elaborazione stateful in tempo reale con applicazione dei modelli ML, e un feature store per arricchire gli eventi con feature comportamentali pre-calcolate.
Architettura Real-Time Fraud Detection
| Componente | Tecnologia | Ruolo | Latenza Tipica |
|---|---|---|---|
| Event Ingestion | Apache Kafka | Buffer transazioni in arrivo | <5ms |
| Feature Enrichment | Redis + Feature Store | Aggiungi feature comportamentali | <10ms |
| ML Inference | Apache Flink + ONNX | Scoring con modello GBM | <20ms |
| Decision Engine | Custom Rules + ML Score | APPROVE / REVIEW / BLOCK | <5ms |
| Alert & Case Mgmt | Elasticsearch + Kibana | Dashboard e alert investigatori | Near real-time |
| Feedback Loop | Kafka + MLflow | Retraining incrementale | Giornaliero/settimanale |
# 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 FalseAnti-Money Laundering con Graph Neural Networks
Il riciclaggio di denaro e strutturalmente diverso dalla frode su carta di credito: coinvolge reti di entità (persone, aziende, conti bancari) connesse da transazioni complesse. I sistemi rule-based tradizionali rilevano schemi noti (strutturazione, layering, integrazione), ma mancano pattern emergenti e connessioni indirette.
Le Graph Neural Networks (GNN) rappresentano l'evoluzione naturale per l'AML: modellano il sistema finanziario come un grafo dove i nodi sono entità (conti, clienti, aziende) e gli archi sono transazioni. Le GNN imparano a classificare i nodi come sospetti o normali aggregando informazioni dai loro vicini nel grafo, catturando esattamente il tipo di pattern che i money launderer cercano di nascondere nella complessità della rete.
# 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 Sistemi Rule-Based per AML: Confronto
| Metrica | Rule-Based | GNN |
|---|---|---|
| False Positive Rate | Alto (40-60%) | Basso (10-20%) |
| Nuovi schemi | Non rilevati (zero-day) | Rilevabili (generalizzazione) |
| Connessioni indirette | Massimo 1-2 hop | Multi-hop (fino a 5+ gradi) |
| Manutenzione | Alta (regole manuali) | Bassa (auto-aggiorna con retraining) |
| Explainability | Alta (regole leggibili) | Media (attention weights) |
| Costo implementazione | Basso | Alto (dati, infrastruttura) |
RegTech: AI Act, PSD3 e Compliance Automatizzata
Il settore finanziario europeo sta attraversando una trasformazione normativa senza precedenti. Tre regolamenti si sovrappongono e interagiscono con implicazioni dirette per i sistemi AI bancari:
Framework Normativo EU per AI nel Finance 2025-2027
| Normativa | Entrata in vigore | Impatto AI Finance |
|---|---|---|
| AI Act EU | Feb 2025 (graduale fino Ago 2027) | Credit scoring = High-Risk. Obblighi: registro, audit, supervisione umana, trasparenza |
| PSD3 | Adozione 2025-2026, recepimento 2027 | SCA rafforzata, open banking API, fraud liability shift, data sharing ampliato |
| DORA (Digital Operational Resilience Act) | Gen 2025 (in vigore) | Resilienza ICT obbligatoria, incident reporting 4h, test penetrazione AI sistemi |
| AMLD6 (Anti-Money Laundering Directive) | Applicazione progressiva 2025 | KYC digitale, beneficial ownership, transaction monitoring obbligatorio |
AI Act EU: Implicazioni Pratiche per le Banche
L'AI Act EU classifica i sistemi bancari AI in tre categorie principali con obblighi crescenti. Il credit scoring automatizzato e esplicitamente elencato come High-Risk AI nell'Allegato III, il che significa che le banche devono:
Checklist Conformità AI Act per Banche (High-Risk AI)
- Registro sistemi AI: Documentare ogni sistema AI usato per decisioni creditizie nel registro EU dei sistemi ad alto rischio (obbligatorio da Agosto 2026)
- Valutazione di conformità: Assessment pre-deployment con analisi di rischio, bias testing, performance su dati demografici protetti
- Supervisione umana: Processo obbligatorio di revisione umana per decisioni ad alto impatto (rifiuto prestiti sopra soglie definite)
- Trasparenza: Il cliente deve essere informato che la decisione e stata presa (o assistita) da un sistema AI
- Accuracy e robustezza: Monitoraggio continuo delle performance, test adversarial, data drift detection
- Documentazione tecnica: Schema del modello, dati di training, metriche di performance, bias analysis (Art. 11)
- Log e audit trail: Conservazione dei log di ogni decisione AI per almeno 5 anni (Art. 12)
- Diritto di spiegazione: Il cliente ha diritto a una spiegazione meaningful della decisione AI (GDPR Art. 22 + AI Act)
PSD3 e Open Banking: Nuovi Dati per Modelli Migliori
PSD3, la revisione della Payment Services Directive, introduce un framework di open banking significativamente più robusto di PSD2. Per i sistemi di credit scoring, questo apre possibilità concrete: con il consenso del cliente, una banca potra accedere ai dati di conto presso altri istituti, ottenendo un quadro finanziario molto più completo rispetto ai soli dati interni.
Un cliente che ha sempre pagato puntualmente il mutuo in un'altra banca, ma e un nuovo cliente per la banca che richiede il prestito, potra ora portare quella storia positiva con se. Questo e l'Open Finance: dati finanziari condivisi con il consenso dell'utente per migliorare l'accuratezza dei modelli di rischio.
Case Study: Fraud Detection per una Banca Retail Italiana
Presentiamo un case study ispirato a implementazioni reali nel contesto bancario italiano: una banca di medie dimensioni (circa 500.000 clienti retail) che ha migrato da un sistema rule-based legacy a un sistema ibrido ML + regole per la fraud detection su transazioni con carta di debito e credito.
Scenario di Partenza
| Metrica | Prima (Rule-Based) | Dopo (ML Ibrido) | Miglioramento |
|---|---|---|---|
| Frodi rilevate (%) | 72% | 91% | +19 punti |
| Falsi positivi (blocchi errati) | 5.2% | 0.8% | -84% |
| Latenza media decisione | 450ms | 85ms | -81% |
| Reclami clienti per blocco errato | 1.200/mese | 190/mese | -84% |
| Costo frodi netto annuale | €4.2M | €1.1M | -74% |
| Operatori fraud team | 22 FTE | 14 FTE | -36% |
Architettura Implementata
Il sistema e stato costruito in tre fasi su un arco di 14 mesi:
Roadmap di Implementazione (14 mesi)
- Fase 1 - Foundation (Mesi 1-4): Consolidamento dati storici di frodi (5 anni, 120M transazioni), feature engineering, training dataset con labeling da investigatori. Infrastruttura Kafka per streaming delle transazioni. Feature store su Redis per feature real-time. Modello baseline Logistic Regression (AUC 0.78).
- Fase 2 - ML Core (Mesi 5-10): Training XGBoost + SMOTE (AUC 0.94). Implementazione pipeline Flink per scoring real-time. Integrazione con sistema core banking. A/B test: 20% traffico sul nuovo sistema, 80% sul vecchio. Monitoring dashboard su Grafana. Tuning soglie per bilanciare FP/FN.
- Fase 3 - Production & Compliance (Mesi 11-14): Rollout completo al 100% del traffico. Implementazione SHAP per logging delle spiegazioni. Documentazione tecnica per conformità AI Act (anticipo obblighi 2026). Retraining pipeline mensile con MLflow. Integrazione con sistema di case management investigatori.
Sfide Affrontate e Soluzioni
L'implementazione ha incontrato diverse sfide tipiche del contesto bancario italiano:
Sfide e Come Sono State Risolte
- Data quality scarsa nei dati storici: Il dataset di 5 anni conteneva label di frode imprecise (non tutte le frodi erano state correttamente marchiate). Soluzione: campagna di re-labeling manuale su un campione stratificato + regole di cleaning automatico per casi evidenti (es. chargeback approvati = frode certa).
- Latenza sistema core banking: Il sistema Temenos T24 legacy non era progettato per integrazioni real-time sub-100ms. Soluzione: pattern di pre-autorizzazione asincrona con Kafka per disaccoppiare il sistema di scoring dall'approvazione finale.
- Resistenza operatori fraud team: Il team teme che l'AI sostituisca i loro lavori. Soluzione: riposizionamento come "AI-assisted investigators" con workflow che porta gli operatori a revisionare i casi REVIEW, non a decidere su APPROVE/BLOCK automatici. Il team si e concentrato su casi complessi e investigazioni AML.
- Bias demografico: Il modello iniziale mostrava un tasso di falsi positivi più alto per transazioni in certe regioni del Sud Italia (minor storico transazionale). Soluzione: fairness constraints durante il training, soglie differenziate per segmenti con storico limitato, monitoring bias mensile.
MLOps per Modelli Finanziari: Monitoring e Retraining
I modelli di fraud detection e credit scoring non sono statici: il comportamento dei fraudatori evolve continuamente, i pattern economici cambiano, nuovi prodotti finanziari introducono nuovi rischi. Il data drift e la concept drift sono fenomeni comuni nei sistemi finanziari 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)Best Practices e Anti-Pattern per AI nel Finance
Best Practices Consolidate
- Inizia con la data quality: Prima di costruire qualsiasi modello ML, investi 2-3 mesi nella comprensione e pulizia dei dati storici. Un modello su dati sporchi e peggio di un sistema rule-based. La garbage-in-garbage-out rule vale doppio nel finance, dove le decisioni errate hanno impatti legali.
- Metriche corrette per classi sbilanciate: Non usare mai l'accuracy come metrica principale per fraud detection. Usa AUC-ROC, Precision-Recall AUC, F1-score. Definisci esplicitamente il costo relativo di FP (cliente bloccato inutilmente) e FN (frode non rilevata) per calibrare la soglia ottimale.
- Modello ibrido regole + ML: Non eliminare mai completamente le regole deterministiche. Le regole catturano pattern noti con alta precision e sono auditabili. Il ML eccelle su pattern complessi e nuovi. L'architettura ottimale usa entrambi in strati separati.
- Feature store centralizzato: Calcola le feature comportamentali (velocity, medie rolling, aggregazioni) una volta sola e memorizzale in un feature store (Redis, Feast, Tecton). Questo garantisce coerenza tra training e inference e riduce la latenza in produzione.
- Champion-challenger testing: Non fare big bang deployments. Usa sempre A/B test con traffic splitting. Il modello "challenger" riceve il 10-20% del traffico e viene promosso solo se le metriche superano il "champion" attuale.
- Logging decisionale completo: Conserva ogni predizione con il vettore di feature completo, i SHAP values, la decisione finale e l'outcome verificato. Questo e fondamentale per audit, debugging e retraining.
Anti-Pattern da Evitare
- Data leakage: Non includere nelle feature informazioni disponibili solo al momento dell'etichettatura (es: "questo conto e stato bloccato" come feature del training). Il modello impara a predire l'etichetta invece del fenomeno reale.
- Overfitting a frodi note: Un modello addestrato solo su schemi di frode storici fallisce sui nuovi pattern. Includi sempre negative sampling robusto e testa su window temporali future (forward validation).
- Ignore the feedback loop: Se il modello blocca una transazione, non sapra mai se era davvero fraudolenta. Questo selection bias distorce progressivamente il training set. Implementa un sistema di revisione campionaria su transazioni bloccate.
- Score senza calibrazione: XGBoost e LightGBM non producono probabilità calibrate di default. Un fraud_score di 0.8 non significa "80% di probabilità di frode". Usa Platt scaling o isotonic regression per calibrare i probability outputs.
- Deployare senza monitoring: I modelli finanziari degradano velocemente. Un modello eccellente in giugno può essere mediocre in dicembre per i cambiamenti stagionali e l'evoluzione dei pattern di frode.
Conclusioni e Prossimi Passi
L'AI nel settore finanziario e maturata rapidamente da semplice anomaly detection a sistemi complessi che combinano gradient boosting, graph neural networks, streaming real-time e explainability per soddisfare requisiti normativi sempre più stringenti. Il 2025-2026 sarà un periodo cruciale: l'AI Act EU porta i sistemi di credit scoring nell'ambito della regolamentazione formale, mentre PSD3 e DORA ridefiniscono i requisiti di sicurezza e resilienza.
Per le banche e fintech italiane, il messaggio e chiaro: chi non ha ancora modernizzato i propri sistemi di fraud detection e credit scoring ha una finestra di opportunità nel 2025-2026 per farlo in anticipazione degli obblighi normativi, trasformando la compliance in un vantaggio competitivo.
Punti Chiave da Ricordare
- Il 99% delle istituzioni finanziarie usa già ML per fraud detection, ma la qualità delle implementazioni varia enormemente
- XGBoost e LightGBM con SMOTE dominano il credit scoring: performance >95% AUC con SHAP explainability per la compliance
- L'architettura real-time Kafka + Flink + Redis e lo standard per transaction monitoring sub-100ms
- Le Graph Neural Networks stanno rivoluzionando l'AML: rilevamento di pattern di riciclaggio in reti complesse impossibili da catturare con regole
- L'AI Act EU classifica il credit scoring come High-Risk AI: obblighi formali dal 2026, ma meglio prepararsi ora
- Il modello ibrido regole + ML e la scelta migliore in produzione: performance ML con la robustezza e auditabilita delle regole deterministiche
Il prossimo articolo della serie esplora l'AI nel Retail, dove affronteremo demand forecasting con serie temporali, recommendation engine con collaborative filtering e matrix factorization, e dynamic pricing con reinforcement learning. Tecniche differenti rispetto al finance, ma molti pattern architetturali in comune: feature store, A/B testing e MLOps per modelli in produzione.
Risorse per Approfondire
- Collegamento con la serie MLOps: "MLOps per Business: Modelli AI in Produzione con MLflow" per la gestione del ciclo di vita dei modelli finanziari
- Collegamento con la serie AI Engineering: "LLM in Azienda: RAG Enterprise" per chatbot di customer care bancario con guardrails e compliance
- Documentazione ufficiale: EBA Guidelines on Internal Governance (2021) e ECB Guide on Climate-related and Environmental Risks per il contesto normativo bancario EU
- Dataset pubblici per sperimentare: IEEE-CIS Fraud Detection su Kaggle (590k transazioni, 1% frodi) e Home Credit Default Risk per credit scoring