"""E8 · CÁLCULO SELLADO, implementación 1 (pyarrow + Python; DuckDB solo para leer). Ver DEFINICIONES_PREVIAS_E8.md §3-§6.
Uso: c2_calculo.py [base.parquet] [carpeta_salida] [carpeta_referencias]
  por defecto: /p/priv/e8_base.parquet, /p (escribe priv/ y salidas/), /p/salidas (referencias).
Salidas: priv/e8_contratos_impl1.csv (un contrato por fila, con estado y razón), salidas/e8_resumen_impl1.json,
salidas/e8_organos_impl1.csv (órgano [× producto] con N ≥ 10: K de N por encima, mediana con IC, hecho), salidas/e8_anio_impl1.csv."""
import sys, csv, json, math, random, re, datetime as dt, collections, statistics as st
sys.path.insert(0, '/a1/scripts')
import pyarrow.parquet as pq
from comun_a1 import provincia_ine
BASE = sys.argv[1] if len(sys.argv) > 1 else '/p/priv/e8_base.parquet'
OUT = sys.argv[2] if len(sys.argv) > 2 else '/p'
REF = sys.argv[3] if len(sys.argv) > 3 else '/p/salidas'
MARGEN = 1.15
FUERA_PROV = {'35', '38', '51', '52'}; FUERA_NUTS = ('ES70', 'ES63', 'ES64')
VENTANA = {'carburantes': ('2015-01', '2026-10'), 'electricidad': ('2015-01', '2026-06'), 'papel': ('2020-05', '2027-03')}
VALIDEZ = {'carburantes': (0.5, 2.0), 'electricidad': (0.5, 2.0), 'papel': (0.5, 2.5)}
PROD_OK = {'gasoleo_a', 'gasoleo_b', 'gasoleo_c', 'gasolina95'}
RX_GRUPO_REF = re.compile(r"(acuerdo marco|a\.?\s?m\.?)\s*(n.?\s*)?20/(20)?(19|21|23|25)\b")
# ------------------------------------------------------------------ referencias
ref_carb = {(r['mes'], r['prov_ine'], r['producto']): float(r['ref_sin_iva_eur_l']) for r in csv.DictReader(open(f'{REF}/ref_carb_mensual.csv'))}
ref_elec = {(r['semestre'], r['banda']): float(r['eur_kwh']) for r in csv.DictReader(open(f'{REF}/ref_electricidad_eurostat.csv')) if r['impuestos'] == 'X_VAT'}
ref_pap = [dict(r, lote=int(r['lote']), gramaje=int(r['gramaje']), med=float(r['mediana_eur_caja'])) for r in csv.DictReader(open(f'{REF}/ref_papel_am.csv'))]
BANDAS = [(20, 'MWH_LT20'), (500, 'MWH20-499'), (2000, 'MWH500-1999'), (20000, 'MWH2000-19999'), (70000, 'MWH20000-69999'),
          (150000, 'MWH70000-149999'), (float('inf'), 'MWH_GE150000')]


def banda(mwh):
    return next(b for lim, b in BANDAS if mwh < lim)


def ref_papel(mes, fibra, gramaje, sin_fibra_a='1'):
    lote = 2 if fibra == 'reciclado' else (int(sin_fibra_a) if fibra == 'sin_fibra' else 1)
    d = mes + '-01'
    for r in ref_pap:
        if r['lote'] == lote and r['gramaje'] == gramaje and r['vigencia_desde'] <= d <= r['vigencia_hasta']:
            return r['med'], r['am'], lote
    return None, None, lote


def mes_de(row):
    f0 = row['fecha_adj'] or row['fecha_formalizacion'] or row['fecha_pub']
    if f0 and dt.date(2015, 1, 1) <= f0 <= dt.date(2026, 10, 31):
        return f'{f0:%Y-%m}'
    return f"{row['anio_ref']}-07" if row['anio_ref'] else None


