"""MP1 · paso 22: desenlace (d), recurso estimado contra el pliego (pre-registro §8, §12 d). Imagen bquantagents-worker.
Lee /w/interno/recurso.csv y /w/salidas/12_recurso.json; escribe /w/interno/mp1_rec/ y /w/salidas/22_rec.json.
"""
import sys, json, os, time
import numpy as np, pandas as pd
import sklearn
from scipy.special import logit, expit
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GroupKFold
import statsmodels.api as sm

T0 = time.time(); SEM = 20261026; B = 2000
OUT = "/w/interno/mp1_rec"; os.makedirs(OUT, exist_ok=True)
d = pd.read_csv("/w/interno/recurso.csv", dtype={"licitacion_id": "string", "organo_clave": "string", "tipo_organo": "string",
                "ccaa": "string", "cpv_div": "string", "proc_nat": "string", "tram": "string", "tipo_contrato": "string",
                "tribunal": "string", "sentido": "string"}, parse_dates=["fecha_pub", "ffo", "fecha_res"], low_memory=False)
i12 = json.load(open("/w/salidas/12_recurso.json"))
frac = {k: v[2] for k, v in i12["fraccion_muestreo"].items()}
for c in ["sara", "ue", "suman100", "pcap", "ppt"]:
    d[c] = d[c].map({True: True, False: False, "true": True, "false": False})
d["ventana"] = np.where(d["fecha_pub"] < pd.Timestamp("2026-01-01"), "train", "test")
def peso(row):
    if row["y"] != 1:
        return 1.0
    s = frac.get(f"{row['tribunal']}|{row['sentido']}|{row['fecha_res'].year}") if pd.notna(row["fecha_res"]) else None
    return 1.0 / s if s else 1.0
d["w"] = d.apply(peso, axis=1)
d["mes_pub"] = d["fecha_pub"].dt.month
TR = (d["ventana"] == "train").to_numpy(); TE = ~TR
inf = {"versiones": {"sklearn": sklearn.__version__, "numpy": np.__version__, "pandas": pd.__version__},
       "n": {"train": int(TR.sum()), "test": int(TE.sum())},
       "positivos": {"train": int(d.loc[TR, "y"].sum()), "test": int(d.loc[TE, "y"].sum())},
       "positivos_ponderados": {"train": float((d["w"] * d["y"])[TR].sum()), "test": float((d["w"] * d["y"])[TE].sum())},
       "banco_casa": i12["banco_casa"]}
bc = i12["banco_casa"]; inf["banco_casa_ok"] = bc["en_unified_20261025"] / bc["casadas"] >= 0.90
print(inf, flush=True)

def num(x):
    f = pd.DataFrame(index=x.index)
    # PBL del expediente (a nivel de licitación; el de la fila de un lote solo existe si hay resultado: fuga, corrección 26-oct)
    pbl = x["pbl_exp"].fillna(x["pbl_lote"]); ve = x["ve_exp"]
    f["log_pbl"] = np.log10(np.maximum(pbl.fillna(1), 1)); f["pbl_na"] = pbl.isna().astype(float)
    f["log_ve"] = np.log10(np.maximum(ve.fillna(1), 1)); f["ve_na"] = ve.isna().astype(float)
    f["dur"] = np.where(x["dur"].isna(), -1, np.minimum(x["dur"].fillna(0), 120)); f["dur_na"] = x["dur"].isna().astype(float)
    f["plazo"] = np.minimum(x["plazo"].fillna(-1), 120); f["plazo_lt15"] = (x["plazo"] < 15).astype(float)
    f["log_nlotes"] = np.log1p(x["n_lotes_an"].fillna(1))   # sin «con_lotes»: no estaba en §4 y depende del resultado
    f["sara"] = x["sara"].fillna(False).astype(float); f["sara_na"] = x["sara"].isna().astype(float)
    f["ue"] = x["ue"].fillna(False).astype(float)
    f["cpv_n_div"] = x["cpv_n_div"].fillna(-1)
    ok_tipo = x["suman100"].fillna(False) & (x["n_sin_tipo"].fillna(1) == 0)
    ps = x["peso_suma"].replace(0, np.nan)
    f["sin_crit"] = x["n_crit"].isna().astype(float); f["n_crit"] = np.minimum(x["n_crit"].fillna(0), 15)
    f["un_solo_crit"] = (x["n_crit"] == 1).astype(float)
    f["peso_precio"] = np.where(ok_tipo, (x["peso_precio"].fillna(0) / ps).fillna(0), -1)
    pjv = np.where(ok_tipo, (x["peso_jv"].fillna(0) / ps).fillna(0), -1)
    f["peso_jv"] = pjv; f["peso_jv_gt50"] = (pjv > 0.5).astype(float)
    f["pcap"] = x["pcap"].fillna(False).astype(float); f["ppt"] = x["ppt"].fillna(False).astype(float)
    return f
