"""app.py — the Perfect Audit Crime Challenge (Gradio Space). Participants download the challenge ledgers (Track A = ledger only; Track B = ledger + the ISA-520 / ISA-505 evidence layer), submit the journal entries they flag as fraudulent, and are scored against the held-out per-JE ground truth. The leaderboard reports the metric and — the scientific payoff — the recall per observability class, so the **mimetic perfect-crime family** is seen to be (near-)unbeatable in Track A and recoverable only with evidence in Track B. This is a research benchmark on **purely synthetic** data. A high score is not a deployable fraud detector; the un-catchable entries are precisely why audit must combine analytics with external evidence. It is not, and must not be read as, a recipe for committing fraud. Run locally: python app.py (reads private/all_labels.csv, writes leaderboard.csv) The private labels ship as a Space secret / private dataset — never in the public repo. """ from __future__ import annotations import os import re import threading from datetime import datetime, timezone from pathlib import Path import gradio as gr import pandas as pd from score import load_labels, load_submission, score # serialize the leaderboard read-modify-write within this process (single-replica Space) _SUBMIT_LOCK = threading.Lock() # team names are a join key (cap) and persisted to CSV → strict charset closes CSV/formula injection # and trailing-variant cap-evasion in one go (first char alphanumeric, so no leading =+-@/.). _TEAM_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.\- ]{0,39}$") HERE = Path(__file__).parent LABELS_PATH = Path(os.environ.get("CHALLENGE_LABELS", HERE / "private" / "all_labels.csv")) LEADERBOARD = Path(os.environ.get("CHALLENGE_LEADERBOARD", HERE / "leaderboard.csv")) DAILY_CAP = int(os.environ.get("CHALLENGE_DAILY_CAP", "20")) # On a public Space the held-out labels + the leaderboard live in a PRIVATE dataset the Space reads via # its HF_TOKEN secret (never in the public Space repo). Locally, both fall back to the paths above. PRIVATE_REPO = os.environ.get("CHALLENGE_PRIVATE_REPO") # e.g. "VynFi/perfect-audit-crime-private" HF_TOKEN = os.environ.get("HF_TOKEN") LB_COLUMNS = ["team", "track", "pr_auc", "f1", "recall", "precision", "mimetic_recall", "submitted_at"] def _resolve_labels_path() -> Path: if PRIVATE_REPO: from huggingface_hub import hf_hub_download return Path(hf_hub_download(PRIVATE_REPO, "all_labels.csv", repo_type="dataset", token=HF_TOKEN)) return LABELS_PATH def _save_leaderboard(lb: pd.DataFrame): lb.to_csv(LEADERBOARD, index=False) if PRIVATE_REPO: from huggingface_hub import upload_file upload_file(path_or_fileobj=str(LEADERBOARD), path_in_repo="leaderboard.csv", repo_id=PRIVATE_REPO, repo_type="dataset", token=HF_TOKEN) INTRO = """ # 🕵️ The Perfect Audit Crime Challenge Synthetic general ledgers with **unlabeled** planted fraud. Flag the fraudulent journal entries; the scorer ranks you on the held-out ground truth. **Two tracks, same ledgers:** - **Track A — ledger only.** Structured, relational and memory frauds are catchable. The *mimetic perfect crime* — fraud drawn from the ledger's own normal distribution — provably is **not**. - **Track B — ledger + evidence.** You also get the ISA-520 *expectations* and ISA-505 *evidence anchors*. Now the aggregate-inflation and fabricated-counterparty crimes become catchable. The gap between the two tracks on the mimetic family is the point: it maps *where ledger analytics ends and external evidence must begin*. *Research benchmark on purely synthetic data. A high score is not a deployable detector, and this is not a how-to for fraud — the un-catchable items are why assurance needs external evidence.* **Submission:** a CSV/JSON listing the `document_id`s you flag as fraudulent (optional `score` column for ranking). Metric: **PR-AUC** on the held-out labels (plus precision / recall / F1). """ def _labels() -> pd.DataFrame: path = _resolve_labels_path() if not Path(path).exists(): raise gr.Error(f"labels not found at {path} (set CHALLENGE_LABELS or CHALLENGE_PRIVATE_REPO)") return load_labels(path) def _leaderboard_df() -> pd.DataFrame: if PRIVATE_REPO: from huggingface_hub import hf_hub_download from huggingface_hub.utils import EntryNotFoundError try: p = hf_hub_download(PRIVATE_REPO, "leaderboard.csv", repo_type="dataset", token=HF_TOKEN, force_download=True) return pd.read_csv(p) except EntryNotFoundError: return pd.DataFrame(columns=LB_COLUMNS) # genuinely empty (first submission) except Exception as e: # a fetch error must NOT be read as "empty" — else the subsequent overwrite would wipe the # whole leaderboard. Fall back to the local cache if present, else refuse the submission. if LEADERBOARD.exists(): return pd.read_csv(LEADERBOARD) raise gr.Error(f"leaderboard temporarily unavailable; please retry ({type(e).__name__})") if LEADERBOARD.exists(): return pd.read_csv(LEADERBOARD) return pd.DataFrame(columns=LB_COLUMNS) def submit(team: str, track: str, file): team = (team or "").strip() if not _TEAM_RE.match(team): raise gr.Error("Team name must be 1–40 chars, start alphanumeric, and use only letters, " "digits, space, '_', '-', '.'.") if file is None: raise gr.Error("Please upload a submission file.") labels = _labels() res = score(load_submission(file.name), labels) track_code = "A" if track.startswith("A") else "B" mimetic = res["recall_by_family"].get("MimeticPerfectCrime", {}).get("recall", 0.0) row = { "team": team, "track": track_code, "pr_auc": res.get("pr_auc", float("nan")), "f1": res["f1"], "recall": res["recall"], "precision": res["precision"], "mimetic_recall": mimetic, "submitted_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), } # serialize the read-modify-write so concurrent submissions can't lost-update the leaderboard # (single-replica Space; multi-replica would need a revision-checked commit — noted in DESIGN). with _SUBMIT_LOCK: prev = _leaderboard_df() if not prev.empty: today = datetime.now(timezone.utc).strftime("%Y-%m-%d") if int(((prev["team"] == team) & prev["submitted_at"].astype(str).str.startswith(today)).sum()) >= DAILY_CAP: raise gr.Error(f"Daily submission cap ({DAILY_CAP}) reached for team '{team}'.") lb = pd.concat([prev, pd.DataFrame([row])], ignore_index=True) if not prev.empty else pd.DataFrame([row]) _save_leaderboard(lb) obs = "\n".join(f"- `{k}`: recall **{v['recall']}** ({v['caught']}/{v['total']})" for k, v in sorted(res["recall_by_observability"].items())) summary = ( f"### Scored — team **{team}**, Track **{track_code}**\n" f"- PR-AUC: **{res.get('pr_auc', 'n/a')}** | F1 {res['f1']} | recall {res['recall']} | " f"precision {res['precision']}\n" f"- flagged {res['submitted']} of {res['ledger_entries']} entries; " f"TP {res['true_positives']}, FP {res['false_positives']}\n" f"- **mimetic perfect-crime recall: {mimetic}** " f"(Track A near-zero by the theorem; Track B recoverable via evidence)\n\n" f"**Recall by observability class:**\n{obs}" ) return summary, _best_leaderboard() def _best_leaderboard() -> pd.DataFrame: lb = _leaderboard_df() if lb.empty: return lb lb = lb.sort_values("pr_auc", ascending=False, na_position="last") best = lb.groupby(["team", "track"], as_index=False).first() return best.sort_values(["track", "pr_auc"], ascending=[True, False]) def build(): with gr.Blocks(title="Perfect Audit Crime Challenge") as demo: gr.Markdown(INTRO) with gr.Row(): team = gr.Textbox(label="Team / handle") track = gr.Radio(["A — ledger only", "B — ledger + evidence"], value="A — ledger only", label="Track") file = gr.File(label="Submission (CSV/JSON of flagged document_ids)", file_types=[".csv", ".json"]) btn = gr.Button("Score submission", variant="primary") out = gr.Markdown() gr.Markdown("## 🏆 Leaderboard (best PR-AUC per team / track)") lb = gr.Dataframe(value=_best_leaderboard, interactive=False) btn.click(submit, [team, track, file], [out, lb]) return demo if __name__ == "__main__": build().launch()