Detección de fraude: anomalías vs. supervisado

"""#12 Detección de fraude: anomalías (Isolation Forest y autoencoder) vs. modelo supervisado.

Datos: Credit Card Fraud Detection (ULB, 284,807 transacciones de 2 días, 492 fraudes = 0.17 %), espejo
público en OpenML (id 1597). Variables V1–V28 anonimizadas por PCA, más monto y tiempo.
Pregunta: ¿cuánto fraude detectamos con un presupuesto fijo de alertas que un analista puede revisar?
"""
import numpy as np
import pandas as pd
from lightgbm import LGBMClassifier
from sklearn.ensemble import IsolationForest
from sklearn.metrics import average_precision_score, precision_recall_curve, roc_auc_score
from sklearn.neural_network import MLPRegressor
from sklearn.preprocessing import StandardScaler

from common import SEED, bars, download, export, line, table
from validacion import bootstrap_ci, explicacion_local, model_card

URL = "https://api.openml.org/data/v1/download/1673544"
COLS = ["Time"] + [f"V{i}" for i in range(1, 29)] + ["Amount", "Class"]
COSTO_ALERTA = 5.0          # USD por alerta revisada por un analista
ALERTAS_POR_MIL = 2.0       # capacidad del equipo: 2 alertas por cada 1,000 transacciones


def load() -> pd.DataFrame:
    ruta = download(URL, "creditcard.arff")
    cache = ruta.with_suffix(".parquet")
    if cache.exists():
        return pd.read_parquet(cache)
    with open(ruta, encoding="utf-8") as f:
        salto = next(i for i, l in enumerate(f) if l.strip().lower() == "@data") + 1
    df = pd.read_csv(ruta, skiprows=salto, header=None, names=COLS, quotechar="'")
    df["Class"] = df["Class"].astype(int)
    df = df.sort_values("Time").reset_index(drop=True)
    df.to_parquet(cache)
    return df


def recall_presupuesto(y, score, monto, por_mil=ALERTAS_POR_MIL):
    """% de fraudes y de monto defraudado capturados revisando solo las top-k alertas."""
    k = max(1, int(len(y) * por_mil / 1000))
    top = np.argsort(-score)[:k]
    return y[top].sum() / y.sum(), (monto[top] * y[top]).sum() / (monto * y).sum(), y[top].mean()


