Scorecard de originación de crédito

"""#4 Scorecard de originación de crédito (WoE / IV + regresión logística → puntos).

Datos: Statlog German Credit (UCI), 1,000 solicitudes con 20 atributos y etiqueta buen/mal pagador.
Se construye el scorecard "a la antigua", como lo usan las áreas de riesgo y cobranza.
"""
import io

import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score
from lightgbm import LGBMClassifier
from sklearn.model_selection import RepeatedStratifiedKFold, train_test_split

from common import SEED, bars, download, export, gini, ks_stat, line, psi, read_from_zip, table
from validacion import bootstrap_ci, graficas_validacion, model_card, t_test_corregido

URL = "https://archive.ics.uci.edu/static/public/144/statlog+german+credit+data.zip"

COLS = ["estatus_cuenta", "plazo_meses", "historial_credito", "proposito", "monto",
        "ahorros", "antiguedad_empleo", "tasa_pago_pct_ingreso", "estado_civil_sexo",
        "otros_deudores", "antiguedad_domicilio", "propiedad", "edad", "otros_planes",
        "vivienda", "creditos_existentes", "empleo", "dependientes", "telefono",
        "trabajador_extranjero", "malo"]
NUMERICAS = ["plazo_meses", "monto", "edad"]
# Atributos protegidos: no se usan para decidir crédito (no discriminación / equidad)
SENSIBLES = ["estado_civil_sexo", "trabajador_extranjero"]

# Diccionario de códigos UCI → etiquetas legibles para el scorecard
ETIQUETAS = {
    "A11": "< 0 DM", "A12": "0–200 DM", "A13": "≥ 200 DM", "A14": "sin cuenta",
    "A30": "sin créditos", "A31": "todo pagado (este banco)", "A32": "al corriente",
    "A33": "atrasos previos", "A34": "cuenta crítica",
    "A40": "auto nuevo", "A41": "auto usado", "A42": "muebles", "A43": "radio/TV",
    "A44": "electrodomésticos", "A45": "reparaciones", "A46": "educación", "A48": "capacitación",
    "A49": "negocio", "A410": "otros",
    "A61": "< 100 DM", "A62": "100–500 DM", "A63": "500–1000 DM", "A64": "≥ 1000 DM", "A65": "sin ahorro",
    "A71": "desempleado", "A72": "< 1 año", "A73": "1–4 años", "A74": "4–7 años", "A75": "≥ 7 años",
    "A91": "H divorciado", "A92": "M div./casada", "A93": "H soltero", "A94": "H casado",
    "A101": "ninguno", "A102": "co-solicitante", "A103": "garante",
    "A121": "inmueble", "A122": "seguro de vida", "A123": "auto/otro", "A124": "sin propiedad",
    "A141": "banco", "A142": "tiendas", "A143": "ninguno",
    "A151": "renta", "A152": "propia", "A153": "gratuita",
    "A171": "no calificado (no residente)", "A172": "no calificado", "A173": "calificado",
    "A174": "directivo/independiente",
    "A191": "no", "A192": "sí", "A201": "sí", "A202": "no",
}

# Escala del scorecard: 600 puntos = momios 50:1 (buenos:malos); cada 20 puntos duplican los momios
BASE, MOMIOS_BASE, PDO = 600, 50, 20


def load() -> pd.DataFrame:
    raw = read_from_zip(download(URL, "german_credit.zip"), "german.data")
    df = pd.read_csv(io.BytesIO(raw), sep=" ", header=None, names=COLS)
    df["malo"] = (df["malo"] == 2).astype(int)
    return df


def woe_table(x: pd.Series, y: pd.Series) -> pd.DataFrame:
    t = pd.crosstab(x, y).reindex(columns=[0, 1], fill_value=0) + 0.5  # suavizado
    t.columns = ["buenos", "malos"]
    t["dist_b"], t["dist_m"] = t.buenos / t.buenos.sum(), t.malos / t.malos.sum()
    t["woe"] = np.log(t.dist_b / t.dist_m)
    t["iv"] = (t.dist_b - t.dist_m) * t.woe
    return t


