build_mainnet_taker_fate.py

#!/usr/bin/env python3
"""Build the slim mainnet taker-fate snapshot for the feasibility study.

This is a one-off data-preparation script (the JoinMarket counterpart of
``fetch_wasabi_rounds.py``). It distills two large, non-portable mainnet
attack corpora into a single small, privacy-clean, columnar snapshot that is
committed to ``data/`` and consumed by
:mod:`coinjoin_simulator.mainnet_taker_exposure` and its tests.

Inputs (all external to the repo; produced by the ``joinmarket-ng`` crawler
and the ILP forward-attribution / chain-elimination analyzers):

* ``deanon_eq.json`` -- per-CoinJoin **single-hop forward-attribution** records.
  For each analyzed CoinJoin ``T`` it records, for every equal output, the tx
  that spent it (within the corpus) and the maker cluster the successor was
  attributed to, plus ``n_makers_attributed`` (equal outputs strictly matched
  to one of ``T``'s maker slots -- the realized taker anon-set reduction).
* ``anon_chain_v5.json`` -- per-CoinJoin **multi-hop chain-elimination** records
  (``S_eff``: the taker's effective anon set after following maker coins across
  rounds and cross-validating the 5-mixdepth ladder).
* ``graph4_bwd_ckpt.json`` -- the JM-induced tx graph; its ``is_jm`` flag is the
  authoritative CoinJoin discriminator and its ``meta.block_height`` supplies
  block heights. We cache the derived ``txid -> block_height`` JM map to
  ``--jm-txids`` so the heavy graph is loaded at most once.

The equal-output forward fate (single hop) is classified as:

* ``certified_maker``   -- spent into a CoinJoin whose change-fee-fingerprint
  cluster reappears (the leak the attack exploits);
* ``cj_spend_undecoded`` -- spent into another CoinJoin but not attributable
  (the chain-elimination attack can keep following these);
* ``organic_in_corpus`` -- spent into a non-CoinJoin tx inside the corpus;
* ``none`` -- no spend recorded inside the corpus: either still unspent or
  spent out of the JM graph (genuine cover the single-hop attack cannot follow).

Note: the spend map only covers spends *into* the corpus, so ``none`` conflates
"still unspent" with "left the graph". The block-height maturation buckets are
the corpus-density control for this (see the study module).

Usage::

    python build_mainnet_taker_fate.py \
        --deanon-eq tmp/deanon_eq.json \
        --anon-chain /path/to/jm/anon_chain_v5.json \
        --graph /path/to/jm/graph4_bwd_ckpt.json \
        --output data/mainnet_taker_fate.json
"""

from __future__ import annotations

import argparse
import json
import time
from pathlib import Path
from typing import Any

DEFAULT_DEANON_EQ = Path("tmp/deanon_eq.json")
DEFAULT_ANON_CHAIN = Path("/home/m0wer/code/bitcoin/joinmarket-ng/tmp/jm/anon_chain_v5.json")
DEFAULT_GRAPH = Path("/home/m0wer/code/bitcoin/joinmarket-ng/tmp/jm/graph4_bwd_ckpt.json")
DEFAULT_JM_TXIDS = Path("tmp/jm_txids.json")
DEFAULT_OUTPUT = Path("data/mainnet_taker_fate.json")

FORWARD_COLUMNS = [
    "block_height",
    "n_eq",
    "n_makers",
    "n_attributed",
    "certified_maker",
    "none",
    "cj_spend_undecoded",
    "organic_in_corpus",
]
CHAIN_COLUMNS = [
    "block_height",
    "n_eq",
    "matched",
    "exposed_prior",
    "exposed_back_elim",
    "s_eff",
]


def load_jm_block_heights(graph_path: Path, cache_path: Path) -> dict[str, int]:
    """Return the ``txid -> block_height`` map for every ``is_jm`` transaction.

    Uses the cached ``cache_path`` if present; otherwise loads the (large) graph
    once, extracts the JM set, and writes the cache.
    """
    if cache_path.exists():
        raw = json.loads(cache_path.read_text())
        return {k: v for k, v in raw.items() if v is not None}
    t0 = time.monotonic()
    graph: dict[str, Any] = json.loads(graph_path.read_text())
    jm: dict[str, int] = {}
    for txid, node in graph.items():
        if not node.get("is_jm"):
            continue
        bh = (node.get("meta") or {}).get("block_height")
        if bh is not None:
            jm[txid] = int(bh)
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    cache_path.write_text(json.dumps(jm))
    print(f"  extracted {len(jm):,} JM block heights in {time.monotonic() - t0:.1f}s")
    return jm


def classify_forward(
    record: dict[str, Any],
    jm_txids: set[str],
) -> tuple[int, int, int, int, int, int]:
    """Return ``(n_attributed, certified, none, cj_undecoded, organic, n_eq)``."""
    certified = none = undecoded = organic = 0
    attribution = record.get("eq_attribution", [])
    for entry in attribution:
        spent_by = entry.get("spent_by")
        successor_cluster = entry.get("successor_cluster")
        if spent_by is None:
            none += 1
        elif spent_by in jm_txids:
            if successor_cluster is not None:
                certified += 1
            else:
                undecoded += 1
        else:
            organic += 1
    n_attributed = int(record.get("n_makers_attributed", 0))
    return n_attributed, certified, none, undecoded, organic, len(attribution)