F = num(d); NUM = list(F.columns)
CATS = ["proc_nat", "tram", "tipo_contrato", "cpv_div", "tipo_organo", "ccaa", "mes_pub"]
CAT_NIV = {}; C = pd.DataFrame(index=d.index)
for c in CATS:
    v = d[c].astype("string").fillna("∅")
    cnt = v[TR].value_counts(); niv = sorted([x for x, n in cnt.items() if n >= 30]); CAT_NIV[c] = niv
    C[c] = v.where(v.isin(niv), "otros")
MU = F.loc[TR].mean().to_numpy(); SD = F.loc[TR].std(ddof=0).to_numpy(); SD[SD == 0] = 1
def X(mask):
    Z = (F.loc[mask, NUM].to_numpy(float) - MU) / SD
    Dm = [np.column_stack([(C.loc[mask, c].to_numpy() == lv).astype(float) for lv in CAT_NIV[c] + ["otros"]]) for c in CATS]
    return np.column_stack([Z] + Dm)
y = d["y"].to_numpy(); w = d["w"].to_numpy()
Xtr, Xte = X(TR), X(TE)
grp = d["organo_clave"].fillna("__" + d["licitacion_id"]).to_numpy()
def ajusta(Xa, ya, wa):
    return LogisticRegression(C=0.1, solver="lbfgs", max_iter=3000).fit(Xa, ya, sample_weight=wa)
m = ajusta(Xtr, y[TR], w[TR])
# Platt con validación cruzada 5 por órgano dentro del entrenamiento
oof = np.zeros(TR.sum())
for a, b in GroupKFold(n_splits=5).split(Xtr, y[TR], grp[TR]):
    mm = ajusta(Xtr[a], y[TR][a], w[TR][a]); oof[b] = mm.predict_proba(Xtr[b])[:, 1]
platt = LogisticRegression(C=1e6).fit(logit(np.clip(oof, 1e-6, 1 - 1e-6))[:, None], y[TR], sample_weight=w[TR])
def cal(p): return platt.predict_proba(logit(np.clip(p, 1e-6, 1 - 1e-6))[:, None])[:, 1]
p_raw = m.predict_proba(Xte)[:, 1]; p = cal(p_raw)
# B3: solo importe
xi = lambda mask: np.log10(np.maximum(d.loc[mask, "ve_exp"].fillna(d.loc[mask, "pbl_exp"]).fillna(1), 1)).to_numpy()[:, None]
b3 = LogisticRegression(C=1e6).fit(xi(TR), y[TR], sample_weight=w[TR]).predict_proba(xi(TE))[:, 1]
# intervalos: 50 reajustes por bootstrap de órganos del entrenamiento (informativo)
rng = np.random.default_rng(SEM)
gtr = pd.factorize(grp[TR])[0]; G = gtr.max() + 1
PB = np.empty((TE.sum(), 50))
for bb in range(50):
    mult = np.bincount(rng.integers(0, G, G), minlength=G)[gtr].astype(float)
    mb = ajusta(Xtr, y[TR], w[TR] * mult); PB[:, bb] = cal(mb.predict_proba(Xte)[:, 1])
te = d.loc[TE, ["licitacion_id", "organo_clave", "ccaa", "tipo_contrato", "fecha_pub", "y", "w", "tribunal", "sentido"]].copy()
te["p"] = p; te["p_raw"] = p_raw; te["b3"] = b3; te["p_lo"] = np.quantile(PB, 0.05, axis=1); te["p_hi"] = np.quantile(PB, 0.95, axis=1)
te.to_csv(f"{OUT}/predicciones.csv", index=False)
import pickle
with open(f"{OUT}/modelo_d_v2.pkl", "wb") as fh:   # modelo congelado para la repetición prospectiva 2026-S2 (prereg/DEFINICIONES_MP1_D_2026S2.md)
    pickle.dump({"modelo": m, "platt": platt, "NUM": NUM, "CATS": CATS, "CAT_NIV": CAT_NIV, "MU": MU, "SD": SD,
                 "versiones": {"sklearn": sklearn.__version__, "numpy": np.__version__, "pandas": pd.__version__}}, fh)