def run():
    df = load()
    feats = [c for c in COLS if c not in ("Class", "Time")]
    X, y, monto = df[feats], df["Class"].values, df["Amount"].values
    # Validación temporal: entrenamiento 0–60 %, validación 60–80 % (umbral), prueba 80–100 %
    n = len(df)
    a, b = int(n * 0.6), int(n * 0.8)
    esc = StandardScaler().fit(X.iloc[:a])
    Z = esc.transform(X)

    # 1) Isolation Forest: no usa etiquetas, aísla puntos raros con cortes aleatorios
    iso = IsolationForest(n_estimators=300, max_samples=4096, random_state=SEED, n_jobs=-1).fit(Z[:a])
    s_iso = -iso.score_samples(Z)
    # 2) Autoencoder (red neuronal que aprende a reconstruir transacciones NORMALES): el fraude se reconstruye mal
    normales = np.where(y[:a] == 0)[0]
    rng = np.random.default_rng(SEED)
    idx = rng.choice(normales, min(60_000, len(normales)), replace=False)
    ae = MLPRegressor(hidden_layer_sizes=(20, 8, 20), activation="relu", max_iter=60, early_stopping=True,
                      random_state=SEED).fit(Z[idx], Z[idx])
    recon = ae.predict(Z)
    err_var = (Z - recon) ** 2
    s_ae = err_var.mean(axis=1)
    # 3) Supervisado: con etiquetas disponibles, el estándar de la industria
    gbm = LGBMClassifier(n_estimators=400, learning_rate=0.03, num_leaves=15, min_child_samples=50,
                         subsample=0.8, subsample_freq=1, scale_pos_weight=5, random_state=SEED, verbose=-1)
    gbm.fit(X.iloc[:a], y[:a])
    s_gbm = gbm.predict_proba(X)[:, 1]
    # 4) Híbrido: el error del autoencoder y el score de aislamiento como variables del supervisado
    Xh = X.assign(err_ae=s_ae, score_iso=s_iso)
    hib = LGBMClassifier(n_estimators=400, learning_rate=0.03, num_leaves=15, min_child_samples=50,
                         subsample=0.8, subsample_freq=1, scale_pos_weight=5, random_state=SEED, verbose=-1)
    hib.fit(Xh.iloc[:a], y[:a])
    s_hib = hib.predict_proba(Xh)[:, 1]

    te = slice(b, n)
    y_te, m_te = y[te], monto[te]
    modelos = {"Isolation Forest (sin etiquetas)": s_iso, "Autoencoder (sin etiquetas)": s_ae,
               "LightGBM supervisado": s_gbm, "Híbrido: LightGBM + anomalías": s_hib}
    filas, curvas = [], {}
    for nombre, s in modelos.items():
        ap, lo, hi = bootstrap_ci(y_te, s[te], metrica=average_precision_score, n=300)
        rec, rec_m, prec = recall_presupuesto(y_te, s[te], m_te)
        filas.append({"Modelo": nombre, "PR-AUC": round(ap, 4), "IC 95 %": f"{lo:.3f}–{hi:.3f}",
                      "ROC-AUC": round(roc_auc_score(y_te, s[te]), 4),
                      f"Fraudes capturados @{ALERTAS_POR_MIL:g}‰": f"{rec:.0%}",
                      "Monto capturado": f"{rec_m:.0%}", "Precisión de alertas": f"{prec:.0%}"})
        p, r, _ = precision_recall_curve(y_te, s[te])
        # Precisión interpolada (la mejor alcanzable con al menos ese recall) en una malla fija de recall: muestrear
        # por umbral deja casi todos los puntos en la cola de baja precisión y aplana el tramo que importa.
        malla = np.linspace(0, 1, 51)
        curvas[nombre] = [[round(float(x) * 100, 1), round(float(p[r >= x].max()) * 100, 2)] for x in malla]
    res = pd.DataFrame(filas)
    mejor = res.loc[res["PR-AUC"].idxmax(), "Modelo"]
    s_best = modelos[mejor]

    # Presupuesto de alertas: captura de monto según alertas por cada 1,000 transacciones
    presupuestos = [0.5, 1, 1.5, 2, 3, 5, 8, 12, 20]
    curva_pres = {nombre: [[p_, round(recall_presupuesto(y_te, s[te], m_te, p_)[1] * 100, 2)] for p_ in presupuestos]
                  for nombre, s in modelos.items()}
    # Umbral económico: elegido en validación, aplicado en prueba (ahorro = fraude evitado − costo de revisar)
    va = slice(a, b)
    def ahorro(s, yy, mm, u):
        alerta = s >= u
        return (mm[alerta] * yy[alerta]).sum() - alerta.sum() * COSTO_ALERTA
    cand = np.quantile(s_best[va], np.linspace(0.95, 0.9999, 200))
    u = cand[int(np.argmax([ahorro(s_best[va], y[va], monto[va], c) for c in cand]))]
    ahorro_te = ahorro(s_best[te], y_te, m_te, u)
    alertas_te = int((s_best[te] >= u).sum())
    fraude_total = float((m_te * y_te).sum())

    # Distribución del error de reconstrucción (log) normales vs. fraude
    bordes = np.linspace(np.log10(s_ae[te].min() + 1e-6), np.log10(np.percentile(s_ae[te], 99.99)), 40)
    def hist(mask):
        h = np.histogram(np.log10(s_ae[te][mask] + 1e-6), bordes)[0] / mask.sum() * 100
        return [[round(float(c), 2), round(float(v), 3)] for c, v in zip((bordes[:-1] + bordes[1:]) / 2, h)]
    # Explicabilidad: qué variables reconstruye peor el autoencoder en la alerta más alta; SHAP del supervisado
    i_top = b + int(np.argmax(s_ae[te]))
    contrib_ae = pd.Series(err_var[i_top], index=feats).sort_values(ascending=False)[:8]
    contrib = gbm.predict(X.iloc[te], pred_contrib=True)[:, :-1]
    shap = pd.Series(np.abs(contrib).mean(0), index=feats).sort_values(ascending=False)[:10]
    j = int(np.argmax(s_gbm[te]))
    etiquetas, valores = explicacion_local(contrib[j], feats, X.iloc[b + j].round(2))

    r = res.set_index("Modelo")
    result = {
        "title": "Detección de fraude: anomalías vs. supervisado",
        "summary": [
            f"{n:,} transacciones con {y.sum()} fraudes ({y.mean():.2%}). Con validación temporal (entrenar con las primeras horas, "
            "evaluar en las últimas), la métrica correcta es PR-AUC: con 0.17 % de positivos, el ROC-AUC engaña.",
            f"Sin etiquetas, el autoencoder logra PR-AUC {r.loc['Autoencoder (sin etiquetas)', 'PR-AUC']:.3f} y Isolation Forest "
            f"{r.loc['Isolation Forest (sin etiquetas)', 'PR-AUC']:.3f}: sirven para arrancar cuando aún no hay fraudes "
            "etiquetados o para detectar patrones nuevos.",
            f"Con etiquetas gana {mejor} (PR-AUC {r.loc[mejor, 'PR-AUC']:.3f}). Revisando solo {ALERTAS_POR_MIL:g} de cada 1,000 "
            f"transacciones captura {r.loc[mejor, f'Fraudes capturados @{ALERTAS_POR_MIL:g}‰']} de los fraudes y "
            f"{r.loc[mejor, 'Monto capturado']} del monto. Frente al supervisado puro "
            f"(PR-AUC {r.loc['LightGBM supervisado', 'PR-AUC']:.3f}) los IC se traslapan: agregar las anomalías no cambia el "
            "resultado de forma concluyente; su valor está en detectar patrones que aún no tienen etiqueta.",
            f"Umbral elegido en validación por ahorro neto (${COSTO_ALERTA:.0f} por alerta): en prueba genera {alertas_te} alertas "
            f"y ahorra ${ahorro_te:,.0f} de ${fraude_total:,.0f} en fraude del periodo.",
        ],
        "metrics": [
            {"label": f"PR-AUC {mejor.split(' ')[0]}", "value": f"{r.loc[mejor, 'PR-AUC']:.3f}"},
            {"label": "PR-AUC autoencoder", "value": f"{r.loc['Autoencoder (sin etiquetas)', 'PR-AUC']:.3f}"},
            {"label": f"Fraudes @{ALERTAS_POR_MIL:g}‰ alertas", "value": r.loc[mejor, f"Fraudes capturados @{ALERTAS_POR_MIL:g}‰"]},
            {"label": "Ahorro neto en prueba", "value": f"${ahorro_te:,.0f}"},
        ],
        "charts": [
            line("pr", "Curvas precisión–exhaustividad (prueba)", curvas, x_label="% de fraudes detectados (recall)",
                 y_label="% de alertas que son fraude (precisión)", fmt="pct"),
            line("presupuesto", "Monto de fraude capturado según el presupuesto de alertas", curva_pres,
                 x_label="Alertas por cada 1,000 transacciones", y_label="% del monto defraudado capturado", fmt="pct",
                 markers=[[ALERTAS_POR_MIL, "Capacidad"]]),
            line("err", "Autoencoder: error de reconstrucción (log10), normales vs. fraude",
                 {"Normales": hist(y_te == 0), "Fraude": hist(y_te == 1)}, x_label="log10(error)",
                 y_label="% de transacciones", fmt="pct"),
            bars("ae_var", "¿Por qué alertó el autoencoder? Error por variable en la alerta más alta", contrib_ae.index,
                 contrib_ae.values.round(2), fmt="dec"),
            bars("shap", "Supervisado: |SHAP| promedio", shap.index, shap.values.round(3), fmt="dec"),
            bars("local", "Supervisado: explicación de la alerta más alta", etiquetas, valores, fmt="dec",
                 subtitle="Contribución SHAP en log-odds"),
        ],
        "tables": [table("Comparación de modelos en el periodo de prueba", res)],
        "card": model_card(
            "Isolation Forest y autoencoder (MLP 20-8-20 entrenado solo con transacciones normales) contra LightGBM "
            "supervisado y un híbrido que usa los scores de anomalía como variables.",
            ["Partición temporal 60/20/20: el umbral se elige en validación y se aplica una vez en prueba.",
             "PR-AUC con IC bootstrap; captura de fraudes y de monto con presupuesto fijo de alertas."],
            [f"Cada alerta cuesta ${COSTO_ALERTA:.0f} de revisión; el fraude detectado se evita al 100 %.",
             f"El equipo revisa {ALERTAS_POR_MIL:g} alertas por cada 1,000 transacciones."],
            ["Solo 2 días de transacciones europeas de 2013 y variables anonimizadas (no se interpretan de negocio).",
             "En producción el fraude se adapta: hay que monitorear drift (proyecto #18)."]),
    }
    muestra = df.iloc[b:].groupby("Class", group_keys=False).apply(
        lambda g: g.sample(min(len(g), 2900 if g.name == 0 else 100), random_state=SEED))
    export("p12-fraude", result, muestra[["V14", "V17", "V12", "V10", "Amount", "Class"]].round(4))
    return result


if __name__ == "__main__":
    run()