| |
| """ |
| 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) |
|
|
| 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) |
| fn = sum(1 for x in e if x["target"] == 1) |
| 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") |
|
|
