Mini Data Warehouse + reporte automatizado

"""#16 Mini Data Warehouse de una financiera + reporte ejecutivo automatizado.

1) Genera datos SINTÉTICOS realistas (clientes, créditos, pagos mensuales con mora, pólizas).
2) Los carga a un modelo estrella con SQL compatible con Snowflake (sql/dwh_financiera.sql),
   ejecutado aquí en DuckDB; con `snowflake-connector-python` corre igual en Snowflake.
3) Produce el reporte ejecutivo (Excel) y los KPIs para el portafolio.
Nunca se usa información real de ninguna empresa.
"""
from pathlib import Path

import duckdb
import numpy as np
import pandas as pd
from scipy.stats import chi2_contingency

from common import ROOT, SEED, bars, export, line, table
from validacion import model_card

DB = ROOT / "data" / "dwh_financiera.duckdb"
SALIDA = ROOT / "output"
CORTE_FINAL = pd.Timestamp("2026-09-01")

ESTADOS = {"Ciudad de México": 18, "Estado de México": 14, "Jalisco": 10, "Nuevo León": 10,
           "Aguascalientes": 6, "Guanajuato": 7, "Puebla": 6, "Querétaro": 5, "Veracruz": 6,
           "Chihuahua": 5, "Sonora": 4, "Yucatán": 4, "Baja California": 5}
NOMBRES = ["María", "José", "Juan", "Guadalupe", "Luis", "Ana", "Carlos", "Laura", "Jorge", "Sofía",
           "Miguel", "Fernanda", "Ricardo", "Daniela", "Alejandro", "Valeria"]
APELLIDOS = ["Hernández", "García", "Martínez", "López", "González", "Pérez", "Rodríguez", "Sánchez",
             "Ramírez", "Cruz", "Flores", "Gómez", "Luna", "Díaz", "Torres", "Vázquez"]
PRODUCTOS = pd.DataFrame({
    "id_producto": [1, 2, 3, 4],
    "producto": ["Crédito auto nuevo", "Crédito seminuevo", "Arrendamiento", "Crédito flotilla"],
    "tipo_producto": ["Auto nuevo", "Seminuevo", "Arrendamiento", "Flotilla"],
    "tasa_anual": [0.139, 0.169, 0.149, 0.129],
    "monto_medio": [380_000, 240_000, 520_000, 1_400_000],
    "peso": [0.55, 0.25, 0.12, 0.08],
})


