"""07. SEGUNDA IMPLEMENTACIÓN (independiente), escrita desde DEFINICIONES_PREVIAS.md sin reutilizar 02/03/05/06.
Entrada: solo la base b (01_base.sql: lotes PLACSP con identidad), la etiqueta lab (c06 de P8), la exclusión excl y
ute_socios. Todo lo demás (población, PBL fiable, mercados, screens, percentiles, compuestos, AUC, bootstrap) se recalcula
aquí en Python puro con otros algoritmos: Gini por la fórmula ordenada, AUC por conteo explícito de pares, bootstrap con
bisect sobre pesos acumulados. Salida: interno/sc_impl2.parquet (pyarrow) e interno/07_impl2.json."""
import duckdb, re, unicodedata, random, bisect, json, time, statistics
import pyarrow as pa, pyarrow.parquet as pq
T0 = time.time()
con = duckdb.connect('/c/interno/col.duckdb', read_only=True)
cols = ['id_canonical', 'licitacion_id', 'organo_clave', 'organo_nombre', 'tipo', 'cpv', 'anio', 'tipo_contrato',
        'persona_fisica', 'num_ofertas', 'mn', 'mx', 'adj', 'pbl', 'rep', 'empresa_clave', 'adjudicatario_nombre', 'nuts',
        'provincia', 'fref']
B = con.execute(f"SELECT {', '.join(cols)} FROM b").fetchall()
I = {c: i for i, c in enumerate(cols)}
lab = con.execute("SELECT exp, var, id_canonical FROM lab").fetchall()
excl = {r[0] for r in con.execute("SELECT id_canonical FROM excl").fetchall()}
uso = con.execute("SELECT ute_clave, socio_nif, socio_persona_fisica FROM read_parquet('/d/unified_20261018.identidad/ute_socios.parquet')").fetchall()
print('leído', len(B), round(time.time() - T0), flush=True)
OSS = {'Obras', 'Servicios', 'Suministros'}

# número de adjudicaciones por licitación (para el PBL fiable)
nadj = {}
for r in B:
    nadj[r[I['licitacion_id']]] = nadj.get(r[I['licitacion_id']], 0) + 1


def pbl_fiable(r):
    p = r[I['pbl']]
    if p is not None and p > 0 and (r[I['rep']] == 'por_lote' or nadj[r[I['licitacion_id']]] == 1):
        return p
    return None


# ---- población P (orden de exclusión de §1)
P = []
embudo = {}
for r in B:
    a = r[I['anio']]
    mn, mx, adj = r[I['mn']], r[I['mx']], r[I['adj']]
    pb = pbl_fiable(r)
    if a is None or a < 2018 or a > 2025:
        m = 'fuera_ventana'
    elif r[I['tipo_contrato']] not in OSS:
        m = 'al_alza_tipo'
    elif r[I['persona_fisica']]:
        m = 'persona_fisica'
    elif r[I['num_ofertas']] is None or r[I['num_ofertas']] < 3:
        m = 'menos_de_3'
    elif mn is None or mx is None or mn <= 0 or mx <= 0:
        m = 'sin_min_max'
    elif mn > mx or adj is None or adj <= 0 or adj < mn * 0.995 or adj > mx * 1.005:
        m = 'incoherente'
    elif pb is not None and mn > pb * 1.001:
        m = 'sobre_pbl'
    else:
        m = 'ok'
    embudo[m] = embudo.get(m, 0) + 1
    if m == 'ok':
        P.append(r)
print('P', len(P), embudo, flush=True)

lot = {}
for r in P:
    pb = pbl_fiable(r)
    mn, mx, adj, n = r[I['mn']], r[I['mx']], r[I['adj']], r[I['num_ofertas']]
    d = dict(id=r[I['id_canonical']], lic=r[I['licitacion_id']], org=r[I['organo_clave']], anio=r[I['anio']], cpv=r[I['cpv']],
             tipo=r[I['tipo']], emp=r[I['empresa_clave']], fref=r[I['fref']], tc=r[I['tipo_contrato']],
             on=r[I['organo_nombre']], nuts=r[I['nuts']], prov=r[I['provincia']])
    d['pnc'] = None if pb is None else (1 if abs(mn - pb) <= 0.001 * pb else 0)
    d['pc'] = (mn != mx) and d['pnc'] != 1
    d['rango'] = (mx - mn) / mn
    d['sep'] = (mx - adj) / adj
    d['baja'] = (1 - adj / pb) if (pb is not None and adj / pb <= 1.25) else None
    d['tramo'] = '3' if n == 3 else ('4-5' if n <= 5 else ('6-9' if n <= 9 else '10+'))
    lot[d['id']] = d
