Cobranza: probabilidad de pago tras mora

"""#5 Modelo de cobranza: probabilidad de pago tras mora.

Datos: Default of Credit Card Clients (UCI, Taiwán, 30,000 tarjetahabientes, 6 meses de historial).
Objetivo: dentro de la cartera vencida, priorizar la gestión según la probabilidad de que el cliente
regularice el próximo mes y el saldo en juego.
"""
import io

import numpy as np
import pandas as pd
from lightgbm import LGBMClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

from common import SEED, bars, download, export, ks_stat, line, read_from_zip, table
from validacion import (bootstrap_ci, calibracion_lineal, cv_clasificacion, explicacion_local,
                        graficas_validacion, model_card, resumen_cv, t_test_corregido)

URL = "https://archive.ics.uci.edu/static/public/350/default+of+credit+card+clients.zip"
PAY = ["PAY_0", "PAY_2", "PAY_3", "PAY_4", "PAY_5", "PAY_6"]   # estatus de pago, mes t … t-5
BILL = [f"BILL_AMT{i}" for i in range(1, 7)]
PAYAMT = [f"PAY_AMT{i}" for i in range(1, 7)]
CAPACIDAD = 0.30  # el call center alcanza a gestionar 30 % de la cartera vencida al mes


def load() -> pd.DataFrame:
    raw = read_from_zip(download(URL, "default_credit_card.zip"), ".xls")
    df = pd.read_excel(io.BytesIO(raw), header=1)
    return df.rename(columns={"default payment next month": "default"})


def features(df: pd.DataFrame) -> pd.DataFrame:
    f = pd.DataFrame(index=df.index)
    pay = df[PAY].clip(lower=0).values                       # meses de atraso por periodo
    f["atraso_actual"] = pay[:, 0]
    f["atraso_max_6m"] = pay.max(1)
    f["meses_en_mora_6m"] = (pay > 0).sum(1)
    # Racha: meses consecutivos en mora hasta hoy
    f["racha_mora"] = np.argmin(np.c_[pay > 0, np.zeros(len(df), bool)], axis=1)
    f["tendencia_atraso"] = pay[:, 0] - pay[:, 2]             # ¿empeora o mejora en 3 meses?
    bill, pagos = df[BILL].values, df[PAYAMT].values
    f["saldo"] = bill[:, 0]
    f["utilizacion"] = bill[:, 0] / df["LIMIT_BAL"].values
    f["ratio_pago_saldo_1m"] = pagos[:, 0] / np.maximum(bill[:, 1], 1)
    f["ratio_pago_saldo_3m"] = pagos[:, :3].sum(1) / np.maximum(bill[:, 1:4].sum(1), 1)
    f["meses_sin_pago"] = (pagos == 0).sum(1)
    f["crecimiento_saldo"] = (bill[:, 0] - bill[:, 3]) / np.maximum(np.abs(bill[:, 3]), 1)
    f["limite"] = df["LIMIT_BAL"]
    f["edad"] = df["AGE"]
    f["escolaridad"] = df["EDUCATION"]
    return f.replace([np.inf, -np.inf], 0)


# Variables cuya relación con la probabilidad de pago tiene sentido de negocio conocido:
# más atraso / más meses sin pagar → menos probable que regularice (restricción monótona −1).
MONOTONAS = {"atraso_actual": -1, "atraso_max_6m": -1, "meses_en_mora_6m": -1, "racha_mora": -1,
             "meses_sin_pago": -1, "ratio_pago_saldo_1m": 1, "ratio_pago_saldo_3m": 1}


def modelos(columnas):
    base = dict(n_estimators=300, learning_rate=0.03, num_leaves=15, min_child_samples=50,
                subsample=0.8, subsample_freq=1, random_state=SEED, verbose=-1)
    mono = [MONOTONAS.get(c, 0) for c in columnas]
    return {
        "Logística (baseline)": make_pipeline(StandardScaler(), LogisticRegression(max_iter=2000)),
        "LightGBM": LGBMClassifier(**base),
        "LightGBM monótono": LGBMClassifier(**base, monotone_constraints=mono),
    }


