run_maker_clustering_study.py

#!/usr/bin/env python3
"""Run the maker mixdepth-clustering study.

Drives the parameterized JoinMarket ecosystem through the strong global matching
clusterer to chart three things the maker-clustering paper left open:

* M1 -- the heterogeneous-fee threat scales (the fee fingerprint resolves the
  ``m -> m+1`` forward edge at every population size);
* M2 -- fee homogenization is percolation-bounded (the sound solver fully
  reconstructs wallets below a co-occurrence threshold and collapses above it,
  with the threshold rising in the counterparties-per-CoinJoin);
* M3 -- the recycle-change route decides the swap defense (same/next mixdepth
  change re-links the recycled equal output; only swapping the change severs it).

Writes ``maker_clustering_results.json``.

Usage::

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

from __future__ import annotations

import argparse
import json
import time
from pathlib import Path

from coinjoin_simulator.maker_clustering_study import ClusteringStudyConfig, build_study

OUTPUT_PATH = Path("maker_clustering_results.json")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--quick", action="store_true", help="tiny smoke run")
    parser.add_argument("--seeds", type=int, default=2, help="number of seeds per cell")
    args = parser.parse_args()

    seed_pool = (101, 202, 303, 404, 505)
    if args.quick:
        cfg = ClusteringStudyConfig(
            populations=(20, 40),
            makers_per_cj_values=(5,),
            seeds=(101,),
        )
    else:
        cfg = ClusteringStudyConfig(seeds=tuple(seed_pool[: max(1, args.seeds)]))

    start = time.time()
    print(
        f"Running maker-clustering study: populations {cfg.populations}, "
        f"mpc {cfg.makers_per_cj_values}, {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 m1m2(title: str, rows: list[dict]) -> None:
        print(f"\n{title}")
        print(
            f"  {'cell':<28} {'wallet_recall':>13} {'fwd_prec':>9} "
            f"{'weak_recall':>11} {'cover_flr':>9} {'cover_x':>8} {'seeds':>6}"
        )
        for r in rows:
            print(
                f"  {r['label'][:28]:<28} "
                f"{r['strong_wallet_recall_mean']:>13.2f} "
                f"{r['strong_forward_precision_mean']:>9.2f} "
                f"{r['weak_wallet_recall_mean']:>11.2f} "
                f"{r['cover_floor_mean']:>9.2f} "
                f"{r['cover_floor_ratio_mean']:>8.2f} "
                f"{r['n_seeds_mean']:>6.0f}"
            )

    m1m2("M1 - heterogeneous-fee scaling", payload["m1_heterogeneous_fee_scaling"])
    m1m2("M2 - uniform-fee percolation", payload["m2_uniform_percolation"])

    print("\nM3 - recycle-change route")
    print(
        f"  {'route':<8} {'recycled_eq':>11} {'relink_frac':>11} "
        f"{'hard_bridge_x':>13} {'wallet_recall':>13}"
    )
    for r in payload["m3_recycle_change_route"]:
        print(
            f"  {r['change_route']:<8} "
            f"{r['recycled_equal_mean']:>11.0f} "
            f"{r['recycle_relink_fraction_mean']:>11.2f} "
            f"{r['hard_bridge_cross_mean']:>13.0f} "
            f"{r['strong_wallet_recall_mean']:>13.2f}"
        )

    h = payload["headline"]
    print("\nHeadline:")

    def pct(v: object) -> str:
        return f"{v:.0%}" if isinstance(v, (int, float)) else "n/a"

    print(
        f"  M1 distinct (smallest n) wallet recall {pct(h['m1_distinct_smallest_recall'])}; "
        f"scale does not save heterogeneous-fee makers"
    )
    below = h["m2_uniform_mpc9_below_threshold_recall"]
    above = h["m2_uniform_mpc9_above_threshold_recall"]
    print(
        f"  M2 uniform mpc=9: recall {pct(below)} below threshold (n=30) vs "
        f"{pct(above)} above it (n=120)"
    )
    m1_cover = h["m1_distinct_cover_floor_ratio"]
    m2_cover = h["m2_uniform_cover_floor_ratio"]
    print(
        f"  Taker cover-floor ratio: distinct {pct(m1_cover)} (fee channel open) vs "
        f"uniform {pct(m2_cover)} (fee channel closed)"
    )
    print(
        f"  M3 relink: none {pct(h['m3_route_none_relink'])}, "
        f"same {pct(h['m3_route_same_relink'])}, "
        f"next {pct(h['m3_route_next_relink'])}, "
        f"swap {pct(h['m3_route_swap_relink'])}"
    )
    same_x = h["m3_route_same_hard_bridge_cross"]
    next_x = h["m3_route_next_hard_bridge_cross"]
    print(
        f"  M3 cross-mixdepth hard bridges: same {same_x}, next {next_x} "
        f"(next violates mixdepth isolation; same does not)"
    )
    print(f"\n  {h['verdict']}")


if __name__ == "__main__":
    main()