#!/usr/bin/env python3 """ sentiqra-audit.py — production quality audit: stratified sampling plans and honest quality reports for analyzer traffic. The problem this prevents (the selective-labels problem): when only flagged cases, complaints, or successful appeals receive labels, an evaluation learns only about the cases somebody chose to look at. Reviewed alerts are not a representative sample of all traffic, and "no complaint received" is not "correct classification". A quality report built on that intake silently describes the flagged stratum, not production. This tool is customer-local — nothing leaves your environment. plan — draw a stratified audit sample over your own case records: Input (JSONL, one case per line): {"case_id": "c1", "stratum": "flagged", "signals": [...]} {"case_id": "c2", "stratum": "unflagged", "signals": []} {"case_id": "c3", "stratum": "abstained", "signals": []} {"case_id": "c4", "stratum": "failed", "error": "timeout"} strata: flagged / unflagged / abstained / failed — every production case belongs to exactly one; derive it from your stored responses (flagged = emitted a signal, abstained = completed with no signal, failed = analyzer did not complete). An optional "importance" field marks rare-but-decisive cases (e.g. a specific signal code or a language you care about) for deliberate oversampling. The plan is representative ACROSS strata first (each stratum is sampled), then oversamples marked-important cases if they remain rare in the base allocation. Every selected row records its selection_probability and weight = 1/p, so population estimates can account for the deliberate oversampling. Weighting corrects what was sampled — it cannot recover cases or outcomes never observed at all. --blind writes the review file WITHOUT the analyzer's prediction fields (signals/error) — reviewers form an initial judgment before seeing what the model said. The held-aside predictions go to --truth-out for later comparison. report — turn a labeled sample into an honest quality report: Sample rows evolve: reviewer fills "label" (immediate content judgment: correct / incorrect) and optionally "outcome" (delayed business result). Rows stay "pending" until judged — a pending row is never counted as negative, and a missing label is reported as missing, not folded into a denominator. The report discloses the sampling design, labeled/missing/pending counts per stratum, the population it can speak for, and — hard rule — emits full-population estimates ONLY when every stratum contributed labels. Labels covering only the flagged stratum yield flagged-conditional precision, not production accuracy. Feedback channels (reviewed alerts, customer complaints, successful appeals) are reported as intake — a self-selected, biased channel — and are never pooled into the population estimate. "No complaint" is recorded, not counted as correct. Usage: python3 sentiqra-audit.py plan --cases cases.jsonl \ --sample-size 400 --seed 7 \ --oversample-field importance --oversample-min 40 \ --blind --out sample.jsonl --truth-out held.jsonl # ... reviewers label rows (label: correct|incorrect; outcome later) python3 sentiqra-audit.py report --sample labeled.jsonl \ --intake complaints.jsonl --out quality-report.json Requires: Python 3.8+, no third-party packages. """ import argparse import hashlib import json import sys from collections import defaultdict SCHEMA_PLAN = "sentiqra/audit-plan/v1" SCHEMA_SAMPLE = "sentiqra/audit-sample/v1" SCHEMA_REPORT = "sentiqra/audit-report/v1" STRATA = ("flagged", "unflagged", "abstained", "failed") REVIEW_FIELDS = ("signals", "prediction", "error") # hidden under --blind def read_jsonl(path): rows = [] with open(path, encoding="utf-8") as fh: for i, line in enumerate(fh, 1): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError as e: sys.exit(f"{path}:{i}: invalid JSON: {e}") return rows def write_jsonl(path, rows): with open(path, "w", encoding="utf-8") as fh: for r in rows: fh.write(json.dumps(r, ensure_ascii=False) + "\n") def rank_key(seed: str, case_id: str) -> str: """Deterministic pseudo-random rank — same seed + same case set => the same sample, no hidden RNG state.""" return hashlib.sha256(f"{seed}\x00{case_id}".encode()).hexdigest() # --------------------------------------------------------------------------- # plan # --------------------------------------------------------------------------- def allocate(counts, sample_size, importance_idx, oversample_min): """Proportional allocation across all strata; then ensure marked- important cases (rare but decisive) reach a minimum allocation by stealing from the largest non-important slices. Returns {stratum_key: n}.""" strata = sorted(counts) total = sum(counts.values()) if sample_size > total: sample_size = total base = {s: max(1 if counts[s] and strata else 0, round(sample_size * counts[s] / total)) for s in strata} # every non-empty stratum must be represented — a zero allocation # would make that stratum's quality permanently unobservable for s in strata: if counts[s] and base[s] == 0: base[s] = 1 # importance oversampling: guarantee a floor for marked cases need = 0 for s in strata: imp = importance_idx.get(s, set()) if imp and base[s] < min(oversample_min, counts[s]): need += min(oversample_min, counts[s]) - base[s] while need > 0: donor = max((s for s in strata if base[s] > 1 and not importance_idx.get(s)), key=lambda s: base[s], default=None) if donor is None: break base[donor] -= 1 need -= 1 for s in strata: imp = importance_idx.get(s, set()) if imp: base[s] = max(base[s], min(oversample_min, counts[s])) return base def plan(cases, sample_size, seed, oversample_field, oversample_min, blind, out_path, truth_path): for i, c in enumerate(cases): if not c.get("case_id") or c.get("stratum") not in STRATA: sys.exit(f"case {i + 1}: needs 'case_id' and a stratum in " f"{list(STRATA)} (got {c.get('stratum')!r})") # every stratum present in the data must be samplable counts = defaultdict(int) importance_idx = defaultdict(set) by_stratum = defaultdict(list) for c in cases: s = c["stratum"] counts[s] += 1 if oversample_field and c.get(oversample_field): importance_idx[s].add(c["case_id"]) by_stratum[s].append(c) alloc = allocate(dict(counts), sample_size, importance_idx, oversample_min) sample, truth = [], [] design = {} for s in STRATA: pool = by_stratum.get(s, []) if not pool: design[s] = {"population": 0, "sampled": 0, "selection_probability": None} continue # important cases rank first within the stratum — deliberate # oversampling — then deterministic-random order imp = importance_idx.get(s, set()) ranked = sorted(pool, key=lambda c: ( 0 if c["case_id"] in imp else 1, rank_key(seed, c["case_id"]))) take = min(alloc[s], len(ranked)) p = take / len(ranked) design[s] = {"population": len(ranked), "sampled": take, "selection_probability": round(p, 8)} for c in ranked[:take]: row = {"schema": SCHEMA_SAMPLE, "case_id": c["case_id"], "stratum": s, "selection_probability": p, "weight": round(1 / p, 4), "label": "pending", "outcome": "pending"} if not blind: for f in REVIEW_FIELDS: if f in c: row[f] = c[f] else: row["blinded"] = True sample.append(row) held = {"case_id": c["case_id"], "stratum": s} for f in REVIEW_FIELDS: if f in c: held[f] = c[f] truth.append(held) header = {"schema": SCHEMA_PLAN, "seed": seed, "population": sum(counts.values()), "sample_size": sum(d["sampled"] for d in design.values()), "design": design, "oversample_field": oversample_field, "oversample_min": oversample_min, "blinded_review": blind, "notes": [ "Every stratum is sampled — flagged, unflagged, " "abstained, and failed cases can each be wrong.", "selection_probability records the deliberate " "oversampling; weight = 1/p lets population estimates " "correct for it. Weighting cannot recover cases that " "were never observable.", "Rows start pending — an unlabeled or awaiting-outcome " "case is never counted as a negative.", ]} if out_path: # the plan header is a single JSONL record so the file stays # valid line-delimited JSON end to end (report reads it back) with open(out_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(header, ensure_ascii=False) + "\n") for r in sample: fh.write(json.dumps(r, ensure_ascii=False) + "\n") if truth_path: write_jsonl(truth_path, truth) return {"header": header, "sample": sample, "truth": truth} # --------------------------------------------------------------------------- # report # --------------------------------------------------------------------------- def report(sample_rows, intake_rows, out_path): """Honest quality report. Emits full-population estimates ONLY when every sampled stratum contributed at least one immediate label — otherwise the report says which population it can speak for.""" rows = [r for r in sample_rows if r.get("schema") == SCHEMA_SAMPLE] if not rows: sys.exit("no audit-sample rows found — run `plan` first") strata = defaultdict(list) for r in rows: strata[r["stratum"]].append(r) per_stratum = {} labeled_total = pending = missing = 0 for s, rs in strata.items(): lab = [r for r in rs if r.get("label") in ("correct", "incorrect")] pend = [r for r in rs if r.get("label") == "pending"] miss = [r for r in rs if r.get("label") not in ("correct", "incorrect", "pending")] # weighted precision among labeled rows only — pending/missing # stay pending/missing, never counted as negative or positive w = sum(r["weight"] for r in lab) w_ok = sum(r["weight"] for r in lab if r["label"] == "correct") per_stratum[s] = { "sampled": len(rs), "labeled": len(lab), "pending": len(pend), "missing": len(miss), "precision_weighted": round(w_ok / w, 6) if w else None, "coverage": round(len(lab) / len(rs), 4), } labeled_total += len(lab) pending += len(pend) missing += len(miss) unlabeled_strata = [s for s, d in per_stratum.items() if d["labeled"] == 0] # delayed outcomes: tracked separately, never collapsed into labels outcome_labeled = [r for r in rows if r.get("outcome") in ("confirmed", "refuted")] outcome_pending = [r for r in rows if r.get("outcome") == "pending"] full_population = not unlabeled_strata population_estimate = None if full_population: w = sum(r["weight"] for r in rows if r.get("label") in ("correct", "incorrect")) w_ok = sum(r["weight"] for r in rows if r.get("label") == "correct") population_estimate = { "accuracy_weighted": round(w_ok / w, 6) if w else None, "basis": "weighted estimate over all sampled strata", } rep = { "schema": SCHEMA_REPORT, "population_represented": ( "all production strata (flagged, unflagged, abstained, " "failed)" if full_population else "only strata with completed labels — NOT all production " "traffic: no labels exist for " + ", ".join(unlabeled_strata)), "sampling": {s: d for s, d in per_stratum.items()}, "label_completion": { "labeled": labeled_total, "pending": pending, "missing": missing, "note": "pending and missing rows are disclosed, never " "counted as negatives or silently dropped", }, "outcome_completion": { "adjudicated": len(outcome_labeled), "pending": len(outcome_pending), "note": "delayed business outcomes are reported " "separately from immediate content labels; a case " "awaiting an outcome stays pending", }, "per_stratum_precision": { s: d["precision_weighted"] for s, d in per_stratum.items()}, "population_estimate": population_estimate, "not_reportable": (None if population_estimate else "full-traffic accuracy cannot be computed: " "strata without labels are unobservable, and " "weighting cannot recover cases never " "observed"), "intake_channels": { "rows": len(intake_rows), "role": "self-selected feedback (reviewed alerts, " "complaints, successful appeals) — disclosed as a " "biased intake channel and NEVER pooled into the " "population estimate; 'no complaint received' is " "not evidence of correct classification", }, "weighting_caveat": ( "Selection probabilities correct deliberate oversampling. " "They do not fix outcomes that remain fundamentally " "unobserved — an appeal, complaint, or label that never " "arrives is still missing."), } if out_path: with open(out_path, "w", encoding="utf-8") as fh: json.dump(rep, fh, indent=2) return rep # --------------------------------------------------------------------------- # Selftest # --------------------------------------------------------------------------- def _selftest() -> int: import tempfile, os failures = [] def check(name, cond, detail=""): if cond: print(f" ok {name}") else: failures.append(name) print(f" FAIL {name}: {detail}") # population: 1000 unflagged, 100 flagged, 30 abstained, 20 failed cases = ([{"case_id": f"u{i}", "stratum": "unflagged"} for i in range(1000)] + [{"case_id": f"f{i}", "stratum": "flagged", "signals": [{"code": "x"}]} for i in range(100)] + [{"case_id": f"a{i}", "stratum": "abstained"} for i in range(30)] + [{"case_id": f"e{i}", "stratum": "failed", "error": "timeout"} for i in range(20)] + [{"case_id": "rare1", "stratum": "flagged", "signals": [{"code": "x"}], "importance": "critical"}]) cases += [{"case_id": "rare_u", "stratum": "unflagged", "importance": "critical"}] res = plan(cases, sample_size=200, seed="7", oversample_field="importance", oversample_min=0, blind=False, out_path=None, truth_path=None) hdr, sample = res["header"], res["sample"] # 1. every stratum is sampled — including the rare failed/abstained by_s = defaultdict(int) for r in sample: by_s[r["stratum"]] += 1 check("all_strata_sampled", all(by_s[s] > 0 for s in STRATA), dict(by_s)) # 2. selection probabilities + weights recorded and consistent check("probabilities_recorded", all(0 < r["selection_probability"] <= 1 and abs(r["weight"] * r["selection_probability"] - 1) < 0.01 for r in sample)) # unflagged stratum is proportionally undersampled vs flagged pu = next(r["selection_probability"] for r in sample if r["stratum"] == "unflagged") pf = next(r["selection_probability"] for r in sample if r["stratum"] == "flagged") check("selection_probs_differ_by_stratum", pf > pu, (pf, pu)) # 3. deterministic — same seed reproduces the same sample res2 = plan(cases, 200, "7", "importance", 0, False, None, None) check("deterministic_seed", [r["case_id"] for r in res2["sample"]] == [r["case_id"] for r in sample]) # 4. oversampling: marked-important cases land first res3 = plan(cases, 50, "7", "importance", 5, False, None, None) marked = [r for r in res3["sample"] if r["case_id"] in ("rare1", "rare_u")] check("importance_oversampled", len(marked) == 2, [r["case_id"] for r in res3["sample"]][:10]) # 5. blind mode strips predictions into the truth file res4 = plan(cases, 20, "7", None, 0, True, None, None) check("blind_strips_predictions", all("signals" not in r and "error" not in r and r["blinded"] for r in res4["sample"])) check("truth_kept_aside", any(r["case_id"].startswith("f") for r in res4["truth"] if "signals" in r), res4["truth"][:3]) # 6. pending rows are never negatives: fully pending sample → # zero labeled, no precision, no population estimate rep = report(sample, [], None) check("pending_not_negative", rep["label_completion"]["pending"] == len(sample) and rep["label_completion"]["labeled"] == 0) check("no_labels_no_estimate", rep["population_estimate"] is None and rep["not_reportable"] is not None) # 7. flagged-only labels → flagged-conditional precision, NOT # full-traffic accuracy flagged_only = [dict(r, label="correct") if r["stratum"] == "flagged" else r for r in sample] rep2 = report(flagged_only, [], None) check("flagged_only_not_full_accuracy", rep2["population_estimate"] is None and "unflagged" in rep2["population_represented"] and rep2["per_stratum_precision"]["flagged"] == 1.0, rep2) # 8. full labeling → weighted population estimate exists full = [dict(r, label="correct") for r in sample] rep3 = report(full, [], None) check("full_labels_population_estimate", rep3["population_estimate"]["accuracy_weighted"] == 1.0 and rep3["population_represented"].startswith( "all production strata"), rep3["population_estimate"]) # 9. weighting recovers the population rate: inject labels so # flagged precision 0.5, unflagged others all correct — the # weighted average must NOT equal the flagged-only number mixed = [] for i, r in enumerate(sample): lab = "correct" if r["stratum"] == "flagged" and i % 2 == 0: lab = "incorrect" mixed.append(dict(r, label=lab)) rep4 = report(mixed, [], None) acc = rep4["population_estimate"]["accuracy_weighted"] check("weighted_estimate_not_flagged_only", acc > 0.5 and acc < 1.0, acc) # 10. outcomes tracked separately: label everything, leave outcomes # pending → adjudicated 0, pending N rep5 = report(full, [], None) check("outcomes_separate_and_pending", rep5["outcome_completion"]["adjudicated"] == 0 and rep5["outcome_completion"]["pending"] == len(sample)) # 11. intake channel disclosed, never pooled rep6 = report(full, [{"case_id": "x", "channel": "complaint"}] * 3, None) check("intake_disclosed_not_pooled", rep6["intake_channels"]["rows"] == 3 and "biased" in rep6["intake_channels"]["role"] and rep6["population_estimate"]["accuracy_weighted"] == 1.0) # 12. missing labels disclosed, not dropped some_missing = [dict(r, label=("correct" if i % 3 else None)) for i, r in enumerate(sample)] rep7 = report(some_missing, [], None) check("missing_labels_disclosed", rep7["label_completion"]["missing"] > 0, rep7["label_completion"]) # 13. weighting caveat present — oversampling math cannot recover # the unobservable check("weighting_caveat_present", "unobserved" in rep["weighting_caveat"] or "never" in rep["weighting_caveat"]) print(f"\n{len(failures)} failure(s)" if failures else "\nall selftest checks passed") return 1 if failures else 0 # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> int: ap = argparse.ArgumentParser( description="Production audit: stratified sampling plans and " "quality reports that cannot describe only flagged " "cases as 'production accuracy'") sub = ap.add_subparsers(dest="cmd") p = sub.add_parser("plan", help="draw a stratified audit sample") p.add_argument("--cases", required=True, help="JSONL cases: case_id + stratum " "(flagged/unflagged/abstained/failed) + optional " "importance flag") p.add_argument("--sample-size", type=int, required=True) p.add_argument("--seed", default="1", help="deterministic draw seed — same seed + cases = " "same sample") p.add_argument("--oversample-field", help="case field marking rare/important cases for " "deliberate oversampling (e.g. importance)") p.add_argument("--oversample-min", type=int, default=40, help="minimum sample for strata containing marked " "cases (default 40)") p.add_argument("--blind", action="store_true", help="write the review file WITHOUT predictions — " "reviewers judge first; predictions go to " "--truth-out") p.add_argument("--out", required=True) p.add_argument("--truth-out", help="held-aside predictions for post-judgment " "comparison (required with --blind)") r = sub.add_parser("report", help="turn a labeled sample into a quality report") r.add_argument("--sample", required=True, help="the labeled audit sample (JSONL)") r.add_argument("--intake", help="optional JSONL of self-selected feedback " "(complaints, appeals) — disclosed, never pooled") r.add_argument("--out") ap.add_argument("--selftest", action="store_true") args = ap.parse_args() if args.selftest: return _selftest() if args.cmd == "plan": if args.blind and not args.truth_out: ap.error("--blind requires --truth-out for the held-aside " "predictions") res = plan(read_jsonl(args.cases), args.sample_size, args.seed, args.oversample_field, args.oversample_min, args.blind, args.out, args.truth_out) hdr = res["header"] print(json.dumps({k: v for k, v in hdr.items() if k != "notes"}, indent=2)) print(f"\nwrote {args.out} ({hdr['sample_size']} rows from " f"{hdr['population']} cases)", file=sys.stderr) return 0 if args.cmd == "report": rep = report(read_jsonl(args.sample), read_jsonl(args.intake) if args.intake else [], args.out) print(json.dumps(rep, indent=2)) return 0 ap.error("a subcommand is required (plan, report)") if __name__ == "__main__": sys.exit(main())