def generar(n_clientes=8000, n_creditos=11000, rng=None):
    rng = rng or np.random.default_rng(SEED)
    est = list(ESTADOS)
    w = np.array(list(ESTADOS.values()), float)
    clientes = pd.DataFrame({
        "id_cliente": np.arange(1, n_clientes + 1),
        "nombre": [f"{rng.choice(NOMBRES)} {rng.choice(APELLIDOS)} {rng.choice(APELLIDOS)}"
                   for _ in range(n_clientes)],
        "email": [f"cliente{i}@correo-ficticio.mx" for i in range(1, n_clientes + 1)],
        "telefono": [f"55{rng.integers(10_000_000, 99_999_999)}" for _ in range(n_clientes)],
        "estado": rng.choice(est, n_clientes, p=w / w.sum()),
        "edad": rng.integers(21, 72, n_clientes),
        "ingreso_mensual": np.round(rng.lognormal(10.3, 0.55, n_clientes), -2),
        "score_buro": np.clip(rng.normal(680, 70, n_clientes), 450, 850).round(),
    })

    # Créditos originados ene-2024 → sep-2026 con estacionalidad de fin de año
    meses = pd.date_range("2024-01-01", CORTE_FINAL, freq="MS")
    peso_mes = 1 + 0.25 * np.isin(meses.month, [11, 12]) + 0.002 * np.arange(len(meses))
    prod = rng.choice(PRODUCTOS.id_producto, n_creditos, p=PRODUCTOS.peso)
    medio = PRODUCTOS.set_index("id_producto").monto_medio.reindex(prod).values
    creditos = pd.DataFrame({
        "id_credito": np.arange(100_001, 100_001 + n_creditos),
        "id_cliente": rng.integers(1, n_clientes + 1, n_creditos),
        "id_producto": prod,
        "fecha_originacion": rng.choice(meses, n_creditos, p=peso_mes / peso_mes.sum()),
        "monto_original": np.round(medio * rng.lognormal(0, 0.3, n_creditos), -3),
        "plazo_meses": rng.choice([24, 36, 48, 60], n_creditos, p=[0.15, 0.35, 0.3, 0.2]),
    })
    creditos["fecha_originacion"] = pd.to_datetime(creditos["fecha_originacion"])

    # Pagos: foto mensual del saldo y días de atraso (cadena de Markov por tramos de 30 días)
    cli = clientes.set_index("id_cliente")
    riesgo = (1 / (1 + np.exp((cli.score_buro.reindex(creditos.id_cliente).values - 600) / 45))) * 0.09 + 0.004
    filas = []
    tramo = np.zeros(n_creditos, int)
    tasa_m = PRODUCTOS.set_index("id_producto").tasa_anual.reindex(prod).values / 12
    n = creditos.plazo_meses.values
    cuota = creditos.monto_original.values * tasa_m / (1 - (1 + tasa_m) ** -n)
    saldo = creditos.monto_original.values.astype(float)
    orig = creditos.fecha_originacion.values
    for mes in meses:
        activo = (orig < mes.to_datetime64()) & (saldo > 1)
        u = rng.random(n_creditos)
        sube = activo & (((tramo == 0) & (u < riesgo)) | ((tramo > 0) & (u < 0.45)))
        cura = activo & (tramo > 0) & (u > 0.55)
        tramo = np.where(sube, np.minimum(tramo + 1, 5), np.where(cura, 0, tramo))
        paga = activo & (tramo == 0)
        interes = saldo * tasa_m
        saldo = np.where(paga, np.maximum(saldo - (cuota - interes), 0), saldo)
        idx = np.where(activo)[0]
        filas.append(pd.DataFrame({
            "id_credito": creditos.id_credito.values[idx], "fecha_corte": mes,
            "saldo_insoluto": saldo[idx].round(2), "pago_realizado": np.where(paga[idx], cuota[idx], 0).round(2),
            "dias_atraso": tramo[idx] * 30}))
    pagos = pd.concat(filas, ignore_index=True)

    # Pólizas de seguro ligadas al crédito
    aut = creditos.sample(frac=0.86, random_state=SEED)
    vida = creditos.sample(frac=0.58, random_state=SEED + 1)
    polizas = pd.concat([
        pd.DataFrame({"id_credito": aut.id_credito, "ramo": "Auto",
                      "prima_anual": np.round(aut.monto_original * rng.uniform(0.035, 0.06, len(aut)), 2)}),
        pd.DataFrame({"id_credito": vida.id_credito, "ramo": "Vida",
                      "prima_anual": np.round(vida.monto_original * 0.008, 2)})], ignore_index=True)
    polizas.insert(0, "id_poliza", np.arange(1, len(polizas) + 1))
    return clientes, PRODUCTOS.drop(columns=["peso", "monto_medio"]), creditos, pagos, polizas


def ensuciar(clientes, creditos, pagos, rng):
    """Simula problemas típicos de los sistemas fuente (los detecta el proyecto #17)."""
    c = clientes.copy()
    c.loc[c.sample(frac=0.012, random_state=1).index, "email"] = None
    c.loc[c.sample(frac=0.004, random_state=2).index, "edad"] = rng.choice([0, 130, -5], 1)[0]
    c = pd.concat([c, c.sample(15, random_state=3)], ignore_index=True)             # duplicados
    cr = creditos.copy()
    cr.loc[cr.sample(8, random_state=4).index, "monto_original"] *= -1               # signo invertido
    cr.loc[cr.sample(12, random_state=5).index, "id_cliente"] = 999_999               # huérfanos
    p = pagos.copy()
    p.loc[p.sample(20, random_state=6).index, "dias_atraso"] = 15                     # fuera de catálogo
    return c, cr, p


