Fundamentos del Riesgo
Crediticio Moderno

# ═══════════════════════════════════════════════════════════════
# MÓDULO 1 · Comparativa de Modelos de Credit Scoring
# Autor: Víctor Miranda | Especialista en Riesgos Financieros
# Ejecutar en Google Colab — requiere no instalación adicional
# ═══════════════════════════════════════════════════════════════

# ── Instalación (ejecutar solo en Colab) ──────────────────────
import subprocess
subprocess.run(["pip", "install", "lightgbm", "xgboost", "--quiet"])

# ── Imports ────────────────────────────────────────────────────
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (roc_auc_score, f1_score,
                                 roc_curve, confusion_matrix)
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import xgboost as xgb
import lightgbm as lgb
import warnings
warnings.filterwarnings('ignore')

np.random.seed(42)

# ═══════════════════════════════════════════════════════════════
# 1. GENERACIÓN DE DATOS SINTÉTICOS
#    Distribuciones calibradas sobre carteras latinoamericanas
#    de crédito de consumo (20-60 meses de plazo)
# ═══════════════════════════════════════════════════════════════

def generar_cartera_sintetica(n: int = 50_000,
                                tasa_default: float = 0.12) -> pd.DataFrame:
    """
    Genera una cartera sintética de crédito de consumo.

    Parámetros:
    -----------
    n             : Número de solicitudes a simular
    tasa_default  : Proporción base de defaults (12% = realista para LATAM)

    Retorna:
    --------
    DataFrame con features crediticios y etiqueta de default
    """
    # ── Características demográficas ───────────────────────────
    edad           = np.random.normal(38, 11, n).clip(18, 75).astype(int)
    ingreso_mensual = np.random.lognormal(8.5, 0.65, n).clip(800, 150_000)

    # ── Historia crediticia ────────────────────────────────────
    meses_en_buro   = np.random.exponential(36, n).clip(0, 240).astype(int)
    num_cuentas_act = np.random.poisson(2.8, n).clip(0, 12)
    max_atraso_hist = np.random.choice([0,0,0,30,60,90,120], n,
                                       p=[0.55,0.10,0.10,0.12,0.07,0.04,0.02])

    # ── Características del crédito solicitado ─────────────────
    monto_solicitado = np.random.lognormal(10.2, 0.8, n).clip(500, 500_000)
    plazo_meses      = np.random.choice([12,24,36,48,60], n,
                                        p=[0.15,0.30,0.30,0.15,0.10])

    # ── Ratios financieros ─────────────────────────────────────
    utilizacion_cc   = np.random.beta(2.0, 3.0, n).clip(0, 1.0)
    dti              = (monto_solicitado / plazo_meses) / ingreso_mensual
    dti              = dti.clip(0, 2.0)

    # ── Variable objetivo: default (etiqueta) ──────────────────
    # Modelo logístico sintético con señal realista:
    log_odds = (
        -3.5
        + 0.025 * max_atraso_hist / 30
        + 1.2  * utilizacion_cc
        + 2.0  * dti
        - 0.015 * np.log1p(meses_en_buro)
        - 0.008 * np.log1p(ingreso_mensual)
        + np.random.normal(0, 0.5, n)  # ruido irreducible
    )
    prob_default = 1 / (1 + np.exp(-log_odds))
    default      = (np.random.uniform(0, 1, n) < prob_default).astype(int)

    # ── Ajustar tasa de default al target especificado ─────────
    actual_rate = default.mean()
    if abs(actual_rate - tasa_default) > 0.01:
        umbral = np.percentile(prob_default, (1 - tasa_default) * 100)
        default = (prob_default >= umbral).astype(int)

    df = pd.DataFrame({
        'edad'            : edad,
        'ingreso_mensual'  : ingreso_mensual.round(2),
        'meses_en_buro'   : meses_en_buro,
        'num_cuentas_act'  : num_cuentas_act,
        'max_atraso_hist'  : max_atraso_hist,
        'monto_solicitado': monto_solicitado.round(2),
        'plazo_meses'      : plazo_meses,
        'utilizacion_cc'   : utilizacion_cc.round(4),
        'dti'              : dti.round(4),
        'default'          : default
    })
    return df

# Generar datos
df = generar_cartera_sintetica(n=50_000, tasa_default=0.12)
print(f"Cartera generada: {len(df):,} registros")
print(f"Tasa de default: {df['default'].mean():.2%}")
print(f"\nPrimeras 5 filas:\n{df.head()}")

