experiments/error-audit/error_analysis.py

#!/usr/bin/env python
"""
Etapa A — análisis de errores sobre los 1.740 (modelo publicado v1).

No usa GPU: trabaja sobre las predicciones y confianzas ya guardadas.

Salidas:
  results/errors_v1.jsonl        todos los errores con features, ordenados por confianza
  results/error_summary.json     agregados por idioma/clase/confianza
  results/error_samples_<lang>.txt   volcado legible para revisión manual

Uso: <venv>/python etapa4/error_analysis.py
"""
import json
import math
import re
from collections import Counter, defaultdict
from pathlib import Path

import numpy as np

HERE = Path(__file__).resolve().parent
TEST = HERE / "test_extended.jsonl"
PREDS = HERE / "results/predictions_v1_v2.jsonl"
NPZ = HERE.parent / "paper/figures/reliability_raw.npz"
OUT = HERE / "results"

recs = [json.loads(l) for l in TEST.read_text(encoding="utf-8").splitlines() if l.strip()]
preds = [json.loads(l) for l in PREDS.read_text(encoding="utf-8").splitlines() if l.strip()]
z = np.load(NPZ)
conf = z["v1|1.0|conf"]
correct = z["v1|1.0|correct"].astype(bool)
conf_v2 = z["v2|1.35|conf"]
pred_v2 = (z["v2|1.35|conf"] > 0.5).astype(int)  # solo para comparar errores; ver nota abajo

assert len(recs) == len(preds) == len(conf), (len(recs), len(preds), len(conf))

NEG = {
    "en": r"\b(no|not|never|n't|cannot|can't|don't|doesn't|didn't|isn't|wasn't|won't|nothing|nobody)\b",
    "es": r"\b(no|nunca|jamas|jamás|nada|nadie|sin|tampoco|ni)\b",
    "de": r"\b(nicht|kein|keine|keinen|keinem|keiner|nie|niemals|nichts|niemand|ohne)\b",
}
CONTRAST = {
    "en": r"\b(but|however|although|though|yet|still)\b",
    "es": r"\b(pero|sin embargo|aunque|no obstante|sino)\b",
    "de": r"\b(aber|jedoch|obwohl|trotzdem|sondern)\b",
}
POSEMO = "😊😍🥰❤️💕😁😂🤣👍🎉🙌✨😘😻💪🔥"
NEGEMO = "😢😭😡🤬👎💔😞😔😠🙁☹️"


def feats(text, lang):
    t = text.strip()
    words = t.split()
    tl = t.lower()
    return {
        "chars": len(t),
        "words": len(words),
        "neg_cue": bool(re.search(NEG[lang], tl)),
        "contrast": bool(re.search(CONTRAST[lang], tl)),
        "pos_emoji": any(e in t for e in POSEMO),
        "neg_emoji": any(e in t for e in NEGEMO),
        "has_user": "@user" in tl,
        "has_url": "http" in tl,
        "has_hashtag": "#" in t,
        "allcaps": sum(1 for w in words if len(w) > 3 and w.isupper()) >= 2,
        "question": "?" in t,
        "exclaim": "!" in t,
    }


def wilson(k, n, z_=1.96):
    if n == 0:
        return (float("nan"), float("nan"))
    ph = k / n
    d = 1 + z_ * z_ / n
    c = (ph + z_ * z_ / (2 * n)) / d
    h = z_ * math.sqrt(ph * (1 - ph) / n + z_ * z_ / (4 * n * n)) / d
    return c - h, c + h


rows = []
for i, r in enumerate(recs):
    lang = r["lang"]
    y = int(r["target"])
    p = int(recs[i].get("pred_v1", preds[i]["pred_v1"]))
    p = int(preds[i]["pred_v1"])
    c = float(conf[i])
    f = feats(r["state"], lang)
    rows.append({"i": i, "id": r["id"], "lang": lang, "target": y, "pred": p,
                 "conf": c, "correct": bool(p == y), "p_v2": int(preds[i]["pred_v2"]),
                 "text": r["state"], **f})

errs = [x for x in rows if not x["correct"]]
n = len(rows)
print("=" * 78)
print(f"ANÁLISIS DE ERRORES — v1 sobre {n} ejemplos")
print("=" * 78)
print(f"  errores totales: {len(errs)} ({len(errs)/n*100:.2f}%)")