def evaluar(row, sin_fibra_a='1', validez=None, solo_a4=False):
    """Devuelve dict con estado, valido, razon, cantidad_total, ref, etc. Reglas en el orden sellado (§3)."""
    m = row['mercado']; o = {}
    es_copia = row['copia_de'] is not None or row['correccion_imp_adj'] == 'copia_otra_fuente' or row['cuenta'] is False
    prov = provincia_ine(row['nuts'], row['provincia'])
    mes = mes_de(row)
    o.update(mes=mes, prov=prov)
    grupo_ref = m == 'papel' and (bool(RX_GRUPO_REF.search((row['objeto'] or '').lower())) or
                                  'racionalizaci' in (row['organo_nombre'] or '').lower())
    if es_copia: return dict(o, estado='copia')
    if not (row['imp'] or 0) > 0: return dict(o, estado='sin_importe')
    if (row['rol_am'] or '') in ('acuerdo', 'sda') and not grupo_ref: return dict(o, estado='acuerdo_marco')
    if prov in FUERA_PROV or (row['nuts'] or '')[:4] in FUERA_NUTS: return dict(o, estado='fuera_territorio')
    v0, v1 = VENTANA[m]
    if not mes or not (v0 <= mes <= v1): return dict(o, estado='fuera_periodo')
    if grupo_ref: o['grupo_referencia'] = True
    if m == 'carburantes' and row['x_excl_v2']: return dict(o, estado='excluido_v2')
    if row['x_motivo'] != 'ok': return dict(o, estado=row['x_motivo'])
    if not (row['x_cantidad'] or 0) > 0: return dict(o, estado='sin_cantidad')   # desvío 1 (26-oct 20:40): «10 000;00 litros» leía 0 y dividía por cero
    imp = row['imp']
    if m == 'carburantes':
        p = row['x_producto']
        if p not in PROD_OK: return dict(o, estado='sin_producto_comparable', producto=p)
        tot = row['x_cantidad']
        ref = ref_carb.get((mes, '00', p)) if p == 'gasoleo_c' else (ref_carb.get((mes, prov, p)) or ref_carb.get((mes, '00', p)))
        o.update(producto=p, cantidad_total=tot)
        if ref is None: return dict(o, estado='sin_referencia')
        nucleo = tot >= 200 and row['x_n_distintas'] == 1
    elif m == 'electricidad':
        dur = row['duracion_meses'] if (row['duracion_meses'] or 0) > 0 else None
        d = dur or 12.0
        k = row['x_cantidad']
        tot = k * d / 12 if row['x_anual'] else k
        mwh = (k if row['x_anual'] else k * 12 / d) / 1000
        sem = f"{mes[:4]}-S{1 if int(mes[5:]) <= 6 else 2}"
        ref = ref_elec.get((sem, banda(mwh)))
        o.update(producto='electricidad', cantidad_total=tot, banda=banda(mwh), duracion_supuesta=dur is None)
        if ref is None: return dict(o, estado='sin_referencia')
        nucleo = tot >= 1000
    else:
        if solo_a4 and not row['x_formato_explicito']: return dict(o, estado='formato_supuesto')
        tot = row['x_cantidad']
        ref, am, lote = ref_papel(mes, row['x_producto'], row['x_gramaje'], sin_fibra_a)
        o.update(producto=row['x_producto'], cantidad_total=tot, am=am, lote=lote)
        if ref is None: return dict(o, estado='sin_referencia')
        nucleo = tot >= 2
    razon = (imp / tot) / ref
    o.update(ref=ref, razon=razon, precio_unitario=imp / tot)
    if not nucleo: return dict(o, estado='fuera_nucleo')
    lo, hi = validez or VALIDEZ[m]
    if not (lo <= razon <= hi): return dict(o, estado='dudoso')
    return dict(o, estado='valido', valido=True, por_encima=razon > MARGEN)


def mediana_ic(xs, B=10000, semilla=20261026):
    if not xs: return None, None, None
    rnd = random.Random(semilla); n = len(xs); meds = []
    for _ in range(B):
        meds.append(st.median([xs[rnd.randrange(n)] for _ in range(n)]))
    meds.sort()
    return st.median(xs), meds[int(0.025 * B)], meds[int(0.975 * B) - 1]


def wilson(k, n, z=1.959964):
    if n == 0: return None, None
    p = k / n; d = 1 + z * z / n; c = p + z * z / (2 * n); r = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n))
    return (c - r) / d, (c + r) / d


def cola_sup(k, n, p):
    if k <= 0: return 1.0
    return min(1.0, sum(math.comb(n, i) * p ** i * (1 - p) ** (n - i) for i in range(k, n + 1)))


def cuantil(xs, q):
    xs = sorted(xs); h = (len(xs) - 1) * q; f = math.floor(h)
    return xs[f] + (h - f) * (xs[min(f + 1, len(xs) - 1)] - xs[f])