# ═══════════════════════════════════════════════════════════════
# 2. PARTICIÓN TEMPORAL (crítica en crédito — sin data leakage)
# ═══════════════════════════════════════════════════════════════
features = ['edad', 'ingreso_mensual', 'meses_en_buro',
            'num_cuentas_act', 'max_atraso_hist',
            'monto_solicitado', 'plazo_meses',
            'utilizacion_cc', 'dti']
target = 'default'

# Últimos 20% como test (simulación temporal)
split_idx = int(len(df) * 0.8)
X_train = df.iloc[:split_idx][features]
X_test  = df.iloc[split_idx:][features]
y_train = df.iloc[:split_idx][target]
y_test  = df.iloc[split_idx:][target]

print(f"Train: {len(X_train):,} | Test: {len(X_test):,}")
print(f"Default rate — Train: {y_train.mean():.2%} | Test: {y_test.mean():.2%}")

# ═══════════════════════════════════════════════════════════════
# 3. ENTRENAMIENTO DE MODELOS
# ═══════════════════════════════════════════════════════════════

def ks_statistic(y_true, y_prob) -> float:
    """Calcula el estadístico KS (Kolmogorov-Smirnov) para scoring crediticio."""
    from scipy import stats
    buenos  = y_prob[y_true == 0]
    malos   = y_prob[y_true == 1]
    ks_stat = stats.ks_2samp(buenos, malos).statistic
    return ks_stat

def gini_coefficient(auc: float) -> float:
    """Gini = 2*AUC - 1  (normalización estándar en banca)."""
    return 2 * auc - 1

def evaluar_modelo(nombre: str, modelo, X_tr, y_tr, X_te, y_te) -> dict:
    """Entrena, predice y calcula métricas estándar de scoring."""
    import time
    t0 = time.perf_counter()
    modelo.fit(X_tr, y_tr)
    t1 = time.perf_counter()

    y_prob = modelo.predict_proba(X_te)[:, 1]

    auc  = roc_auc_score(y_te, y_prob)
    ks   = ks_statistic(y_te, y_prob)
    gini = gini_coefficient(auc)
    f1   = f1_score(y_te, (y_prob > 0.5).astype(int))

    # Latencia de inferencia (1000 predicciones)
    t_inf_0 = time.perf_counter()
    for _ in range(1000):
        modelo.predict_proba(X_te.iloc[:1])
    latency_ms = (time.perf_counter() - t_inf_0) * 1000 / 1000

    print(f"{nombre:30s} AUC={auc:.4f}  KS={ks:.4f}  Gini={gini:.4f}  "
          f"F1={f1:.4f}  Latencia={latency_ms:.1f}ms")
    return {'nombre': nombre, 'modelo': modelo,
            'auc': auc, 'ks': ks, 'gini': gini,
            'f1': f1, 'latency': latency_ms, 'y_prob': y_prob}

# ── Modelos ────────────────────────────────────────────────
resultados = []

# 1. Regresión Logística (con escalado)
lr = Pipeline([
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression(C=0.1, max_iter=500, class_weight='balanced'))
])
resultados.append(evaluar_modelo("Regresión Logística", lr,
                                  X_train, y_train, X_test, y_test))

# 2. Random Forest
rf = RandomForestClassifier(n_estimators=500, max_depth=8,
                             class_weight='balanced', n_jobs=-1, random_state=42)
resultados.append(evaluar_modelo("Random Forest (500 árboles)", rf,
                                  X_train, y_train, X_test, y_test))

# 3. XGBoost
xgb_model = xgb.XGBClassifier(
    n_estimators=200, learning_rate=0.05, max_depth=6,
    subsample=0.8, colsample_bytree=0.8,
    scale_pos_weight=y_train.value_counts()[0]/y_train.value_counts()[1],
    use_label_encoder=False, eval_metric='auc',
    verbosity=0, random_state=42
)
resultados.append(evaluar_modelo("XGBoost (200 árboles)", xgb_model,
                                  X_train, y_train, X_test, y_test))

# 4. LightGBM
lgb_model = lgb.LGBMClassifier(
    n_estimators=300, learning_rate=0.05, num_leaves=63,
    subsample=0.8, colsample_bytree=0.8,
    class_weight='balanced', random_state=42, verbose=-1
)
resultados.append(evaluar_modelo("LightGBM (300 árboles)", lgb_model,
                                  X_train, y_train, X_test, y_test))

