| """Deney sonuçlarının istatistiksel analizi (yalnızca standart kütüphane). |
| |
| * Öğe bazında tekrarların ortalaması alınır; karşılaştırmalar öğeler üzerinde eşleştirilmiş yapılır. |
| * Wilcoxon işaretli sıra testi: sıfır farklar atılır, bağlı sıralara ortalama sıra verilir; p değeri |
| n ≤ 25 için kesin dağılımdan (dinamik programlama), daha büyük n için normal yaklaşımla hesaplanır. |
| * Vargha–Delaney A12: B yapılandırmasının A'dan büyük değer üretme olasılığı (0,5 = fark yok; |
| |A12 − 0,5| ≥ 0,06 / 0,14 / 0,21 → küçük / orta / büyük etki). |
| """ |
| from __future__ import annotations |
|
|
| import math |
| from collections import defaultdict |
| from statistics import mean, stdev |
| from typing import Any, Iterable, Optional |
|
|
| from .benchmark import EvalState, ItemResult |
|
|
| METRICS = { |
| "final_compiled": "Derlenme (%)", |
| "final_pass_rate": "Geçen test (%)", |
| "final_coverage": "Satır kapsama (%)", |
| "mutation_score": "Mutasyon skoru (%)", |
| "seeded_rate": "Yerleştirilmiş hata yakalama (%)", |
| } |
|
|
|
|
| def _val(i: ItemResult, metric: str) -> Optional[float]: |
| if i.status != "done": |
| return None |
| if metric == "final_compiled": |
| return 100.0 if i.final_compiled else 0.0 |
| if metric == "seeded_rate": |
| return 100.0 * i.seeded_caught / i.seeded_total if i.seeded_total else None |
| if metric == "first_seeded_rate": |
| return None |
| return getattr(i, metric) |
|
|
|
|
| def per_item(ev: EvalState, metric: str, reps: Optional[set[int]] = None) -> dict[str, float]: |
| """Öğe → tekrarların ortalaması (değeri olmayan tekrarlar atlanır; `reps` verilirse yalnızca o tekrarlar).""" |
| acc: dict[str, list[float]] = defaultdict(list) |
| for i in ev.items: |
| if reps is not None and i.rep not in reps: |
| continue |
| v = _val(i, metric) |
| if v is not None: |
| acc[i.id].append(v) |
| return {k: mean(v) for k, v in acc.items() if v} |
|
|
|
|
| def rep_means(ev: EvalState, metric: str) -> list[float]: |
| """Her tekrarın (tüm öğeler üzerinden) ortalaması — tekrarlar arası yayılım için.""" |
| acc: dict[int, list[float]] = defaultdict(list) |
| for i in ev.items: |
| v = _val(i, metric) |
| if v is not None: |
| acc[i.rep].append(v) |
| return [mean(v) for _, v in sorted(acc.items()) if v] |
|
|
|
|
| def _ranks(values: list[float]) -> list[float]: |
| order = sorted(range(len(values)), key=lambda k: values[k]) |
| ranks = [0.0] * len(values) |
| i = 0 |
| while i < len(order): |
| j = i |
| while j + 1 < len(order) and values[order[j + 1]] == values[order[i]]: |
| j += 1 |
| for k in range(i, j + 1): |
| ranks[order[k]] = (i + j) / 2 + 1 |
| i = j + 1 |
| return ranks |
|
|
|
|
| def wilcoxon(a: list[float], b: list[float]) -> dict[str, Any]: |
| """Eşleştirilmiş iki taraflı Wilcoxon işaretli sıra testi (b − a).""" |
| d = [y - x for x, y in zip(a, b) if y - x != 0] |
| n = len(d) |
| if n == 0: |
| return {"n": 0, "w_plus": 0.0, "p": 1.0} |
| r = _ranks([abs(x) for x in d]) |
| w_plus = sum(rk for rk, x in zip(r, d) if x > 0) |
| if n <= 25: |
| |
| r2 = [int(round(2 * x)) for x in r] |
| total = sum(r2) |
| counts = [0] * (total + 1) |
| counts[0] = 1 |
| for x in r2: |
| for s in range(total, x - 1, -1): |
| counts[s] += counts[s - x] |
| w2 = int(round(2 * w_plus)) |
| denom = 2 ** n |
| lo = sum(counts[: w2 + 1]) / denom |
| hi = sum(counts[w2:]) / denom |
| p = min(1.0, 2 * min(lo, hi)) |
| else: |
| mu = n * (n + 1) / 4 |
| ties: dict[float, int] = defaultdict(int) |
| for rk in r: |
| ties[rk] += 1 |
| sigma = math.sqrt(n * (n + 1) * (2 * n + 1) / 24 - sum(t ** 3 - t for t in ties.values()) / 48) |
| z = (w_plus - mu) / sigma if sigma else 0.0 |
| p = math.erfc(abs(z) / math.sqrt(2)) |
| return {"n": n, "w_plus": w_plus, "p": round(p, 4)} |
|
|
|
|
| def a12(a: Iterable[float], b: Iterable[float]) -> Optional[float]: |
| """Vargha–Delaney A12: P(B > A) + 0,5·P(B = A).""" |
| a, b = list(a), list(b) |
| if not a or not b: |
| return None |
| gt = sum(1 for y in b for x in a if y > x) |
| eq = sum(1 for y in b for x in a if y == x) |
| return round((gt + 0.5 * eq) / (len(a) * len(b)), 3) |
|
|
|
|
| def magnitude(v: Optional[float]) -> str: |
| if v is None: |
| return "—" |
| d = abs(v - 0.5) |
| return "ihmal" if d < 0.06 else "küçük" if d < 0.14 else "orta" if d < 0.21 else "büyük" |
|
|
|
|
| def compare(a: EvalState, b: EvalState, metric: str) -> dict[str, Any]: |
| |
| |
| ra, rb = {i.rep for i in a.items}, {i.rep for i in b.items} |
| reps = None if ra == rb else (ra & rb) |
| pa, pb = per_item(a, metric, reps), per_item(b, metric, reps) |
| common = sorted(set(pa) & set(pb)) |
| xa, xb = [pa[k] for k in common], [pb[k] for k in common] |
| w = wilcoxon(xa, xb) |
| eff = a12(xa, xb) |
| return {"metric": metric, "items": len(common), "mean_a": round(mean(xa), 1) if xa else None, |
| "mean_b": round(mean(xb), 1) if xb else None, |
| "delta": round(mean(xb) - mean(xa), 1) if xa else None, |
| "p": w["p"], "a12": eff, "effect": magnitude(eff)} |
|
|
|
|
| def _fmt(v: Optional[float], suf: str = "") -> str: |
| return "—" if v is None else f"{v:.1f}{suf}" |
|
|
|
|
| def _msd(vals: list[float]) -> str: |
| if not vals: |
| return "—" |
| return f"{mean(vals):.1f}" + (f" ± {stdev(vals):.1f}" if len(vals) > 1 else "") |
|
|
|
|
| def _gpu(done: list[ItemResult], lang: str = "tr") -> str: |
| vals = [i.gpu_percent for i in done if i.gpu_percent is not None] |
| if not vals: |
| return "—" |
| low = min(vals) |
| if lang == "en": |
| return f"{low}%" + ("" if low == 100 else " ⚠ some runs spilled to CPU") |
| return f"%{low}" + ("" if low == 100 else " ⚠ CPU'ya taşan koşu var") |
|
|
|
|
| METRICS_EN = { |
| "final_compiled": "Compiled (%)", |
| "final_pass_rate": "Passing tests (%)", |
| "final_coverage": "Line coverage (%)", |
| "mutation_score": "Mutation score (%)", |
| "seeded_rate": "Seeded-bug detection (%)", |
| } |
| EFFECT_EN = {"ihmal": "negligible", "küçük": "small", "orta": "medium", "büyük": "large", "—": "—"} |
| A_TEXT = { |
| "tr": {"title": "# Deney analizi", |
| "intro": "Değerler, tekrarların ortalaması ± tekrarlar arası standart sapmadır. Karşılaştırmalar öğe bazında " |
| "(tekrar ortalamaları) eşleştirilmiş iki taraflı Wilcoxon işaretli sıra testi ve Vargha–Delaney A12 " |
| "etki büyüklüğü ile yapılmıştır.", |
| "s1": "## 1. Yapılandırmalar", "s2": "## 2. Eşleştirilmiş karşılaştırmalar", |
| "s3": "## 3. Bölümlere göre: geliştirme (dev), ayrılmış test ve bağımsız (QuixBugs)", |
| "sum_head": "| Yapılandırma | Model | Tur | Ablasyon | Bölüm | Tekrar | ", |
| "sum_tail": " | Ayıklanan (ort.) | Hata bildirimi kabul / ret | Ort. süre (sn) | GPU payı (en düşük) | Hız (token/sn) | Enerji (Wh/koşu) |", |
| "cmp_head": "| Karşılaştırma (A → B) | Ölçüt | Öğe | A | B | Δ (B − A) | Wilcoxon p | A12 | Etki |", |
| "split_head": "| Yapılandırma | Bölüm | Öğe | ", "round": "tur", "full": "tam"}, |
| "en": {"title": "# Experiment analysis", |
| "intro": "Values are the mean over repetitions ± the standard deviation across repetitions. Comparisons use " |
| "the paired two-sided Wilcoxon signed-rank test over items (repetition means) and the " |
| "Vargha–Delaney A12 effect size.", |
| "s1": "## 1. Configurations", "s2": "## 2. Paired comparisons", |
| "s3": "## 3. By split: development (dev), held-out test and independent (QuixBugs)", |
| "sum_head": "| Configuration | Model | Rounds | Ablation | Split | Repeats | ", |
| "sum_tail": " | Pruned (avg.) | Bug reports accepted / rejected | Avg. time (s) | GPU share (lowest) | Speed (tokens/s) | Energy (Wh/run) |", |
| "cmp_head": "| Comparison (A → B) | Metric | Items | A | B | Δ (B − A) | Wilcoxon p | A12 | Effect |", |
| "split_head": "| Configuration | Split | Items | ", "round": "round", "full": "full"}, |
| } |
|
|
|
|
| def _metrics(lang: str) -> dict[str, str]: |
| return METRICS_EN if lang == "en" else METRICS |
|
|
|
|
| def summary_rows(evals: list[EvalState], lang: str = "tr") -> list[str]: |
| A = A_TEXT["en" if lang == "en" else "tr"] |
| out = [A["sum_head"] + " | ".join(_metrics(lang).values()) + A["sum_tail"], |
| "|---|---|---|---|---|---|" + "---|" * len(METRICS) + "---|---|---|---|---|---|"] |
| for ev in evals: |
| done = [i for i in ev.items if i.status == "done"] |
| cells = [_msd(rep_means(ev, m)) for m in METRICS] |
| out.append( |
| f"| {ev.label or ev.id} | `{ev.model}` | {ev.config.get('max_iterations', '—')} | " |
| f"{', '.join(ev.config.get('ablate') or []) or '—'} | {ev.config.get('split', 'all')} | " |
| f"{ev.config.get('repeats', 1)} | " + " | ".join(cells) + " | " |
| f"{_fmt(mean([i.pruned_tests for i in done]) if done else None)} | " |
| f"{sum(i.suspected_bugs for i in done)} / {sum(i.rejected_bug_reports for i in done)} | " |
| f"{_fmt(mean([i.duration_s for i in done]) if done else None)} | " |
| f"{_gpu(done, lang)} | {_fmt(mean(tps) if (tps := [i.tokens_per_s for i in done if i.tokens_per_s]) else None)} | " |
| f"{_fmt(mean(en), '') if (en := [i.energy_wh for i in done if i.energy_wh is not None]) else '—'} |") |
| return out |
|
|
|
|
| def comparison_rows(pairs: list[tuple[str, EvalState, EvalState]], lang: str = "tr") -> list[str]: |
| out = [A_TEXT["en" if lang == "en" else "tr"]["cmp_head"], |
| "|---|---|---|---|---|---|---|---|---|"] |
| for name, a, b in pairs: |
| for m, label in _metrics(lang).items(): |
| c = compare(a, b, m) |
| out.append(f"| {name} | {label} | {c['items']} | {_fmt(c['mean_a'])} | {_fmt(c['mean_b'])} | " |
| f"{_fmt(c['delta'])} | {c['p']} | {c['a12'] if c['a12'] is not None else '—'} | " |
| f"{EFFECT_EN.get(c['effect'], c['effect']) if lang == 'en' else c['effect']} |") |
| return out |
|
|
|
|
| def find_pairs(evals: list[EvalState], lang: str = "tr") -> list[tuple[str, EvalState, EvalState]]: |
| """Otomatik eşleştirme: aynı model için tur 1 → tur>1 (ablasyonsuz) ve tam yapılandırma → her ablasyon.""" |
| def key(ev): |
| return ev.model, ev.config.get("split", "all") |
| full: dict[tuple, EvalState] = {} |
| base: dict[tuple, EvalState] = {} |
| abl: list[EvalState] = [] |
| for ev in sorted(evals, key=lambda e: e.created_at): |
| k = ev.config.get("max_iterations") or 1 |
| if ev.config.get("ablate"): |
| abl.append(ev) |
| elif k == 1: |
| base[key(ev)] = ev |
| else: |
| full[key(ev)] = ev |
| pairs = [] |
| for kk, b in full.items(): |
| if kk in base: |
| r = A_TEXT["en" if lang == "en" else "tr"]["round"] |
| pairs.append((f"{kk[0]}: {r} 1 → {r} {b.config.get('max_iterations')}", base[kk], b)) |
| for ev in abl: |
| f = full.get(key(ev)) |
| if f: |
| full_w = A_TEXT["en" if lang == "en" else "tr"]["full"] |
| pairs.append((f"{ev.model}: {full_w} → {'−' + ', −'.join(ev.config['ablate'])}", f, ev)) |
| return pairs |
|
|
|
|
| def split_rows(evals: list[EvalState], lang: str = "tr") -> list[str]: |
| out = [A_TEXT["en" if lang == "en" else "tr"]["split_head"] + " | ".join(_metrics(lang).values()) + " |", |
| "|---|---|---|" + "---|" * len(METRICS)] |
| for ev in evals: |
| if ev.config.get("ablate"): |
| continue |
| for split in ("dev", "test", "bagimsiz"): |
| sub = EvalState(id=ev.id, created_at=ev.created_at, model=ev.model, label=ev.label, config=ev.config, |
| items=[i for i in ev.items if i.split == split]) |
| if not sub.items: |
| continue |
| cells = [_fmt(mean(list(per_item(sub, m).values())) if per_item(sub, m) else None) for m in METRICS] |
| out.append(f"| {ev.label or ev.id} | {split} | {len({i.id for i in sub.items})} | " + " | ".join(cells) + " |") |
| return out |
|
|
|
|
| def render_analysis(evals: list[EvalState], lang: str = "tr") -> str: |
| A = A_TEXT["en" if lang == "en" else "tr"] |
| evals = sorted(evals, key=lambda e: e.created_at) |
| lines = [A["title"], "", A["intro"], "", A["s1"], ""] |
| lines += summary_rows(evals, lang) |
| pairs = find_pairs(evals, lang) |
| if pairs: |
| lines += ["", A["s2"], ""] + comparison_rows(pairs, lang) |
| lines += ["", A["s3"], ""] + split_rows(evals, lang) |
| return "\n".join(lines) + "\n" |
|
|