summary = {"n": n, "errors": len(errs), "by_lang": {}, "by_class": {}, "features": {}}
print("\n  Por idioma:")
for lang in ("en", "de", "es"):
    sub = [x for x in rows if x["lang"] == lang]
    e = [x for x in sub if not x["correct"]]
    lo, hi = wilson(len(e), len(sub))
    conf_err = float(np.mean([x["conf"] for x in e])) if e else float("nan")
    hi_conf = sum(1 for x in e if x["conf"] >= 0.9)
    vhi = sum(1 for x in e if x["conf"] >= 0.95)
    fp = sum(1 for x in e if x["target"] == 0)   # negativo predicho positivo
    fn = sum(1 for x in e if x["target"] == 1)   # positivo predicho negativo
    print(f"    {lang}: {len(e):3d}/{len(sub):3d} = {len(e)/len(sub)*100:5.1f}%  IC95 [{lo*100:.1f},{hi*100:.1f}]"
          f" | conf media de los errores {conf_err:.3f} | conf>=0.9: {hi_conf:3d} | >=0.95: {vhi:3d}"
          f" | falso pos {fp} / falso neg {fn}")
    summary["by_lang"][lang] = {"n": len(sub), "errors": len(e), "err_rate": len(e) / len(sub),
                                "wilson": [lo, hi], "mean_conf_of_errors": conf_err,
                                "errors_conf_ge_090": hi_conf, "errors_conf_ge_095": vhi,
                                "false_pos": fp, "false_neg": fn}

print("\n  Por clase real:")
for cl, nm in ((1, "positivo"), (0, "negativo")):
    sub = [x for x in rows if x["target"] == cl]
    e = [x for x in sub if not x["correct"]]
    print(f"    {nm:9s}: {len(e):3d}/{len(sub):3d} = {len(e)/len(sub)*100:5.1f}%")
    summary["by_class"][nm] = {"n": len(sub), "errors": len(e), "err_rate": len(e) / len(sub)}

print("\n  Features: errores vs aciertos (proporción presente)")
ces = [x for x in rows if x["correct"]]
for k in ("neg_cue", "contrast", "pos_emoji", "neg_emoji", "has_user", "has_url",
          "has_hashtag", "allcaps", "question", "exclaim"):
    pe = sum(x[k] for x in errs) / max(1, len(errs))
    pc = sum(x[k] for x in ces) / max(1, len(ces))
    flag = "  <-- " if abs(pe - pc) > 0.08 else ""
    print(f"    {k:12s} error {pe*100:5.1f}%  acierto {pc*100:5.1f}%{flag}")
    summary["features"][k] = {"errors": pe, "correct": pc}
for k in ("chars", "words"):
    print(f"    {k:12s} error {np.mean([x[k] for x in errs]):7.1f}  acierto {np.mean([x[k] for x in ces]):7.1f}")
    summary["features"][k] = {"errors": float(np.mean([x[k] for x in errs])),
                              "correct": float(np.mean([x[k] for x in ces]))}

both_wrong = sum(1 for x in errs if x["p_v2"] != x["target"])
print(f"\n  Errores compartidos con v2 (otro entrenamiento): {both_wrong}/{len(errs)} "
      f"({both_wrong/len(errs)*100:.1f}%) -> candidatos a ambigüedad del dato, no del modelo")
summary["both_models_wrong"] = both_wrong

print("\n  Distribución de confianza de los errores:")
buckets = Counter()
for x in errs:
    c = x["conf"]
    b = ">=0.95" if c >= 0.95 else "0.90-0.95" if c >= 0.9 else "0.75-0.90" if c >= 0.75 else "0.50-0.75"
    buckets[b] += 1
for b in (">=0.95", "0.90-0.95", "0.75-0.90", "0.50-0.75"):
    print(f"    {b:10s} {buckets[b]:3d}  ({buckets[b]/len(errs)*100:4.1f}%)")
summary["conf_buckets"] = dict(buckets)

errs_sorted = sorted(errs, key=lambda x: -x["conf"])
with open(OUT / "errors_v1.jsonl", "w", encoding="utf-8") as f:
    for x in errs_sorted:
        f.write(json.dumps({k: v for k, v in x.items() if k != "correct"}) + "\n")

for lang in ("es", "de", "en"):
    sub = [x for x in errs_sorted if x["lang"] == lang]
    with open(OUT / f"error_samples_{lang}.txt", "w", encoding="utf-8") as f:
        f.write(f"# {len(sub)} errores en {lang}, ordenados por confianza del modelo\n")
        for x in sub:
            f.write(f"\n[{x['id']}] conf={x['conf']:.3f} label={'POS' if x['target'] else 'NEG'} "
                    f"pred={'POS' if x['pred'] else 'NEG'} v2={'POS' if x['p_v2'] else 'NEG'}\n")
            f.write(f"  {x['text']}\n")

(OUT / "error_summary.json").write_text(json.dumps(summary, indent=2))
print(f"\n  guardado: {OUT/'errors_v1.jsonl'}, error_summary.json y error_samples_<lang>.txt")