185 lines
9.2 KiB
Python
185 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Полигон, пакет-7, фаза B: агрегация Best-Worst Scaling и ВЕРДИКТЫ F1–F4.
|
||
|
||
Счёт BWS (стандартный): по каждому критерию лучший вариант +1, худший −1; сумма по термам,
|
||
нормировка на число термов. Ноль означает «выбирают не чаще и не реже прочих».
|
||
|
||
Критерии провала названы ДО трат (§A5.6) и здесь только ПРОВЕРЯЮТСЯ, а не переформулируются:
|
||
F1 P0b (без контекстов) неотличим от P1 ⇒ контексты не работают
|
||
F1b P0 (СЛУЧАЙНЫЙ глоссарий) неотличим от P2 ⇒ прирост даёт факт подсказки, а не её правильность
|
||
F2 ни один арм не бьёт подписанный владельцем dst хотя бы вничью по K1 ⇒ роль не готова к авто
|
||
F3 арм проваливает ловушку S5 чаще, чем в 1 случае из 5 ⇒ авто-режим опасен
|
||
F4 согласие судей по K1 ниже порога ИЛИ leave-one-out по судье меняет победителя ⇒ мерить нечем
|
||
|
||
«Неотличим» операционализирован ДО просмотра данных: |ΔBWS| < 0.10 по K1 на общей выборке.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import itertools
|
||
import json
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
CRITS = ["K1", "K2", "K5", "K6", "K9"]
|
||
INDIST = 0.10 # порог «неотличимости» по BWS
|
||
ALPHA_FLOOR = 0.60 # порог согласия (F4)
|
||
|
||
|
||
def load_rows(paths: list[Path]) -> list[dict]:
|
||
rows = []
|
||
for p in paths:
|
||
rows.extend(json.loads(p.read_text(encoding="utf-8"))["rows"])
|
||
return rows
|
||
|
||
|
||
def bws(rows: list[dict], crit: str, samples: set[str] | None = None,
|
||
judges: set[str] | None = None) -> dict[str, float]:
|
||
plus, minus, seen = defaultdict(int), defaultdict(int), defaultdict(int)
|
||
for r in rows:
|
||
if samples and r["sample"] not in samples:
|
||
continue
|
||
if judges and r["judge"] not in judges:
|
||
continue
|
||
v = (r.get("verdicts") or {}).get(crit)
|
||
if not isinstance(v, dict):
|
||
continue
|
||
l2a = r["letter2arm"]
|
||
for arm in l2a.values():
|
||
seen[arm] += 1
|
||
b, w = l2a.get(str(v.get("best"))), l2a.get(str(v.get("worst")))
|
||
if b:
|
||
plus[b] += 1
|
||
if w:
|
||
minus[w] += 1
|
||
return {a: (plus[a] - minus[a]) / n for a, n in seen.items() if n}
|
||
|
||
|
||
def agreement_k1(rows: list[dict]) -> tuple[float, int]:
|
||
"""Доля совпадений выбора ЛУЧШЕГО по K1 между парами вердиктов на один терм.
|
||
|
||
Считаем простое попарное согласие по номинальной категории (какой АРМ признан лучшим):
|
||
оно интерпретируемо и не требует допущений, которых мы не проверяли. Krippendorff α по
|
||
номинальной шкале на двух оценщиках вырождается в ту же величину с поправкой на случайность,
|
||
поэтому рядом печатается и она.
|
||
"""
|
||
by_term = defaultdict(list)
|
||
for r in rows:
|
||
v = (r.get("verdicts") or {}).get("K1")
|
||
if isinstance(v, dict):
|
||
arm = r["letter2arm"].get(str(v.get("best")))
|
||
if arm:
|
||
by_term[r["id"]].append(arm)
|
||
agree = tot = 0
|
||
counts = defaultdict(int)
|
||
for _, arms in by_term.items():
|
||
for x, y in itertools.combinations(arms, 2):
|
||
tot += 1
|
||
agree += int(x == y)
|
||
for x in arms:
|
||
counts[x] += 1
|
||
n = sum(counts.values())
|
||
pe = sum((c / n) ** 2 for c in counts.values()) if n else 0
|
||
po = agree / tot if tot else 0
|
||
kappa = (po - pe) / (1 - pe) if tot and pe < 1 else float("nan")
|
||
return po, tot, kappa
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--judge", action="append", required=True, type=Path)
|
||
ap.add_argument("--termset", required=True, type=Path)
|
||
ap.add_argument("--out", required=True, type=Path)
|
||
a = ap.parse_args()
|
||
|
||
rows = load_rows(a.judge)
|
||
items = {i["id"]: i for i in json.loads(a.termset.read_text(encoding="utf-8"))}
|
||
fwd = [r for r in rows if not r.get("reverse")]
|
||
judges = sorted({r["judge"] for r in rows})
|
||
|
||
report: dict = {"n_verdict_rows": len(rows), "judges": judges}
|
||
|
||
print("=== BWS по критериям (все выборки, прямой порядок) ===")
|
||
table = {}
|
||
for c in CRITS:
|
||
s = bws(fwd, c)
|
||
table[c] = s
|
||
print(f" {c}: " + " ".join(f"{k}={v:+.3f}" for k, v in sorted(s.items(), key=lambda x: -x[1])))
|
||
report["bws"] = table
|
||
|
||
print("\n=== BWS по K1 в разрезе выборок ===")
|
||
per_sample = {}
|
||
for s in sorted({r["sample"] for r in fwd}):
|
||
d = bws(fwd, "K1", samples={s})
|
||
per_sample[s] = d
|
||
print(f" {s}: " + " ".join(f"{k}={v:+.3f}" for k, v in sorted(d.items(), key=lambda x: -x[1])))
|
||
report["bws_k1_by_sample"] = per_sample
|
||
|
||
k1 = table["K1"]
|
||
# --- F1 / F1b ---------------------------------------------------------------------------
|
||
d1 = abs(k1.get("P1", 0) - k1.get("P0b", 0))
|
||
d1b = abs(k1.get("P2", 0) - k1.get("P0", 0))
|
||
f1 = d1 < INDIST
|
||
f1b = d1b < INDIST
|
||
|
||
# --- F2: бьёт ли лучший арм подпись владельца (только термы, где GOLD участвовал) --------
|
||
gold_ids = {i for i, it in items.items() if it.get("gold")}
|
||
gold_rows = [r for r in fwd if r["id"] in gold_ids and "GOLD" in r["letter2arm"].values()]
|
||
k1g = bws(gold_rows, "K1")
|
||
best_arm = max((x for x in k1g.items() if x[0] != "GOLD"), key=lambda x: x[1], default=("—", -9))
|
||
f2 = best_arm[1] < k1g.get("GOLD", 0)
|
||
|
||
# --- F3: ловушка S5 — арм назван катастрофой или проиграл GOLD -----------------------------
|
||
s5 = [r for r in fwd if r["sample"] == "S5"]
|
||
cat = defaultdict(int)
|
||
for r in s5:
|
||
for L in r.get("catastrophic") or []:
|
||
arm = r["letter2arm"].get(str(L))
|
||
if arm:
|
||
cat[arm] += 1
|
||
n_s5 = len({r["id"] for r in s5})
|
||
f3 = any(v > 1 for v in cat.values())
|
||
|
||
# --- F4: согласие + leave-one-out по судье --------------------------------------------------
|
||
po, npairs, kappa = agreement_k1(fwd)
|
||
loo_winners = {}
|
||
for j in judges:
|
||
rest = [r for r in fwd if r["judge"] != j]
|
||
if rest:
|
||
d = bws(rest, "K1")
|
||
loo_winners[j] = max(d.items(), key=lambda x: x[1])[0] if d else None
|
||
overall_winner = max(k1.items(), key=lambda x: x[1])[0] if k1 else None
|
||
loo_flip = any(w != overall_winner for w in loo_winners.values() if w)
|
||
f4 = (kappa == kappa and kappa < ALPHA_FLOOR) or loo_flip
|
||
|
||
# --- своп ------------------------------------------------------------------------------------
|
||
rev = [r for r in rows if r.get("reverse")]
|
||
swap = {}
|
||
if rev:
|
||
kf, kr = bws(fwd, "K1"), bws(rev, "K1")
|
||
swap = {a: round(kr.get(a, 0) - kf.get(a, 0), 3) for a in set(kf) | set(kr)}
|
||
|
||
print("\n=== ВЕРДИКТЫ КРИТЕРИЕВ ПРОВАЛА (пороги назывались до трат) ===")
|
||
print(f" F1 P0b vs P1 по K1: Δ={d1:.3f} (порог {INDIST}) → {'ПРОВАЛ' if f1 else 'НЕ провал'}")
|
||
print(f" F1b P0 vs P2 по K1: Δ={d1b:.3f} (порог {INDIST}) → {'ПРОВАЛ' if f1b else 'НЕ провал'}")
|
||
print(f" F2 лучший арм {best_arm[0]}={best_arm[1]:+.3f} против GOLD={k1g.get('GOLD', float('nan')):+.3f} "
|
||
f"→ {'ПРОВАЛ' if f2 else 'НЕ провал'}")
|
||
print(f" F3 катастрофы на S5 ({n_s5} термов): {dict(cat) or 'нет'} → {'ПРОВАЛ' if f3 else 'НЕ провал'}")
|
||
print(f" F4 согласие по K1: po={po:.3f}, κ={kappa:.3f} на {npairs} парах; "
|
||
f"leave-one-out победитель {loo_winners} при общем {overall_winner} → {'ПРОВАЛ' if f4 else 'НЕ провал'}")
|
||
if swap:
|
||
print(f" своп (обратный порядок − прямой) по K1: {swap}")
|
||
|
||
report.update({"F1": {"delta": d1, "fail": f1}, "F1b": {"delta": d1b, "fail": f1b},
|
||
"F2": {"best_arm": best_arm[0], "best": best_arm[1],
|
||
"gold": k1g.get("GOLD"), "fail": f2, "n_terms": len({r['id'] for r in gold_rows})},
|
||
"F3": {"catastrophic": dict(cat), "n_terms": n_s5, "fail": f3},
|
||
"F4": {"po": po, "kappa": kappa, "pairs": npairs,
|
||
"loo": loo_winners, "winner": overall_winner, "fail": f4},
|
||
"swap_effect_k1": swap})
|
||
a.out.write_text(json.dumps(report, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|