"""MP1 · paso 32: test ex ante dinámico (pre-registro §12 (6)) para (a), (b) y (c). Compara, para los lotes de las pruebas
(anuncio 2025-01-01…2026-04-30), las columnas brutas que leen las variables en lotes.csv y en lotes_perturbado.csv (todo lo
posterior al anuncio de los lotes de 2026 alterado: nº de ofertas, resultado, fecha y año de adjudicación, modificaciones de
2026). Además, el test estático: ninguna columna leída está en la lista negra. Imagen bquantagents-worker.
"""
import json, sys
import numpy as np, pandas as pd
ANIO = int(sys.argv[1])
VENT = {2026: ("2026-01-01", "2026-04-30"), 2025: ("2025-01-01", "2025-04-15")}[ANIO]

NEGRA = {"num_ofertas", "imp_adj_eur", "baja", "adjudicatario_nombre", "adjudicatario_nif", "es_pyme", "ofertas_anormales",
         "oferta_min_eur", "oferta_max_eur", "fecha_adj", "fecha_adj_es", "fecha_formalizacion", "estado", "estado_native", "unidad",
         "anio_ref", "fpub_ok", "plazo_fable", "desierto", "des", "desistido", "adjudicado", "resuelto", "dias_res", "y_mod", "y_mod10", "y"}
inf = {}
leidas = set()
for h in ("ou", "des", "mod"):
    leidas |= set(json.load(open(f"/w/salidas/20_{h}.json"))["columnas_leidas"])
inf["estatico_columnas_leidas"] = sorted(leidas)
inf["estatico_en_lista_negra"] = sorted(leidas & NEGRA)
k = ["licitacion_id", "lote_u"]
dt = {c: "string" for c in k + ["organo_clave", "cpv_cel", "cpv_div", "proc_nat", "tram", "tipo_contrato", "tipo_organo", "ccaa"]}
A = pd.read_csv("/w/interno/lotes.csv", dtype=dt, parse_dates=["fecha_pub"], low_memory=False)
B = pd.read_csv(f"/w/interno/lotes_perturbado_{ANIO}.csv", dtype=dt, parse_dates=["fecha_pub"], low_memory=False)
# unidades de la prueba del año (las de P: anuncio con plazo y fecha_pub <= fecha_adj, en la ventana)
sel = lambda X: X[(X["conjunto"] == "hist") & (X["fecha_pub"] >= VENT[0]) & (X["fecha_pub"] <= VENT[1]) & X["fpub_ok"].astype(str).str.lower().eq("true")]
A, B = sel(A), sel(B)
M = A.merge(B, on=k, suffixes=("_a", "_b"))
inf["anio"] = ANIO; inf["lotes_comparados"] = int(len(M))
cols = sorted(c for c in leidas if f"{c}_a" in M)
dif = {}
for c in cols:
    a, b = M[f"{c}_a"], M[f"{c}_b"]
    if pd.api.types.is_numeric_dtype(a) and pd.api.types.is_numeric_dtype(b):
        ne = ~((a == b) | (a.isna() & b.isna()) | (np.abs(a.astype(float) - b.astype(float)) <= 1e-9))
    else:
        ne = ~((a.astype(str) == b.astype(str)))
    dif[c] = int(ne.sum())
inf["diferencias_por_columna"] = dif
inf["total_diferencias"] = int(sum(dif.values()))
# control positivo: las columnas de desenlace SÍ cambian en los lotes de 2026 (si no, la perturbación no se aplicó)
inf["control_positivo_num_ofertas_cambia"] = int((M["num_ofertas_a"].fillna(-1) != M["num_ofertas_b"].fillna(-1)).sum())
inf["pasa"] = inf["total_diferencias"] == 0 and not inf["estatico_en_lista_negra"] and inf["control_positivo_num_ofertas_cambia"] > 0
json.dump(inf, open(f"/w/salidas/32_test_dinamico_{ANIO}.json", "w"), indent=1)
print(json.dumps({x: inf[x] for x in ("anio", "lotes_comparados", "total_diferencias", "estatico_en_lista_negra",
                                      "control_positivo_num_ofertas_cambia", "pasa")}))