AI nel Finance: Fraud Detection, Credit Scoring e Risk

# fraud_detection_pipeline.py
# Pipeline completa di fraud detection con gestione sbilanciamento

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, StratifiedKFold
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import IsolationForest, RandomForestClassifier
from sklearn.metrics import (
    classification_report, roc_auc_score,
    precision_recall_curve, average_precision_score,
    confusion_matrix
)
from sklearn.pipeline import Pipeline
import xgboost as xgb
from imblearn.over_sampling import SMOTE
from imblearn.pipeline import Pipeline as ImbPipeline
import warnings
warnings.filterwarnings('ignore')

# ============================================================
# 1. FEATURE ENGINEERING PER TRANSAZIONI FINANZIARIE
# ============================================================

def engineer_transaction_features(df: pd.DataFrame) -> pd.DataFrame:
    """
    Costruisce feature comportamentali e di contesto
    dalle transazioni grezze.
    """
    df = df.copy()

    # Feature temporali
    df['ora_del_giorno'] = pd.to_datetime(df['timestamp']).dt.hour
    df['giorno_settimana'] = pd.to_datetime(df['timestamp']).dt.dayofweek
    df['e_weekend'] = df['giorno_settimana'].isin([5, 6]).astype(int)
    df['e_notte'] = df['ora_del_giorno'].between(0, 6).astype(int)

    # Feature di velocità (velocity) - window scorrevole
    df = df.sort_values(['account_id', 'timestamp'])
    df['importo_medio_7d'] = (
        df.groupby('account_id')['importo']
        .transform(lambda x: x.rolling('7D', min_periods=1).mean())
    )
    df['num_transazioni_1h'] = (
        df.groupby('account_id')['importo']
        .transform(lambda x: x.rolling('1H', min_periods=1).count())
    )
    df['num_transazioni_24h'] = (
        df.groupby('account_id')['importo']
        .transform(lambda x: x.rolling('24H', min_periods=1).count())
    )

    # Deviazione dall'importo tipico del cliente
    df['z_score_importo'] = (
        df.groupby('account_id')['importo']
        .transform(lambda x: (x - x.mean()) / (x.std() + 1e-8))
    )

    # Feature geografiche
    df['distanza_km'] = calcola_distanza_ultima_transazione(df)

    # Cambio di paese rispetto alla transazione precedente
    df['cambio_paese'] = (
        df.groupby('account_id')['paese']
        .transform(lambda x: x != x.shift(1))
        .astype(int)
    )

    return df


