File size: 4,397 Bytes
3fccb94 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | #!/usr/bin/env python3
"""Audit a MaizeGuard CSV manifest for group leakage and source concentration.
The manifest is expected to contain ``split``, ``label``, ``source_id``,
``record_id`` and ``group_id`` columns. The command prints a deterministic JSON
report and exits non-zero when a biological/collection group or record appears
in more than one split.
"""
from __future__ import annotations
import argparse
import csv
import json
from collections import Counter, defaultdict
from pathlib import Path
REQUIRED_COLUMNS = {"split", "label", "source_id", "record_id", "group_id"}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("manifest", type=Path, help="CSV split manifest to audit")
parser.add_argument(
"--max-source-share",
type=float,
default=0.80,
help="Flag classes where one source supplies more than this share (default: 0.80)",
)
return parser.parse_args()
def audit(path: Path, max_source_share: float) -> dict[str, object]:
groups: dict[tuple[str, str], set[str]] = defaultdict(set)
records: dict[str, set[str]] = defaultdict(set)
split_counts: Counter[str] = Counter()
label_counts: Counter[str] = Counter()
label_sources: dict[str, Counter[str]] = defaultdict(Counter)
with path.open(newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
missing = REQUIRED_COLUMNS - set(reader.fieldnames or ())
if missing:
raise ValueError(f"manifest is missing required columns: {sorted(missing)}")
for row in reader:
split = row["split"].strip()
label = row["label"].strip()
source = row["source_id"].strip()
record = row["record_id"].strip()
group = row["group_id"].strip()
if not all((split, label, source, record, group)):
raise ValueError("manifest contains an empty required field")
groups[(source, group)].add(split)
records[record].add(split)
split_counts[split] += 1
label_counts[label] += 1
label_sources[label][source] += 1
cross_split_groups = sorted(
(source, group, sorted(splits))
for (source, group), splits in groups.items()
if len(splits) > 1
)
cross_split_records = sorted(
(record, sorted(splits)) for record, splits in records.items() if len(splits) > 1
)
concentration: dict[str, object] = {}
concentrated_labels: list[str] = []
for label in sorted(label_sources):
total = label_counts[label]
source, count = label_sources[label].most_common(1)[0]
share = count / total
concentration[label] = {
"record_count": total,
"largest_source": source,
"largest_source_count": count,
"largest_source_share": round(share, 6),
}
if share > max_source_share:
concentrated_labels.append(label)
return {
"schema_version": "1",
"manifest": path.name,
"record_count": sum(split_counts.values()),
"group_count": len(groups),
"split_counts": dict(sorted(split_counts.items())),
"cross_split_group_count": len(cross_split_groups),
"cross_split_group_examples": [
{"source_id": source, "group_id": group, "splits": splits}
for source, group, splits in cross_split_groups[:20]
],
"cross_split_record_count": len(cross_split_records),
"cross_split_record_examples": [
{"record_id": record, "splits": splits}
for record, splits in cross_split_records[:20]
],
"class_source_concentration": concentration,
"classes_above_source_share_threshold": concentrated_labels,
"max_source_share_threshold": max_source_share,
"leakage_gate_passed": not cross_split_groups and not cross_split_records,
}
def main() -> int:
args = parse_args()
if not 0 < args.max_source_share <= 1:
raise SystemExit("--max-source-share must be greater than 0 and at most 1")
report = audit(args.manifest, args.max_source_share)
print(json.dumps(report, indent=2, sort_keys=True))
return 0 if report["leakage_gate_passed"] else 2
if __name__ == "__main__":
raise SystemExit(main())
|