# ═══════════════════════════════════════════════════════════════
# VINTAGE ANALYSIS — Módulo 1, Sección 1.3
# Autor: Víctor Miranda
# ═══════════════════════════════════════════════════════════════

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
from datetime import datetime, timedelta

np.random.seed(2024)

def simular_vintages(
    num_vintages: int = 24,
    creditos_por_vintage: int = 800,
    max_meses_vida: int = 24
) -> pd.DataFrame:
    """
    Simula datos de vintage analysis con tres patrones realistas:
    - Vintages estables (año 1)
    - Vintages en deterioro (año 2 — política crediticia más laxa)
    - Shock macro en el mes 18 (incremento generalizado)
    """
    registros = []
    fecha_base = pd.Timestamp('2022-01-01')

    for v in range(num_vintages):
        fecha_origen = fecha_base + pd.DateOffset(months=v)

        # Calidad del vintage: los últimos 12 son de menor calidad
        calidad = 1.0 if v < 12 else 1.35  # 35% más riesgo en vintages recientes

        for _ in range(creditos_por_vintage):
            # Probabilidad base de default mensual (hazard rate)
            hazard_base = np.random.beta(1.5, 18) * calidad

            mes_default = None
            for mes in range(1, max_meses_vida + 1):
                fecha_obs = fecha_origen + pd.DateOffset(months=mes)

                # Shock macro: a partir de 2023-07 todas las PD suben 50%
                factor_macro = 1.5 if fecha_obs >= pd.Timestamp('2023-07-01') else 1.0

                # Hazard rate aumenta ligeramente con la madurez (forma de campana)
                hazard_mensual = hazard_base * factor_macro * (1 + 0.02 * mes)

                if np.random.uniform() < hazard_mensual:
                    mes_default = mes
                    break

            registros.append({
                'vintage'    : fecha_origen.strftime('%Y-%m'),
                'mes_default': mes_default
            })

    return pd.DataFrame(registros)

# Simular
df_v = simular_vintages()

def calcular_cdr(df: pd.DataFrame, max_meses: int = 24) -> pd.DataFrame:
    """Calcula Cumulative Default Rate por vintage y mes de vida."""
    vintages = df['vintage'].unique()
    resultados = []

    for v in vintages:
        sub = df[df['vintage'] == v]
        n_total = len(sub)

        for m in range(1, max_meses + 1):
            n_defaults = sub['mes_default'].dropna()
            cdr = (n_defaults <= m).sum() / n_total

            resultados.append({
                'vintage'     : v,
                'mes_vida'    : m,
                'cdr'         : cdr,
                'n_creditos'  : n_total
            })

    return pd.DataFrame(resultados)

cdr_df = calcular_cdr(df_v)

# ── Visualización ──────────────────────────────────────────────
fig, axes = plt.subplots(1, 2, figsize=(16, 7))
fig.patch.set_facecolor('#FAF7F0')

vintages_list = sorted(cdr_df['vintage'].unique())

# Paleta: vintages estables=azul oscuro, recientes=naranja/rojo
colors_stable = plt.cm.Blues(np.linspace(0.4, 0.9, 12))
colors_risky  = plt.cm.Oranges(np.linspace(0.4, 0.9, 12))

ax = axes[0]
ax.set_facecolor('#FAF7F0')

for i, v in enumerate(vintages_list):
    sub = cdr_df[cdr_df['vintage'] == v]
    color = colors_stable[i] if i < 12 else colors_risky[i - 12]
    lw    = 1.5 if i not in [0, 11, 12, 23] else 2.5
    ax.plot(sub['mes_vida'], sub['cdr'] * 100,
            color=color, linewidth=lw, alpha=0.85)

ax.set_xlabel('Mes de Vida del Crédito', fontsize=11)
ax.set_ylabel('CDR — Tasa de Default Acumulada (%)', fontsize=11)
ax.set_title('Vintage Analysis: CDR por Cohorte\n(azul=estables, naranja=deterioro)',
             fontsize=12, fontweight='bold')
ax.yaxis.set_major_formatter(mticker.FormatStrFormatter('%.1f%%'))
ax.grid(True, alpha=0.3, linestyle='--')

plt.tight_layout()
plt.savefig('vintage_analysis.png', dpi=150, bbox_inches='tight')
plt.show()
print("✓ Gráfico de vintage analysis generado")