def calcola_distanza_ultima_transazione(df: pd.DataFrame) -> pd.Series:
    """Calcola distanza in km dalla transazione precedente dello stesso account."""
    from math import radians, sin, cos, sqrt, atan2

    def haversine(lat1, lon1, lat2, lon2):
        R = 6371  # raggio Terra in km
        dlat = radians(lat2 - lat1)
        dlon = radians(lon2 - lon1)
        a = sin(dlat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2
        return R * 2 * atan2(sqrt(a), sqrt(1-a))

    distanze = []
    for account_id, group in df.groupby('account_id'):
        group = group.sort_values('timestamp')
        d = [0.0]  # prima transazione: distanza 0
        for i in range(1, len(group)):
            d.append(haversine(
                group.iloc[i-1]['lat'], group.iloc[i-1]['lon'],
                group.iloc[i]['lat'], group.iloc[i]['lon']
            ))
        distanze.extend(d)
    return pd.Series(distanze, index=df.index)


# ============================================================
# 2. MODELLO XGBOOST CON SMOTE E CROSS-VALIDATION
# ============================================================

def train_fraud_detector(df: pd.DataFrame):
    """
    Allena un rilevatore di frodi con XGBoost + SMOTE.
    Restituisce il modello, lo scaler e i risultati.
    """
    feature_cols = [
        'importo', 'ora_del_giorno', 'giorno_settimana', 'e_weekend',
        'e_notte', 'importo_medio_7d', 'num_transazioni_1h',
        'num_transazioni_24h', 'z_score_importo', 'distanza_km',
        'cambio_paese', 'categoria_merchant', 'canale'
    ]

    X = df[feature_cols].fillna(0)
    y = df['is_frode'].astype(int)

    print(f"Distribuzione classi: {y.value_counts().to_dict()}")
    print(f"Rapporto frodi: {y.mean():.4%}")

    # Split stratificato per mantenere proporzione frodi
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    # Pipeline: SMOTE -> StandardScaler -> XGBoost
    pipeline = ImbPipeline([
        ('smote', SMOTE(
            sampling_strategy=0.1,  # porta le frodi al 10% del totale
            random_state=42,
            k_neighbors=5
        )),
        ('scaler', StandardScaler()),
        ('classifier', xgb.XGBClassifier(
            n_estimators=500,
            max_depth=6,
            learning_rate=0.05,
            subsample=0.8,
            colsample_bytree=0.8,
            # Peso classi: penalizza errori su frodi
            scale_pos_weight=len(y_train[y_train==0]) / len(y_train[y_train==1]),
            eval_metric='aucpr',  # Precision-Recall AUC per classi sbilanciate
            tree_method='hist',
            device='cpu',
            random_state=42
        ))
    ])

    # Cross-validation stratificata (5 fold)
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    cv_scores = []

    for fold, (train_idx, val_idx) in enumerate(cv.split(X_train, y_train)):
        X_cv_train, X_cv_val = X_train.iloc[train_idx], X_train.iloc[val_idx]
        y_cv_train, y_cv_val = y_train.iloc[train_idx], y_train.iloc[val_idx]

        pipeline.fit(X_cv_train, y_cv_train)
        y_pred_proba = pipeline.predict_proba(X_cv_val)[:, 1]
        score = roc_auc_score(y_cv_val, y_pred_proba)
        cv_scores.append(score)
        print(f"Fold {fold+1} AUC-ROC: {score:.4f}")

    print(f"\nAUC-ROC medio CV: {np.mean(cv_scores):.4f} +/- {np.std(cv_scores):.4f}")

    # Training finale su tutto il training set
    pipeline.fit(X_train, y_train)

    # Valutazione sul test set
    y_pred = pipeline.predict(X_test)
    y_pred_proba = pipeline.predict_proba(X_test)[:, 1]

    print("\n--- Report di Classificazione ---")
    print(classification_report(y_test, y_pred, target_names=['Legittima', 'Frode']))
    print(f"AUC-ROC: {roc_auc_score(y_test, y_pred_proba):.4f}")
    print(f"Average Precision (PR-AUC): {average_precision_score(y_test, y_pred_proba):.4f}")

    return pipeline, X_test, y_test, y_pred_proba

# anomaly_detection.py
# Isolation Forest per fraud detection unsupervised

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np

def detect_anomalies_isolation_forest(
    df: pd.DataFrame,
    feature_cols: list,
    contamination: float = 0.02  # stima del 2% di frodi nel dataset
) -> pd.DataFrame:
    """
    Rileva anomalie con Isolation Forest.
    contamination: percentuale attesa di anomalie (prior)
    """
    X = df[feature_cols].fillna(0)

    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    iso_forest = IsolationForest(
        n_estimators=200,
        max_samples='auto',
        contamination=contamination,
        max_features=1.0,
        bootstrap=False,
        n_jobs=-1,
        random_state=42
    )

    # -1 = anomalia, 1 = normale
    df = df.copy()
    df['anomaly_flag'] = iso_forest.fit_predict(X_scaled)
    df['anomaly_score'] = iso_forest.score_samples(X_scaled)

    # Normalizza score in [0, 1] dove 1 = alta probabilità frode
    min_score = df['anomaly_score'].min()
    max_score = df['anomaly_score'].max()
    df['fraud_probability'] = 1 - (df['anomaly_score'] - min_score) / (max_score - min_score)

    anomalie = df[df['anomaly_flag'] == -1]
    print(f"Transazioni anomale rilevate: {len(anomalie)} ({len(anomalie)/len(df):.2%})")
    print(f"Importo medio anomalie: €{anomalie['importo'].mean():.2f}")
    print(f"Importo medio normali: €{df[df['anomaly_flag']==1]['importo'].mean():.2f}")

    return df


# Uso combinato: Isolation Forest per pre-screening + XGBoost per scoring
def dual_model_fraud_pipeline(transaction: dict, iso_model, xgb_pipeline) -> dict:
    """
    Pipeline a due stadi:
    1. Isolation Forest: filtro rapido (latenza <1ms)
    2. XGBoost: scoring preciso solo per sospetti (latenza <10ms)
    """
    features = extract_features(transaction)

    # Stadio 1: fast pre-screening
    anomaly_score = iso_model.score_samples([features])[0]

    if anomaly_score > -0.1:  # threshold per "normale"
        return {
            'fraud_probability': 0.01,
            'decision': 'APPROVE',
            'model': 'isolation_forest_fast_path'
        }

    # Stadio 2: scoring dettagliato solo per sospetti
    fraud_prob = xgb_pipeline.predict_proba([features])[0][1]

    decision = 'BLOCK' if fraud_prob > 0.7 else 'REVIEW' if fraud_prob > 0.3 else 'APPROVE'

    return {
        'fraud_probability': float(fraud_prob),
        'decision': decision,
        'model': 'xgboost_detailed',
        'anomaly_score': float(anomaly_score)
    }

# 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 features

# credit_scoring_model.py
# Training LightGBM per credit scoring con threshold optimization

import lightgbm as lgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, precision_recall_curve
import shap
import matplotlib.pyplot as plt
import numpy as np

def train_credit_scoring_model(df: pd.DataFrame):
    """
    Allena un modello LightGBM per credit scoring.
    Target: 1 = default (non rimborsa), 0 = bono pagatore
    """
    feature_cols = [
        'saldo_medio_12m', 'saldo_min_12m', 'volatilita_saldo',
        'giorni_saldo_negativo', 'reddito_medio_mensile',
        'stabilita_reddito', 'num_fonti_reddito', 'spesa_media_mensile',
        'ratio_risparmio', 'utilizzo_fido_medio', 'num_prestiti_attivi',
        'num_rate_scadute', 'max_giorni_ritardo', 'ratio_rimborso',
        'anzianita_relazione_banca', 'eta', 'anni_impiego_attuale',
        'ratio_debt_income', 'tipo_impiego_score'
    ]

    X = df[feature_cols].fillna(df[feature_cols].median())
    y = df['default_flag'].astype(int)

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    # Dataset LightGBM nativo (più efficiente)
    dtrain = lgb.Dataset(X_train, label=y_train)
    dval = lgb.Dataset(X_test, label=y_test, reference=dtrain)

    params = {
        'objective': 'binary',
        'metric': ['binary_logloss', 'auc'],
        'boosting_type': 'gbdt',
        'num_leaves': 63,
        'max_depth': 8,
        'learning_rate': 0.03,
        'n_estimators': 1000,
        'feature_fraction': 0.8,
        'bagging_fraction': 0.8,
        'bagging_freq': 5,
        'min_child_samples': 30,  # min campioni per foglia (regularizzazione)
        'lambda_l1': 0.1,
        'lambda_l2': 0.1,
        'is_unbalance': True,  # gestisce classi sbilanciate
        'verbose': -1,
        'random_state': 42
    }

    callbacks = [
        lgb.early_stopping(stopping_rounds=50, verbose=False),
        lgb.log_evaluation(period=100)
    ]

    model = lgb.train(
        params,
        dtrain,
        valid_sets=[dtrain, dval],
        valid_names=['train', 'val'],
        callbacks=callbacks
    )

    # Ottimizzazione soglia di decisione
    y_pred_proba = model.predict(X_test)
    threshold = optimize_threshold(y_test, y_pred_proba)

    y_pred = (y_pred_proba >= threshold).astype(int)
    auc = roc_auc_score(y_test, y_pred_proba)
    print(f"AUC-ROC: {auc:.4f}")
    print(f"Soglia ottimale: {threshold:.3f}")

    return model, threshold, X_test, y_test


def optimize_threshold(y_true, y_prob) -> float:
    """
    Trova la soglia che massimizza l'F1-score.
    In credit scoring si può calibrare per bilanciare
    tra costo di un default (FN) e costo di un rifiuto errato (FP).
    """
    precisions, recalls, thresholds = precision_recall_curve(y_true, y_prob)
    f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-8)
    optimal_idx = np.argmax(f1_scores)
    return thresholds[optimal_idx] if optimal_idx < len(thresholds) else 0.5