if os.path.exists("/w/interno/recurso_pred.csv"):
    q = pd.read_csv("/w/interno/recurso_pred.csv", dtype={"licitacion_id": "string", "organo_clave": "string", "tipo_organo": "string",
                    "ccaa": "string", "cpv_div": "string", "proc_nat": "string", "tram": "string", "tipo_contrato": "string"},
                    parse_dates=["fecha_pub", "ffo"], low_memory=False)
    for c in ["sara", "ue", "suman100", "pcap", "ppt"]:
        q[c] = q[c].map({True: True, False: False, "true": True, "false": False})
    q["mes_pub"] = q["fecha_pub"].dt.month
    Fq = num(q); Cq = pd.DataFrame(index=q.index)
    for c in CATS:
        v = q[c].astype("string").fillna("∅"); Cq[c] = v.where(v.isin(CAT_NIV[c]), "otros")
    Zq = (Fq[NUM].to_numpy(float) - MU) / SD
    Xq = np.column_stack([Zq] + [np.column_stack([(Cq[c].to_numpy() == lv).astype(float) for lv in CAT_NIV[c] + ["otros"]]) for c in CATS])
    q["p"] = cal(m.predict_proba(Xq)[:, 1])
    q[["licitacion_id", "organo_clave", "ccaa", "tipo_contrato", "fecha_pub", "ffo", "ve_exp", "p"]].to_csv(f"{OUT}/prediccion_en_plazo.csv", index=False)

# métricas (implementación 1): AUC ponderada con bootstrap por órgano
def auc_w(s, yy, ww):
    u, g = np.unique(s, return_inverse=True); Gn = len(u)
    wp = np.bincount(g, ww * yy, Gn); wn = np.bincount(g, ww * (1 - yy), Gn); below = np.cumsum(wn) - wn
    return float(np.sum(wp * (below + 0.5 * wn)) / (wp.sum() * wn.sum()))
yt = te["y"].to_numpy(float); wt = te["w"].to_numpy(float)
cl = pd.factorize(te["organo_clave"].fillna("__" + te["licitacion_id"]))[0]; Cn = cl.max() + 1
pt = {"p": auc_w(p, yt, wt), "p_raw": auc_w(p_raw, yt, wt), "b3": auc_w(b3, yt, wt)}
rep = {k: [] for k in pt}
rng = np.random.default_rng(SEM)
for bb in range(B):
    mult = np.bincount(rng.integers(0, Cn, Cn), minlength=Cn)[cl].astype(float)
    if (yt * mult).sum() == 0:
        continue
    for k, s in (("p", p), ("p_raw", p_raw), ("b3", b3)):
        rep[k].append(auc_w(s, yt, wt * mult))
rep = {k: np.array(v) for k, v in rep.items()}
ic = {k: [float(np.percentile(v, 2.5)), float(np.percentile(v, 97.5))] for k, v in rep.items()}
dd = rep["p"] - rep["b3"]
o_e = float((wt * yt).sum() / p.sum())
sl = sm.GLM(yt, sm.add_constant(logit(np.clip(p, 1e-6, 1 - 1e-6))), family=sm.families.Binomial(), freq_weights=wt).fit()
q = pd.qcut(p, 10, duplicates="drop")
dec = pd.DataFrame({"p": p, "y": yt * wt, "q": q}).groupby("q", observed=True).agg(n=("p", "size"), E=("p", "sum"), O=("y", "sum")).reset_index(drop=True)
inf.update({"auc": pt, "auc_ic": ic, "delta_b3": {"delta": pt["p"] - pt["b3"], "ic": [float(np.percentile(dd, 2.5)), float(np.percentile(dd, 97.5))]},
            "o_e": o_e, "pendiente": float(sl.params[1]), "deciles": dec.to_dict(orient="records"),
            "intervalo_ancho_mediano": float(np.median(te["p_hi"] - te["p_lo"])), "tasa_test_ponderada": float((wt * yt).sum() / len(yt)),
            "coeficientes": dict(zip(NUM + [f"{c}={lv}" for c in CATS for lv in CAT_NIV[c] + ["otros"]], map(float, m.coef_[0])))})
pot = inf["positivos"]["test"] >= 30 and inf["positivos"]["train"] >= 50 and inf["banco_casa_ok"]
if not pot:
    ver = "NO CONFIRMADA (sin potencia o banco sin casar)"
elif not (pt["p"] >= 0.65 and ic["p"][0] > 0.55 and inf["delta_b3"]["delta"] >= 0.02 and inf["delta_b3"]["ic"][0] > 0):
    ver = "NO PASA"
elif not (0.67 <= o_e <= 1.5):
    ver = "NO CONFIRMADA (calibración)"
else:
    ver = "PASA a falta de 7-8"
inf["veredicto_provisional"] = ver; inf["segundos"] = round(time.time() - T0, 1)
json.dump(inf, open("/w/salidas/22_rec.json", "w"), indent=1, ensure_ascii=False, default=float)
print(json.dumps({k: inf[k] for k in ("n", "positivos", "auc", "auc_ic", "delta_b3", "o_e", "pendiente", "veredicto_provisional")},
                 ensure_ascii=False, default=float, indent=1))