AI în finanțe: detectarea fraudelor, scorarea creditelor și managementul riscului
În 2024, pierderile globale din frauda financiară au depășit 12,5 miliarde de dolari, cu o creştere de 25% faţă de anul precedent. În același an, sistemele de detectare a fraudei bazat pe inteligența artificială au a prevenit pierderi estimate la 25,5 miliarde de dolari la nivel global. Concluzia este clară: AI în sectorul financiar nu mai reprezintă un avantaj competitiv, și o necesitate operațională.
Astăzi cel 99% din organizațiile financiare folosește o formă de învățare automată sau AI pentru a combate frauda și 93% cred că AI va revoluționa capacitățile de detectare a fraudei în anii următori. Dar cum funcționează de fapt aceste sisteme? Cum să construiți un model de credit un punctaj care depășește cerințele de reglementare europene? Cum se implementează un sistem AML capabil să fie detectat scheme de spălare a banilor în rețele de mii de tranzacții?
Acest articol răspunde la aceste întrebări cu cod real, arhitecturi concrete și un studiu de caz inspirat din contextul bancar italian. Vom explora detectarea fraudelor în timp real cu Kafka și Flink, cel scorul de credit interpretabil cu XGBoost și SHAP, Detectarea AML cu rețelele neuronale grafice și conformitatea cuAI Act EU e PSD3.
Ce veți învăța în acest articol
- Cum funcționează detectarea fraudelor cu ML: pădure de izolare, creșterea gradientului, rețele neuronale
- Scor de credit AI cu XGBoost și LightGBM: inginerie de caracteristici și performanță reală (precizie > 95%)
- AI explicabilă (SHAP și LIME) pentru decizii financiare conforme cu reglementările
- Arhitectură în timp real pentru monitorizarea tranzacțiilor cu Kafka și Flink
- Anti-spălarea banilor cu rețele neuronale grafice: detectează tipare pe care sistemele bazate pe reguli le scapă
- RegTech: AI Act EU, PSD3, DORA și implicațiile pentru băncile italiene
- Studiu de caz: implementarea detectării fraudei pentru o bancă de retail italiană
Seria Data Warehouse, AI și Transformare digitală
| # | Articol | Concentrează-te |
|---|---|---|
| 1 | Evoluția depozitului de date | De la SQL Server la Data Lakehouse |
| 2 | Mesh de date și arhitectură descentralizată | Proprietatea domeniului datelor |
| 3 | ETL vs ELT modern | dbt, Airbyte și Fivetran |
| 4 | Pipeline Orchestration | Flux de aer, Dagster și Prefect |
| 5 | AI în producție | Întreținere predictivă și Digital Twin |
| 6 | Sunteți aici - AI în finanțe | Detectarea fraudelor și scorarea creditelor |
| 7 | AI în retail | Prognoza cererii și recomandare |
| 8 | AI în asistența medicală | Diagnosticare și descoperire de medicamente |
| 9 | AI în logistică | Optimizarea rutelor și automatizarea depozitelor |
| 10 | LLM în afaceri | RAG Enterprise și balustrade |
Peisajul AI în finanțe: date și tendințe 2025
Sectorul financiar a fost unul dintre primii care a adoptat tehnici de învățare automată la scară largă, iar astăzi rămâne printre cele mai avansate. Motivele sunt structurale: date abundente și istorice, stimulente costuri economice enorme (prevenirea unei fraude de 10.000 de euro are un ROI imediat) și un cadru de reglementare care, deși complexă, necesită în mod explicit sisteme de monitorizare automatizate.
Piață și date cheie 2025
| Indicator | Valoare 2025 | Tendințe |
|---|---|---|
| Pierderile globale din fraude financiare | 12,5 miliarde USD (2024) | +25% an la an |
| Frauda prevenită de AI | 25,5 miliarde de dolari | Estimată 2025 |
| Instituțiile financiare care folosesc IA | 99% | Pentru detectarea fraudei |
| Fraudă care implică IA generativă | >50% | Identități deepfake, sintetice |
| Acuratețea punctajului de credit LightGBM | 98,13% | Cu explicabilitate SHAP |
| Piața RegTech AI | 21,7 miliarde USD (CAGR) | Creștere anuală așteptată |
O schimbare fundamentală a 2024-2025 și aparițiaAI generativ ca instrument ofensator: 50% din fraudele de astăzi implică utilizarea AI de către infractori, care creează deepfake video, identități sintetice și campanii de phishing personalizate la scară. Aceasta fenomenul a accelerat adoptarea unor sisteme de apărare mai sofisticate, mișcând ștacheta de la detectarea simplă a anomaliilor până la modele capabile să detecteze modele comportamentale complexe.
Cele trei domenii principale de aplicare
AI în finanțe este structurată în jurul a trei domenii distincte, dar interconectate, fiecare cu cerințe tehnice și de reglementare specifice:
Domeniu AI în Finanțe
| Domeniu | Principalele tehnici | Latența necesară | Standard relevant |
|---|---|---|---|
| Detectarea fraudelor | Pădure de izolare, GBM, Învățare profundă | <100 ms (timp real) | PSD2/PSD3, AI Act |
| Scorul de credit | XGBoost, LightGBM, SHAP | Secunde-minute | CRR/CRD IV, AI Act High-Risk |
| AML (anti-spălare de bani) | Rețele neuronale grafice, NLP | Ore (lot) + alertă în timp real | AMLD6, orientări FATF |
Detectarea fraudelor cu învățare automată
Detectarea fraudei este cel mai matur caz de utilizare ML în finanțe. Principala provocare nu este și tehnice, dar statistice: seturile de date privind fraudele sunt foarte dezechilibrat. Din 10.000 tranzacțiile, de obicei doar 1-3 sunt frauduloase (clasa pozitivă), restul sunt legitime (clasa negativă). Un model care prezice „totul legitim” ar atinge deja o acuratețe de 99,9%, dar ar fi complet inutil.
Problema dezechilibrului de clasă
Pentru a face față dezechilibrului, sunt utilizate diferite tehnici: SMOTE (Minoritatea sintetică Tehnica de supraeșantionare) pentru a genera mostre sintetice din clasa minoritară, ponderi de clasă pentru penalizează erorile din clasa rară și valorile de evaluare adecvate, cum ar fi AUC-ROC, Precizie-Recall AUC e Scor F1 în loc de precizie.
# 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_probaPădure de izolare pentru detectarea nesupravegheată a anomaliilor
Când etichetele de fraudă nu sunt disponibile (scenariu comun pentru sisteme noi sau domenii fără istoric), folosimPădure de izolare: un algoritm nesupravegheat care identifică izolarea anomaliilor mai rapid decât observațiile normale prin arbori de decizie aleatoriu. Tranzacțiile anormale sunt „izolate” cu mai puține reduceri decât cele normale.
# 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: de la regresia logistică la LightGBM
Scorul de credit tradițional se bazează pe modele statistice liniare, cum ar fi regresie logistica, preferată istoric de bănci pentru interpretabilitatea sa. Dar degradeurile mașinile moderne de amplificare (GBM), în special XGBoost e LightGBM, au demonstrat performanțe semnificativ superioare, cu o acuratețe de până la 98,13% în benchmark-uri recente faţă de 70-75% din regresia logistică clasică.
Bariera în calea adoptării GBM-urilor nu a fost tehnică, ci de reglementare: cum o explicați unui client (și altora). de reglementare) de ce a fost respins împrumutul dacă modelul este o „cutie neagră”? Răspunsul esteIA explicabilă (XAI) cu Valorile SHAP.
Inginerie de caracteristici pentru notarea creditelor
Calitatea punctajului de credit depinde în mod critic de caracteristicile utilizate. Băncile se combină date interne (istoric tranzacțional, comportamentul contului) cu date externe (Bank Risk Center al Italiei, CRIF) pentru a construi un profil de risc multidimensional.
# 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 featuresAntrenament LightGBM și optimizare prag
# 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.5AI explicabilă (XAI): SHAP și LIME pentru decizii financiare
L'AI Act EU clasifică sistemele de notare a creditelor ca AI cu risc ridicat (Anexa III punctul 5). Aceasta implică obligații stricte: documentație tehnică detaliată, evaluarea conformității, supravegherea umană și, mai ales, transparență în decizii automatizate. GDPR (Art. 22) și regulamentul bancar CRD IV impun deja asta astăzi deciziile automate de creditare sunt explicabile clientului.
SHAP (explicații despre aditivi Shapley) și cel mai folosit instrument pentru a se conforma acestora cerințe. Se bazează pe teoria jocurilor cooperative: pentru fiecare predicție, calculați contribuția marginală a fiecărei caracteristici până la rezultatul final, într-un mod riguros și coerent din punct de vedere matematic.
# 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: Când să folosiți Care
| astept | SHAP | LĂMÂIE VERDE |
|---|---|---|
| Fundamentul teoretic | Valori Shapley (teoria jocurilor) | Aproximație liniară locală |
| Consecvență | Ridicat (garantii matematice) | Medie (stochastică) |
| Viteză | Mai lent (dar TreeSHAP și rapid pentru GBM) | Mai repede |
| Tipul explicației | Global + local | Numai local |
| Utilizare în producția bancară | Standard de facto (pregătit pentru audit) | Complementar pentru depanare |
| Conformitatea AI Act | Adecvat (cu documentație) | Insuficient de la sine |
Monitorizarea tranzacțiilor în timp real cu Kafka și Flink
Detectarea fraudei la tranzacțiile de plată nu poate aștepta: o tranzacție frauduloasă prin card de credit trebuie blocat Înainte de finalizare, nu după. Acest lucru necesită o arhitectură de streaming capabilă să proceseze milioane de evenimente pe secundă cu o latență mai mică de 100 de milisecunde.
Arhitectura standard din 2025 se combină Apache Kafka ca broker de mesaje pentru ingerarea de evenimente, Apache Flink pentru procesarea cu stare în timp real cu aplicarea modelelor ML și un magazin de funcții pentru a îmbogăți evenimentele cu caracteristici comportamentale precalculate.
Arhitectură de detectare a fraudelor în timp real
| Componentă | Tehnologie | Rol | Latența tipică |
|---|---|---|---|
| Ingestie de evenimente | Apache Kafka | Salvați tranzacțiile de intrare | <5 ms |
| Îmbogățirea caracteristicilor | Redis + Magazin de funcții | Adăugați caracteristici comportamentale | <10 ms |
| Inferență ML | Apache Flink + ONNX | Punctajul cu modelul GBM | <20 ms |
| Motor de decizie | Reguli personalizate + Scor ML | APROBĂ / REVIZUĂ / BLOCĂ | <5 ms |
| Alertă și administrare de caz | Elasticsearch + Kibana | Tabloul de bord și alertele investigatorilor | Aproape în timp real |
| Bucla de feedback | Kafka + MLflow | Recalificare incrementală | Zilnic/săptămânal |
# 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-Spălarea Banilor cu Rețele Neurale Graph
Spălarea banilor este structural diferită de frauda cu cardul de credit: implică rețele de entități (persoane, companii, conturi bancare) legate prin tranzacții complex. Sistemele tradiționale bazate pe reguli detectează modele cunoscute (structurare, stratificare, integrare), dar lipsesc modelele emergente și conexiunile indirecte.
Le Rețele neuronale grafice (GNN) reprezintă evoluția naturală pentru AML: modelează sistemul financiar ca un grafic în care nodurile sunt entități (conturi, clienți, companii) iar arcurile sunt tranzacții. GNN-urile învață să clasifice nodurile ca fiind suspecte sau normale agregând informații de la vecinii lor în grafic, captând exact tipul de model pe care spalatorii de bani incearca sa le ascunda in complexitatea retelei.
# 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 sisteme bazate pe reguli pentru AML: comparație
| Metric | Bazat pe reguli | GNN |
|---|---|---|
| Rata fals pozitivă | Ridicat (40-60%) | Scăzut (10-20%) |
| Modele noi | Nedetectat (zero zi) | Detectabil (generalizare) |
| Conexiuni indirecte | Maxim 1-2 hamei | Multi-hop (până la 5+ grade) |
| Întreţinere | Ridicat (reguli manuale) | Scăzut (actualizări automate cu reinstruire) |
| Explicabilitate | Ridicat (reguli lizibile) | Medie (greutăți de atenție) |
| Costul de implementare | Bas | Ridicat (date, infrastructură) |
RegTech: AI Act, PSD3 și conformitate automată
Sectorul financiar european trece printr-o transformare de reglementare fără precedent. Trei reglementări se suprapun și interacționează cu implicații directe pentru sistemele AI bancar:
Cadrul de reglementare al UE pentru IA în finanțe 2025-2027
| Regulamente | Intrare în vigoare | Impact AI Finance |
|---|---|---|
| AI Act EU | februarie 2025 (treptat până în august 2027) | Scorul de credit = Risc ridicat. Obligații: registru, audit, supraveghere umană, transparență |
| PSD3 | Adopție 2025-2026, transpunere 2027 | SCA consolidat, open banking API, schimbare a răspunderii pentru fraudă, partajare extinsă a datelor |
| DORA (Legea privind rezistența operațională digitală) | ianuarie 2025 (în vigoare) | Reziliența TIC obligatorie, raportarea incidentelor de 4 ore, test de penetrare a sistemelor AI |
| AMLD6 (Directiva împotriva spălării banilor) | Aplicare progresivă 2025 | KYC digital, proprietate efectivă, monitorizare obligatorie a tranzacțiilor |
AI Act EU: Implicații practice pentru bănci
L'AI Act EU clasifică sistemele bancare AI în trei categorii principale cu obligații în creștere. Scorul de credit automat este enumerat în mod explicit ca AI cu risc ridicat în anexa III, ceea ce înseamnă că băncile trebuie:
Lista de verificare a conformității cu AI Act pentru bănci (AI cu risc ridicat)
- Jurnalul sistemelor AI: Documentați fiecare sistem AI utilizat pentru deciziile de creditare în registrul UE al sistemelor cu risc ridicat (obligatoriu din august 2026)
- Evaluarea conformității: Evaluare înainte de implementare cu analiză de risc, testare de părtinire, performanță pe date demografice protejate
- Supraveghere umană: Proces obligatoriu de revizuire umană pentru deciziile cu impact mare (respingerea împrumutului peste pragurile definite)
- Transparenţă: Clientul trebuie să fie informat că decizia a fost luată (sau asistat) de un sistem AI
- Acuratețe și robustețe: Monitorizare continuă a performanței, testare adversară, detectarea derivării datelor
- Documentatie tehnica: Schemă model, date de antrenament, metrici de performanță, analiză părtinire (Art. 11)
- Jurnalele și traseele de audit: Păstrarea jurnalelor fiecărei decizii AI timp de cel puțin 5 ani (Art. 12)
- Dreptul la explicație: Clientul are dreptul la o explicație semnificativă a deciziei AI (GDPR Art. 22 + AI Act)
PSD3 și Open Banking: date noi pentru modele mai bune
PSD3, revizuirea Directivei privind serviciile de plată, introduce un cadru de open banking semnificativ mai robust decât PSD2. Pentru sisteme credit scoring, aceasta deschide posibilități concrete: cu acordul clientului, o bancă va putea accesa datele contului la alte instituții, obținând o imagine de ansamblu financiară mult mai complet decât numai datele interne.
Un client care a plătit întotdeauna ipoteca la timp într-o altă bancă, dar este unul nou client pentru banca care solicită împrumutul, va putea acum să aducă acea poveste pozitivă cu tine. Acesta esteFinanțe deschise: Date financiare partajate cu consimțământ utilizator pentru a îmbunătăți acuratețea modelelor de risc.
Studiu de caz: Detectarea fraudelor pentru o bancă italiană de retail
Prezentăm un studiu de caz inspirat din implementări reale în contextul bancar italian: o bancă mijlocie (aproximativ 500.000 de clienți retail) care a migrat dintr-un sistem moștenire bazată pe reguli la un sistem hibrid ML + reguli pentru detectarea fraudelor în tranzacții prin card de debit și credit.
Scenariul de pornire
| Metric | Înainte (bazat pe reguli) | După (ML hibrid) | Îmbunătăţire |
|---|---|---|---|
| Fraudă detectată (%) | 72% | 91% | +19 puncte |
| False pozitive (blocuri proaste) | 5,2% | 0,8% | -84% |
| Latența medie a deciziei | 450 ms | 85 ms | -81% |
| Reclamațiile clienților pentru blocare incorectă | 1.200/luna | 190/luna | -84% |
| Costul net anual al fraudei | 4,2 milioane de euro | 1,1 milioane de euro | -74% |
| Operatorii echipei de fraudă | 22 FTE | 14 FTE | -36% |
Arhitectură implementată
Sistemul a fost construit în trei faze pe o perioadă de 14 luni:
Foaia de parcurs de implementare (14 luni)
- Faza 1 - Fundație (Lunile 1-4): Consolidarea datelor istorice privind fraudele (5 ani, 120 de milioane de tranzacții), inginerie de caracteristici, set de date de instruire cu etichetare de la anchetatori. Infrastructura Kafka pentru fluxul de tranzacții. Magazine de caracteristici pe Redis pentru funcții în timp real. Model de regresie logistică de referință (AUC 0,78).
- Faza 2 - ML Core (Lunile 5-10): Antrenament XGBoost + SMOTE (AUC 0,94). Implementarea pipeline Flink pentru scor în timp real. Integrare cu sistemul de bază bancar. Test A/B: 20% trafic pe noul sistem, 80% pe cel vechi. Monitorizare Tabloul de bord Grafana. Reglarea pragurilor pentru a echilibra FP/FN.
- Faza 3 - Producție și conformitate (Lunile 11-14): Lansare completă către 100% din trafic. Implementarea SHAP pentru înregistrarea explicațiilor. Documentare tehnică pentru conformitatea cu AI Act (anticiparea obligațiilor din 2026). Recalificare lunară a conductelor cu MLflow. Integrare cu sistemul de management al cazului investigatorului.
Provocări întâmpinate și soluții
Implementarea a întâmpinat câteva provocări tipice contextului bancar italian:
Provocări și cum au fost rezolvate
- Calitatea slabă a datelor în datele istorice: Setul de date de 5 ani conținut etichete de fraudă inexacte (nu toate fraudele au fost etichetate corect). Soluție: campanie de reetichetare manuală pe un eșantion stratificat + reguli curățare automată pentru cazuri evidente (de exemplu, rambursări aprobate = anumite fraude).
- Latența sistemului bancar de bază: Vechiul sistem Temenos T24 nu era conceput pentru integrări în timp real sub 100 ms. Soluție: model de preautorizare asincron cu Kafka pentru a decupla sistemul de punctare de aprobarea finală.
- Rezistenta la frauda operatorilor echipei: Echipa se teme că AI va înlocui locurile de muncă ale acestora. Soluție: Repoziționarea ca „investigatori asistați de inteligență artificială” cu flux de lucru ceea ce determină operatorii să revizuiască cazurile, nu să decidă asupra APROBARE/BLOCARE automată. Echipa sa concentrat pe cazuri complexe și pe investigațiile AML.
- Prejudecăți demografice: Modelul inițial a arătat o rată falsă pozitive mai mari pentru tranzacțiile din anumite regiuni din sudul Italiei (mai scăzute istorice tranzacționale). Soluție: constrângeri de corectitudine în timpul antrenamentului, praguri diferențiate pentru segmentele cu istoric limitat, părtinire de monitorizare lunară.
MLOps pentru modele financiare: monitorizare și recalificare
Modelele de detectare a fraudei și de scoring al creditelor nu sunt statice: comportamentul fraudătorilor evoluează continuu, modelele economice se schimbă, sunt introduse noi produse financiare noi riscuri. The deriva de date iar cel deriva conceptului sunt fenomene comune în sistemele financiare 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)Cele mai bune practici și anti-modele pentru AI în finanțe
Cele mai bune practici consolidate
- Începeți cu calitatea datelor: Înainte de a construi orice model ML, investiți 2-3 luni în înțelegerea și curățarea datelor istorice. Un model pe date murdar și mai rău decât un sistem bazat pe reguli. Regula garbage-in-garbage-out este dublă în finanțe, unde deciziile incorecte au impact juridic.
- Valori corecte pentru clasele neechilibrate: Nu folosiți niciodată precizia ca principală măsură pentru detectarea fraudelor. Utilizați AUC-ROC, Precision-Recall AUC, Scor F1. Definiți în mod explicit costul relativ al FP (client blocat inutil) și FN (fraudă nedetectată) pentru a calibra pragul optim.
- Reguli hibride + model ML: Never eliminate the rules completely determinist. Regulile captează tipare cunoscute cu mare precizie și sunt auditabile. ML excels on complex and novel patterns. The optimal architecture uses both in straturi separate.
- Magazin de caracteristici centralizat: Calculați caracteristicile comportamentale (viteză, medii rulante, agregare) o singură dată și stocați-le într-o caracteristică magazine (Redis, Feast, Tecton). Acest lucru asigură coerența între antrenament și inferență și reduce latența producției.
- Testarea campion-provocator: Nu face implementări big bang. Utilizați întotdeauna testare A/B cu împărțirea traficului. Modelul „challenger” primește 10-20% de trafic si este promovata doar daca metricile depasesc actualul „campion”.
- Înregistrare cuprinzătoare a deciziilor: Păstrați fiecare predicție cu vectorul caracteristic complet, valorile SHAP, decizia finală și rezultatul verificat. Acest lucru este esențial pentru auditare, depanare și reinstruire.
Anti-modele de evitat
- Data scurgerii: Nu includeți informațiile disponibile în funcții numai în momentul etichetării (de exemplu, „acest cont a fost blocat” ca caracteristică a instruirii). Modelul învață să prezică eticheta în loc de fenomenul real.
- Supraadaptarea fraudelor cunoscute: Un model antrenat doar pe modele Frauda istorică eșuează pe noi modele. Includeți întotdeauna eșantionare negativă robustă și testați pe ferestre de timp viitoare (validare înainte).
- Ignorați bucla de feedback: Dacă modelul blochează o tranzacție, nu o face nu va ști niciodată dacă a fost cu adevărat frauduloasă. Această tendință de selecție denaturează progresiv setul de antrenament. Implementați un exemplu de sistem de examinare a tranzacțiilor blocate.
- Scor fără calibrare: XGBoost și LightGBM nu produc probabilități calibrat implicit. Un scor_fraudă de 0,8 nu înseamnă „80% șanse de fraudă”. Utilizați scalarea Platt sau regresia izotonică pentru a calibra rezultatele probabilității.
- Implementați fără monitorizare: Modelele financiare se degradează rapid. Un model excelent în iunie poate fi mediocru în decembrie din cauza schimbărilor sezonier și evoluția tiparelor de fraudă.
Concluzii și pașii următori
AI din sectorul financiar s-a maturizat rapid de la simpla detectare a anomaliilor la sisteme complexe care combină creșterea gradientului, rețele neuronale grafice, streaming în timp real și explicabilitate pentru a îndeplini cerințele de reglementare din ce în ce mai stricte. 2025-2026 va fi o perioadă crucială: celAI Act EU aduce în domeniu sistemele de notare a creditelor de reglementare formală, în timp ce PSD3 e DORA ei redefinesc cerințele de securitate și rezistență.
Pentru băncile și fintech-urile italiene, mesajul este clar: cei care nu s-au modernizat încă sistemele sale de detectare a fraudelor și de scorare a creditelor au o fereastră de oportunitate în 2025-2026 să facă acest lucru în anticiparea obligațiilor de reglementare, transformând conformitatea într-un avantaj competitiv.
Puncte cheie de reținut
- Il 99% dintre instituțiile financiare folosește deja ML pentru detectarea fraudelor, dar calitatea implementărilor variază foarte mult
- XGBoost și LightGBM cu SMOTE domină scorul de credit: performanță >95% AUC cu explicabilitate SHAP pentru conformitate
- Arhitectura în timp real Kafka + Flink + Redis și standardul pentru monitorizarea tranzacțiilor sub 100 ms
- Le Rețele neuronale grafice revoluționează AML: detectează modele de spălare a banilor în rețele complexe care sunt imposibil de capturat cu reguli
- L'AI Act EU clasifică scorul de credit ca IA cu risc ridicat: obligații formale din 2026, dar pregătiți-vă mai bine acum
- Il reguli hibride + model ML și cea mai bună alegere în producție: performanță ML cu robustețea și auditabilitatea regulilor deterministe
Următorul articol din serie exploreazăAI în retail, unde ne vom confrunta prognoza cererii cu serii de timp, motor de recomandare cu filtrare colaborativa și factorizare matrice și stabilire a prețurilor dinamice cu învățare prin consolidare. Tehnici diferite în comparație cu finanțele, dar multe modele arhitecturale în comun: magazine de caracteristici, testare A/B și MLOps pentru modelele aflate în producție.
Resurse pentru a afla mai multe
- Legatura cu seria MLOps: „MLOps for Business: AI Models in Production with MLflow” pentru gestionarea ciclului de viață al modelelor financiare
- Legatura cu seria Inginerie AI: „LLM în companie: RAG Enterprise” pentru chatbot bancar de asistență clienți cu balustrade și conformitate
- Documentație oficială: Orientări ABE privind guvernanța internă (2021) e Ghidul BCE privind riscurile legate de climă și de mediu pentru contextul reglementărilor bancare din UE
- Seturi de date publice pentru a experimenta: Detectarea fraudelor IEEE-CIS pe Kaggle (590.000 tranzacții, 1% fraudă) e Riscul de neplată a creditului la domiciliu pentru scorul de credit