247 lines
14 KiB
Python
247 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""ОСЕВОЙ КАЛИБРАТОР СУДЕЙСКИХ СЕМЕЙСТВ. $0 — только чтение уже купленного сырья.
|
||
|
||
Отвечает на вопрос владельца «как синхронизировать судей разных семейств»: шкалу синхронизировать
|
||
нечем и не надо (риг уже нормирован контрастом внутри единицы, порогом из своего пола и декоем как
|
||
единицей измерения), а сверять надо ЧУВСТВИТЕЛЬНОСТЬ ПО ОСЯМ — долю пойманного дефекта ИЗВЕСТНОГО
|
||
типа и правильность отнесения его на ось.
|
||
|
||
Прибор состоит из трёх замеров, и все три считаются здесь:
|
||
|
||
1. АТРИБУЦИЯ. Посадка декоя восстанавливается побайтно (сверка sha1 с ключом), каждый посаженный
|
||
дефект локализуется в тексте, и по цитатам судьи проверяется: процитирован ли он и назван ли
|
||
под ТОЙ осью, к которой правило посадки относится. Числа по осям для этого не годятся —
|
||
прирост числа не доказывает, что судья увидел именно посаженное.
|
||
2. ДИНАМИЧЕСКИЙ ДИАПАЗОН. Уровень оси на боевых клетках и шум пола ПО ОСИ. Ось, у которой на
|
||
реальном тексте уровень ~0 и шум ровно 0, контраст армов нести не может, даже если посаженное
|
||
на ней ловится 16/16.
|
||
3. ЗАРАЖЕНИЕ. Ошибки в декой-клетке ВНЕ посаженных мест против той же клетки без порчи: сколько
|
||
семейство доначисляет тексту просто за то, что он уже пойман испорченным.
|
||
|
||
⚠ ЧЕГО ПРИБОР НЕ МЕРЯЕТ — см. вывод `границы`: посадка даёт ГРУБЫЕ дефекты, а спор семейств живёт
|
||
в маргинальном режиме, и на грубых обе семьи упираются в потолок.
|
||
|
||
Запуск: eval/.venv/bin/python eval/editor_tier/axiscal.py [tier]
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import re
|
||
import statistics as st
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ZONE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(ZONE))
|
||
|
||
import judge as J # noqa: E402
|
||
|
||
OUT = Path.home() / "books" / "editor-tier"
|
||
AXES = ("ВЕРНОСТЬ", "ТЕРМИН", "ЯЗЫК", "ФОРМА")
|
||
# Ось, к которой ПО ЗАМЫСЛУ относится каждое правило `blind-keys/PLANTS.json`. Порядок правил —
|
||
# порядок файла; правило, которого в файле нет, сюда не попадает.
|
||
RULE_AXIS = {0: "ВЕРНОСТЬ", 1: "ВЕРНОСТЬ", 2: "ВЕРНОСТЬ", 3: "ЯЗЫК", 4: "ФОРМА", 5: "ФОРМА"}
|
||
CTX = 40 # окно контекста вокруг посаженного места, в котором цитата считается попавшей
|
||
|
||
|
||
def plants() -> list:
|
||
return json.loads((OUT / "blind-keys" / "PLANTS.json").read_text(encoding="utf-8"))
|
||
|
||
|
||
def trace(text: str) -> tuple[str, list[dict]]:
|
||
"""Посадка с отслеживанием ФИНАЛЬНЫХ смещений: каждая следующая замена сдвигает прежние.
|
||
Без пересчёта смещений метка «где посажено» указывает не туда, и атрибуция врёт."""
|
||
out, marks = text, []
|
||
for i, (pat, rep, _c) in enumerate(plants()):
|
||
m = re.search(pat, out)
|
||
if not m:
|
||
continue
|
||
new = re.sub(pat, rep, m.group(0), count=1)
|
||
delta = len(new) - len(m.group(0))
|
||
for mk in marks:
|
||
if mk["beg"] >= m.end():
|
||
mk["beg"] += delta
|
||
mk["end"] += delta
|
||
marks.append(dict(rule=i, beg=m.start(), end=m.start() + len(new)))
|
||
out = out[:m.start()] + new + out[m.end():]
|
||
return out, marks
|
||
|
||
|
||
def quotes(why: str) -> list[tuple[str, str]]:
|
||
"""(ось, цитата). Ось — последняя названная перед цитатой: так пишут обе семьи, и на этом же
|
||
стоит приёмка ответа (`solscore.votes`), поэтому разбор не вводит нового допущения."""
|
||
res, cur = [], None
|
||
for tk in re.finditer(r"(ВЕРНОСТЬ|ТЕРМИН|ЯЗЫК|ФОРМА)|[«\"“„](.+?)[»\"”]", why):
|
||
if tk.group(1):
|
||
cur = tk.group(1)
|
||
elif cur:
|
||
res.append((cur, tk.group(2)))
|
||
return res
|
||
|
||
|
||
def norm(s: str) -> str:
|
||
return re.sub(r"\s+", " ", s).strip().lower().replace("ё", "е")
|
||
|
||
|
||
def votes_claude(key: dict) -> dict:
|
||
out = {}
|
||
for f in sorted((OUT / "aj-tier-votes").glob("*.json")):
|
||
v = json.loads(f.read_text(encoding="utf-8"))
|
||
out[next(iter(v.values()))["uid"]] = {m["arm"]: m for m in v.values()}
|
||
return out
|
||
|
||
|
||
def votes_sol(key: dict) -> dict:
|
||
"""Разбор ответов внешнего судьи. Числа и ПОЧЕМУ читаются теми же выражениями, что в
|
||
`solscore.votes`; здесь дополнительно СОХРАНЯЕТСЯ разбивка по осям, которую тот схлопывает."""
|
||
out = {}
|
||
for f in sorted((OUT / "sol-tier").glob("*.txt")):
|
||
k = key.get(f.stem)
|
||
if not k:
|
||
continue
|
||
txt, unit = f.read_text(encoding="utf-8"), {}
|
||
for lab, meta in k.items():
|
||
ax = {}
|
||
for a in AXES:
|
||
m = re.search(rf"^{lab}-{a}[ \t]*:[ \t]*(\d+)[ \t]*$", txt, re.M)
|
||
if not m:
|
||
break
|
||
ax[a] = int(m.group(1))
|
||
if len(ax) != len(AXES):
|
||
break
|
||
wm = re.search(rf"^{lab}-ПОЧЕМУ[ \t]*:[ \t]*(.*?)(?=^\S+-(?:{'|'.join(AXES)}|ПОЧЕМУ)"
|
||
rf"[ \t]*:|\Z)", txt, re.M | re.S)
|
||
unit[meta["arm"]] = dict(meta, why=(wm.group(1) if wm else "").strip(), **ax)
|
||
if len(unit) == len(k):
|
||
out[next(iter(k.values()))["uid"]] = unit
|
||
return out
|
||
|
||
|
||
def attribute(data: dict, hook, units: dict) -> tuple[dict, dict]:
|
||
"""(uid, правило) -> поймано-с-верной-осью · плюс счётчики цитат вне посадки."""
|
||
hit, tally = {}, dict(quotes=0, on=0, off=0, nomatch=0, base=0, decoy=0)
|
||
for uid, d in data.items():
|
||
if "CTRLdecoy" not in d:
|
||
continue
|
||
donor = d["CTRLdecoy"].get("decoy_from") or "R0"
|
||
planted, marks = trace(hook(donor, units[uid]))
|
||
sig = hashlib.sha1(planted.encode()).hexdigest()[:12] # noqa: S324
|
||
if sig != d["CTRLdecoy"]["sig"]:
|
||
raise SystemExit(f"⛔ посадка не восстановлена для {uid}: {sig} ≠ {d['CTRLdecoy']['sig']}")
|
||
pn = norm(planted)
|
||
tally["base"] += sum(d[donor][a] for a in AXES)
|
||
tally["decoy"] += sum(d["CTRLdecoy"][a] for a in AXES)
|
||
spans = [(norm(planted[max(0, mk["beg"] - CTX):mk["end"] + CTX]), mk) for mk in marks]
|
||
qs = []
|
||
for ax, q in quotes(d["CTRLdecoy"]["why"]):
|
||
tally["quotes"] += 1
|
||
nq = norm(q)
|
||
i = pn.find(nq)
|
||
if i < 0:
|
||
tally["nomatch"] += 1
|
||
continue
|
||
qs.append((ax, i, i + len(nq)))
|
||
for ctx, mk in spans:
|
||
c0 = pn.find(ctx)
|
||
over = [ax for ax, b, e in qs if c0 >= 0 and b < c0 + len(ctx) and e > c0]
|
||
hit[(uid, mk["rule"])] = int(RULE_AXIS[mk["rule"]] in over)
|
||
for _ax, b, e in qs:
|
||
on = any(pn.find(ctx) >= 0 and b < pn.find(ctx) + len(ctx) and e > pn.find(ctx)
|
||
for ctx, _ in spans)
|
||
tally["on" if on else "off"] += 1
|
||
return hit, tally
|
||
|
||
|
||
def mcnemar(b: int, c: int) -> float:
|
||
from math import comb # noqa: PLC0415
|
||
n = b + c
|
||
if n == 0:
|
||
return 1.0
|
||
return min(1.0, 2 * sum(comb(n, i) for i in range(min(b, c) + 1)) / 2 ** n)
|
||
|
||
|
||
def need_pairs(p1: float, p2: float, alpha: float = 0.0125, power: float = 0.80) -> int:
|
||
"""Парная схема (один и тот же посаженный дефект читают оба семейства) — McNemar.
|
||
Независимость промахов между семьями даёт МАКСИМУМ доли расхождений, то есть оценка сверху."""
|
||
za = {0.05: 1.96, 0.025: 2.2414, 0.0167: 2.394, 0.0125: 2.4977}[alpha]
|
||
zb = {0.80: 0.8416, 0.90: 1.2816}[power]
|
||
p01, p10 = p1 * (1 - p2), (1 - p1) * p2
|
||
psi, d = p01 + p10, abs(p01 - p10)
|
||
if d <= 0:
|
||
return 10 ** 9
|
||
return math.ceil((za * math.sqrt(psi) + zb * math.sqrt(psi - d * d)) ** 2 / d ** 2)
|
||
|
||
|
||
def main() -> int:
|
||
pass_ = (sys.argv[1:] or ["tier"])[0]
|
||
J.setup(pass_)
|
||
key = json.loads((OUT / "blind-keys" / f"{pass_}-KEY.json").read_text(encoding="utf-8"))
|
||
hook, units = J.text_of(pass_), {u["uid"]: u for u in J.units(pass_)}
|
||
fam = {"Claude": votes_claude(key), "Sol 5-6": votes_sol(key)}
|
||
uids = sorted(set.intersection(*(set(v) for v in fam.values())))
|
||
print(f"проход {pass_} · единиц у обоих семейств {len(uids)} · правил посадки {len(plants())}")
|
||
print(f"⚠ на ось ТЕРМИН посадки НЕТ: {sorted(set(RULE_AXIS.values()))}")
|
||
|
||
H, T = {}, {}
|
||
for n, d in fam.items():
|
||
H[n], T[n] = attribute({u: d[u] for u in uids}, hook, units)
|
||
print("\n1. АТРИБУЦИЯ ПОСАЖЕННОГО (поймано И названо верной осью)")
|
||
print(f" {'правило':<36} {'ось':<9} {'n':>3} " + " ".join(f"{n:>8}" for n in fam))
|
||
for i, (_p, _r, cmt) in enumerate(plants()):
|
||
n_i = sum(1 for (u, r) in H[next(iter(fam))] if r == i)
|
||
if not n_i:
|
||
continue
|
||
cols = [sum(v for (u, r), v in H[n].items() if r == i) / n_i for n in fam]
|
||
print(f" R{i} {cmt:<32} {RULE_AXIS[i]:<9} {n_i:>3} " + " ".join(f"{c:>8.2f}" for c in cols))
|
||
print(f"\n {'ось':<9} {'дефектов':>8} " + " ".join(f"{n:>8}" for n in fam) + " McNemar p")
|
||
names = list(fam)
|
||
for a in AXES:
|
||
keys = [k for k in H[names[0]] if RULE_AXIS[k[1]] == a]
|
||
if not keys:
|
||
print(f" {a:<9} {'—':>8} посадок нет — ось прибором НЕ ПОКРЫТА")
|
||
continue
|
||
b = sum(1 for k in keys if H[names[0]][k] and not H[names[1]][k])
|
||
c = sum(1 for k in keys if not H[names[0]][k] and H[names[1]][k])
|
||
cols = [sum(H[n][k] for k in keys) / len(keys) for n in names]
|
||
print(f" {a:<9} {len(keys):>8} " + " ".join(f"{x:>8.2f}" for x in cols)
|
||
+ f" {mcnemar(b, c):.3f} (расхождений {b}/{c})")
|
||
|
||
print("\n2. ДИНАМИЧЕСКИЙ ДИАПАЗОН ОСИ (уровень боевых клеток · шум пола · порог 2.8·sd/√n)")
|
||
for n, d in fam.items():
|
||
print(f" {n}:")
|
||
arms = [a for a in J.arms(pass_)]
|
||
for a in AXES:
|
||
lv = [d[u][x][a] for u in uids for x in arms if x in d[u]]
|
||
fl = [d[u]["CTRLfloorA"][a] - d[u]["CTRLfloorB"][a] for u in uids]
|
||
thr = 2.8 * st.pstdev(fl) / math.sqrt(len(fl))
|
||
dead = " ⛔ ОСЬ МЁРТВАЯ: контраст армов нести не может" if st.pstdev(fl) == 0 else ""
|
||
print(f" {a:<9} уровень {st.mean(lv):6.2f}/клетку · шум пола sd {st.pstdev(fl):4.2f}"
|
||
f" · порог {thr:4.2f}{dead}")
|
||
|
||
print("\n3. ЗАРАЖЕНИЕ (цитаты в декой-клетке ВНЕ посадки против той же клетки без порчи)")
|
||
for n in fam:
|
||
t = T[n]
|
||
print(f" {n:<8} база {t['base']:>4} · декой {t['decoy']:>4} · цитат {t['quotes']:>4} "
|
||
f"(на посадке {t['on']:>3} · вне {t['off']:>3} · не найдено в тексте {t['nomatch']:>3})"
|
||
f" → заражение {t['off'] - t['base']:+d}")
|
||
print(f" сигнал/фон посадки: {(t['decoy'] - t['base']) / max(t['base'], 1):.2f}")
|
||
|
||
print("\n4. ГРАНИЦЫ ПРИБОРА ПРИ ТЕКУЩЕМ ОБЪЁМЕ (парный McNemar, α=0.0125 Холм, мощность 0.80)")
|
||
for a in AXES:
|
||
keys = [k for k in H[names[0]] if RULE_AXIS[k[1]] == a]
|
||
if not keys:
|
||
continue
|
||
best = min(((d / 100, p / 100) for d in range(1, 100) for p in range(1, 100 - d)
|
||
if need_pairs(p / 100 + d / 100, p / 100) <= len(keys)), default=None)
|
||
got = f"{best[0]:.2f}" if best else ">0.99"
|
||
print(f" {a:<9} дефектов {len(keys):>3} → различим разрыв долей ≥ {got}")
|
||
print(" надо дефектов на ось: " + " · ".join(
|
||
f"Δ{p1 - p2:.2f} ({p1:.2f}/{p2:.2f}) → {need_pairs(p1, p2)}"
|
||
for p1, p2 in ((0.90, 0.60), (0.90, 0.70), (0.90, 0.80))))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|