PC = [d for d in lot.values() if d['pc']]

# ---- BAJA_COND: mediana de la baja en PC por mercado × tramo (celdas con >= 10)
cel = {}
for d in PC:
    if d['baja'] is not None:
        cel.setdefault((d['cpv'], d['tipo'], d['anio'], d['tramo']), []).append(d['baja'])
med = {k: statistics.median(v) for k, v in cel.items() if len(v) >= 10}
for d in PC:
    k = (d['cpv'], d['tipo'], d['anio'], d['tramo'])
    d['baja_cond'] = d['baja'] - med[k] if (d['baja'] is not None and k in med) else None

# ---- REP_IGUAL (Gini por la fórmula ordenada)
porlic = {}
for r in B:
    if r[I['anio']] is not None and 2018 <= r[I['anio']] <= 2025 and r[I['tipo_contrato']] in OSS:
        porlic.setdefault(r[I['licitacion_id']], []).append(r)
rep = {}
for l, rs in porlic.items():
    if len(rs) < 3:
        continue
    g = {}
    for r in rs:
        k = r[I['empresa_clave']] if r[I['empresa_clave']] is not None else r[I['adjudicatario_nombre']]
        if k is None:
            continue
        g[k] = g.get(k, 0.0) + (r[I['adj']] or 0.0)
    tot = sum(g.values())
    if len(g) < 2 or tot <= 0:
        continue
    x = sorted(v / tot for v in g.values()); K = len(x)
    gini = 2 * sum((i + 1) * xi for i, xi in enumerate(x)) / (K * sum(x)) - (K + 1) / K
    rep[l] = 1 - gini
# ---- ESTAB
win = {}
for r in B:
    a = r[I['anio']]
    if a is None or a < 2017 or a > 2025 or r[I['tipo_contrato']] not in OSS or r[I['persona_fisica']]:
        continue
    k = (r[I['organo_clave']], r[I['cpv']], a)
    e = win.setdefault(k, [0, set()])
    e[0] += 1
    g = r[I['empresa_clave']] if r[I['empresa_clave']] is not None else r[I['adjudicatario_nombre']]
    if g is not None:
        e[1].add(g)
est = {}
for (o, c, a), (n, W) in win.items():
    prev = win.get((o, c, a - 1))
    if n >= 3 and prev and prev[0] >= 3:
        uni = W | prev[1]
        est[(o, c, a)] = len(W & prev[1]) / len(uni) if uni else None
# ---- UTE_RIVALES
socios = {}
for u, s, pf in uso:
    if s is not None and not pf:
        socios.setdefault(u, set()).add(s)
solos = set()
for r in B:
    if r[I['empresa_clave']] is not None:
        solos.add((r[I['empresa_clave']], r[I['cpv']], r[I['tipo']]))
for d in PC:
    d['rep_igual'] = rep.get(d['lic'])
    d['estab'] = est.get((d['org'], d['cpv'], d['anio']))
    ss = socios.get(d['emp'], set())
    d['ute_rivales'] = 1 if len(ss) >= 2 and sum(1 for s in ss if (s, d['cpv'], d['tipo']) in solos) >= 2 else 0

# ---- percentiles orientados (rango medio de empates) dentro de mercado, mercados con >= 20 lotes de PC
merc = {}
for d in PC:
    merc.setdefault((d['cpv'], d['tipo'], d['anio']), []).append(d)


def pct_mid(ds, key, sgn, dest):
    v = [(sgn * d[key], d) for d in ds if d[key] is not None]
    v.sort(key=lambda t: t[0]); N = len(v); i = 0
    while i < N:
        j = i
        while j + 1 < N and v[j + 1][0] == v[i][0]:
            j += 1
        rm = (i + 1 + j + 1) / 2
        for t in range(i, j + 1):
            v[t][1][dest] = (rm - 0.5) / N
        i = j + 1


ORI = [('rango', -1, 'p_rango'), ('sep', -1, 'p_sep'), ('baja_cond', -1, 'p_baja'), ('rep_igual', 1, 'p_rep'), ('estab', 1, 'p_est')]
for k, ds in merc.items():
    for d in ds:
        for _, _, dest in ORI:
            d[dest] = None
        d['en_c'] = len(ds) >= 20
    if len(ds) < 20:
        continue
    for key, sg, dest in ORI:
        pct_mid(ds, key, sg, dest)