# ═══════════════════════════════════════════════════════════════
# SIMULACIÓN DE ORQUESTADOR DE DECISIÓN CREDITICIA
# Módulo 1.4 — Ecosistema FinTech
# Autor: Víctor Miranda
#
# Este código simula un orquestador de decisión crediticia
# real-time con llamadas asíncronas a APIs externas (simuladas).
# Ejecutar directamente en Google Colab.
# ═══════════════════════════════════════════════════════════════

import asyncio
import json
import random
import time
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from enum import Enum

# ── Tipos de datos ─────────────────────────────────────────────

class DecisionResult(Enum):
    APPROVED    = "APPROVED"
    DECLINED    = "DECLINED"
    MANUAL_REV  = "MANUAL_REVIEW"

@dataclass
class CreditApplication:
    app_id: str
    curp: str
    monto_solicitado: float
    plazo_meses: int
    ingreso_declarado: float

@dataclass
class DecisionContext:
    application: CreditApplication
    bureau_score: Optional[int]    = None
    bureau_data: Optional[Dict]    = None
    open_banking: Optional[Dict]   = None
    ml_score: Optional[float]      = None
    decision: Optional[str]        = None
    monto_aprobado: Optional[float]= None
    tasa_asignada: Optional[float] = None
    latency_ms: Dict[str, float]   = field(default_factory=dict)
    errors: list                   = field(default_factory=list)

# ── Simulación de APIs externas ────────────────────────────────

async def call_bureau_api(curp: str) -> Dict[str, Any]:
    """Simula llamada a buró de crédito (~800ms latencia típica)."""
    await asyncio.sleep(0.8 + random.uniform(-0.2, 0.4))

    # Score sintético basado en hash del CURP (reproducible)
    seed = sum(ord(c) for c in curp)
    random.seed(seed)
    score = random.randint(550, 810)
    random.seed()  # reset seed

    return {
        'score'          : score,
        'max_atraso_hist': random.choice([0, 0, 0, 30, 60]),
        'num_cuentas_act': random.randint(1, 6),
        'meses_en_buro'  : random.randint(6, 120),
        'consultas_6m'   : random.randint(0, 5),
        'fuente'         : 'CIRCULO_CREDITO'
    }

async def call_open_banking_api(curp: str) -> Dict[str, Any]:
    """Simula llamada a Open Banking/Belvo (~1.2s latencia)."""
    await asyncio.sleep(1.2 + random.uniform(-0.3, 0.5))
    return {
        'ingreso_promedio_3m': random.uniform(8000, 45000),
        'saldo_promedio'     : random.uniform(500, 25000),
        'num_transacciones'  : random.randint(15, 80),
        'volatilidad_ingreso': random.uniform(0.05, 0.35),
        'fuente'             : 'BELVO'
    }

async def call_ml_scoring_service(features: Dict) -> float:
    """Simula inferencia del modelo ML en el feature store (<50ms)."""
    await asyncio.sleep(0.045)  # 45ms — modelo LightGBM servido via ONNX

    # Modelo sintético: función determinista de los features
    score = features.get('bureau_score', 650) / 850
    score -= features.get('dti', 0.3) * 0.4
    score -= features.get('max_atraso_hist', 0) / 300
    score += features.get('meses_en_buro', 24) / 1000
    score  = max(0.02, min(0.98, score + random.uniform(-0.05, 0.05)))
    return round(score, 4)

# ── Motor de decisión con política crediticia ─────────────────

def aplicar_politica_credito(ctx: DecisionContext) -> DecisionContext:
    """
    Aplica reglas de política crediticia sobre el contexto enriquecido.
    Combina score ML con reglas regulatorias y de negocio.
    """
    score = ctx.ml_score
    bureau = ctx.bureau_data
    ob = ctx.open_banking
    app = ctx.application

    # ── REGLAS DE RECHAZO AUTOMÁTICO (KO rules) ───────────────
    if bureau['max_atraso_hist'] >= 90:
        ctx.decision = DecisionResult.DECLINED.value
        ctx.monto_aprobado = 0
        return ctx

    if score < 0.25:  # PD implícita > 75%: rechazo automático
        ctx.decision = DecisionResult.DECLINED.value
        ctx.monto_aprobado = 0
        return ctx

    # ── REVISIÓN MANUAL ───────────────────────────────────────
    if 0.25 <= score < 0.40:
        ctx.decision = DecisionResult.MANUAL_REV.value
        ctx.monto_aprobado = app.monto_solicitado * 0.5
        ctx.tasa_asignada = 0.48  # 48% anual
        return ctx

    # ── APROBACIÓN: calcular monto y tasa ────────────────────
    ingreso_verificado = ob['ingreso_promedio_3m'] if ob else app.ingreso_declarado
    dti_cuota = (app.monto_solicitado / app.plazo_meses) / ingreso_verificado
    monto_max_dti = ingreso_verificado * 0.35 * app.plazo_meses  # DTI máx 35%
    monto_aprobado = min(app.monto_solicitado, monto_max_dti)

    # Tasa por riesgo: escala entre 18% y 42% anual
    tasa = 0.18 + (1 - score) * 0.35

    ctx.decision = DecisionResult.APPROVED.value
    ctx.monto_aprobado = round(monto_aprobado, 2)
    ctx.tasa_asignada = round(tasa, 4)
    return ctx