def run():
    df = load()
    X_all, y_default = features(df), df["default"].values
    # Universo de cobranza: clientes con algún atraso este mes
    mora = X_all["atraso_actual"] >= 1
    X, y = X_all[mora], 1 - y_default[mora]            # y = 1 → paga / regulariza
    saldo = X["saldo"].clip(lower=0).values
    X_tr, X_te, y_tr, y_te, s_tr, s_te = train_test_split(X, y, saldo, test_size=0.3, stratify=y,
                                                          random_state=SEED)
    mods = modelos(list(X.columns))

    # 1) CV 5×2 con los mismos folds para los tres modelos
    cv = cv_clasificacion(mods, X_tr, y_tr, folds=5, repeticiones=2)
    auc_cv = cv.groupby("modelo", sort=False).AUC
    d_mono, p_mono = t_test_corregido(cv[cv.modelo == "LightGBM monótono"].AUC.values,
                                      cv[cv.modelo == "LightGBM"].AUC.values,
                                      int(len(X_tr) * 0.8), int(len(X_tr) * 0.2))

    # 2) Modelo final: LightGBM monótono (mismo desempeño, comportamiento explicable y auditable)
    fit = {k: m.fit(X_tr, y_tr) for k, m in mods.items()}
    probs = {k: m.predict_proba(X_te)[:, 1] for k, m in fit.items()}
    gbm = fit["LightGBM monótono"]
    p = probs["LightGBM monótono"]
    auc, auc_lo, auc_hi = bootstrap_ci(y_te, p)
    ks = ks_stat(y_te, p)
    pend, interc = calibracion_lineal(y_te, p)

    # 3) ¿El modelo solo reproduce el tramo de mora? AUC DENTRO de cada tramo
    tramo_te = pd.cut(X_te["atraso_actual"], [0, 1, 2, 9], labels=["1 mes", "2 meses", "3+ meses"])
    auc_tramo = []
    for t_, m in pd.Series(tramo_te).groupby(tramo_te, observed=True).groups.items():
        idx = X_te.index.get_indexer(m)
        if len(np.unique(y_te[idx])) == 2:
            auc_tramo.append([str(t_), len(idx), round(roc_auc_score(y_te[idx], p[idx]), 3),
                              round(roc_auc_score(y_te[idx], -X_te["atraso_actual"].values[idx]), 3)])
    auc_tramo = pd.DataFrame(auc_tramo, columns=["Tramo", "Cuentas", "AUC modelo", "AUC regla (días de mora)"])

    # 4) Priorización vs. regla tradicional y azar
    def capturado(orden):
        no_paga = (1 - y_te)[orden]
        acum = np.cumsum(no_paga) / no_paga.sum()
        idx = np.linspace(0, len(acum) - 1, 40).astype(int)
        return [[0, 0]] + [[round((i + 1) / len(acum) * 100, 1), round(acum[i] * 100, 2)] for i in idx]
    rng = np.random.default_rng(SEED)
    estrategias = {
        "Modelo (prob. de impago)": np.argsort(p),
        "Más días de mora primero": np.lexsort((-s_te, -X_te["atraso_actual"].values)),
        "Al azar": rng.permutation(len(p)),
    }
    curvas = {k: capturado(v) for k, v in estrategias.items()}
    cap = {n: next(v for x, v in reversed(c) if x <= CAPACIDAD * 100) for n, c in curvas.items()}
    # IC bootstrap de la ganancia del modelo sobre la regla a la capacidad dada
    k = int(len(p) * CAPACIDAD)
    gan = []
    for _ in range(300):
        i = rng.integers(0, len(p), len(p))
        yy, pp, at, ss = y_te[i], p[i], X_te["atraso_actual"].values[i], s_te[i]
        imp = (1 - yy).sum()
        m_ = (1 - yy)[np.argsort(pp)[:k]].sum() / imp
        r_ = (1 - yy)[np.lexsort((-ss, -at))[:k]].sum() / imp
        gan.append((m_ - r_) * 100)
    g_lo, g_hi = np.percentile(gan, [2.5, 97.5])

    # 5) Explicabilidad: SHAP global y local (cuenta con menor probabilidad de pago)
    contrib = gbm.predict(X_te, pred_contrib=True)
    shap = pd.Series(np.abs(contrib[:, :-1]).mean(0), index=X.columns).sort_values(ascending=False)
    i0 = int(np.argmin(p))
    etiquetas, valores = explicacion_local(contrib[i0, :-1], list(X.columns), X_te.iloc[i0].round(2))
    # Efecto parcial (dependencia) de la variable con más peso, verificando la monotonía
    var = shap.index[0]
    grid = np.unique(np.quantile(X_te[var], np.linspace(0.02, 0.98, 15)))
    pdp = {}
    for nombre in ["LightGBM", "LightGBM monótono"]:
        pts = []
        for v in grid:
            Xg = X_te.copy()
            Xg[var] = v
            pts.append([round(float(v), 3), round(float(fit[nombre].predict_proba(Xg)[:, 1].mean() * 100), 2)])
        pdp[nombre] = pts

    tramo = pd.cut(X["atraso_actual"], [0, 1, 2, 3, 9], labels=["1 mes", "2 meses", "3 meses", "4+ meses"])
    por_tramo = pd.DataFrame({"tramo": tramo, "paga": y}).groupby("tramo", observed=True).agg(
        clientes=("paga", "size"), tasa_pago=("paga", "mean")).reset_index()
    por_tramo["tasa_pago"] = (por_tramo["tasa_pago"] * 100).round(1)
    por_tramo.columns = ["Tramo de mora", "Clientes", "Regularizan %"]

    modelo, regla = "Modelo (prob. de impago)", "Más días de mora primero"
    result = {
        "title": "Cobranza: probabilidad de pago tras mora",
        "summary": [
            f"Universo: {mora.sum():,} cuentas con atraso; {y.mean():.0%} regulariza al mes siguiente.",
            "CV 5×2 (AUC): " + ", ".join(f"{k} {v.mean():.3f} ± {v.std():.3f}" for k, v in auc_cv) +
            f". Las restricciones monótonas cambian el AUC en {d_mono:+.3f} (p = {p_mono:.2f}) y garantizan que "
            "más atraso nunca suba la probabilidad de pago: modelo auditable.",
            f"En prueba: AUC {auc:.3f} (IC 95 % {auc_lo:.3f}–{auc_hi:.3f}), KS {ks:.3f}; calibración pendiente "
            f"{pend:.2f} (< 1: probabilidades algo extremas; se corrige con recalibración Platt si se usan como "
            f"probabilidad), intercepto {interc:+.2f}.",
            f"Con capacidad para {CAPACIDAD:.0%} de las cuentas, el modelo alcanza {cap[modelo]:.0f} % de los clientes que no "
            f"regularizarían vs. {cap[regla]:.0f} % de la regla de días de mora (ganancia IC 95 % {g_lo:+.1f} a {g_hi:+.1f} pp). "
            "El valor real está dentro de cada tramo, donde la regla no distingue (ver AUC por tramo).",
        ],
        "metrics": [
            {"label": "AUC CV (monótono)", "value": f"{auc_cv.mean()['LightGBM monótono']:.3f} ± {auc_cv.std()['LightGBM monótono']:.3f}"},
            {"label": "AUC prueba (IC 95 %)", "value": f"{auc:.3f} [{auc_lo:.2f}–{auc_hi:.2f}]"},
            {"label": f"Impagos alcanzados @{CAPACIDAD:.0%}", "value": f"{cap[modelo]:.0f} %"},
            {"label": "Ganancia vs. regla", "value": f"{cap[modelo] - cap[regla]:+.0f} pp"},
        ],
        "charts": [
            line("prior", "Cuentas que no regularizarían solas, alcanzadas según estrategia", curvas,
                 x_label="% de cuentas gestionadas", y_label="% de impagos alcanzados", fmt="pct",
                 markers=[[CAPACIDAD * 100, "Capacidad"]]),
            bars("cv", "AUC en CV 5×2 por modelo (media)", auc_cv.mean().index, auc_cv.mean().round(4).values, fmt="dec"),
            bars("auct", "AUC dentro de cada tramo de mora: modelo", auc_tramo["Tramo"], auc_tramo["AUC modelo"].values,
                 fmt="dec", subtitle="La regla de días de mora tiene AUC ≈ 0.5 dentro de un mismo tramo"),
            *graficas_validacion(y_te, {"LightGBM monótono": p, "Logística": probs["Logística (baseline)"]}),
            line("pdp", f"Efecto parcial de {var}: ¿se respeta la lógica de negocio?", pdp,
                 x_label=var, y_label="Prob. de pago promedio %", fmt="pct"),
            bars("shap", "Explicabilidad global: |SHAP| promedio", shap.index[:10], shap.values[:10].round(3), fmt="dec"),
            bars("local", "Explicabilidad local: cuenta con menor probabilidad de pago", etiquetas, valores, fmt="dec",
                 subtitle=f"Contribución SHAP en log-odds · prob. de pago {p[i0]:.0%}"),
        ],
        "tables": [table("Validación cruzada 5×2", resumen_cv(cv)),
                   table("AUC dentro de cada tramo de mora", auc_tramo),
                   table("Regularización por tramo", por_tramo)],
        "card": model_card(
            "LightGBM con restricciones monótonas sobre variables de comportamiento de pago, comparado con "
            "logística y LightGBM libre.",
            ["Split 70/30 estratificado; CV 5×2 en entrenamiento con t-test corregido.",
             "AUC con IC bootstrap, calibración y AUC estratificado por tramo de mora.",
             "IC bootstrap de la ganancia operativa frente a la regla tradicional.",
             "Efectos parciales para verificar que el modelo respeta la lógica de negocio."],
            ["\"Regularizar\" = no caer en default el mes siguiente (definición del dataset).",
             f"El call center gestiona {CAPACIDAD:.0%} de las cuentas en mora al mes."],
            ["Tarjetas de crédito en Taiwán (2005); sin fechas de corte, por lo que no hay validación fuera de tiempo.",
             "El modelo predice pago, no el efecto de la llamada: para eso se necesita un experimento (ver #3)."]),
    }
    sample = pd.DataFrame({"p_pago": p.round(4), "paga": y_te, "saldo": s_te,
                           "atraso": X_te["atraso_actual"].values})
    export("p05-cobranza", result, sample.sample(min(3000, len(sample)), random_state=1))
    return result


if __name__ == "__main__":
    run()