for d in PC:
    if not d['en_c']:
        d['s3'] = d['s5'] = None; continue
    a3 = [d[x] for x in ('p_rango', 'p_sep', 'p_baja') if d[x] is not None]
    a5 = [d[x] for x in ('p_rango', 'p_sep', 'p_baja', 'p_rep', 'p_est') if d[x] is not None]
    d['s3'] = sum(a3) / len(a3); d['s5'] = sum(a5) / len(a5)
for k, ds in merc.items():
    for d in ds:
        d['pct_s3'] = None
    if len(ds) >= 20:
        for d in ds:
            d['s3r'] = None if d['s3'] is None else round(d['s3'], 12)  # empates numéricos (desvío técnico declarado)
        pct_mid(ds, 's3r', 1, 'pct_s3')
print('screens', round(time.time() - T0), flush=True)

# ---- etiqueta y mercados afectados (reglas de P8 c06 reescritas)
def norm(s):
    return unicodedata.normalize('NFKD', (s or '').lower()).encode('ascii', 'ignore').decode()
RX_MIL = re.compile('defensa|ejercito|jaemale|mando de apoyo logistico|parque y centro|armada|intendencia|jefatura de asuntos economicos|direccion general de armamento|mando de ingenieros|acuartelamiento|base aerea|regimiento|brigada')
EXP = {'alimentos': 'S/0016/21', 'militar': 'S/0008/21', 'carreteras': 'S/0013/19'}
pos = {c: {i for e, v, i in lab if e == x and v in ('V2', 'V3')} for c, x in EXP.items()}
pos_cant = {i for e, v, i in lab if e == 'S/0004/25' and v in ('V2', 'V3')}
def mercado_afectado(c, d):
    if c == 'alimentos':
        return d['cpv'] in ('15', '03')
    if c == 'carreteras':
        return d['tipo'] == 'age' and 'carreteras' in norm(d['on'])
    if c == 'militar':
        return d['tipo'] == 'age' and RX_MIL.search(norm(d['on'])) is not None


def conjunto_unidades(c, key, pob='c'):
    base = [d for d in (lot.values() if pob == 'p' else PC) if d.get(key) is not None and (pob == 'p' or d['en_c'])]
    P1 = [d for d in base if d['id'] in pos[c]]
    anios = {d['anio'] for d in P1}
    C0 = [d for d in base if d['id'] not in pos[c] and d['id'] not in excl and mercado_afectado(c, d) and d['anio'] in anios]
    return P1, C0


def auc_pares(P1, C0, key, w=None):
    """Conteo explícito de pares dentro de estrato (año); w: peso por licitación."""
    num = den = 0.0
    porA = {}
    for d in C0:
        porA.setdefault(d['anio'], []).append(d)
    for p in P1:
        wp = 1 if w is None else w.get(p['lic'], 0)
        if not wp:
            continue
        for q in porA.get(p['anio'], []):
            wq = 1 if w is None else w.get(q['lic'], 0)
            if not wq:
                continue
            s = 1.0 if p[key] > q[key] else (0.5 if p[key] == q[key] else 0.0)
            num += wp * wq * s; den += wp * wq
    return num / den if den else None


def auc_bisect(grupos, key, w):
    """Para el bootstrap: por estrato, controles ordenados con pesos acumulados y bisect."""
    num = den = 0.0
    for P1, C0 in grupos:
        porA = {}
        for q in C0:
            ww = w.get(q['lic'], 0)
            if ww:
                porA.setdefault(q['anio'], []).append((q[key], ww))
        cum = {}
        for a, lst in porA.items():
            lst.sort(); vals = [x for x, _ in lst]; acc = [0.0]
            for _, ww in lst:
                acc.append(acc[-1] + ww)
            cum[a] = (vals, acc)
        for p in P1:
            wp = w.get(p['lic'], 0)
            if not wp or p['anio'] not in cum:
                continue
            vals, acc = cum[p['anio']]
            lo = bisect.bisect_left(vals, p[key]); hi = bisect.bisect_right(vals, p[key])
            num += wp * (acc[lo] + 0.5 * (acc[hi] - acc[lo])); den += wp * acc[-1]
    return num / den if den else None


CANDS = [('S3', 's3', 'c', 1), ('S5', 's5', 'c', 1), ('RANGO', 'p_rango', 'c', 1), ('SEP', 'p_sep', 'c', 1),
         ('BAJA_COND', 'p_baja', 'c', 1), ('PNC', 'pnc', 'p', 1), ('REP_IGUAL', 'p_rep', 'c', 1), ('ESTAB', 'p_est', 'c', 1),
         ('UTE_RIVALES', 'ute_rivales', 'c', 1)]
