#!/usr/bin/env python3 """Полигон, пакет-7, фаза B′: подсчёт по пре-регистрации §0 (T · J · R) и вердикты по её порогам. Пороги названы ДО трат и здесь только ПРОВЕРЯЮТСЯ: T1 `春秋蝉` и `Silversaint`: P5 (без правила транскрипции) против P2 T2 прокси «целиком несловарный dst»: |P5 − P2| ≥ 0.05 ⇒ эффект правила есть T3 совпадение с подписью владельца: P5 ≥ 17/19 ⇒ снятие правила не ломает другое C1 судья отбраковывается, если ДЕКОЙ назван лучшим по K1 более чем в 10% термов C2 судья отбраковывается, если катастроф на GOLD ≥ чем на декое C3 воспроизводит ли судья объективный слой: P0b в нижней трети по K1 R межарменный разрыв P0b↔P1 против внутриармового разброса по трём повторам BWS при дедупликации: вариант получает +1/−1, и этот балл достаётся КАЖДОМУ арму-источнику варианта (одинаковые строки схлопнуты, значит и заслуга у них общая). Декой считается отдельной «армой» и в ранжирование арм не входит. """ from __future__ import annotations import argparse import collections import json import re import statistics from pathlib import Path CRITS = ["K1", "K2", "K5", "K6", "K9"] ARMS = ["P0b", "P0", "P1", "P2", "P3", "P4", "P5"] T2_MIN = 0.05 T3_MIN = 17 C1_MAX = 0.10 STOP = {"и", "в", "на", "с", "из", "по", "о", "у", "за", "от", "для"} def norm(s: str | None) -> str: return re.sub(r"\s+", " ", (s or "").strip().lower()) def load_arms(paths: list[Path]) -> dict[str, dict[str, str]]: by: dict[str, dict[str, str]] = {} for p in paths: for r in json.loads(p.read_text(encoding="utf-8"))["results"]: if r.get("dst"): by.setdefault(r["id"], {})[r["arm"]] = r["dst"].strip() return by def nondict(dst: str, morph) -> bool: ws = [w for w in re.findall(r"[А-Яа-яЁё-]+", dst or "") if w.lower() not in STOP] return bool(ws) and all(not any(x.is_known for x in morph.parse(w)) for w in ws) def verdicts_of(row: dict) -> dict: """Ключи критериев приводятся к коду: модель иногда отдаёт «K1 верность концепту» или «K1: …». Это разбор, а не додумывание: код критерия берётся из начала ключа, содержимое не трогается. Без нормализации терялись вердикты, которые модель прислала (урок `repair_judge.py` фазы B — чинить разбор из сохранённого сырья, а не покупать повторные вызовы). """ out = {} for k, v in (row.get("verdicts") or {}).items(): m = re.match(r"\s*(K\d+)\b", str(k)) if m and isinstance(v, dict) and "best" in v: out.setdefault(m.group(1), v) return out def bws_dedup(rows: list[dict], crit: str) -> tuple[dict[str, float], int]: plus, minus, seen = collections.Counter(), collections.Counter(), collections.Counter() n = 0 for r in rows: v = verdicts_of(r).get(crit) if not isinstance(v, dict): continue l2s = r["letter2sources"] n += 1 for srcs in l2s.values(): for s in srcs: seen[s] += 1 for src in l2s.get(str(v.get("best")), []): plus[src] += 1 for src in l2s.get(str(v.get("worst")), []): minus[src] += 1 return {a: (plus[a] - minus[a]) / c for a, c in seen.items() if c}, n def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--termset", required=True, type=Path) ap.add_argument("--arms", action="append", required=True, type=Path) ap.add_argument("--judge", action="append", default=[], type=Path) ap.add_argument("--repeat", action="append", default=[], type=Path, help="файлы повторных прогонов для замера R") ap.add_argument("--out", required=True, type=Path) a = ap.parse_args() items = {i["id"]: i for i in json.loads(a.termset.read_text(encoding="utf-8"))} by = load_arms(a.arms) rep: dict = {} # ---------- T: контрольный арм без правила транскрипции --------------------------------- print("=== T. КОНТРОЛЬ ПРАВИЛА ТРАНСКРИПЦИИ (P5 = P2 минус одна строка) ===") print("\n T1 качественно — термы, на которых строилась рекомендация Z-B3:") t1 = {} for tid, it in items.items(): if it["src"] in ("春秋蝉",) or "ilversaint" in it["src"]: v = by.get(tid, {}) t1[it["src"]] = {k: v.get(k) for k in ("GOLD", "P2", "P5")} t1[it["src"]]["GOLD"] = it.get("gold") print(f" {it['src']:14s} эталон={it.get('gold')!r} P2={v.get('P2')!r} P5={v.get('P5')!r}") try: import pymorphy3 morph = pymorphy3.MorphAnalyzer() except Exception: morph = None t2 = {} if morph: per = collections.defaultdict(list) for tid, v in by.items(): for arm, dst in v.items(): if arm in ARMS: per[arm].append(nondict(dst, morph)) t2 = {arm: sum(x) / len(x) for arm, x in per.items() if x} d = t2.get("P2", 0) - t2.get("P5", 0) print(f"\n T2 прокси «целиком несловарный dst» по всем термам: " + " · ".join(f"{k} {t2[k]:.3f}" for k in ARMS if k in t2)) print(f" P2 − P5 = {d:+.3f} (порог {T2_MIN}) → " f"{'ЭФФЕКТ ЕСТЬ' if d >= T2_MIN else 'ЭФФЕКТА НЕТ'}") gold_ids = [tid for tid, it in items.items() if it.get("gold")] t3 = {} for arm in ARMS: hit = sum(1 for tid in gold_ids if norm(by.get(tid, {}).get(arm)) == norm(items[tid]["gold"])) have = sum(1 for tid in gold_ids if by.get(tid, {}).get(arm)) if have: t3[arm] = (hit, have) print("\n T3 совпадение с подписью владельца: " + " · ".join(f"{k} {v[0]}/{v[1]}" for k, v in t3.items())) p5hit = t3.get("P5", (0, 0))[0] print(f" P5 = {p5hit}/19 (порог ≥{T3_MIN}) → {'НЕ ХУЖЕ' if p5hit >= T3_MIN else 'ХУЖЕ, снятие правила вредит'}") rep["T"] = {"T1": t1, "T2": t2, "T3": {k: list(v) for k, v in t3.items()}} # ---------- J: слепая оценка на починенном наборе ----------------------------------------- if a.judge: print("\n=== J. СЛЕПАЯ ОЦЕНКА НА ПОЧИНЕННОМ НАБОРЕ ===") rep["J"] = {} for jf in a.judge: rows = json.loads(jf.read_text(encoding="utf-8"))["rows"] name = rows[0]["judge"] if rows else jf.stem k1, n1 = bws_dedup(rows, "K1") arms_only = {k: v for k, v in k1.items() if k in ARMS} order = sorted(arms_only.items(), key=lambda x: -x[1]) spread = (order[0][1] - order[-1][1]) if order else 0 print(f"\n судья {name}: термов с вердиктом {n1}") print(" K1: " + " · ".join(f"{k} {v:+.3f}" for k, v in order) + f" [DECOY {k1.get('DECOY', float('nan')):+.3f}] [GOLD {k1.get('GOLD', float('nan')):+.3f}]") print(f" разброс между армами: {spread:.3f} (в фазе B было 0.088 на семи одинаковых строках)") # C1 — декой лучшим dec_best = tot = 0 for r in rows: v = verdicts_of(r).get("K1") if not isinstance(v, dict): continue if "DECOY" not in {s for ss in r["letter2sources"].values() for s in ss}: continue tot += 1 if "DECOY" in r["letter2sources"].get(str(v.get("best")), []): dec_best += 1 c1 = dec_best / tot if tot else float("nan") # C2 — катастрофы: декой против подписи cat = collections.Counter() for r in rows: for L in r.get("catastrophic") or []: for s in r["letter2sources"].get(str(L), []): cat[s] += 1 # C3 — воспроизводит ли объективный слой names = [k for k, _ in order] c3 = names.index("P0b") >= len(names) - max(1, len(names) // 3) if "P0b" in names else False print(f" C1 декой назван ЛУЧШИМ: {dec_best}/{tot} = {c1:.3f} (порог ≤{C1_MAX}) → " f"{'ПРОШЁЛ' if c1 <= C1_MAX else 'ОТБРАКОВАН'}") print(f" C2 катастрофы: декой {cat.get('DECOY', 0)} против GOLD {cat.get('GOLD', 0)} → " f"{'ПРОШЁЛ' if cat.get('DECOY', 0) > cat.get('GOLD', 0) else 'ОТБРАКОВАН'}") print(f" C3 P0b в нижней трети по K1: {'ДА' if c3 else 'НЕТ'} (объективный слой: P0b худший)") # C3-бис: ранговое согласие с объективным слоем целиком, а не по одному арму. # Бинарный C3 отвечает «да/нет» про P0b и молчит о том, совпадает ли остальной порядок. common = [x for x in ARMS if x in arms_only and x in t3] if len(common) >= 3: jr = {a: k for k, a in enumerate(sorted(common, key=lambda x: -arms_only[x]))} orr = {a: k for k, a in enumerate(sorted(common, key=lambda x: -t3[x][0]))} n = len(common) d2 = sum((jr[a] - orr[a]) ** 2 for a in common) rho = 1 - 6 * d2 / (n * (n * n - 1)) print(f" C3-бис ранговая корреляция с объективным слоем (Спирмен): ρ = {rho:+.3f} " f"на {n} армах; судья: {sorted(common, key=lambda x: -arms_only[x])}; " f"объективно: {sorted(common, key=lambda x: -t3[x][0])}") rep["J"][name] = {"k1": k1, "spread": spread, "C1": c1, "C1_n": tot, "cat": dict(cat), "C3": bool(c3), "pass": bool(c1 <= C1_MAX and cat.get("DECOY", 0) > cat.get("GOLD", 0))} # H7 — грубая ошибка против тонкой разницы for name, d in rep["J"].items(): arms_only = {k: v for k, v in d["k1"].items() if k in ARMS} if arms_only and "DECOY" in d["k1"]: gap_arms = max(arms_only.values()) - min(arms_only.values()) gap_dec = statistics.median(arms_only.values()) - d["k1"]["DECOY"] print(f"\n H7 [{name}]: разрыв «медиана армов − декой» = {gap_dec:+.3f}; " f"разрыв «лучший − худший арм» = {gap_arms:.3f} → " f"{'подтверждается' if gap_dec > gap_arms else 'не подтверждается'}") # ---------- R: разброс прогона ------------------------------------------------------------- if a.repeat: print("\n=== R. РАЗБРОС ПРОГОНА (три повтора) ===") runs = collections.defaultdict(list) for f in a.repeat: b = load_arms([f]) for arm in ("P0b", "P1", "P2"): hit = sum(1 for tid in gold_ids if norm(b.get(tid, {}).get(arm)) == norm(items[tid]["gold"])) have = sum(1 for tid in gold_ids if b.get(tid, {}).get(arm)) if have: runs[arm].append(hit) for arm, xs in runs.items(): print(f" {arm:4s} повторы {xs} из 19 → медиана {statistics.median(xs):.1f}, " f"размах {max(xs)-min(xs)}") if "P0b" in runs and "P1" in runs: gap = statistics.median(runs["P1"]) - statistics.median(runs["P0b"]) spread = max(max(runs["P0b"]) - min(runs["P0b"]), max(runs["P1"]) - min(runs["P1"])) print(f" разрыв P0b↔P1 = {gap:.1f} терма; внутриармовый размах = {spread} терма → " f"{'ВЫВОД УСТОЯЛ' if gap > spread else 'ВЫВОД НЕ УСТАНОВЛЕН'}") rep["R"] = {"runs": dict(runs), "gap": gap, "spread": spread, "holds": bool(gap > spread)} # ---------- H6 ------------------------------------------------------------------------------- deg = collections.Counter() tot = collections.Counter() for tid, v in by.items(): s = items[tid]["sample"] tot[s] += 1 cands = {norm(x) for x in v.values()} if items[tid].get("gold"): cands.add(norm(items[tid]["gold"])) if len(cands) == 1: deg[s] += 1 print("\n=== H6. доля вырожденных термов по выборкам ===") for s in sorted(tot): print(f" {s}: {deg[s]}/{tot[s]} = {deg[s]/tot[s]:.2f}") rep["H6"] = {s: [deg[s], tot[s]] for s in tot} a.out.write_text(json.dumps(rep, ensure_ascii=False, indent=1), encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())