#!/usr/bin/env python3 """Wave 46 (L'Arbitro Verificato): independent referee reproduction. Proof that the referee report is chain-derivable, not "trust our DB": 1. Downloads the public report from GET {base}/v1/referee. 2. For every merchant whose verdict is "flagged", queries the PUBLIC Algorand indexer independently (keyless REST, same endpoint family the scanner uses) for the merchant's inbound USDC transfers of the last 30 days. 3. Recomputes single_payer_dominance and burst_pattern from that independent data and prints an agreement/disagreement table. Exit code 0 when every recomputed heuristic agrees with the published flag, 1 when at least one disagrees (or the report/indexer is unreachable). Stdlib + httpx only. Usage: python3 scripts/referee_reproduce.py [base_url] [--indexer URL] [--asa ID] [--timeout SECONDS] """ from __future__ import annotations import argparse import sys from collections import Counter from datetime import datetime, timedelta import httpx DEFAULT_BASE_URL = "https://legit.gonna.bond" DEFAULT_INDEXER_URL = "https://mainnet-idx.algonode.cloud" DEFAULT_USDC_ASA = 31566704 # Mirrors of app.referee constants (methodology v1.0.0). Kept as literals # on purpose: the reproduction must recompute from the PUBLISHED # thresholds in the report, and these are the fallback when a field is # absent. When they drift from app.referee, METHODOLOGY_VERSION drifts too. FALLBACK_SINGLE_PAYER_SHARE = 0.8 FALLBACK_SINGLE_PAYER_MIN = 10 FALLBACK_BURST_SHARE = 0.8 FALLBACK_BURST_WINDOW_HOURS = 24 FALLBACK_BURST_MIN = 10 RECEIPTS_WINDOW_DAYS = 30 _PAGE_LIMIT = 100 _MAX_PAGES = 20 def fetch_report(base_url: str, timeout: float) -> dict: """Download the published referee report.""" url = f"{base_url.rstrip('/')}/v1/referee" response = httpx.get(url, timeout=timeout) response.raise_for_status() payload = response.json() if not isinstance(payload, dict): raise ValueError("referee report is not a JSON object") return payload def fetch_inbound_axfers(indexer_url: str, address: str, asa_id: int, after: datetime, timeout: float) -> list[dict]: """The merchant's inbound USDC transfers since ``after``, independently. Same filters as indexer.algorand_chain: receiver == address, amount > 0, sender != address. Paginates via next-token. Raises on HTTP failure (the caller reports the merchant as not reproducible). """ url = f"{indexer_url.rstrip('/')}/v2/accounts/{address}/transactions" params: dict = { "asset-id": asa_id, "tx-type": "axfer", "after-time": after.strftime("%Y-%m-%dT%H:%M:%SZ"), "limit": _PAGE_LIMIT, } hits: list[dict] = [] with httpx.Client(timeout=timeout) as client: for _ in range(_MAX_PAGES): response = client.get(url, params=params) response.raise_for_status() payload = response.json() for tx in payload.get("transactions") or []: axfer = tx.get("asset-transfer-transaction") or {} receiver = str(axfer.get("receiver") or "") sender = str(tx.get("sender") or "") amount = int(axfer.get("amount") or 0) if receiver != address or amount <= 0 or sender == address: continue round_time = int(tx.get("round-time") or 0) if round_time <= 0: continue hits.append({"payer": sender, "ts": round_time}) token = payload.get("next-token") if not token: break params["next"] = token return hits def recompute_heuristics(hits: list[dict], thresholds: dict) -> dict: """Recompute single_payer_dominance and burst_pattern from raw hits. Pure function (unit-tested with fixtures): ``hits`` is a list of {"payer": str, "ts": epoch_seconds}; ``thresholds`` comes from the published report so the reproduction follows the same methodology. """ dominance_cfg = thresholds.get("single_payer_dominance") or {} burst_cfg = thresholds.get("burst_pattern") or {} dom_share = float(dominance_cfg.get("share", FALLBACK_SINGLE_PAYER_SHARE)) dom_min = int(dominance_cfg.get("min_receipts", FALLBACK_SINGLE_PAYER_MIN)) burst_share_threshold = float(burst_cfg.get("share", FALLBACK_BURST_SHARE)) burst_hours = int(burst_cfg.get("window_hours", FALLBACK_BURST_WINDOW_HOURS)) burst_min = int(burst_cfg.get("min_receipts", FALLBACK_BURST_MIN)) total = len(hits) result = { "receipts_30d": total, "single_payer_dominance": {"flagged": False, "share": 0.0, "payer": None}, "burst_pattern": {"flagged": False, "share": 0.0}, } if total >= dom_min: payer, count = Counter(h["payer"] for h in hits).most_common(1)[0] share = count / total result["single_payer_dominance"] = { "flagged": share > dom_share, "share": round(share, 4), "payer": payer, } if total >= burst_min: ordered = sorted(int(h["ts"]) for h in hits) window = burst_hours * 3600 best = 0 left = 0 for right, ts in enumerate(ordered): while ordered[left] < ts - window: left += 1 best = max(best, right - left + 1) share = best / total result["burst_pattern"] = { "flagged": share >= burst_share_threshold, "share": round(share, 4), } return result def _published_flags(entry: dict) -> set[str]: return { str(ev.get("heuristic")) for ev in entry.get("evidence", []) if isinstance(ev, dict) and ev.get("heuristic") } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("base_url", nargs="?", default=DEFAULT_BASE_URL) parser.add_argument("--indexer", default=DEFAULT_INDEXER_URL) parser.add_argument("--asa", type=int, default=DEFAULT_USDC_ASA) parser.add_argument("--timeout", type=float, default=15.0) args = parser.parse_args(argv) try: report = fetch_report(args.base_url, args.timeout) except Exception as exc: print(f"ERROR: could not fetch {args.base_url}/v1/referee: {exc}") return 1 thresholds = report.get("thresholds") or {} flagged = [ m for m in report.get("merchants", []) if m.get("verdict") == "flagged" and m.get("network") == "algorand-mainnet" ] print(f"Report methodology v{report.get('methodology_version')}, " f"generated {report.get('generated_at')}") print(f"Flagged algorand-mainnet merchants to reproduce: {len(flagged)}") if not flagged: print("Nothing to reproduce.") return 0 after = datetime.utcnow() - timedelta(days=RECEIPTS_WINDOW_DAYS) header = ( f"{'address':<20} {'heuristic':<24} {'published':<10} " f"{'recomputed':<22} {'agreement'}" ) print(header) print("-" * len(header)) disagreements = 0 for entry in flagged: address = str(entry.get("address") or "") short = address[:18] + ".." if len(address) > 20 else address published = _published_flags(entry) try: hits = fetch_inbound_axfers( args.indexer, address, args.asa, after, args.timeout ) except Exception as exc: print(f"{short:<20} {'(fetch failed)':<24} {'-':<10} " f"{str(exc)[:20]:<22} UNVERIFIABLE") disagreements += 1 continue recomputed = recompute_heuristics(hits, thresholds) for heuristic in ("single_payer_dominance", "burst_pattern"): was_published = heuristic in published re = recomputed[heuristic] agree = was_published == bool(re["flagged"]) if not agree: disagreements += 1 print( f"{short:<20} {heuristic:<24} " f"{'FLAGGED' if was_published else 'not flagged':<10} " f"{('flagged share=' + str(re['share'])):<22} " f"{'AGREE' if agree else 'DISAGREE'}" ) print("-" * len(header)) if disagreements: print(f"DISAGREEMENTS: {disagreements} (investigate before citing)") return 1 print("All recomputed heuristics agree with the published report.") return 0 if __name__ == "__main__": sys.exit(main())