def describir(rs, B=10000):
    xs = [r['razon'] for r in rs]
    if not xs: return dict(n=0)
    med, lo, hi = mediana_ic(xs, B)
    k = sum(1 for r in rs if r['por_encima'])
    wl, wh = wilson(k, len(xs))
    return dict(n=len(xs), mediana=med, ic95=[lo, hi], p10=cuantil(xs, .1), p90=cuantil(xs, .9),
                pct_dentro_15=100 * sum(1 for x in xs if 0.85 <= x <= 1.15) / len(xs),
                k_por_encima=k, pct_por_encima=100 * k / len(xs), ic95_pct_por_encima=[100 * wl, 100 * wh],
                pct_encima_110=100 * sum(1 for x in xs if x > 1.10) / len(xs), pct_encima_125=100 * sum(1 for x in xs if x > 1.25) / len(xs))


NIF_PERSONA = re.compile(r"^[0-9XYZKLM]")


def main():
    T = pq.read_table(BASE).to_pylist()
    filas = []
    for row in T:
        e = evaluar(row)
        e.update({k: row[k] for k in ('source', 'id_canonical', 'mercado', 'organo_clave', 'organo_nombre', 'organo_nif', 'tipo_organo',
                                      'menor', 'anio_ref', 'imp', 'rol_am', 'x_literal', 'x_cantidad', 'x_producto', 'x_gramaje', 'duracion_meses')})
        filas.append(e)
    campos = ['source', 'id_canonical', 'mercado', 'estado', 'valido', 'razon', 'por_encima', 'ref', 'precio_unitario', 'cantidad_total',
              'producto', 'x_gramaje', 'banda', 'am', 'lote', 'mes', 'prov', 'grupo_referencia', 'duracion_supuesta', 'organo_clave',
              'organo_nombre', 'organo_nif', 'tipo_organo', 'menor', 'anio_ref', 'imp', 'rol_am', 'x_literal', 'x_cantidad', 'duracion_meses']
    with open(f'{OUT}/priv/e8_contratos_impl1.csv', 'w', newline='') as fh:
        w = csv.DictWriter(fh, fieldnames=campos, extrasaction='ignore'); w.writeheader(); w.writerows(filas)
    res = {}
    UNIV_FUERA = {'copia', 'sin_importe', 'acuerdo_marco', 'fuera_territorio', 'fuera_periodo'}
    for m in ('carburantes', 'electricidad', 'papel'):
        F = [f for f in filas if f['mercado'] == m]
        U = [f for f in F if f['estado'] not in UNIV_FUERA and not f.get('grupo_referencia')]
        V = [f for f in U if f['estado'] == 'valido']
        r = dict(filas_base=len(F), estados_base=collections.Counter(f['estado'] for f in F if not f.get('grupo_referencia')),
                 universo=len(U), universo_importe=sum(f['imp'] for f in U), validos=len(V), validos_importe=sum(f['imp'] for f in V),
                 cobertura_pct=100 * len(V) / len(U) if U else None, cobertura_importe_pct=100 * sum(f['imp'] for f in V) / sum(f['imp'] for f in U) if U else None,
                 estados_universo=collections.Counter(f['estado'] for f in U), todos=describir(V))
        clave = 'producto'
        r['por_producto'] = {p: describir([v for v in V if v.get(clave) == p], 2000) for p in sorted({v.get(clave) for v in V})}
        r['por_anio'] = {a: describir([v for v in V if v['mes'][:4] == a], 2000) for a in sorted({v['mes'][:4] for v in V})}
        r['por_tipo_organo'] = {t: describir([v for v in V if v['tipo_organo'] == t], 2000) for t in sorted({v['tipo_organo'] for v in V})}
        r['por_menor'] = {str(t): describir([v for v in V if v['menor'] == t], 2000) for t in (True, False)}
        r['por_rol_am'] = {str(t): describir([v for v in V if (v['rol_am'] or 'ninguno') == t], 2000) for t in sorted({v['rol_am'] or 'ninguno' for v in V})}
        # sensibilidades (§6)
        sens = {}
        if m == 'carburantes':
            sens['sin_bonificacion_2022'] = describir([v for v in V if not ('2022-04' <= v['mes'] <= '2022-12')], 2000)
            sens['gasoleo_c_dentro_15_pct'] = r['por_producto'].get('gasoleo_c', {}).get('pct_dentro_15')
        if m == 'papel':
            for nombre, kw in (('sin_fibra_lote2', dict(sin_fibra_a='2')), ('validez_05_2', dict(validez=(0.5, 2.0))), ('solo_a4_explicito', dict(solo_a4=True))):
                ee = [dict(evaluar(row, **kw), menor=row['menor']) for row in T if row['mercado'] == 'papel']
                sens[nombre] = describir([x for x in ee if x['estado'] == 'valido' and not x.get('grupo_referencia')], 2000)
            G = [f for f in F if f.get('grupo_referencia') and f['estado'] == 'valido']
            gx = [g['razon'] for g in G]
            r['grupo_referencia'] = dict(describir(G, 2000), dentro_085_110_pct=100 * sum(1 for x in gx if 0.85 <= x <= 1.10) / len(gx) if gx else None,
                                         estados=collections.Counter(f['estado'] for f in F if f.get('grupo_referencia')))
        if m == 'electricidad':
            sens['duracion_conocida'] = describir([v for v in V if not v.get('duracion_supuesta')], 2000)
        sens['solo_menores'] = r['por_menor']['True']; sens['sin_menores'] = r['por_menor']['False']
        r['sensibilidad'] = sens
        res[m] = r
    # hechos por órgano (§5): carburantes órgano × producto; papel órgano; electricidad ninguno
    org_rows = []
    for m, por_prod in (('carburantes', True), ('papel', False)):
        V = [f for f in filas if f['mercado'] == m and f['estado'] == 'valido' and not f.get('grupo_referencia')]
        grupos = collections.defaultdict(list)
        for v in V:
            if v['organo_clave']:
                grupos[(v['organo_clave'], v['producto'] if por_prod else 'todos')].append(v)
        tot = collections.Counter(); totk = collections.Counter()
        for v in V:
            g = v['producto'] if por_prod else 'todos'; tot[g] += 1; totk[g] += v['por_encima']
        unidades = []
        for (oc, g), vs in grupos.items():
            n = len(vs); k = sum(v['por_encima'] for v in vs)
            if n < 10: continue
            n0, k0 = tot[g] - n, totk[g] - k
            p0 = k0 / n0 if n0 > 0 else None
            pv = cola_sup(k, n, p0) if p0 not in (None, 0.0) else (0.0 if k > 0 else 1.0)
            med, lo, hi = mediana_ic([v['razon'] for v in vs], 2000, 20261026)
            unidades.append(dict(mercado=m, organo_clave=oc, organo_nombre=vs[0]['organo_nombre'], tipo_organo=vs[0]['tipo_organo'],
                                 producto=g, n=n, k_por_encima=k, p0_resto=p0, p_valor=pv, mediana=med, ic95_lo=lo, ic95_hi=hi,
                                 nif_persona=bool(NIF_PERSONA.match(vs[0]['organo_nif'] or ''))))
        unidades.sort(key=lambda u: u['p_valor'])
        mm = len(unidades); rech = True
        for i, u in enumerate(unidades):
            umbral = 0.05 / (mm - i)
            rech = rech and u['p_valor'] <= umbral
            u['hecho'] = 'por_encima_mas_frecuente' if rech else 'ninguno'
        org_rows += unidades
        res[m]['organos_evaluables'] = mm
        res[m]['organos_con_hecho'] = sum(1 for u in unidades if u['hecho'] != 'ninguno')
    with open(f'{OUT}/salidas/e8_organos_impl1.csv', 'w', newline='') as fh:
        cw = ['mercado', 'organo_clave', 'organo_nombre', 'tipo_organo', 'producto', 'n', 'k_por_encima', 'p0_resto', 'p_valor', 'mediana', 'ic95_lo', 'ic95_hi', 'hecho', 'nif_persona']
        w = csv.DictWriter(fh, fieldnames=cw); w.writeheader(); w.writerows(org_rows)
    json.dump(res, open(f'{OUT}/salidas/e8_resumen_impl1.json', 'w'), indent=1, ensure_ascii=False, default=str)
    for m, r in res.items():
        t = r['todos']
        print(m, 'universo', r['universo'], 'válidos', r['validos'], f"cobertura {r['cobertura_pct'] or 0:.2f} %",
              'mediana', t.get('mediana'), 'IC', t.get('ic95'), '% encima', t.get('pct_por_encima'), 'órganos', r.get('organos_evaluables'), 'hechos', r.get('organos_con_hecho'))


if __name__ == '__main__':
    main()