| |
| """Audit 5: precision held-out des regles de TRAIT D'UNION + confusion matrix point final.""" |
| import csv, json, os |
| from collections import Counter |
|
|
| OUT = [] |
| def P(*a): |
| s = " ".join(str(x) for x in a); OUT.append(s); print(s) |
|
|
| sub = {} |
| with open("/root/sub_LMA06.csv", encoding="utf-8") as f: |
| for row in csv.DictReader(f): sub[row["ID"]] = row["Target"] |
| lang = json.load(open("/root/test_lang.json", encoding="utf-8")) |
| hyp = {"lin": [], "sna": []} |
| for k, v in sub.items(): |
| if lang.get(k) in hyp: hyp[lang[k]].append(v) |
| train, val = {"lin": [], "sna": []}, {"lin": [], "sna": []} |
| for L in ("lin", "sna"): |
| for sp, st in (("train", train), ("validation", val)): |
| with open("/scratch/prep/manifests/waxal_%s_%s.jsonl" % (L, sp), encoding="utf-8") as f: |
| for line in f: |
| if line.strip(): st[L].append(json.loads(line)["text"]) |
| tot_w = sum(len(t.split()) for L in hyp for t in hyp[L]) |
| tot_c = sum(len(t) for L in hyp for t in hyp[L]) |
| W_ERR, C_ERR = 0.5/tot_w, 0.5/tot_c |
| PUNCT = ".,;:!?\"()«»" |
| def toks(t): return [x for x in (w.strip(PUNCT) for w in t.split()) if x] |
|
|
| P("=== TRAIT D'UNION : la regle 'a b' -> 'a-b' est-elle fiable ? (VAL held-out) ===") |
| P("1 fix correct = -2 err-mot -1 err-car ; 1 fix errone = +2 err-mot +1 err-car") |
| UNIT = 2*W_ERR + C_ERR |
| P("valeur d'un fix = %.3e score ; ecart au 2e = %.1f fix nets" % (UNIT, 0.000199/UNIT)) |
| for L in ("lin", "sna"): |
| def counts(texts): |
| u = Counter(); b = Counter() |
| for t in texts: |
| ws = [w.lower() for w in toks(t)] |
| u.update(ws); b.update(zip(ws, ws[1:])) |
| return u, b |
| u_tr, b_tr = counts(train[L]) |
| u_va, b_va = counts(val[L]) |
| hyp_bi = Counter() |
| for t in hyp[L]: |
| ws = [w.lower() for w in toks(t)]; hyp_bi.update(zip(ws, ws[1:])) |
|
|
| P("") |
| P("--- %s ---" % L) |
| P(" %-30s %6s %6s %7s | %6s %6s %7s | %5s" % ("regle a b -> a-b", "trHyph", "trSep", "trPrec", "vaHyph", "vaSep", "vaPrec", "hyp")) |
| tot_fire = 0; tot_tp = 0; tot_fp = 0 |
| rows = [] |
| for (a, b), c in hyp_bi.items(): |
| h = a + "-" + b |
| th, ts = u_tr.get(h, 0), b_tr.get((a, b), 0) |
| vh, vs = u_va.get(h, 0), b_va.get((a, b), 0) |
| if th + vh < 3: continue |
| rows.append((c, a, b, th, ts, vh, vs)) |
| rows.sort(reverse=True) |
| for c, a, b, th, ts, vh, vs in rows: |
| tp = th/max(th+ts, 1); vp = vh/max(vh+vs, 1) if (vh+vs) else float("nan") |
| P(" %-30s %6d %6d %6.1f%% | %6d %6d %6s | %5d" |
| % (a+" "+b, th, ts, 100*tp, vh, vs, ("%.1f%%" % (100*vp)) if (vh+vs) else "n/a", c)) |
| tot_fire += c; tot_tp += vh; tot_fp += vs |
| prec = tot_tp/max(tot_tp+tot_fp, 1) |
| P(" TOTAL: %d declenchements | precision VAL agregee = %.2f%% (%d/%d)" % (tot_fire, 100*prec, tot_tp, tot_tp+tot_fp)) |
| net = tot_fire*(2*prec-1) |
| P(" GAIN = %d x (2p-1)=%.3f = %.1f fix nets = %+.5f score" % (tot_fire, 2*prec-1, net, net*UNIT)) |
| |
| ff = tt = fpp = 0 |
| for c, a, b, th, ts, vh, vs in rows: |
| if th/max(th+ts,1) >= 0.90: |
| ff += c; tt += vh; fpp += vs |
| if ff: |
| pr2 = tt/max(tt+fpp,1); net2 = ff*(2*pr2-1) |
| P(" CONSERVATEUR (precision train>=90%%): %d declench., precision VAL=%.2f%% -> %.1f fix nets = %+.5f score" |
| % (ff, 100*pr2, net2, net2*UNIT)) |
| |
| nh = sum(1 for t in hyp[L] for w in toks(t) if "-" in w) |
| ntr = sum(1 for t in train[L] for w in toks(t) if "-" in w) |
| P(" mots contenant '-' : hyp=%d (%.3f%% des mots) | train=%d (%.3f%%)" |
| % (nh, 100.0*nh/sum(len(toks(t)) for t in hyp[L]), ntr, 100.0*ntr/sum(len(toks(t)) for t in train[L]))) |
|
|
| |
| P("") |
| P("=== POINT FINAL : matrice de confusion devhard-lin (joint_cont_best) ===") |
| allh = json.load(open("/root/devhard_allhyps.json", encoding="utf-8")) |
| refs = {} |
| for fn in os.listdir("/root/devhard"): |
| with open("/root/devhard/"+fn, encoding="utf-8") as f: |
| for line in f: |
| if line.strip(): |
| d = json.loads(line); refs[d["id"]] = d.get("text", "") |
| hyps = allh["joint_cont_best"] |
| pairs = [(h.strip().endswith("."), refs[i].strip().endswith(".")) |
| for i, h in hyps.items() if i.startswith("lin_") and i in refs and refs[i].strip()] |
| n11 = sum(1 for a,b in pairs if a and b); n10 = sum(1 for a,b in pairs if a and not b) |
| n01 = sum(1 for a,b in pairs if not a and b); n00 = sum(1 for a,b in pairs if not a and not b) |
| P("n=%d n11=%d n10=%d n01=%d n00=%d" % (len(pairs), n11, n10, n01, n00)) |
| P("P(ref='.' | nous emettons '.') = %.3f" % (n11/max(n11+n10,1))) |
| P("P(ref='.' | nous n'emettons PAS) = %.3f <- si > 0.5, forcer '.' partout gagne" % (n01/max(n01+n00,1))) |
| ntest_no = sum(1 for t in hyp["lin"] if not t.strip().endswith(".")) |
| p = n01/max(n01+n00,1) |
| P("clips lingala du TEST sans point final = %d" % ntest_no) |
| P("gain si on force le point: %d x (2p-1)=%.3f = %.1f clips nets = %+.5f score" |
| % (ntest_no, 2*p-1, ntest_no*(2*p-1), ntest_no*(2*p-1)*(W_ERR+C_ERR))) |
| ntest_no_s = sum(1 for t in hyp["sna"] if not t.strip().endswith(".")) |
| P("clips shona du TEST sans point final = %d (marge negligeable)" % ntest_no_s) |
|
|
| with open("/root/audit_STATSLING5_out.txt","w",encoding="utf-8") as f: f.write("\n".join(OUT)) |
| print("\n[OK] /root/audit_STATSLING5_out.txt") |
|
|
