"""Post-experiment analysis: stability, compliance, ANOVA, charts, export. Usage: python analysis_export.py - stability_score: mean pairwise token-Jaccard of the 3 repeat outputs per (model, level, sample) group, written back to each result row. - line-number compliance: share of runs whose review reported line numbers. - ANOVA + paired t-tests on detection_rate across model-level groups. - Charts (heatmap / boxplot / grouped bar) saved as PNG. - group_summary.csv + raw run-level export for thesis chapter 5. """ import base64 import csv import json import re import sys import uuid from collections import defaultdict from pathlib import Path from statistics import mean from app.analysis.service import AnalysisService from app.db import SessionLocal, init_db from app.models import ExperimentRun def token_set(text: str) -> set: return set(re.findall(r"[a-zA-Z_][a-zA-Z_0-9]*", (text or "").lower())) def jaccard(a: set, b: set) -> float: union = a | b return len(a & b) / len(union) if union else 1.0 def main() -> None: experiment_id = sys.argv[1] out_dir = Path(sys.argv[2]) out_dir.mkdir(parents=True, exist_ok=True) init_db() db = SessionLocal() svc = AnalysisService(db) runs = ( db.query(ExperimentRun) .filter(ExperimentRun.experiment_id == uuid.UUID(experiment_id), ExperimentRun.status == "done") .all() ) # --- stability: token-Jaccard across repeats --- groups = defaultdict(list) for run in runs: level = run.template_version.template.level groups[(run.model_id, level, str(run.sample_id))].append(run) for key, group_runs in groups.items(): texts = [r.result.raw_output for r in group_runs if r.result and r.result.raw_output] if len(texts) < 2: continue sets = [token_set(t) for t in texts] pairs = [(i, j) for i in range(len(sets)) for j in range(i + 1, len(sets))] score = mean([jaccard(sets[i], sets[j]) for i, j in pairs]) for r in group_runs: if r.result: r.result.stability_score = round(score, 4) db.commit() # --- per model-level summary --- summary = defaultdict(lambda: defaultdict(list)) compliance = defaultdict(lambda: [0, 0]) for run in runs: level = run.template_version.template.level key = (run.model_id, level) res = run.result if not res: continue if res.detection_rate is not None: summary[key]["detection_rate"].append(res.detection_rate) if res.false_positive_rate is not None: summary[key]["false_positive_rate"].append(res.false_positive_rate) if res.coverage_rate is not None: summary[key]["coverage_rate"].append(res.coverage_rate) if res.stability_score is not None: summary[key]["stability"].append(res.stability_score) pf = res.parsed_findings or {} verdict = pf.get("judge_verdict") if isinstance(pf, dict) else None if verdict: compliance[key][1] += 1 if verdict.get("lines_reported"): compliance[key][0] += 1 rows = [] for (model, level) in sorted(summary): g = summary[(model, level)] c = compliance[(model, level)] row = { "model": model, "level": level, "n": len(g["detection_rate"]), "detection_rate": round(mean(g["detection_rate"]), 4) if g["detection_rate"] else None, "false_positive_rate": round(mean(g["false_positive_rate"]), 4) if g["false_positive_rate"] else None, "coverage_rate": round(mean(g["coverage_rate"]), 4) if g["coverage_rate"] else None, "coverage_n": len(g["coverage_rate"]), "stability": round(mean(g["stability"]), 4) if g["stability"] else None, "line_compliance": round(c[0] / c[1], 4) if c[1] else None, } rows.append(row) print(row) with open(out_dir / "group_summary.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows) # --- per-language breakdown --- lang_summary = defaultdict(list) for run in runs: if run.result and run.result.detection_rate is not None: lang_summary[(run.model_id, run.sample.language)].append(run.result.detection_rate) lang_rows = [ {"model": m, "language": lang, "n": len(v), "detection_rate": round(mean(v), 4)} for (m, lang), v in sorted(lang_summary.items()) ] with open(out_dir / "language_summary.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=["model", "language", "n", "detection_rate"]) writer.writeheader() writer.writerows(lang_rows) # --- ANOVA + paired t-tests --- stats = {"anova": svc.run_anova(experiment_id), "paired_t_tests": {}} models = sorted({r.model_id for r in runs}) levels = sorted({r.template_version.template.level for r in runs}) for m in models: for a, b in [("L1", "L2"), ("L2", "L3"), ("L1", "L3")]: if a in levels and b in levels: key = f"{m}:{a} vs {b}" stats["paired_t_tests"][key] = svc.run_paired_t_test(f"{m}-{a}", f"{m}-{b}", experiment_id) with open(out_dir / "statistics.json", "w", encoding="utf-8") as f: json.dump(stats, f, ensure_ascii=False, indent=2) print("ANOVA:", stats["anova"]) # --- charts --- charts = svc.generate_charts(experiment_id) for name, b64 in charts.items(): try: (out_dir / f"{name}.png").write_bytes(base64.b64decode(b64)) print("chart saved:", name) except Exception as e: # noqa: BLE001 print(f"chart {name} failed: {e}") # --- raw run-level export --- with open(out_dir / "runs.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow(["run_id", "model", "level", "language", "sample_id", "repeat", "detection_rate", "false_positive_rate", "coverage_rate", "stability_score", "latency_ms", "total_tokens"]) for run in runs: res = run.result writer.writerow([ str(run.id), run.model_id, run.template_version.template.level, run.sample.language, str(run.sample_id), run.repeat_index, res.detection_rate, res.false_positive_rate, res.coverage_rate, res.stability_score, round(res.latency_ms or 0), (res.token_usage or {}).get("total_tokens"), ]) print("exports written to", out_dir) if __name__ == "__main__": main()