def build_forward(deanon_path: Path, jm_txids: set[str]) -> dict[str, Any]:
    """Build the single-hop forward-attribution block of the snapshot."""
    data = json.loads(deanon_path.read_text())
    records = [r for r in data["records"] if r.get("ok") and r.get("block_height")]
    rows: list[list[int]] = []
    fate_totals = {"certified_maker": 0, "none": 0, "cj_spend_undecoded": 0, "organic_in_corpus": 0}
    n_eq_total = 0
    n_any_reduction = 0
    n_full_deanon = 0
    for r in records:
        n_attr, cert, none, undec, org, n_eq = classify_forward(r, jm_txids)
        rows.append(
            [
                int(r["block_height"]),
                n_eq,
                int(r.get("n_makers", 0)),
                n_attr,
                cert,
                none,
                undec,
                org,
            ]
        )
        fate_totals["certified_maker"] += cert
        fate_totals["none"] += none
        fate_totals["cj_spend_undecoded"] += undec
        fate_totals["organic_in_corpus"] += org
        n_eq_total += n_eq
        if n_attr >= 1:
            n_any_reduction += 1
        if n_eq - n_attr <= 1:
            n_full_deanon += 1
    rows.sort(key=lambda row: row[0])
    n = len(rows)
    return {
        "description": (
            "single-hop forward attribution: each CoinJoin's equal outputs are "
            "followed one hop; n_attributed equal outputs are strictly matched to "
            "this CoinJoin's maker slots (the realized taker anon-set reduction)"
        ),
        "columns": FORWARD_COLUMNS,
        "n_cjs": n,
        "n_equal_outputs": n_eq_total,
        "fate_counts": fate_totals,
        "frac_any_reduction": n_any_reduction / n if n else 0.0,
        "frac_full_deanon": n_full_deanon / n if n else 0.0,
        "rows": rows,
    }


def build_chain(anon_chain_path: Path, jm_block_heights: dict[str, int]) -> dict[str, Any]:
    """Build the multi-hop chain-elimination block of the snapshot."""
    data = json.loads(anon_chain_path.read_text())
    per_cj = data["per_cj"]
    rows: list[list[int]] = []
    n_s_eq_1 = n_s_le_2 = n_s_le_half = 0
    for r in per_cj:
        bh = jm_block_heights.get(r["txid"])
        if bh is None:
            continue
        n_eq = int(r["N"])
        s_eff = int(r["S_eff"])
        rows.append(
            [
                bh,
                n_eq,
                int(r["matched"]),
                int(r["exposed_prior"]),
                int(r["exposed_back_elim"]),
                s_eff,
            ]
        )
        if s_eff <= 1:
            n_s_eq_1 += 1
        if s_eff <= 2:
            n_s_le_2 += 1
        if s_eff <= n_eq / 2:
            n_s_le_half += 1
    rows.sort(key=lambda row: row[0])
    n = len(rows)
    return {
        "description": (
            "multi-hop chain elimination: maker coins are followed across rounds "
            "and cross-validated on the 5-mixdepth ladder; s_eff is the taker's "
            "effective anon set after elimination"
        ),
        "columns": CHAIN_COLUMNS,
        "n_cjs": n,
        "frac_s_eq_1": n_s_eq_1 / n if n else 0.0,
        "frac_s_le_2": n_s_le_2 / n if n else 0.0,
        "frac_s_le_half_n": n_s_le_half / n if n else 0.0,
        # Native aggregates from the analyzer (its degree d uses the attack's own
        # posterior, not the uniform-over-survivors approximation; kept for the
        # headline figure since per-CJ d is not stored in the int rows).
        "mean_s_eff_native": float(data.get("mean_S_eff", 0.0)),
        "mean_d_native": float(data.get("mean_d", 0.0)),
        "mean_n_native": float(data.get("mean_N", 0.0)),
        "rows": rows,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--deanon-eq", type=Path, default=DEFAULT_DEANON_EQ)
    parser.add_argument("--anon-chain", type=Path, default=DEFAULT_ANON_CHAIN)
    parser.add_argument("--graph", type=Path, default=DEFAULT_GRAPH)
    parser.add_argument("--jm-txids", type=Path, default=DEFAULT_JM_TXIDS)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    args = parser.parse_args()

    print("Loading JM block heights ...")
    jm_block_heights = load_jm_block_heights(args.graph, args.jm_txids)
    jm_txids = set(jm_block_heights)
    print(f"  {len(jm_txids):,} JM CoinJoins in corpus")

    print("Building single-hop forward-attribution block ...")
    forward = build_forward(args.deanon_eq, jm_txids)
    print(f"  {forward['n_cjs']:,} CoinJoins, {forward['n_equal_outputs']:,} equal outputs")

    print("Building multi-hop chain-elimination block ...")
    chain = build_chain(args.anon_chain, jm_block_heights)
    print(f"  {chain['n_cjs']:,} CoinJoins")

    all_heights = [row[0] for row in forward["rows"]] + [row[0] for row in chain["rows"]]
    payload = {
        "meta": {
            "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "source": (
                "JoinMarket mainnet corpus (JM-induced tx subgraph crawled from "
                "mempool.space); single-hop forward-attribution and multi-hop "
                "chain-elimination ILP analyzers"
            ),
            "jm_corpus_cjs": len(jm_txids),
            "block_height_range": [min(all_heights), max(all_heights)],
            "note": (
                "slim, privacy-clean columnar snapshot (no addresses or txids). "
                "The single-hop spend map only sees spends into the corpus, so the "
                "forward 'none' fate conflates still-unspent with left-the-graph; "
                "block-height buckets control for corpus density."
            ),
        },
        "forward_attribution": forward,
        "chain_elimination": chain,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(payload) + "\n")
    size_mb = args.output.stat().st_size / 1e6
    print(f"\nWrote {args.output} ({size_mb:.2f} MB)")


if __name__ == "__main__":
    main()