"""Replay expanded graph-call endpoints and diagnose fallback/selection effects.""" import collections import hashlib import itertools import json import sys from pathlib import Path B = Path(__file__).resolve().parent E = B.parent CV = E / "pangenome-reference-cv-20261014" sys.path.insert(0, str(CV / "asian-seven-comparison")) from endpoints import canonical_slots, named_pair, unmatched_cores GENES = ["A", "B", "C", "DPA1", "DPB1", "DQA1", "DQB1", "DRB1"] SOURCES = {} def load(path): raw = path.read_bytes() SOURCES[str(path.relative_to(E))] = hashlib.sha256(raw).hexdigest() return json.loads(raw) def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def totals(rows): return {"donors": len({r["donor"] for r in rows}), "eligible": sum(r["four_field"]["eligible"] for r in rows), "correct": sum(r["four_field"]["correct"] for r in rows), "called_on_eligible": sum(r["four_field"]["eligible"] and r["four_field"]["called"] for r in rows), "unnamed": sum(r["unmatched"]["eligible_copies"] for r in rows), "unnamed_recovered": sum(r["unmatched"]["exact_copies_recovered"] for r in rows), "decision_counts": dict(collections.Counter(r.get("reason", "native") for r in rows))} def main(): native = load(E / "track-a-scoring-20261018/TRACK_A_RESULT.json") fold = load(CV / "FOLDS.json") truth = collections.defaultdict(list) for r in load(CV / "SOURCE_RECORDS.json")["records"]: truth[r["donor"], r["gene"]].append(r) roots = [ (CV / comp / "partial-inputs" / CV.name / stage / "runs", "DogoHLA") for comp, stage in [ ("pilot-comparison", "inference-v2"), ("na18940-comparison", "inference-na18940"), ("japanese-comparison-v3", "inference-japanese"), ("asian-seven-comparison", "inference-asian-seven"), ] ] + [(CV / "collected/inference-expansion41/runs", "DogoHLA")] roots += [(CV / "collected/inference-expansion26/runs", "DogoHLA"), (E / "genomic-graph-rebase-20261020/collected/runs", "DogoHLA_genomic"), (E / "genomic-graph-rebase-remaining26-20261020/collected/runs", "DogoHLA_genomic")] rows = [] missing = [] available = set() for root, prefix in roots: if not root.exists(): continue for ddir in sorted(root.iterdir()): if not ddir.is_dir(): continue donor = ddir.name for panel in ("hprc", "hprc_asian"): arm = ddir / panel if not (arm / "COMPLETE.json").exists(): missing.append(str(arm.relative_to(E))) continue receipt = load(arm / "COMPLETE.json") assert receipt["status"] == "CV_inference_complete" assert receipt["donor"] == donor and receipt["panel"] == panel for mode in ("E1", "E2", "E3"): needed = ["calls.json", "decisions.json", "sequence_calls.json"] if not all((arm / mode / n).exists() for n in needed): missing.append(str((arm / mode).relative_to(E))) continue payload = {} for name in needed: path = arm / mode / name assert sha(path) == receipt["output_sha256"][mode + "/" + name] payload[name] = load(path) method = f"{prefix}/{panel}/{mode}" for gene in GENES: copies = truth[donor, gene] assert len(copies) == 2 key = donor, gene, method assert key not in available available.add(key) call = payload["calls.json"].get(gene, {}) slots = canonical_slots(call) if len(slots) != 2: slots = [{"labels": [], "label_unknown": True}] * 2 seq = payload["sequence_calls.json"][gene]["slots"] for s, q in zip(slots, seq): s.update(q) dec = payload["decisions.json"]["decisions"].get(gene, {}) rows.append({ "donor": donor, "gene": gene, "method": method, "population": fold["donors"][donor]["population"], "four_field": named_pair(slots, copies), "unmatched": unmatched_cores(slots, copies), "reason": dec.get("reason", "no_inference_decision"), "score_gap": dec.get("score_gap"), "origin": payload["sequence_calls.json"][gene]["origin"], }) graph_methods = sorted({r["method"] for r in rows}) for r in native["rows"]: rows.append(dict(r, method="T1K/" + r["arm"])) methods = sorted({r["method"] for r in rows}) index = {(r["donor"], r["gene"], r["method"]): r for r in rows} contrasts = [] for method in graph_methods: panel = method.split("/")[1] same_panel = ("genomic_" if method.startswith("DogoHLA_genomic/") else "") + panel for baseline in ("T1K/" + same_panel, "T1K/genomic", "T1K/genomic_core_" + panel): matched = [(r, index[r["donor"], r["gene"], baseline]) for r in rows if r["method"] == method and r["four_field"]["eligible"] and (r["donor"], r["gene"], baseline) in index] reason_counts = collections.defaultdict(collections.Counter) for a, z in matched: ac, zc = a["four_field"]["correct"], z["four_field"]["correct"] reason_counts[a["reason"]][ "both_correct" if ac and zc else "graph_only_correct" if ac else "baseline_only_correct" if zc else "both_wrong"] += 1 gains = sum(a["four_field"]["correct"] and not z["four_field"]["correct"] for a, z in matched) losses = sum(z["four_field"]["correct"] and not a["four_field"]["correct"] for a, z in matched) contrasts.append({ "method": method, "baseline": baseline, "pairs": len(matched), "donors": len({a["donor"] for a, z in matched}), "gains": gains, "losses": losses, "difference_pp": 100 * (gains - losses) / len(matched) if matched else None, "by_decision_reason": dict(reason_counts), }) old = load(CV / "SHORT_READ19_COMPARISON.json") for expected in old["totals"]: if not expected["method"].startswith("DogoHLA_hybrid/"): continue method = expected["method"].replace("DogoHLA_hybrid/", "DogoHLA/") actual = totals([r for r in rows if r["method"] == method and r["donor"] in old["donors"]]) assert (actual["eligible"], actual["correct"]) == ( expected["eligible_genotypes"], expected["correct_genotypes"]) result = { "status": "verified_available_graph_results", "missing_outputs": missing, "summary": {m: totals([r for r in rows if r["method"] == m]) for m in methods}, "by_gene": {g: {m: totals([r for r in rows if r["method"] == m and r["gene"] == g]) for m in methods} for g in GENES}, "contrasts": contrasts, "rows": rows, "original_cv19_named_results_reproduced": True, "track_result_sha256": sha(E / "track-a-scoring-20261018/TRACK_A_RESULT.json"), "sources_sha256": SOURCES, "script_sha256": sha(Path(__file__)), "limitations": [ "DogoHLA uses original.fa/flanked hybrid models; DogoHLA_genomic uses full-genomic/flanked models. Comparison to genomic-only baseline is competitive, not a topology-only ablation.", "Native fallback and graph retrieval are mixed; decision counts identify this conditional behavior.", "Developmental/exposed cohort; no new raw-read truth certification.", "Available completed outputs, not all planned donors; missing runs are distinct from no-calls.", "Descriptive mode comparisons; no selection of a winning mode or inferential p-values.", ], } (B / "GRAPH_RESULTS.json").write_text(json.dumps(result, indent=2) + "\n") print(json.dumps(result["summary"], indent=2)) if __name__ == "__main__": main()