def construir_dwh():
    rng = np.random.default_rng(SEED)
    clientes, productos, creditos, pagos, polizas = generar(rng=rng)
    clientes, creditos, pagos = ensuciar(clientes, creditos, pagos, rng)
    DB.parent.mkdir(parents=True, exist_ok=True)
    con = duckdb.connect(str(DB))
    for nombre, df in {"stg_cliente": clientes, "stg_producto": productos, "stg_credito": creditos,
                       "stg_pago": pagos, "stg_poliza": polizas}.items():
        con.register("tmp", df)
        con.execute(f"CREATE OR REPLACE TABLE {nombre} AS SELECT * FROM tmp")
        con.unregister("tmp")
    sql = (ROOT / "sql" / "dwh_financiera.sql").read_text(encoding="utf-8")
    construccion, pruebas = sql.split("-- ============ Pruebas de datos")
    con.execute(construccion)
    return con, pruebas


def correr_pruebas(con, pruebas: str) -> pd.DataFrame:
    """Ejecuta cada bloque `-- @test nombre` y cuenta las filas devueltas (0 = pasa)."""
    filas = []
    for bloque in pruebas.split("-- @test ")[1:]:
        nombre, consulta = bloque.split("\n", 1)
        n = len(con.execute(consulta.strip().rstrip(";")).fetchall())
        filas.append([nombre.strip(), n, "Pasa" if n == 0 else "Falla"])
    return pd.DataFrame(filas, columns=["Prueba", "Filas que violan", "Resultado"])