res = {}
for nombre, key, pob, _ in CANDS:
    res[nombre] = {}
    allp = allc = 0; num = den = 0.0
    for c in EXP:
        P1, C0 = conjunto_unidades(c, key, pob)
        a = auc_pares(P1, C0, key)
        n1 = sum(1 for p in P1 if any(q['anio'] == p['anio'] for q in C0)); n0 = sum(1 for q in C0)
        res[nombre][c] = dict(auc=a, n_pos=len(P1), n_ctl=n0)
    # conjunto: pares dentro de estrato cártel × año
    for c in EXP:
        P1, C0 = conjunto_unidades(c, key, pob)
        porA = {}
        for q in C0:
            porA.setdefault(q['anio'], []).append(q)
        for p in P1:
            for q in porA.get(p['anio'], []):
                num += 1.0 if p[key] > q[key] else (0.5 if p[key] == q[key] else 0.0); den += 1
    res[nombre]['conjunto'] = dict(auc=num / den if den else None)
print('auc', round(time.time() - T0), flush=True)

# ---- bootstrap de S3 (dos etapas, otro orden de muestreo y otro algoritmo de AUC)
rng = random.Random(20261021)
boot = {}
for nombre_g, cs in [('alimentos', ['alimentos']), ('militar', ['militar']), ('carreteras', ['carreteras']), ('conjunto', list(EXP))]:
    grupos = [conjunto_unidades(c, 's3') for c in cs]
    org = {}
    for P1, C0 in grupos:
        for d in P1 + C0:
            org.setdefault(str(d['org']), set()).add(d['lic'])
    O = sorted(org); L = {o: sorted(org[o], key=str) for o in O}
    vals = []
    for _ in range(2000):
        w = {}
        elegidos = rng.choices(O, k=len(O))
        for o in elegidos:
            for l in rng.choices(L[o], k=len(L[o])):
                w[l] = w.get(l, 0) + 1
        a = auc_bisect(grupos, 's3', w)
        if a is not None:
            vals.append(a)
    vals.sort()
    q = lambda f: vals[int(round(f * (len(vals) - 1)))] if vals else None
    boot[nombre_g] = dict(lo=q(0.025), hi=q(0.975), validas=len(vals))
print('boot', round(time.time() - T0), flush=True)

# ---- Cantabria (D2): cifras agregadas para cotejo
VI, VF = '2019-01-01', '2025-03-31'
def en_cant(d):
    return (d['nuts'] or '').startswith('ES13') or 'cantabria' in norm(d['prov'])
def obras(d):
    return d['tc'] == 'Obras' or d['cpv'] == '45'
amb = [d for d in PC if d['en_c'] and obras(d) and en_cant(d) and d['fref'] is not None and VI <= str(d['fref']) <= VF]
doce = [d for d in amb if d['id'] in pos_cant]
ctl = [d for d in PC if d['en_c'] and obras(d) and not en_cant(d) and d['id'] not in excl and d['fref'] is not None and VI <= str(d['fref']) <= VF]
cant = dict(ambito=len(amb), senal_ambito=sum(d['pct_s3'] >= 0.8 for d in amb), doce=len(doce), senal_doce=sum(d['pct_s3'] >= 0.8 for d in doce),
            ctl=len(ctl), senal_ctl=sum(d['pct_s3'] >= 0.8 for d in ctl),
            lics_senal=len({d['lic'] for d in amb if d['pct_s3'] >= 0.8}))
json.dump(dict(embudo=embudo, n_P=len(P), n_PC=len(PC), n_c=sum(1 for d in PC if d['en_c']), auc=res, boot_s3=boot, cantabria=cant),
          open('/c/interno/07_impl2.json', 'w'), indent=1, default=str)
# tabla por lote para el cotejo (interno)
K = ['id', 'rango', 'sep', 'baja_cond', 'rep_igual', 'estab', 'ute_rivales', 'p_rango', 'p_sep', 'p_baja', 'p_rep', 'p_est', 's3', 's5', 'pct_s3']
tab = {k: [d.get(k) for d in PC] for k in K}
tab['pnc'] = [d['pnc'] for d in PC]
pq.write_table(pa.table(tab), '/c/interno/sc_impl2.parquet')
print('ok', round(time.time() - T0), flush=True)