# 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}")

# real_time_fraud_flink.py
# Pseudo-codice architetturale per Flink fraud detection
# (PyFlink API - Apache Flink 1.18+)

from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors.kafka import (
    FlinkKafkaConsumer, FlinkKafkaProducer
)
import json
import redis
import joblib
import numpy as np

# Modello ML pre-caricato (ONNX o joblib)
MODEL = joblib.load('/models/fraud_detector_v2.pkl')
REDIS_CLIENT = redis.Redis(host='redis', port=6379, db=0)


def enrich_with_behavioral_features(transaction: dict) -> dict:
    """
    Arricchisce la transazione con feature comportamentali
    dal feature store (Redis).
    """
    account_id = transaction['account_id']

    # Feature pre-calcolate aggiornate ogni 5 minuti
    cached = REDIS_CLIENT.hgetall(f"features:{account_id}")

    if cached:
        transaction['importo_medio_7d'] = float(cached.get(b'importo_medio_7d', 0))
        transaction['num_txn_24h'] = int(cached.get(b'num_txn_24h', 0))
        transaction['importo_max_30d'] = float(cached.get(b'importo_max_30d', 0))
        transaction['paesi_visitati_30d'] = int(cached.get(b'paesi_visitati_30d', 1))
    else:
        # Default per account senza storico (nuovi clienti)
        transaction['importo_medio_7d'] = 0.0
        transaction['num_txn_24h'] = 0
        transaction['importo_max_30d'] = 0.0
        transaction['paesi_visitati_30d'] = 1

    return transaction


