#!/usr/bin/env python3 """ sentiqra-prevalence.py — translate measured evaluation rates into expected alert volumes, missed cases, and review workload under YOUR prevalence assumptions. "95% detection" does not answer "how many incorrect alerts will my team triage?" Precision, recall, and false-positive rate answer different questions, and their operational meaning depends on the customer's traffic volume and the relative frequency of the target condition — not just the benchmark score. This tool computes the alert budget explicitly instead of leaving it implied. Inputs are strictly separated: MEASURED — rates from your own labeled evaluation: --eval-report FILE report.json from sentiqra-eval.py — pulls hit_measure (recall over positive items) and fp_measure (flag rate over 'negative' items) with their support counts and Wilson CIs or pass rates directly: --recall R [--recall-n N] --fpr F [--fpr-n N] Rates given without a support count are accepted but marked 'unverifiable': nobody can check them against a labeled set, and no uncertainty range can be computed. Benchmark or ranking numbers must NOT be pasted here — a signal score expresses match strength under a configuration, not a probability. ASSUMED — your traffic and operating conditions: --traffic N items per day --prevalence P estimated base rate of the target condition in YOUR traffic (0.001 or 0.1%) --prevalence-scenarios L alternative prevalences, comma-separated --review-minutes M average staff time per alert --reviewer-hourly C optional loaded cost per review hour Worked example (hypothetical values — not Sentiqra measurements): 100,000 items/day, 0.1% prevalence, 95% recall, 1% FPR -> relevant events 100, correct alerts 95, incorrect alerts 999, total alerts 1,094, alert precision ~8.7%, and at 3 minutes per review ~54.7 staff-hours/day. Run --selftest to verify the math. Usage: python3 sentiqra-prevalence.py \ --eval-report eval-report.json \ --traffic 100000 --prevalence 0.1% \ --prevalence-scenarios "0.05%,0.1%,0.5%" \ --review-minutes 3 --reviewer-hourly 45 --out workload.json Requires: Python 3.8+, no third-party packages. Fully local — no network calls. """ import argparse import json import math import sys SCHEMA_REPORT = "sentiqra/prevalence-report/v1" Z = 1.959963984540054 # 95% Wilson def wilson(k, n): """Wilson score interval -> (rate, lo, hi); (None, None, None) if n==0.""" if n == 0: return None, None, None rate = k / n denom = 1 + Z * Z / n center = (rate + Z * Z / (2 * n)) / denom margin = Z * math.sqrt(rate * (1 - rate) / n + Z * Z / (4 * n * n)) / denom return rate, max(0.0, center - margin), min(1.0, center + margin) def parse_rate(raw: str, name: str) -> float: """Accept '0.001', '0.1%', or '0.1' (fractions only above 1 fails).""" s = raw.strip() if s.endswith("%"): try: v = float(s[:-1]) / 100.0 except ValueError: sys.exit(f"{name} is not a number: {raw!r}") else: try: v = float(s) except ValueError: sys.exit(f"{name} is not a number: {raw!r}") if v > 1.0: sys.exit(f"{name} must be a fraction (0-1) or a percent " f"like 0.1% — got {raw!r}. Scores and counts are not " f"rates.") if not (0.0 <= v <= 1.0): sys.exit(f"{name} must be between 0 and 1 — got {raw!r}") return v def scenario(traffic, prevalence, recall, fpr, review_minutes, reviewer_hourly=None): """One row of the alert budget. All rates are fractions.""" relevant = traffic * prevalence negatives = traffic - relevant tp = relevant * recall # correct alerts fn = relevant * (1 - recall) # missed cases fp = negatives * fpr # incorrect alerts alerts = tp + fp precision = tp / alerts if alerts > 0 else None hours = alerts * review_minutes / 60.0 out = { "prevalence": prevalence, "relevant_per_day": relevant, "correct_alerts": tp, "missed_cases": fn, "incorrect_alerts": fp, "total_alerts": alerts, "alert_precision": precision, "review_hours_per_day": hours, } if reviewer_hourly is not None: out["review_cost_per_day"] = hours * reviewer_hourly return out def scenario_interval(traffic, prevalence, recall_ci, fpr_ci, review_minutes): """Propagate rate intervals through the alert budget using the adverse corners — low recall/high FPR bounds the bad end. Returns {field: [lo, hi]} or None when either interval is unavailable.""" if not recall_ci or not fpr_ci or None in recall_ci or None in fpr_ci: return None corners = [scenario(traffic, prevalence, r, f, review_minutes) for r in recall_ci for f in fpr_ci] fields = ("relevant_per_day", "correct_alerts", "missed_cases", "incorrect_alerts", "total_alerts", "review_hours_per_day") out = {k: [min(c[k] for c in corners), max(c[k] for c in corners)] for k in fields} precs = [c["alert_precision"] for c in corners if c["alert_precision"] is not None] out["alert_precision"] = ([min(precs), max(precs)] if precs else [None, None]) return out def fmt(v, digits=1): return "—" if v is None else f"{v:,.{digits}f}" def fmt_rate(v): return "—" if v is None else f"{v * 100:,.2f}%" def fmt_ci(ci, rate_fmt): return "" if not ci or None in ci else \ f" [{rate_fmt(ci[0])}–{rate_fmt(ci[1])}]" # --------------------------------------------------------------------------- # Measured-rate loading # --------------------------------------------------------------------------- def measured_from_eval_report(path: str): """Pull hit_measure (recall) and fp_measure from a sentiqra-eval report.json. These are rates computed over a labeled set — the only kind of input this tool accepts as 'measured'.""" try: with open(path, encoding="utf-8") as fh: rep = json.load(fh) except (OSError, json.JSONDecodeError) as e: sys.exit(f"cannot read eval report: {e}") if rep.get("report") != "sentiqra-local-eval": sys.exit('eval report must be a sentiqra-eval.py report.json ' '("report": "sentiqra-local-eval") — not a benchmark, ' 'dashboard export, or aggregated score table') def cell_rate(cell): if not isinstance(cell, dict) or cell.get("rate") is None: sys.exit("eval report is missing a usable rate cell") return {"rate": cell["rate"], "n": cell.get("n"), "ci95": cell.get("ci95"), "source": f"eval report ({path})"} return (cell_rate(rep["hit_measure"]), cell_rate(rep["fp_measure"])) def measured_from_cli(rate, n, name): if rate is None: return None entry = {"rate": rate, "n": n, "source": "--" + name} if n: _, lo, hi = wilson(round(rate * n), n) entry["ci95"] = [lo, hi] else: entry["ci95"] = None return entry # --------------------------------------------------------------------------- # Report # --------------------------------------------------------------------------- def build_report(measured, assumed, scenarios): return { "schema": SCHEMA_REPORT, "inputs": { "measured": measured, "assumed": assumed, }, "definitions": { "correct_alerts": "traffic x prevalence x recall", "incorrect_alerts": "traffic x (1 - prevalence) x FPR", "alert_precision": "correct / (correct + incorrect) — the " "share of alerts worth reviewing", "missed_cases": "traffic x prevalence x (1 - recall)", "intervals": "adverse-corner propagation of the measured " "rates' Wilson 95% intervals", }, "warnings": [ "Signal scores express match strength under a pinned " "configuration — they are not probabilities and must never " "be substituted for rates here.", "Alert precision is not recall: a high recall coexists with " "a low alert precision whenever prevalence is low.", "Prevalence, traffic, and review time are assumptions " "supplied by you — run multiple prevalence scenarios before " "committing to a workload estimate.", ], "scenarios": scenarios, } def print_report(rep): m, a = rep["inputs"]["measured"], rep["inputs"]["assumed"] print("\n=== Alert-budget estimate ===") print("measured inputs:") for name in ("recall", "fpr"): e = m[name] n = f" n={e['n']}" if e["n"] else " n=? (unverifiable — no support count)" ci = fmt_ci(e["ci95"], fmt_rate) print(f" {name:8s} {fmt_rate(e['rate'])}{ci} [{e['source']}]{'' if e['n'] else n}") print("assumed inputs (yours to change):") print(f" traffic {a['traffic_per_day']:,} items/day") print(f" review {a['review_minutes_per_alert']:g} min/alert" + (f" @ ${a['reviewer_hourly']:g}/h" if a.get("reviewer_hourly") is not None else "")) print("\n{:>10s} {:>8s} {:>8s} {:>8s} {:>11s} {:>10s} {:>11s}".format( "prevalence", "relevant", "correct", "missed", "incorrect", "precision", "rev hrs/day")) for s in rep["scenarios"]: row = s["point"] ci = s.get("intervals") or {} print("{:>10s} {:>8s} {:>8s} {:>8s} {:>11s} {:>10s} {:>11s}".format( fmt_rate(s["prevalence"]), fmt(row["relevant_per_day"], 0), fmt(row["correct_alerts"], 0), fmt(row["missed_cases"], 0), fmt(row["incorrect_alerts"], 0), fmt_rate(row["alert_precision"]), fmt(row["review_hours_per_day"], 1))) if ci: print("{:>10s} {:>8s} {:>8s} {:>8s} {:>11s} {:>10s} {:>11s}".format( "", "", fmt_ci([ci["correct_alerts"][0], ci["correct_alerts"][1]], lambda v: fmt(v, 0)).strip(), fmt_ci([ci["missed_cases"][0], ci["missed_cases"][1]], lambda v: fmt(v, 0)).strip(), fmt_ci([ci["incorrect_alerts"][0], ci["incorrect_alerts"][1]], lambda v: fmt(v, 0)).strip(), fmt_ci([ci["alert_precision"][0], ci["alert_precision"][1]], fmt_rate).strip(), fmt_ci([ci["review_hours_per_day"][0], ci["review_hours_per_day"][1]], lambda v: fmt(v, 1)).strip())) for w in rep["warnings"]: print(f"\nnote: {w}") # --------------------------------------------------------------------------- # Selftest # --------------------------------------------------------------------------- def _selftest() -> int: failures = [] def check(name, cond, detail=""): if cond: print(f" ok {name}") else: failures.append(name) print(f" FAIL {name}: {detail}") # 1. the documented worked example, exactly s = scenario(traffic=100_000, prevalence=0.001, recall=0.95, fpr=0.01, review_minutes=3) check("worked_example_counts", s["relevant_per_day"] == 100 and round(s["correct_alerts"], 1) == 95.0 and round(s["incorrect_alerts"], 1) == 999.0 and round(s["total_alerts"], 1) == 1094.0, s) check("worked_example_precision", abs(s["alert_precision"] - 0.0868) < 0.001, s["alert_precision"]) check("worked_example_hours", abs(s["review_hours_per_day"] - 54.7) < 0.05, s["review_hours_per_day"]) # 2. precision is not recall — low prevalence can invert the story hi_prev = scenario(100_000, 0.10, 0.95, 0.01, 3) check("prevalence_drives_precision", hi_prev["alert_precision"] > 0.9 and s["alert_precision"] < 0.1, (hi_prev["alert_precision"], s["alert_precision"])) # 3. interval propagation — adverse corners bound the bad end iv = scenario_interval(100_000, 0.001, [0.90, 0.98], [0.005, 0.02], 3) check("interval_brackets_point", iv["incorrect_alerts"][0] <= 999 <= iv["incorrect_alerts"][1] and iv["total_alerts"][0] <= 1094 <= iv["total_alerts"][1], iv) check("precision_interval_uses_adverse_corners", iv["alert_precision"][0] < s["alert_precision"] < iv["alert_precision"][1], iv["alert_precision"]) # 4. no intervals when support is missing check("intervals_none_without_support", scenario_interval(1, 0.1, [None, None], [0.1, 0.2], 3) is None) # 5. measured vs assumed separation survives into the report rep = build_report( measured={ "recall": {"rate": 0.95, "n": 200, "ci95": wilson(190, 200)[1:], "source": "eval report (x.json)"}, "fpr": {"rate": 0.01, "n": 1000, "ci95": wilson(10, 1000)[1:], "source": "eval report (x.json)"}}, assumed={"traffic_per_day": 100000, "review_minutes_per_alert": 3}, scenarios=[{"prevalence": 0.001, "point": s, "intervals": None}]) check("report_schema", rep["schema"] == SCHEMA_REPORT, rep["schema"]) check("inputs_separated", set(rep["inputs"]["measured"]) == {"recall", "fpr"} and "traffic_per_day" in rep["inputs"]["assumed"] and "traffic_per_day" not in rep["inputs"]["measured"]) check("warnings_present", any("not probabilities" in w for w in rep["warnings"]) and any("precision is not recall" in w for w in rep["warnings"])) # 6. unverifiable rate (no n) is flagged, not silently trusted e = measured_from_cli(0.95, None, "recall") check("no_support_flagged", e["ci95"] is None and e["n"] is None, e) # 7. cost appears only when reviewer hourly is given with_cost = scenario(100_000, 0.001, 0.95, 0.01, 3, reviewer_hourly=45) check("cost_only_when_rated", "review_cost_per_day" in with_cost and "review_cost_per_day" not in s, with_cost) check("cost_math", abs(with_cost["review_cost_per_day"] - 54.7 * 45) < 5, with_cost["review_cost_per_day"]) # 8. eval-report ingestion accepts only the measured-rate schema import tempfile, os good = {"report": "sentiqra-local-eval", "hit_measure": {"n": 200, "k": 190, "rate": 0.95, "ci95": [0.91, 0.97]}, "fp_measure": {"n": 1000, "k": 10, "rate": 0.01, "ci95": [0.005, 0.02]}} with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: json.dump(good, fh) good_path = fh.name r, f = measured_from_eval_report(good_path) check("eval_report_ingested", r["rate"] == 0.95 and r["n"] == 200 and f["rate"] == 0.01, (r, f)) bad = {"report": "some-dashboard-export", "rate": 0.95} with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: json.dump(bad, fh) bad_path = fh.name import subprocess try: measured_from_eval_report(bad_path) check("foreign_report_rejected", False, "no exit raised") except SystemExit: check("foreign_report_rejected", True) os.unlink(good_path) os.unlink(bad_path) # 9. rate parsing — percent and fraction, rejects scores > 1 check("parse_percent", parse_rate("0.1%", "x") == 0.001) check("parse_fraction", parse_rate("0.001", "x") == 0.001) for bad_in in ("1.5", "95", "abc"): try: parse_rate(bad_in, "x") check(f"reject_{bad_in}", False, "accepted") except SystemExit: check(f"reject_{bad_in}", True) # 10. zero-prevalence edge — no relevant events, alerts are all FPs z = scenario(50_000, 0.0, 0.95, 0.01, 3) check("zero_prevalence", z["relevant_per_day"] == 0 and z["correct_alerts"] == 0 and z["alert_precision"] == 0.0, z) 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="Translate measured eval rates into expected alerts, " "missed cases, and review workload under your " "prevalence assumptions") ap.add_argument("--selftest", action="store_true") ap.add_argument("--traffic", type=int, help="items per day (assumption)") ap.add_argument("--prevalence", help="estimated base rate of the target condition — " "fraction (0.001) or percent (0.1%%) (assumption)") ap.add_argument("--prevalence-scenarios", help="comma-separated alternative prevalences to " "report alongside the base assumption") ap.add_argument("--eval-report", help="sentiqra-eval.py report.json — supplies measured " "hit_measure (recall) and fp_measure (FPR) with " "support counts and Wilson intervals") ap.add_argument("--recall", help="measured recall over positive items " "(fraction or percent)") ap.add_argument("--recall-n", type=int, help="support count behind --recall (enables the " "uncertainty interval)") ap.add_argument("--fpr", help="measured false-positive rate over " "negative items (fraction or percent)") ap.add_argument("--fpr-n", type=int, help="support count behind --fpr") ap.add_argument("--review-minutes", type=float, help="average staff time per alert (assumption)") ap.add_argument("--reviewer-hourly", type=float, help="optional loaded cost per review hour (assumption)") ap.add_argument("--out", help="write the full report as JSON") args = ap.parse_args() if args.selftest: return _selftest() if args.eval_report and (args.recall or args.fpr): ap.error("--eval-report already supplies both measured rates — " "don't mix it with --recall/--fpr") if args.eval_report: recall_m, fpr_m = measured_from_eval_report(args.eval_report) else: if args.recall is None or args.fpr is None: ap.error("measured rates required: --eval-report FILE, or " "both --recall and --fpr") recall_m = measured_from_cli(parse_rate(args.recall, "--recall"), args.recall_n, "recall") fpr_m = measured_from_cli(parse_rate(args.fpr, "--fpr"), args.fpr_n, "fpr") for e, name in ((recall_m, "recall"), (fpr_m, "fpr")): if e["n"] is None: print(f"warning: {name} has no support count — no uncertainty " f"interval can be computed, and the rate is unverifiable " f"against a labeled set", file=sys.stderr) if not args.traffic or args.traffic <= 0: ap.error("--traffic is required and must be positive") if args.prevalence is None: ap.error("--prevalence is required — it is an assumption about " "YOUR traffic; the tool will not guess it") base_prev = parse_rate(args.prevalence, "--prevalence") if not (0 < base_prev < 1): ap.error("--prevalence must be in (0, 1) — a base rate of 0 or 1 " "makes the alert budget degenerate") if args.review_minutes is None or args.review_minutes <= 0: ap.error("--review-minutes is required and must be positive") prevs = [base_prev] if args.prevalence_scenarios: for raw in args.prevalence_scenarios.split(","): p = parse_rate(raw, "--prevalence-scenarios") if not (0 < p < 1): ap.error("each scenario prevalence must be in (0, 1)") if p not in prevs: prevs.append(p) prevs.sort() scenarios = [] for p in prevs: point = scenario(args.traffic, p, recall_m["rate"], fpr_m["rate"], args.review_minutes, args.reviewer_hourly) scenarios.append({ "prevalence": p, "point": point, "intervals": scenario_interval( args.traffic, p, tuple(recall_m["ci95"]) if recall_m["ci95"] else None, tuple(fpr_m["ci95"]) if fpr_m["ci95"] else None, args.review_minutes), }) rep = build_report( measured={"recall": recall_m, "fpr": fpr_m}, assumed={"traffic_per_day": args.traffic, "prevalence_base": base_prev, "review_minutes_per_alert": args.review_minutes, "reviewer_hourly": args.reviewer_hourly}, scenarios=scenarios) print_report(rep) if args.out: with open(args.out, "w", encoding="utf-8") as fh: json.dump(rep, fh, indent=2) print(f"\nwrote {args.out}") return 0 if __name__ == "__main__": sys.exit(main())