← Source tree

audit_synthetic.py

"""Reproduce one deliberately narrow synthetic audit; no accuracy benchmark."""
import argparse
import copy
import hashlib
import json
import platform
from pathlib import Path

import sklearn
from reviewqueue import review


def make_row(rid, quantity, unit):
    return dict(record_id=rid, currency="USD", amount=str(quantity * unit),
                quantity=str(quantity))


def run_audit():
    reference = [make_row(f"R{q:02}{u:03}", q, u)
                 for q in range(1, 6) for u in range(90, 111, 2)]
    normal = [make_row(f"N{i:04}", 1 + i % 5, 91 + 2 * (i // 5))
              for i in range(25)]
    high = [make_row(f"H{i:04}", 1 + i % 5, 1000 + 100 * i)
            for i in range(10)]
    plausible = [dict(make_row(f"P{i:04}", 1 + i % 5, 100),
                      synthetic_note="plausible-valued but intentionally incorrect business fact; no semantic truth available")
                 for i in range(5)]
    payload = dict(reference=reference, records=normal + high + plausible)
    before = copy.deepcopy(payload)
    result = review(payload)
    assert payload == before, "Input mutated"
    assert result["input_rows"] == result["output_rows"] == 40
    assert [r["original"] for r in result["rows"]] == payload["records"]
    groups = []
    for name, start, end in [("ordinary_by_construction", 0, 25),
                             ("extreme_unit_amount", 25, 35),
                             ("plausible_numbers_with_unverifiable_fact", 35, 40)]:
        group = result["rows"][start:end]
        groups.append(dict(name=name, rows=len(group),
                           model_flags=sum("model_unusual_for_reference" in r["flags"] for r in group)))
    return dict(
        purpose="Reproduction of 15 October 2026 synthetic stress audit, not production accuracy",
        python=platform.python_version(), sklearn=sklearn.__version__,
        engine_sha256=hashlib.sha256(Path(__file__).with_name("reviewqueue.py").read_bytes()).hexdigest(),
        fixture_sha256=hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest(),
        groups=groups,
        preservation=dict(input_unchanged=True, rows=40, original_objects_order_and_text_ids_preserved=True),
        limitations=["All labels are synthetic construction choices, not independent reviewer judgments.",
                     "Counts are not population false-positive rates, recall, or business accuracy.",
                     "A numeric outlier model has no evidence to verify the business fact in a note.",
                     "The 0.6 conservative policy requires both a model outlier and a feature outside reference support; this audit is a development fixture, not independent validation.",
                     "Model flags are review hints, not findings of an error or fraud."],
        report=result)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    data = run_audit()
    encoded = (json.dumps(data, indent=2, ensure_ascii=False, allow_nan=False) + "\n").encode()
    with args.output.open("xb") as target:
        target.write(encoded)
    print(json.dumps({"groups": data["groups"], "preservation": data["preservation"]}, indent=2))


if __name__ == "__main__":
    main()