run_protocol_anonymity_study.py

#!/usr/bin/env python3
"""Run the protocol-anonymity ecosystem study.

Drives a parameterized JoinMarket ecosystem (100 makers by default) through the
probabilistic mixdepth-keyed diffusion clusterer and the taker-anonymity scorer,
sweeping the ecosystem parameters that govern privacy:

* S1 — strict roles, fee-regime sweep (distinct / overlapping / uniform);
* S2 — spending makers (maker_spend_prob sweep);
* S3 — maker clustering vs fee homogeneity (+ realized-fee jitter);
* S4 — mixed roles (fraction of takers that also make).

Writes ``protocol_anonymity_results.json``.

Usage::

    python run_protocol_anonymity_study.py            # full default scale
    python run_protocol_anonymity_study.py --quick    # tiny smoke run
"""

from __future__ import annotations

import argparse
import json
import time
from pathlib import Path

from coinjoin_simulator.protocol_anonymity_study import StudyConfig, build_study

OUTPUT_PATH = Path("protocol_anonymity_results.json")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--quick", action="store_true", help="tiny smoke run")
    parser.add_argument("--makers", type=int, default=100)
    parser.add_argument("--takers", type=int, default=400)
    parser.add_argument("--makers-per-cj", type=int, default=9)
    parser.add_argument("--seeds", type=int, default=3)
    args = parser.parse_args()

    if args.quick:
        cfg = StudyConfig(n_makers=20, n_takers=80, makers_per_cj=4, seeds=(101, 202))
    else:
        seed_pool = (101, 202, 303, 404, 505)
        cfg = StudyConfig(
            n_makers=args.makers,
            n_takers=args.takers,
            makers_per_cj=args.makers_per_cj,
            seeds=tuple(seed_pool[: args.seeds]),
        )

    start = time.time()
    print(
        f"Running protocol-anonymity study: {cfg.n_makers} makers, "
        f"{cfg.n_takers} takers, {len(cfg.seeds)} seeds ..."
    )
    study = build_study(cfg)
    payload = study.payload
    payload["generated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    OUTPUT_PATH.write_text(json.dumps(payload, indent=2) + "\n")
    elapsed = time.time() - start
    print(f"\nWrote {OUTPUT_PATH} in {elapsed:.1f}s\n")

    _print_summary(payload)


def _print_summary(payload: dict) -> None:
    def table(title: str, rows: list[dict]) -> None:
        print(f"\n{title}")
        print(
            f"  {'scenario':<34} {'mkr_rec':>8} {'mkr_frag':>9} "
            f"{'tk_eff':>7} {'tk_deanon':>10} {'done':>6} {'top_acc':>8}"
        )
        for r in rows:
            print(
                f"  {r['label'][:34]:<34} "
                f"{r['maker_recall_mean']:>8.3f} "
                f"{r['maker_fragments_mean']:>9.0f} "
                f"{r['taker_effective_set_mean']:>7.2f} "
                f"{r['taker_deanon_fraction_mean']:>10.3f} "
                f"{r['completed_round_fraction_mean']:>6.2f} "
                f"{r['taker_top_guess_acc_mean']:>8.3f}"
            )

    table("S1 — strict roles (fee-regime sweep)", payload["s1_strict_roles"])
    table("S2 — spending makers", payload["s2_spending_makers"])
    table("S3 — maker clustering vs fee homogeneity", payload["s3_maker_clustering"])
    table("S4 — mixed roles", payload["s4_mixed_roles"])
    _spend_table("S5 — maker-spend anonymity (everyone spends from time to time)",
                 payload["s5_maker_spend_anonymity"])
    table("S6 — protocol-fix comparison", payload["s6_protocol_fixes"])

    h = payload["headline"]
    print("\nHeadline:")
    print(
        f"  N = {h['n_equal_outputs']:.0f} equal outputs; strict-roles taker "
        f"effective set {h['strict_roles_taker_effective_set']:.2f} "
        f"(deanon {h['strict_roles_taker_deanon_fraction']:.0%})"
    )
    print(
        f"  maker clustering recall: distinct fees "
        f"{h['maker_clustering_recall_distinct_fees']:.3f}, "
        f"uniform fees {h['maker_clustering_recall_uniform_fees']:.3f}"
    )
    print(
        f"  spending makers: taker deanon at 10% spend "
        f"{h['spending_makers_taker_deanon_at_10pct']:.0%}, "
        f"at 50% spend {h['spending_makers_taker_deanon_at_50pct']:.0%}"
    )
    print(
        f"  own-spend anonymity: effective set {h['maker_spend_effset_at_1pct']:.2f} "
        f"at 1% spend rate, {h['maker_spend_effset_at_10pct']:.2f} at 10%"
    )
    print(
        f"  tx_extension (2x participants) taker deanon: "
        f"{h['tx_extension_taker_deanon']:.0%} (no help)"
    )
    print(f"\n  {h['verdict']}")


def _spend_table(title: str, rows: list[dict]) -> None:
    print(f"\n{title}")
    print(
        f"  {'scenario':<34} {'n_spends':>9} {'spend_eff':>10} "
        f"{'spend_deanon':>13} {'tk_eff':>7} {'tk_deanon':>10} {'done':>6}"
    )
    for r in rows:
        print(
            f"  {r['label'][:34]:<34} "
            f"{r['n_spends_mean']:>9.0f} "
            f"{r['spend_effective_set_mean']:>10.2f} "
            f"{r['spend_deanon_fraction_mean']:>13.3f} "
            f"{r['taker_effective_set_mean']:>7.2f} "
            f"{r['taker_deanon_fraction_mean']:>10.3f} "
            f"{r['completed_round_fraction_mean']:>6.2f}"
        )


if __name__ == "__main__":
    main()