def run():
    con, pruebas = construir_dwh()
    q = lambda sql: con.execute(sql).df()
    tests = correr_pruebas(con, pruebas)
    # Las mismas pruebas sobre la capa cruda (staging) muestran qué corrigió la capa curada
    crudo = correr_pruebas(con, pruebas.replace("dim_cliente", "stg_cliente").replace("fact_credito", "stg_credito")
                           .replace("fact_pago", "stg_pago"))
    tests.insert(1, "Filas en staging", crudo["Filas que violan"].values)
    cuarentena = q("SELECT motivo, COUNT(*) n FROM cuarentena_credito GROUP BY 1 ORDER BY 2 DESC")
    cosechas = q("""SELECT cosecha, inicio_cosecha, meses_en_libros, pct_30 FROM vw_cosechas
                    WHERE inicio_cosecha <= DATE '2025-10-01' ORDER BY inicio_cosecha, meses_en_libros""")
    roll = q("SELECT * FROM vw_roll_rates")
    # IMOR por estado con IC bootstrap (re-muestreo de créditos) + prueba χ² de independencia
    corte = q(f"""SELECT c.estado, p.saldo_insoluto saldo, (p.dias_atraso >= 30)::INT mora
                  FROM fact_pago p JOIN fact_credito cr USING (id_credito) JOIN dim_cliente c USING (id_cliente)
                  WHERE p.fecha_corte = DATE '{CORTE_FINAL:%Y-%m-%d}' AND p.saldo_insoluto > 0""")
    rng = np.random.default_rng(SEED)
    ic = {}
    for est, g in corte.groupby("estado"):
        sal, mo = g.saldo.values, g.mora.values
        bs = [(sal[i] * mo[i]).sum() / sal[i].sum() * 100 for i in (rng.integers(0, len(g), len(g)) for _ in range(300))]
        ic[est] = np.percentile(bs, [2.5, 97.5])
    chi_p = float(chi2_contingency(pd.crosstab(corte.estado, corte.mora))[1])
    mora = q("""SELECT fecha_corte, SUM(saldo) saldo, SUM(saldo_30)/SUM(saldo) imor30,
                       SUM(saldo_90)/SUM(saldo) imor90
                FROM vw_cartera_mensual GROUP BY 1 ORDER BY 1""")
    estado = q(f"""SELECT estado, SUM(saldo_30)/SUM(saldo)*100 imor30, SUM(saldo) saldo
                  FROM vw_cartera_mensual WHERE fecha_corte = DATE '{CORTE_FINAL:%Y-%m-%d}'
                  GROUP BY 1 ORDER BY imor30 DESC""")
    coloc = q("SELECT mes, SUM(creditos) creditos, SUM(monto) monto FROM vw_colocacion GROUP BY 1 ORDER BY 1")
    seguros = q("SELECT * FROM vw_penetracion_seguros ORDER BY creditos DESC")
    conteos = q("""SELECT 'dim_cliente' tabla, COUNT(*) filas FROM dim_cliente UNION ALL
                   SELECT 'fact_credito', COUNT(*) FROM fact_credito UNION ALL
                   SELECT 'fact_pago', COUNT(*) FROM fact_pago UNION ALL
                   SELECT 'fact_poliza', COUNT(*) FROM fact_poliza""")

    # Reporte ejecutivo automatizado (se programa con cron o GitHub Actions)
    SALIDA.mkdir(exist_ok=True)
    with pd.ExcelWriter(SALIDA / "reporte_ejecutivo.xlsx") as xl:
        mora.to_excel(xl, sheet_name="Mora mensual", index=False)
        estado.to_excel(xl, sheet_name="Mora por estado", index=False)
        coloc.to_excel(xl, sheet_name="Colocación", index=False)
        seguros.to_excel(xl, sheet_name="Seguros", index=False)

    ult, prev = mora.iloc[-1], mora.iloc[-13]
    seg_tbl = seguros.assign(pct_auto=(seguros.pct_auto * 100).round(1), pct_vida=(seguros.pct_vida * 100).round(1))
    seg_tbl.columns = ["Producto", "Créditos", "% con seguro de auto", "% con seguro de vida"]
    est_tbl = estado.assign(imor30=estado.imor30.round(2), saldo=(estado.saldo / 1e6).round(1),
                            lo=estado.estado.map(lambda e: round(ic[e][0], 2)), hi=estado.estado.map(lambda e: round(ic[e][1], 2)))
    est_tbl.columns = ["Estado", "IMOR 30+ %", "Saldo (MDP)", "IC 2.5 %", "IC 97.5 %"]
    # Cosechas: una línea por trimestre de originación (las 5 más recientes con al menos 9 meses en libros)
    sel = cosechas.groupby("cosecha").meses_en_libros.max().loc[lambda x: x >= 9].index
    sel = cosechas[cosechas.cosecha.isin(sel)].drop_duplicates("cosecha").tail(4).cosecha.tolist()
    vint = {c: cosechas[(cosechas.cosecha == c) & (cosechas.meses_en_libros <= 18)][["meses_en_libros", "pct_30"]]
            .assign(pct_30=lambda d: (d.pct_30 * 100).round(2)).values.tolist() for c in sel}
    # Roll rates como matriz % (fila = tramo de origen)
    rr = roll.pivot_table(index="tramo_origen", columns="tramo_destino", values="creditos", aggfunc="sum").fillna(0)
    rr = (rr.div(rr.sum(axis=1), axis=0) * 100).round(1)
    rr.index = [f"{int(i)} días" for i in rr.index]
    rr.columns = [f"→ {int(c)}" for c in rr.columns]
    rr = rr.reset_index().rename(columns={"index": "Desde"})
    cura = float(rr.loc[rr.Desde == "30 días", "→ 0"].iloc[0])
    rueda = float(rr.loc[rr.Desde == "30 días", "→ 60"].iloc[0])
    fmt_mes = lambda s: s.dt.strftime("%Y-%m")
    result = {
        "title": "Mini Data Warehouse + reporte automatizado",
        "summary": [
            f"Modelo estrella con {int(conteos.filas.sum()):,} registros sintéticos: clientes, créditos, "
            "pagos mensuales y pólizas; SQL portable a Snowflake.",
            f"Cartera vigente al corte {CORTE_FINAL:%m/%Y}: ${ult.saldo / 1e9:,.2f} mil millones; "
            f"IMOR 30+ de {ult.imor30:.2%} (hace 12 meses {prev.imor30:.2%}).",
            f"Capa curada con reglas explícitas: {int(cuarentena.n.sum())} créditos en cuarentena "
            f"({', '.join(f'{r.motivo}: {r.n}' for r in cuarentena.itertuples())}); {int((tests.Resultado == 'Pasa').sum())}/"
            f"{len(tests)} pruebas de datos pasan en la capa curada vs. {int((crudo['Filas que violan'] == 0).sum())}/{len(tests)} en staging.",
            f"Por estado, la mora va de {estado.imor30.iloc[-1]:.1f} % a {estado.imor30.iloc[0]:.1f} %, pero los IC bootstrap "
            f"se traslapan y la prueba χ² da p = {chi_p:.2f}: "
            + ("no hay evidencia de que el estado importe (los datos sintéticos no tienen efecto por estado). "
               "Un ranking sin intervalos habría llevado a una conclusión falsa." if chi_p > 0.05 else "las diferencias son significativas."),
            f"Roll rates: de 30 días, {cura:.0f} % se cura y {rueda:.0f} % rueda a 60 días al mes siguiente. "
            "Las cosechas muestran cómo madura la mora por meses en libros. Reporte Excel automático con un comando.",
        ],
        "metrics": [
            {"label": "Registros DWH", "value": f"{int(conteos.filas.sum()):,}"},
            {"label": "Cartera vigente", "value": f"${ult.saldo / 1e9:,.2f} mil M"},
            {"label": "IMOR 30+", "value": f"{ult.imor30:.2%}"},
            {"label": "Pruebas de datos", "value": f"{int((tests.Resultado == 'Pasa').sum())} / {len(tests)}"},
        ],
        "charts": [
            line("imor", "Índice de morosidad (IMOR) de la cartera",
                 {"IMOR 30+": [[m, round(v * 100, 3)] for m, v in zip(fmt_mes(mora.fecha_corte), mora.imor30)],
                  "IMOR 90+": [[m, round(v * 100, 3)] for m, v in zip(fmt_mes(mora.fecha_corte), mora.imor90)]},
                 x_label="Corte", y_label="% del saldo", fmt="pct"),
            line("coloc", "Colocación mensual (millones de pesos)",
                 {"Monto colocado": [[m, round(v / 1e6, 1)] for m, v in zip(fmt_mes(coloc.mes), coloc.monto)]},
                 x_label="Mes", y_label="MDP", fmt="num"),
            line("vint", "Cosechas: % del saldo con 30+ días por meses en libros", vint,
                 x_label="Meses desde la originación", y_label="% del saldo en mora 30+", fmt="pct",
                 subtitle="Una línea por trimestre de originación"),
            bars("estado", f"IMOR 30+ por estado al corte {CORTE_FINAL:%m/%Y}", estado.estado, estado.imor30.round(2),
                 fmt="pct", subtitle=f"IC 95 % en la tabla · χ² p = {chi_p:.2f}"),
        ],
        "tables": [table("Pruebas de datos (estilo dbt): capa curada vs. staging", tests),
                   table("Matriz de roll rates (% de créditos que pasan de un tramo a otro en un mes)", rr),
                   table("Mora por estado con IC 95 % bootstrap", est_tbl),
                   table("Penetración de seguros por producto", seg_tbl)],
        "card": model_card(
            "Staging → capa curada (deduplicación, cuarentena, normalización) → modelo estrella → vistas de consumo.",
            ["Pruebas de datos tipo dbt (unicidad, integridad referencial, rangos, conciliación) en cada corrida.",
             "IC bootstrap e independencia χ² antes de comparar estados.",
             "Cosechas y roll rates como validación del comportamiento de la cartera."],
            ["Datos sintéticos con mora generada por una cadena de Markov dependiente del score de buró."],
            ["Sin información real de ninguna institución; el SQL corre en DuckDB y es portable a Snowflake."]),
        "sql": "sql/dwh_financiera.sql",
    }
    export("p16-dwh-snowflake", result)
    con.close()
    return result


if __name__ == "__main__":
    run()