Sztuczna inteligencja w finansach: wykrywanie oszustw, punktacja kredytowa i zarządzanie ryzykiem
W 2024 r. globalne straty spowodowane oszustwami finansowymi przekroczyły kwotę 12,5 miliarda dolarów, co oznacza wzrost o 25% w porównaniu do roku poprzedniego. W tym samym roku systemy wykrywania nadużyć finansowych oparte na sztucznej inteligencji zapobiegło stratom szacowanym na 25,5 miliarda dolarów globalnie. Konkluzja jest jasna: sztuczna inteligencja w sektorze finansowym nie stanowi już przewagi konkurencyjnej, i konieczność operacyjną.
Dzisiaj 99% organizacji finansowych wykorzystuje jakąś formę uczenia maszynowego lub sztucznej inteligencji w celu zwalczania oszustw, a 93% uważa, że sztuczna inteligencja zrewolucjonizuje możliwości wykrywania oszustw następne lata. Ale jak właściwie działają te systemy? Jak zbudować model kredytowy punktację przekraczającą europejskie wymogi regulacyjne? Jak wdrożyć system AML zdolny do wykrywania programy prania pieniędzy w sieciach tysięcy transakcji?
W tym artykule znajdziesz odpowiedzi na te pytania, korzystając z prawdziwego kodu, konkretnych architektur i studium przypadku inspirowany włoskim kontekstem bankowym. Będziemy eksplorować wykrywanie oszustw w czasie rzeczywistym z Kafką i Flinkiem, interpretowalna punktacja kredytowa z XGBoost i SHAP, Wykrywanie AML z Graph Neural Networks i zgodność zUstawa o sztucznej inteligencji UE e PSD3.
Czego dowiesz się w tym artykule
- Jak wykrywanie oszustw działa w przypadku uczenia maszynowego: las izolacyjny, wzmacnianie gradientu, sieci neuronowe
- Ocena zdolności kredytowej AI za pomocą XGBoost i LightGBM: inżynieria funkcji i rzeczywista wydajność (dokładność> 95%)
- Wyjaśnialna sztuczna inteligencja (SHAP i LIME) umożliwiająca podejmowanie decyzji finansowych zgodnych z przepisami
- Architektura w czasie rzeczywistym do monitorowania transakcji za pomocą Kafki i Flink
- Przeciwdziałanie praniu pieniędzy za pomocą graficznych sieci neuronowych: wykrywa wzorce, które przeoczają systemy oparte na regułach
- RegTech: AI Act EU, PSD3, DORA i konsekwencje dla włoskich banków
- Studium przypadku: wdrożenie wykrywania oszustw dla włoskiego banku detalicznego
Seria hurtowni danych, sztucznej inteligencji i transformacji cyfrowej
| # | Przedmiot | Centrum |
|---|---|---|
| 1 | Ewolucja hurtowni danych | Od SQL Server do Data Lakehouse |
| 2 | Siatka danych i zdecentralizowana architektura | Własność domeny danych |
| 3 | ETL kontra nowoczesny ELT | dbt, Airbyte i Fivetran |
| 4 | Orkiestracja rurociągów | Przepływ powietrza, Dagster i Prefekt |
| 5 | Sztuczna inteligencja w produkcji | Konserwacja predykcyjna i cyfrowy bliźniak |
| 6 | Jesteś tutaj - AI w finansach | Wykrywanie oszustw i scoring kredytowy |
| 7 | Sztuczna inteligencja w handlu detalicznym | Prognozowanie popytu i rekomendacje |
| 8 | AI w służbie zdrowia | Diagnostyka i odkrywanie leków |
| 9 | AI w logistyce | Optymalizacja tras i automatyzacja magazynu |
| 10 | LLM w biznesie | RAG Enterprise i poręcze |
Krajobraz AI w finansach: dane i trendy 2025
Sektor finansowy jako jeden z pierwszych zastosował techniki uczenia maszynowego na dużą skalę, i dziś pozostaje jednym z najbardziej zaawansowanych. Powody są strukturalne: liczne i historyczne dane, zachęty ogromne koszty ekonomiczne (zapobieganie oszustwom na kwotę 10 000 euro zapewnia natychmiastowy zwrot z inwestycji) oraz ramy regulacyjne które, choć złożone, wyraźnie wymaga zautomatyzowanych systemów monitorowania.
Dane rynkowe i kluczowe 2025
| Wskaźnik | Wartość 2025 | Trendy |
|---|---|---|
| Globalne straty związane z oszustwami finansowymi | 12,5 miliarda dolarów (2024) | +25% rok do roku |
| Sztuczna inteligencja zapobiega oszustwom | 25,5 miliarda dolarów | Szacowany rok 2025 |
| Instytucje finansowe wykorzystujące sztuczną inteligencję | 99% | Do wykrywania oszustw |
| Oszustwo związane z generatywną sztuczną inteligencją | >50% | Deepfake, syntetyczne tożsamości |
| Dokładność Punktacja kredytowa LightGBM | 98,13% | Z możliwością wyjaśnienia SHAP |
| Rynek sztucznej inteligencji RegTech | 21,7 miliarda dolarów (CAGR) | Oczekiwany roczny wzrost |
Zasadnicza zmiana lat 2024-2025 i pojawienie sięGeneratywna sztuczna inteligencja jako narzędzie obraźliwe: 50% dzisiejszych oszustw wiąże się z wykorzystaniem sztucznej inteligencji przez przestępców, co tworzą na dużą skalę fałszywe filmy wideo, tożsamości syntetyczne i niestandardowe kampanie phishingowe. To zjawisko przyspieszyło przyjęcie bardziej wyrafinowanych systemów obronnych, przesuwając poprzeczkę od prostego wykrywania anomalii po modele zdolne do wykrywania złożonych wzorców zachowań.
Trzy główne obszary zastosowań
Sztuczna inteligencja w finansach opiera się na trzech odrębnych, ale wzajemnie powiązanych obszarach, z których każda ma wymagania szczegółowe informacje techniczne i regulacyjne:
Domeni AI w finansach
| Domena | Główne techniki | Wymagane opóźnienie | Odpowiedni standard |
|---|---|---|---|
| Wykrywanie oszustw | Izolacyjny las, GBM, głębokie uczenie się | <100 ms (w czasie rzeczywistym) | PSD2/PSD3, ustawa o sztucznej inteligencji |
| Punktacja kredytowa | XGBoost, LightGBM, SHAP | Sekundy-minuty | CRR/CRD IV, ustawa o sztucznej inteligencji obarczona wysokim ryzykiem |
| AML (Przeciwdziałanie praniu pieniędzy) | Wykres sieci neuronowych, NLP | Godziny (partia) + alert w czasie rzeczywistym | AMLD6, wytyczne FATF |
Wykrywanie oszustw za pomocą uczenia maszynowego
Wykrywanie oszustw jest najbardziej dojrzałym przypadkiem użycia ML w finansach. Głównym wyzwaniem nie jest i techniczne, ale statystyczne: zbiory danych na temat oszustw są wysoce niezrównoważony. Z 10 tys transakcji, zazwyczaj tylko 1-3 jest oszukańczych (klasa pozytywna), reszta jest legalna (klasa negatywna). Model przewidujący „wszystko zgodne z prawem” osiągnąłby już dokładność na poziomie 99,9%, ale byłoby to całkowicie bezużyteczne.
Problem braku równowagi klasowej
Aby poradzić sobie z brakiem równowagi, stosuje się różne techniki: PALENIE (Mniejszość syntetyczna Technika nadmiernego próbkowania) w celu wygenerowania syntetycznych próbek klasy mniejszości, wagi klas karać błędy w rzadkich klasach i odpowiednie wskaźniki oceny, takie jak AUC-ROC, Precyzyjne przywołanie AUC e Wynik F1 zamiast dokładności.
# 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_probaLas izolacji do wykrywania anomalii bez nadzoru
Gdy etykiety oszustwa nie są dostępne (typowy scenariusz w przypadku nowych systemów lub domen bez historyczny), używamyIzolacyjny las: Nienadzorowany algorytm identyfikujący izolowanie anomalii szybciej niż normalne obserwacje za pomocą drzew decyzyjnych losowe. Nieprawidłowe transakcje są „izolowane” z mniejszą liczbą cięć niż normalne.
# anomaly_detection.py
# Isolation Forest per fraud detection unsupervised
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
def detect_anomalies_isolation_forest(
df: pd.DataFrame,
feature_cols: list,
contamination: float = 0.02 # stima del 2% di frodi nel dataset
) -> pd.DataFrame:
"""
Rileva anomalie con Isolation Forest.
contamination: percentuale attesa di anomalie (prior)
"""
X = df[feature_cols].fillna(0)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
iso_forest = IsolationForest(
n_estimators=200,
max_samples='auto',
contamination=contamination,
max_features=1.0,
bootstrap=False,
n_jobs=-1,
random_state=42
)
# -1 = anomalia, 1 = normale
df = df.copy()
df['anomaly_flag'] = iso_forest.fit_predict(X_scaled)
df['anomaly_score'] = iso_forest.score_samples(X_scaled)
# Normalizza score in [0, 1] dove 1 = alta probabilità frode
min_score = df['anomaly_score'].min()
max_score = df['anomaly_score'].max()
df['fraud_probability'] = 1 - (df['anomaly_score'] - min_score) / (max_score - min_score)
anomalie = df[df['anomaly_flag'] == -1]
print(f"Transazioni anomale rilevate: {len(anomalie)} ({len(anomalie)/len(df):.2%})")
print(f"Importo medio anomalie: €{anomalie['importo'].mean():.2f}")
print(f"Importo medio normali: €{df[df['anomaly_flag']==1]['importo'].mean():.2f}")
return df
# Uso combinato: Isolation Forest per pre-screening + XGBoost per scoring
def dual_model_fraud_pipeline(transaction: dict, iso_model, xgb_pipeline) -> dict:
"""
Pipeline a due stadi:
1. Isolation Forest: filtro rapido (latenza <1ms)
2. XGBoost: scoring preciso solo per sospetti (latenza <10ms)
"""
features = extract_features(transaction)
# Stadio 1: fast pre-screening
anomaly_score = iso_model.score_samples([features])[0]
if anomaly_score > -0.1: # threshold per "normale"
return {
'fraud_probability': 0.01,
'decision': 'APPROVE',
'model': 'isolation_forest_fast_path'
}
# Stadio 2: scoring dettagliato solo per sospetti
fraud_prob = xgb_pipeline.predict_proba([features])[0][1]
decision = 'BLOCK' if fraud_prob > 0.7 else 'REVIEW' if fraud_prob > 0.3 else 'APPROVE'
return {
'fraud_probability': float(fraud_prob),
'decision': decision,
'model': 'xgboost_detailed',
'anomaly_score': float(anomaly_score)
}AI scoringu kredytowego: od regresji logistycznej do LightGBM
Tradycyjna punktacja kredytowa opiera się na liniowych modelach statystycznych, takich jak regresja logistyka, historycznie preferowane przez banki ze względu na możliwość interpretacji. Ale gradienty w szczególności nowoczesne maszyny wzmacniające (GBM). XGBoost e LekkiGBM, wykazały znacznie lepszą wydajność, z dokładnością do 98,13% w ostatnich testach porównawczych w porównaniu do 70–75% klasycznej regresji logistycznej.
Bariera w przyjęciu GBM nie była natury technicznej, ale regulacyjnej: jak to wyjaśnić klientowi (i innym osobom). regulatora) dlaczego pożyczka została odrzucona, skoro model jest „czarną skrzynką”? Odpowiedź brzmiWyjaśnialna sztuczna inteligencja (XAI) con Wartości SHAP.
Inżynieria funkcji w zakresie scoringu kredytowego
Jakość scoringu kredytowego zależy w dużym stopniu od wykorzystywanych funkcji. Banki łączą dane wewnętrzne (historia transakcji, zachowanie rachunku) z danymi zewnętrznymi (Centrum Ryzyka Banku Włoch, CRIF) w celu zbudowania wielowymiarowego profilu ryzyka.
# 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 featuresSzkolenie LightGBM i optymalizacja progów
# 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.5Wyjaśnialna sztuczna inteligencja (XAI): SHAP i LIME w przypadku decyzji finansowych
L'Ustawa o sztucznej inteligencji UE klasyfikuje systemy scoringu kredytowego jako Sztuczna inteligencja wysokiego ryzyka (Załącznik III, punkt 5). Oznacza to rygorystyczne obowiązki: szczegółową dokumentację techniczną, ocena zgodności, nadzór ludzki, a przede wszystkim przejrzystość decyzji zautomatyzowane. Już dziś wymagają tego RODO (art. 22) i rozporządzenie bankowe CRD IV zautomatyzowane decyzje kredytowe są zrozumiałe dla klienta.
SHAP (wyjaśnienia dodatku SHapleya) i najczęściej używanym narzędziem do ich przestrzegania wymagania. Opiera się na teorii gier kooperacyjnych: dla każdej prognozy oblicz wkład marginesu każdej cechy do wyniku końcowego, w matematycznie rygorystyczny i spójny sposób.
# 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}")KSZTAŁT vs LIME: kiedy użyć którego
| Czekam | KSZTAŁT | WAPNO |
|---|---|---|
| Podstawa teoretyczna | Wartości Shapleya (teoria gier) | Liniowe przybliżenie lokalne |
| Konsystencja | Wysoka (gwarancje matematyczne) | Średnia (stochastyczna) |
| Prędkość | Wolniejsze (ale TreeSHAP i szybkie dla GBM) | Szybciej |
| Typ wyjaśnienia | Globalny + lokalny | Tylko lokalnie |
| Zastosowanie w produkcji bankowej | De facto standard (gotowy do audytu) | Uzupełniające debugowanie |
| Zgodność z ustawą o sztucznej inteligencji | Odpowiednie (z dokumentacją) | Samo w sobie niewystarczające |
Monitorowanie transakcji w czasie rzeczywistym za pomocą Kafki i Flink
Wykrywanie oszustw w transakcjach płatniczych nie może czekać: fałszywa transakcja kartą kredytową muszą zostać zablokowane Zanim zakończenia, a nie później. Wymaga to architektury przesyłania strumieniowego zdolnej do przetwarzania milionów zdarzeń na sekundę z opóźnieniem mniejszym niż 100 milisekund.
Łączy standardowa architektura w 2025 roku Apacz Kafka jako broker wiadomości do przyjmowania zdarzeń, Apache Flink do przetwarzania stanowego w czasie rzeczywistym z zastosowaniem modeli ML oraz magazynem funkcji wzbogacającym wydarzenia z wcześniej obliczonymi cechami behawioralnymi.
Architektura wykrywania oszustw w czasie rzeczywistym
| Część | Technologia | Rola | Typowe opóźnienie |
|---|---|---|---|
| Przetwarzanie zdarzeń | Apacz Kafka | Buforuj transakcje przychodzące | <5 ms |
| Wzbogacanie funkcji | Redis + sklep z funkcjami | Dodaj funkcje behawioralne | <10 ms |
| Wnioskowanie ML | Apache Flink + ONNX | Punktacja w modelu GBM | <20 ms |
| Silnik decyzyjny | Reguły niestandardowe + wynik ML | ZATWIERDŹ / PRZEGLĄD / ZABLOKUJ | <5 ms |
| Zarządzanie alertami i sprawami | Elasticsearch + Kibana | Alerty na pulpicie nawigacyjnym i dla badaczy | Prawie w czasie rzeczywistym |
| Pętla sprzężenia zwrotnego | Kafka + MLflow | Stopniowe przekwalifikowanie | Codziennie/co tydzień |
# 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 FalsePrzeciwdziałanie praniu pieniędzy za pomocą grafowych sieci neuronowych
Pranie pieniędzy różni się strukturalnie od oszustw związanych z kartami kredytowymi: obejmuje sieci podmiotów (ludzie, firmy, rachunki bankowe) powiązane transakcjami złożone. Tradycyjne systemy oparte na regułach wykrywają znane wzorce (strukturyzacja, nakładanie warstw, integracji), ale brakuje wyłaniających się wzorców i pośrednich powiązań.
Le Grafowe sieci neuronowe (GNN) reprezentują naturalną ewolucję AML: modelują system finansowy jako graf, którego węzłami są podmioty (konta, klienci, firmy) i łuki są transakcjami. Sieci GNN uczą się klasyfikować węzły jako podejrzane lub normalne agregując informacje od swoich sąsiadów na wykresie, wychwytując dokładnie typ wzorca że osoby zajmujące się praniem pieniędzy próbują ukryć się w złożoności sieci.
# 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 a systemy oparte na regułach dla AML: porównanie
| Metryczny | Oparte na regułach | GNN |
|---|---|---|
| Fałszywie dodatni współczynnik | Wysoka (40-60%) | Niski (10-20%) |
| Nowe wzory | Nie wykryto (dzień zero) | Wykrywalne (uogólnienie) |
| Połączenia pośrednie | Maksymalnie 1-2 przeskoki | Multi-hop (do 5+ stopni) |
| Konserwacja | Wysoka (reguły ręczne) | Niski (automatyczne aktualizacje z ponownym szkoleniem) |
| Wyjaśnialność | Wysoki (czytelne zasady) | Średnia (wagi uwagi) |
| Koszt wdrożenia | Bas | Wysoki (dane, infrastruktura) |
RegTech: ustawa o sztucznej inteligencji, PSD3 i automatyczna zgodność
Europejski sektor finansowy przechodzi bezprecedensową transformację regulacyjną. Trzy rozporządzenia nakładają się na siebie i oddziałują na siebie, co ma bezpośrednie skutki dla systemów sztucznej inteligencji bankowość:
Ramy regulacyjne UE dotyczące sztucznej inteligencji w finansach na lata 2025–2027
| Regulamin | Wejście w życie | Wpływ AI na finanse |
|---|---|---|
| Ustawa o sztucznej inteligencji UE | luty 2025 r. (stopniowo do sierpnia 2027 r.) | Scoring kredytowy = wysokie ryzyko. Obowiązki: rejestr, audyt, nadzór personalny, przejrzystość |
| PSD3 | Przyjęcie 2025–2026, transpozycja 2027 | Wzmocnione SCA, API otwartej bankowości, zmiana odpowiedzialności za oszustwa, rozszerzone udostępnianie danych |
| DORA (Ustawa o cyfrowej odporności operacyjnej) | styczeń 2025 r. (obowiązuje) | Obowiązkowa odporność ICT, zgłaszanie incydentów w ciągu 4 godzin, test penetracji systemów AI |
| AMLD6 (Dyrektywa w sprawie przeciwdziałania praniu pieniędzy) | Aplikacja progresywna 2025 | Cyfrowe KYC, beneficjent rzeczywisty, obowiązkowy monitoring transakcji |
Ustawa o sztucznej inteligencji UE: praktyczne implikacje dla banków
L'Ustawa o sztucznej inteligencji UE klasyfikuje systemy bankowe AI na trzy główne kategorie rosnące obowiązki. Automatyczna punktacja kredytowa jest wyraźnie wymieniona jako Sztuczna inteligencja wysokiego ryzyka w załączniku III, co oznacza, że banki muszą:
Lista kontrolna zgodności z ustawą o sztucznej inteligencji dla banków (AI wysokiego ryzyka)
- Dziennik systemów AI: Dokumentuj każdy system AI stosowany przy podejmowaniu decyzji kredytowych w unijnym rejestrze systemów wysokiego ryzyka (obowiązkowe od sierpnia 2026 r.)
- Ocena zgodności: Ocena przed wdrożeniem z analizą ryzyka, testami stronniczości, wydajnością na chronionych danych demograficznych
- Nadzór człowieka: Obowiązkowy proces przeglądu ludzkiego w przypadku decyzji o dużym wpływie (odrzucenie pożyczki powyżej określonych progów)
- Przezroczystość: Klient musi zostać poinformowany, że decyzję podjął (lub wspomagał) system sztucznej inteligencji
- Dokładność i solidność: Ciągłe monitorowanie wydajności, testowanie kontradyktoryjne, wykrywanie dryfu danych
- Dokumentacja techniczna: Schemat modelu, dane szkoleniowe, metryki wydajności, analiza błędów (art. 11)
- Dzienniki i ścieżki audytu: Przechowywanie rejestrów każdej decyzji AI przez co najmniej 5 lat (art. 12)
- Prawo do wyjaśnień: Klient ma prawo do merytorycznego wyjaśnienia decyzji AI (art. 22 RODO + ustawa o AI)
PSD3 i otwarta bankowość: nowe dane dla lepszych modeli
PSD3, zmiana dyrektywy w sprawie usług płatniczych, wprowadza ramy z otwarta bankowość znacznie solidniejsze niż PSD2. Dla systemów scoring kredytowy, otwiera to konkretne możliwości: za zgodą klienta, banku będzie mógł uzyskać dostęp do danych rachunków w innych instytucjach, uzyskując przegląd finansowy znacznie pełniejsze niż same dane wewnętrzne.
Klient, który zawsze terminowo spłacał kredyt hipoteczny w innym banku, a jest nowym klientem klient banku wnioskującego o pożyczkę, będzie teraz mógł przedstawić tę pozytywną historię z tobą. To jestOtwarte finanse: Dane finansowe udostępniane za zgodą użytkownika w celu poprawy dokładności modeli ryzyka.
Studium przypadku: Wykrywanie oszustw we włoskim banku detalicznym
Przedstawiamy studium przypadku inspirowane rzeczywistymi wdrożeniami w kontekście włoskiej bankowości: średniej wielkości bank (około 500 000 klientów detalicznych), który przeprowadził migrację z jednego systemu oparte na regułach dziedzictwo hybrydowego systemu uczenia maszynowego + zasady wykrywania oszustw transakcyjnych kartą debetową i kredytową.
Scenariusz startowy
| Metryczny | Przed (w oparciu o reguły) | Po (hybrydowy ML) | Poprawa |
|---|---|---|---|
| Wykryto oszustwo (%) | 72% | 91% | +19 punktów |
| Fałszywe alarmy (złe bloki) | 5,2% | 0,8% | -84% |
| Średnie opóźnienie decyzji | 450 ms | 85 ms | -81% |
| Reklamacje klientów dotyczące nieprawidłowego blokowania | 1200/miesiąc | 190/miesiąc | -84% |
| Roczny koszt netto oszustw | 4,2 mln euro | 1,1 mln euro | -74% |
| Operatorzy zespołów ds. oszustw | 22 etaty | 14 etatów | -36% |
Wdrożona architektura
System został zbudowany w trzech fazach w ciągu 14 miesięcy:
Plan wdrożenia (14 miesięcy)
- Faza 1 – Podstawa (miesiące 1-4): Konsolidacja historycznych danych dotyczących oszustw (5 lat, 120 mln transakcji), inżynieria funkcji, zbiór danych szkoleniowych z etykietami z śledczy. Infrastruktura Kafka do strumieniowego przesyłania transakcji. Sklepy z funkcjami na Redis, aby korzystać z funkcji czasu rzeczywistego. Bazowy model regresji logistycznej (AUC 0,78).
- Faza 2 – ML Core (miesiące 5-10): Trening XGBoost + SMOTE (AUC 0,94). Implementacja potoku Flink do punktacji w czasie rzeczywistym. Integracja z systemem podstawowym bankowość. Test A/B: 20% ruchu w nowym systemie, 80% w starym. Monitorowanie Panel Grafany. Progi strojenia w celu zrównoważenia FP/FN.
- Faza 3 – Produkcja i zgodność (miesiące 11–14): Pełne wdrożenie do 100% ruchu. Implementacja SHAP do rejestrowania wyjaśnień. Dokumentacja technika zgodności z ustawą o sztucznej inteligencji (przewidywanie obowiązków na rok 2026). Comiesięczne przekwalifikowanie rurociągu z MLflowem. Integracja z systemem zarządzania sprawami badaczy.
Wyzwania i rozwiązania
Wdrożenie napotkało kilka wyzwań typowych dla włoskiego kontekstu bankowego:
Wyzwania i sposoby ich rozwiązywania
- Słaba jakość danych historycznych: Zawierał zbiór danych z 5 lat niedokładne etykiety oszustw (nie wszystkie oszustwa zostały prawidłowo oznaczone). Rozwiązanie: ręczna kampania re-etykietowania na próbie warstwowej + zasady automatyczne czyszczenie w oczywistych przypadkach (np. zatwierdzone obciążenia zwrotne = określone oszustwo).
- Opóźnienie podstawowego systemu bankowego: Starszy system Temenos T24 nie był zaprojektowany do integracji w czasie rzeczywistym poniżej 100 ms. Rozwiązanie: Wzór preautoryzacji asynchronicznie z Kafką, aby oddzielić system punktacji od ostatecznego zatwierdzenia.
- Operatorzy zespołów odporności na oszustwa: Zespół obawia się, że sztuczna inteligencja zastąpi ich prace. Rozwiązanie: Zmiana pozycji „badaczy wspomaganych sztuczną inteligencją” dzięki przepływowi pracy co prowadzi operatorów do przeglądu przypadków, a nie do podjęcia decyzji o ZATWIERDZENIU/BLOKOWANIU automatyczne. Zespół skupił się na skomplikowanych sprawach i dochodzeniach w zakresie przeciwdziałania praniu pieniędzy.
- Stronniczość demograficzna: Początkowy model pokazywał fałszywą stopę wyższe dodatnie wartości transakcji w niektórych regionach południowych Włoch (historycznie niższe transakcyjne). Rozwiązanie: ograniczenia uczciwości podczas szkolenia, zróżnicowane progi w przypadku segmentów o ograniczonej historii, błąd monitorowania miesięcznego.
MLOps dla modeli finansowych: monitorowanie i przekwalifikowanie
Modele wykrywania oszustw i scoringu kredytowego nie są statyczne: zachowania oszustów stale ewoluuje, zmieniają się wzorce gospodarcze, wprowadzane są nowe produkty finansowe nowe zagrożenia. The dryf danych e la dryf koncepcji są fenomenem powszechne w systemach finansowych 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)Najlepsze praktyki i antywzorce dotyczące sztucznej inteligencji w finansach
Skonsolidowane najlepsze praktyki
- Zacznij od jakości danych: Przed zbudowaniem dowolnego modelu ML zainwestuj 2-3 miesiące w zrozumienie i oczyszczenie danych historycznych. Model na danych brudny i gorszy niż system oparty na zasadach. Zasada „śmieci w śmieciach i na zewnątrz” jest podwójna w finansach, gdzie błędne decyzje mają skutki prawne.
- Prawidłowe metryki dla klas niezrównoważonych: Nigdy nie używaj dokładności jako główny wskaźnik wykrywania oszustw. Użyj AUC-ROC, AUC z precyzyjnym przywołaniem, Wynik F1. Jawnie zdefiniuj względny koszt FP (klient zablokowany niepotrzebnie) i FN (nie wykryto oszustwa), aby skalibrować optymalny próg.
- Reguły hybrydowe + model ML: Nigdy nie eliminuj zasad całkowicie deterministyczny. Reguły wychwytują znane wzorce z dużą precyzją i podlegają audytowi. ML wyróżnia się złożonymi i nowatorskimi wzorami. Optymalna architektura wykorzystuje zarówno in oddzielne warstwy.
- Scentralizowany sklep z funkcjami: Oblicz cechy behawioralne (prędkość, średnie kroczące, agregacje) tylko raz i zapisz je w funkcji sklepy (Redis, Feast, Tecton). Zapewnia to spójność między szkoleniem a wnioskowaniem i zmniejsza opóźnienia produkcyjne.
- Testowanie mistrza- pretendenta: Nie wykonuj wdrożeń typu „big bang”. Użyj zawsze testy A/B z podziałem ruchu. Model „challenger” otrzymuje 10-20% ruchu i jest promowany tylko wtedy, gdy wskaźniki przekraczają aktualnego „mistrza”.
- Kompleksowe rejestrowanie decyzji: Zachowaj każdą prognozę za pomocą kompletny wektor cech, wartości SHAP, ostateczna decyzja i wynik zweryfikowane. Jest to niezbędne do audytu, debugowania i przekwalifikowania.
Anty-wzorce, których należy unikać
- Data wycieku: Nie uwzględniaj dostępnych informacji w funkcjach tylko w momencie etykietowania (np. „to konto zostało zablokowane” jako funkcja). szkolenia). Model uczy się przewidywać etykietę zamiast rzeczywistego zjawiska.
- Nadmierne dopasowanie do znanych oszustw: Modelka trenowana wyłącznie na wzorach Historyczne oszustwo zawodzi według nowych wzorców. Zawsze uwzględniaj solidną próbkę ujemną i testuj w przyszłych oknach czasowych (walidacja w przód).
- Ignoruj pętlę sprzężenia zwrotnego: Jeśli model blokuje transakcję, tak się nie dzieje nigdy nie dowiemy się, czy rzeczywiście było to oszustwo. Ta stronniczość selekcji stopniowo ulega zniekształceniu zestaw treningowy. Wdrożyć przykładowy system przeglądu zablokowanych transakcji.
- Wynik bez kalibracji: XGBoost i LightGBM nie generują prawdopodobieństw skalibrowany domyślnie. Fraud_score wynoszący 0,8 nie oznacza „80% szans na oszustwo”. Użyj skalowania Platta lub regresji izotonicznej, aby skalibrować wyniki prawdopodobieństwa.
- Wdróż bez monitorowania: Modele finansowe szybko ulegają degradacji. Doskonały model w czerwcu może być przeciętny w grudniu ze względu na zmiany sezonowość i ewolucję wzorców oszustw.
Wnioski i dalsze kroki
Sztuczna inteligencja w sektorze finansowym szybko przekształciła się z prostego wykrywania anomalii w systemy kompleksy łączące wzmacnianie gradientu, grafowe sieci neuronowe, przesyłanie strumieniowe w czasie rzeczywistym i wyjaśnialność w celu spełnienia coraz bardziej rygorystycznych wymogów regulacyjnych. Lata 2025-2026 będą okres kluczowy:Ustawa o sztucznej inteligencji UE wprowadza do zakresu systemy scoringu kredytowego regulacji formalnych, podczas gdy PSD3 e DORA definiują na nowo wymagania dotyczące bezpieczeństwa i odporności.
Dla włoskich banków i fintechów przekaz jest jasny: ci, którzy jeszcze się nie zmodernizowali jego systemy wykrywania oszustw i scoringu kredytowego otwierają przed nami ogromne możliwości w latach 2025-2026, aby to zrobić w oczekiwaniu na obowiązki regulacyjne, przekształcające zgodność w przewagę konkurencyjną.
Kluczowe punkty do zapamiętania
- Il 99% instytucji finansowych wykorzystuje już ML do wykrywania oszustw, ale jakość wdrożeń jest bardzo zróżnicowana
- XGBoost i LightGBM w przypadku SMOTE dominują w punktacji kredytowej: wydajność > 95% AUC z możliwością wyjaśnienia zgodności przez SHAP
- Architektura czasu rzeczywistego Kafka + Flink + Redis oraz standard monitorowania transakcji poniżej 100 ms
- Le Wykres sieci neuronowych rewolucjonizują AML: wykrywają wzorce prania pieniędzy w złożonych sieciach, których nie da się uchwycić za pomocą reguł
- L'Ustawa o sztucznej inteligencji UE klasyfikuje scoring kredytowy jako AI wysokiego ryzyka: formalne zobowiązania od 2026 r., ale lepiej się przygotować już teraz
- Il reguły hybrydowe + model ML i najlepszy wybór w środowisku produkcyjnym: wydajność uczenia maszynowego z solidnością i możliwością audytu reguł deterministycznych
Kolejny artykuł z tej serii poświęcony jestSztuczna inteligencja w handlu detalicznym, gdzie się zmierzymy prognozowanie popytu z szeregami czasowymi, silnik rekomendacji ze wspólnym filtrowaniem i faktoryzacja macierzy oraz dynamiczna wycena z uczeniem się przez wzmacnianie. Różne techniki w porównaniu z finansami, ale wiele wspólnych wzorców architektonicznych: sklepy z funkcjami, testy A/B i MLOps dla modeli w produkcji.
Zasoby, aby dowiedzieć się więcej
- Połączenie z serią MLOps: „MLops for Business: Modele AI w produkcji za pomocą MLflow” do zarządzania cyklem życia modeli finansowych
- Połączenie z serią Inżynieria AI: „LLM w firmie: RAG Enterprise” dla chatbota obsługi klienta bankowego z barierkami i zgodnością
- Oficjalna dokumentacja: Wytyczne EUNB dotyczące zarządzania wewnętrznego (2021 r.) e Przewodnik EBC dotyczący zagrożeń związanych z klimatem i środowiskiem w kontekście regulacyjnym UE w zakresie bankowości
- Publiczne zbiory danych, z którymi można eksperymentować: Wykrywanie oszustw IEEE-CIS na Kaggle (590 tys. transakcji, 1% oszustw) e Ryzyko niespłacenia kredytu w domu dla scoringu kredytowego