def cortes_monotonicos(x: pd.Series, y: pd.Series, n_ini=10, min_frac=0.05) -> np.ndarray:
    """Binning monotónico: parte de deciles y fusiona bins vecinos hasta que la tasa de malos
    sea monótona (en la dirección de la correlación) y cada bin tenga al menos 5 % de los casos.
    Un WoE monótono es requisito regulatorio típico: el score debe tener sentido de negocio."""
    cortes = list(np.unique(np.quantile(x, np.linspace(0, 1, n_ini + 1))))
    cortes[0], cortes[-1] = -np.inf, np.inf
    signo = np.sign(pd.Series(x).corr(pd.Series(y), method="spearman")) or 1
    while len(cortes) > 2:
        b = pd.cut(x, cortes)
        g = pd.DataFrame({"b": b, "y": y}).groupby("b", observed=False).y.agg(["mean", "size"])
        tasas, tam = g["mean"].fillna(0).values, g["size"].values
        malos = [i for i in range(1, len(tasas)) if signo * (tasas[i] - tasas[i - 1]) < 0]
        chicos = [i for i in range(len(tam)) if tam[i] < min_frac * len(x)]
        if not malos and not chicos:
            break
        i = (malos or chicos)[0]
        cortes.pop(i if i > 0 else 1)          # fusiona el bin i con su vecino izquierdo
    return np.array(cortes)


def binning(df: pd.DataFrame, cuts: dict):
    out = df.copy()
    for c in NUMERICAS:
        out[c] = pd.cut(df[c], cuts[c]).astype(str)
    return out


class Scorecard:
    """Pipeline completo (binning → WoE → selección → logística → puntos) ajustado solo con
    los datos que recibe en `fit`, para poder validarlo con CV sin fuga de información."""

    def fit(self, d: pd.DataFrame):
        y = d.malo
        self.cuts = {c: cortes_monotonicos(d[c], y) for c in NUMERICAS}
        b = binning(d, self.cuts)
        feats = [c for c in COLS if c not in ("malo", *SENSIBLES)]
        self.woes = {c: woe_table(b[c], y) for c in feats}
        self.iv = pd.Series({c: self.woes[c].iv.sum() for c in feats}).sort_values(ascending=False)
        cand = self.iv[self.iv >= 0.02].index.tolist()          # IV < 0.02: poder predictivo nulo
        X = self._woe(b, cand)
        # Filtro de colinealidad: si |r| > 0.6 entre dos WoE, se queda la de mayor IV
        corr = X.corr().abs()
        self.sel = []
        for c in cand:
            if all(corr.loc[c, s] <= 0.6 for s in self.sel):
                self.sel.append(c)
        # Control de signos: con WoE = ln(buenos/malos) todo coeficiente debe ser negativo; si alguno no lo es
        # (colinealidad residual), se retira la variable de menor IV y se reajusta.
        self.retiradas = []
        while True:
            self.lr = LogisticRegression(C=0.5, max_iter=2000).fit(self._woe(b, self.sel), y)
            malos = [c for c, beta in zip(self.sel, self.lr.coef_[0]) if beta >= 0]
            if not malos:
                break
            peor = min(malos, key=lambda c: self.iv[c])
            self.sel.remove(peor)
            self.retiradas.append(peor)
        factor = PDO / np.log(2)
        offset = BASE - factor * np.log(MOMIOS_BASE)
        n = len(self.sel)
        filas = []
        for c, beta in zip(self.sel, self.lr.coef_[0]):
            for cat, w in self.woes[c].woe.items():
                pts = -(beta * w + self.lr.intercept_[0] / n) * factor + offset / n
                filas.append([c, str(cat), ETIQUETAS.get(str(cat), str(cat)), round(float(w), 3), int(round(pts))])
        self.card = pd.DataFrame(filas, columns=["Variable", "Atributo", "Descripción", "WoE", "Puntos"])
        return self

    def _woe(self, b, cols):
        return pd.DataFrame({c: b[c].map(self.woes[c].woe).fillna(0) for c in cols})

    def proba(self, d):
        return self.lr.predict_proba(self._woe(binning(d, self.cuts), self.sel))[:, 1]

    def puntos(self, d) -> pd.DataFrame:
        b = binning(d, self.cuts)
        out = {}
        for c in self.sel:
            m = self.card[self.card.Variable == c].set_index("Atributo").Puntos
            out[c] = b[c].astype(str).map(m).fillna(m.median()).values
        return pd.DataFrame(out, index=d.index)


