253 lines
14 KiB
Python
253 lines
14 KiB
Python
#!/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())
|