# ── Orquestador principal ─────────────────────────────────────

async def process_credit_application(app: CreditApplication) -> DecisionContext:
    """Orquesta el flujo completo de decisión crediticia en tiempo real."""
    ctx = DecisionContext(application=app)
    t_start = time.perf_counter()

    # ── ETAPA 1: Enriquecimiento de datos (paralelo) ───────────
    t1 = time.perf_counter()
    bureau_task = asyncio.create_task(call_bureau_api(app.curp))
    ob_task     = asyncio.create_task(call_open_banking_api(app.curp))

    # Esperamos ambas APIs en paralelo (la más lenta dicta el tiempo)
    bureau_data, ob_data = await asyncio.gather(bureau_task, ob_task)
    ctx.latency_ms['data_enrichment'] = (time.perf_counter() - t1) * 1000

    ctx.bureau_score = bureau_data['score']
    ctx.bureau_data  = bureau_data
    ctx.open_banking = ob_data

    # ── ETAPA 2: Inferencia ML ────────────────────────────────
    t2 = time.perf_counter()
    features = {
        'bureau_score'  : bureau_data['score'],
        'max_atraso_hist': bureau_data['max_atraso_hist'],
        'meses_en_buro' : bureau_data['meses_en_buro'],
        'dti'           : (app.monto_solicitado / app.plazo_meses) / ob_data['ingreso_promedio_3m'],
        'volatilidad'   : ob_data['volatilidad_ingreso']
    }
    ctx.ml_score = await call_ml_scoring_service(features)
    ctx.latency_ms['ml_inference'] = (time.perf_counter() - t2) * 1000

    # ── ETAPA 3: Política crediticia ──────────────────────────
    t3 = time.perf_counter()
    ctx = aplicar_politica_credito(ctx)
    ctx.latency_ms['policy_engine'] = (time.perf_counter() - t3) * 1000

    ctx.latency_ms['total'] = (time.perf_counter() - t_start) * 1000
    return ctx

# ── Ejecutar simulación ───────────────────────────────────────

async def main():
    apps = [
        CreditApplication("APP-001", "MIRA850101HDFRNC08", 25000, 24, 18000),
        CreditApplication("APP-002", "ROGL920315MDFDRZ01", 80000, 36, 35000),
        CreditApplication("APP-003", "HERN751220HDFRNR09", 5000,  12, 9000),
    ]

    print(f"\n{'='*65}")
    print(f"{'SIMULADOR DE DECISIÓN CREDITICIA EN TIEMPO REAL':^65}")
    print(f"{'Módulo 1 — Víctor Miranda':^65}")
    print(f"{'='*65}\n")

    for app in apps:
        ctx = await process_credit_application(app)
        print(f"📋 Solicitud: {app.app_id}")
        print(f"   Monto solicitado : ${app.monto_solicitado:,.0f} MXN")
        print(f"   Score de buró    : {ctx.bureau_score}")
        print(f"   ML Score (Proba) : {ctx.ml_score:.4f}")
        print(f"   DECISIÓN         : {ctx.decision}")
        if ctx.monto_aprobado:
            print(f"   Monto aprobado   : ${ctx.monto_aprobado:,.0f} MXN")
            print(f"   Tasa asignada    : {ctx.tasa_asignada:.1%} anual")
        print(f"   Latencias:")
        for k, v in ctx.latency_ms.items():
            print(f"     {k:25s}: {v:6.1f} ms")
        print()

# Ejecutar en Colab/Jupyter
await main()