def benchmark():
    """Modelo retador (challenger) sin restricciones de interpretabilidad."""
    return LGBMClassifier(n_estimators=300, learning_rate=0.03, num_leaves=8, min_child_samples=20,
                          subsample=0.8, subsample_freq=1, colsample_bytree=0.8, random_state=SEED, verbose=-1)


def a_categorias(d):
    d = d.drop(columns=["malo", *SENSIBLES]).copy()
    for c in d.select_dtypes("object"):
        d[c] = d[c].astype("category")
    return d


def run():
    df = load()
    tr, te = train_test_split(df, test_size=0.3, stratify=df.malo, random_state=SEED)

    # 1) CV 5×3 del pipeline completo (WoE recalculado en cada fold) vs. retador LightGBM
    cv = []
    rskf = RepeatedStratifiedKFold(n_splits=5, n_repeats=3, random_state=SEED)
    for k, (a_, b_) in enumerate(rskf.split(tr, tr.malo)):
        f_tr, f_va = tr.iloc[a_], tr.iloc[b_]
        sc = Scorecard().fit(f_tr)
        gb = benchmark().fit(a_categorias(f_tr), f_tr.malo)
        for nombre, pr in [("Scorecard WoE", sc.proba(f_va)), ("LightGBM (retador)", gb.predict_proba(a_categorias(f_va))[:, 1])]:
            cv.append({"modelo": nombre, "fold": k + 1, "Gini": gini(f_va.malo, pr), "KS": ks_stat(f_va.malo, pr)})
    cv = pd.DataFrame(cv)
    cvg = cv.groupby("modelo", sort=False)[["Gini", "KS"]].agg(["mean", "std"])
    d_gini, p_gini = t_test_corregido(cv[cv.modelo == "Scorecard WoE"].Gini.values,
                                      cv[cv.modelo == "LightGBM (retador)"].Gini.values,
                                      n_train=int(len(tr) * 0.8), n_test=int(len(tr) * 0.2))

    # 2) Modelo final y prueba
    sc = Scorecard().fit(tr)
    p_te = sc.proba(te)
    pts_te = sc.puntos(te)
    s_te, s_tr = pts_te.sum(axis=1).values, sc.puntos(tr).sum(axis=1).values
    auc, ks = roc_auc_score(te.malo, p_te), ks_stat(te.malo, p_te)
    g, g_lo, g_hi = bootstrap_ci(te.malo, p_te, metrica=gini)
    estabilidad = psi(s_tr, s_te)
    signos_ok = bool((sc.lr.coef_[0] < 0).all())     # WoE = ln(buenos/malos) → coeficientes negativos
    def woe_ordenado(c):
        orden = [str(x) for x in pd.cut(pd.Series([0]), sc.cuts[c]).cat.categories]
        return sc.woes[c].woe.reindex(orden).dropna()
    monot = {c: bool(woe_ordenado(c).is_monotonic_increasing or woe_ordenado(c).is_monotonic_decreasing)
             for c in NUMERICAS}
    corr_max = float(sc._woe(binning(tr, sc.cuts), sc.sel).corr().abs().where(
        lambda m: ~np.eye(len(sc.sel), dtype=bool)).max().max())

    # 3) Corte óptimo con la matriz de costos oficial del dataset: aprobar a un malo cuesta 5, rechazar a un bueno 1
    malo = te.malo.values == 1
    filas_c = []
    for c in np.arange(np.percentile(s_te, 2), np.percentile(s_te, 98), 2):
        aprob = s_te >= c
        costo = 5 * (aprob & malo).sum() + 1 * (~aprob & ~malo).sum()
        filas_c.append([float(round(c)), aprob.mean() * 100, (malo[aprob].mean() * 100) if aprob.any() else 0, costo / len(te)])
    cortes = pd.DataFrame(filas_c, columns=["corte", "aprob", "malos", "costo"])
    best = cortes.loc[cortes.costo.idxmin()]
    costo_todos = 5 * malo.sum() / len(te)

    # 4) Razones de rechazo (reason codes): las variables donde el solicitante pierde más puntos
    #    respecto al máximo posible de esa variable — así se explica una negativa al cliente.
    maximos = sc.card.groupby("Variable").Puntos.max()
    rechazados = te.index[s_te < best.corte][:5]
    razones = []
    for i in rechazados:
        perdida = (maximos[sc.sel] - pts_te.loc[i, sc.sel]).sort_values(ascending=False)[:3]
        attr = binning(te.loc[[i]], sc.cuts).iloc[0]
        razones.append([int(i), int(pts_te.loc[i].sum()),
                        "; ".join(f"{v} = {ETIQUETAS.get(str(attr[v]), str(attr[v]))} (−{int(pp)})" for v, pp in perdida.items())])
    razones = pd.DataFrame(razones, columns=["Solicitud", "Puntos", "Top 3 razones (puntos perdidos)"])

    edges = np.arange(np.floor(s_te.min() / 20) * 20, s_te.max() + 20, 20)
    def hist(mask):
        h = np.histogram(s_te[mask], edges)[0] / mask.sum() * 100
        return [[float((a + b) / 2), round(float(v), 2)] for a, b, v in zip(edges[:-1], edges[1:], h)]

    woe_mono = {c: [[i + 1, round(float(w), 3)] for i, w in enumerate(woe_ordenado(c).values)] for c in NUMERICAS}
    cv_tbl = pd.DataFrame({"Modelo": cvg.index,
                           "Gini (media ± desv.)": [f"{m:.3f} ± {s:.3f}" for m, s in cvg["Gini"].values],
                           "KS (media ± desv.)": [f"{m:.3f} ± {s:.3f}" for m, s in cvg["KS"].values]})
    corte_tbl = cortes.iloc[::5].round(1).rename(columns={"corte": "Corte", "aprob": "Aprobación %",
                                                          "malos": "Malos aprobados %", "costo": "Costo esperado"})
    g_sc = cvg.loc["Scorecard WoE", ("Gini", "mean")]
    g_gb = cvg.loc["LightGBM (retador)", ("Gini", "mean")]

    result = {
        "title": "Scorecard de originación de crédito",
        "summary": [
            f"Pipeline WoE completo validado con CV 5×3 (binning y WoE recalculados dentro de cada fold): Gini "
            f"{g_sc:.3f} vs. {g_gb:.3f} del retador LightGBM (diferencia {d_gini:+.3f}, p = {p_gini:.2f}). "
            "El scorecard interpretable no pierde poder frente a una caja negra.",
            f"En prueba: Gini {g:.3f} (IC 95 % {g_lo:.3f}–{g_hi:.3f}), KS {ks:.3f}. PSI entrenamiento→prueba "
            f"{estabilidad:.3f} (< 0.1 = estable).",
            f"Controles de modelo: {len(sc.sel)} variables tras IV ≥ 0.02, filtro de colinealidad (|r| máx. "
            f"{corr_max:.2f}) y control de signos ({len(sc.retiradas)} retiradas: {', '.join(sc.retiradas) or 'ninguna'}); "
            f"signos {'consistentes' if signos_ok else 'INCONSISTENTES'}; WoE monótono en {sum(monot.values())}/{len(monot)} "
            "numéricas. Se excluyen atributos protegidos (estado civil/sexo, extranjería).",
            f"Corte por costo esperado (aprobar a un malo cuesta 5× rechazar a un bueno): {int(best.corte)} puntos, "
            f"aprueba {best.aprob:.0f} % con {best.malos:.0f} % de malos; costo {best.costo:.2f} vs. {costo_todos:.2f} aprobando a todos.",
            "Cada rechazo se explica con razones (reason codes): las variables donde el solicitante pierde más puntos.",
        ],
        "metrics": [
            {"label": "Gini CV 5×3", "value": f"{g_sc:.3f} ± {cvg.loc['Scorecard WoE', ('Gini', 'std')]:.3f}"},
            {"label": "Gini prueba (IC 95 %)", "value": f"{g:.3f} [{g_lo:.2f}–{g_hi:.2f}]"},
            {"label": "PSI estabilidad", "value": f"{estabilidad:.3f}"},
            {"label": "Corte óptimo", "value": f"{int(best.corte)} pts"},
        ],
        "charts": [
            bars("iv", "Information Value por variable", sc.iv.index, sc.iv.round(3).values, fmt="dec",
                 subtitle="< 0.02 nulo · 0.02–0.1 débil · 0.1–0.3 medio · > 0.3 fuerte"),
            bars("cvg", "Gini en CV 5×3: scorecard vs. retador", cvg.index, cvg[("Gini", "mean")].round(3).values,
                 fmt="dec"),
            line("woe", "WoE por bin de las variables numéricas (binning monotónico)",
                 {"Plazo": woe_mono["plazo_meses"], "Monto": woe_mono["monto"], "Edad": woe_mono["edad"]},
                 x_label="Bin (de menor a mayor valor)", y_label="WoE", fmt="dec"),
            line("dist", "Distribución del score: buenos vs. malos pagadores",
                 {"Buenos": hist(~malo), "Malos": hist(malo)},
                 x_label="Puntos del scorecard", y_label="% de clientes", fmt="pct",
                 markers=[[float(best.corte), "Corte óptimo"]]),
            line("costo", "Costo esperado por solicitud según el corte",
                 {"Costo": cortes[["corte", "costo"]].round(3).values.tolist()},
                 x_label="Corte (puntos)", y_label="Costo por solicitud", fmt="dec",
                 markers=[[float(best.corte), "Óptimo"]]),
            *graficas_validacion(te.malo.values, {"Scorecard": p_te}),
        ],
        "tables": [table("Validación cruzada 5×3: champion vs. challenger", cv_tbl),
                   table("Razones de rechazo (reason codes) de solicitudes bajo el corte", razones),
                   table("Scorecard (puntos por atributo)", sc.card),
                   table("Tabla de cortes", corte_tbl)],
        "card": model_card(
            "Scorecard clásico: binning monotónico, WoE, selección por IV y colinealidad, regresión logística y "
            f"escala de puntos ({BASE} pts = {MOMIOS_BASE}:1, PDO {PDO}).",
            ["CV estratificada repetida 5×3 del pipeline completo (sin fuga de WoE).",
             "Champion/challenger contra LightGBM con t-test corregido.",
             "Gini con IC bootstrap, PSI de estabilidad, chequeo de signos, monotonía y colinealidad.",
             "Corte elegido por la matriz de costos publicada con el dataset."],
            ["El costo relativo 5:1 viene del dataset; en una financiera real se calcula con pérdida esperada (PD × LGD × EAD).",
             "La muestra está sobre-representada en malos (30 %): las probabilidades deben recalibrarse a la tasa real."],
            ["1,000 solicitudes de un banco alemán de los 90s: el Gini tiene intervalos amplios.",
             "Se excluyeron estado civil/sexo y extranjería por no discriminación; edad se conserva (uso permitido "
             "con justificación de riesgo), pero debe revisarse con el área legal."]),
    }
    sample = tr[["estatus_cuenta", "historial_credito", "ahorros", "plazo_meses", "monto", "edad", "malo"]]
    export("p04-scorecard", result, sample)
    return result


if __name__ == "__main__":
    run()