def score_transaction(transaction: dict) -> dict:
    """
    Applica il modello ML per calcolare il fraud score.
    Questa funzione viene eseguita in Flink per ogni evento.
    """
    features = [
        transaction.get('importo', 0),
        transaction.get('ora_del_giorno', 12),
        transaction.get('e_weekend', 0),
        transaction.get('e_notte', 0),
        transaction.get('importo_medio_7d', 0),
        transaction.get('num_txn_24h', 0),
        transaction.get('z_score_importo', 0),
        transaction.get('distanza_km', 0),
        transaction.get('cambio_paese', 0),
        transaction.get('canale_encoded', 0),
    ]

    fraud_prob = MODEL.predict_proba([features])[0][1]

    # Regole ibride: combina ML score con regole business hard-coded
    # Le regole business catturano pattern noti sempre validi
    hard_block = apply_hard_rules(transaction)

    if hard_block:
        final_decision = 'BLOCK'
        final_score = 1.0
    elif fraud_prob >= 0.70:
        final_decision = 'BLOCK'
        final_score = fraud_prob
    elif fraud_prob >= 0.30:
        final_decision = 'REVIEW'
        final_score = fraud_prob
    else:
        final_decision = 'APPROVE'
        final_score = fraud_prob

    return {
        **transaction,
        'fraud_score': float(final_score),
        'decision': final_decision,
        'scored_at': datetime.utcnow().isoformat()
    }


def apply_hard_rules(txn: dict) -> bool:
    """
    Regole deterministiche ad alta precision.
    Eseguono PRIMA del modello ML per blocchi immediati.
    """
    # Regola 1: importo anomalo (10x superiore alla media del cliente)
    if txn.get('importo', 0) > txn.get('importo_medio_7d', 0) * 10:
        return True

    # Regola 2: transazione da paese blacklist
    BLACKLIST_PAESI = {'COUNTRY_A', 'COUNTRY_B'}  # configurabile
    if txn.get('paese', '') in BLACKLIST_PAESI:
        return True

    # Regola 3: velocity check - più di 5 transazioni in 1 ora
    if txn.get('num_txn_1h', 0) > 5:
        return True

    # Regola 4: impossibile travel (2 paesi diversi in meno di 2 ore)
    if txn.get('distanza_km', 0) > 1000 and txn.get('ore_dalla_precedente', 999) < 2:
        return True

    return False

# 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 model

# 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)