btc-intel / analysis /evaluate.py
Anish-g's picture
Deploy btc-intel on free ZeroGPU Gradio Space
824aa03 verified
Raw History Blame Contribute Delete
25.3 kB
"""The validity layer, measured: detector precision/recall, and each abstention
policy under both the P6 metric and the pre-registered revised metric.
`python -m eval.report` calls `run` and renders `section` into eval/results.md
section 10 and `write_doc` into docs/VALIDITY.md, from one result. The
pre-registration (`## Metric revision`) and the frozen P6 results in
docs/VALIDITY.md are copied through verbatim, never regenerated.
Protocol, as pre-registered: three policies — off, binary (P6's gate),
tiered (the revision) — on the same test transactions, each under both
metrics, each (policy, metric) cutoff chosen on the calibration captures by
`eval.origin.choose_cutoff_for`. Base and variant corpora, within- and
cross-topology. The decision rule is `verdict`'s.
"""
from __future__ import annotations
from pathlib import Path
import pandas as pd
import config
from eval import origin as origin_eval
from origination import corpus
from origination import evaluate as origination_eval
from origination.evaluate import TEST_SETS, label_coinjoins, with_cutoff
from origination.model import KEY
from . import validity
#: The base corpus's omission sentence with its Dandelion clause struck, then
#: what the variant adds and how it simplifies.
OMISSIONS = (corpus.OMISSIONS.replace(
"any Dandelion/Dandelion++ stem phase (every broadcast diffuses from its first "
"hop immediately), ", "") + " " + corpus.VALIDITY_ADDS)
POLICIES = ("off", "binary", "tiered")
PRESERVED = ("## Metric revision", "## P6 results (frozen)", "## Red-team timing (measured)")
def policy(cfg: dict, name: str) -> dict:
v = cfg["validity"]
return {**cfg, "validity": {**v, "enforce": name != "off",
"mode": "binary" if name == "binary" else "tiered"}}
# --- detectors against ground truth -------------------------------------------
def detections(matrix: pd.DataFrame, decided: pd.DataFrame, truth: pd.DataFrame,
cfg: dict) -> pd.DataFrame:
"""Each detector run on its own, per observed transaction, beside the
generator's labels. TOR_ONION is about the peer the model named."""
named = decided.set_index(KEY)["estimated_origin_ip"]
labels = truth.set_index(KEY)
rows = []
for (capture_id, txid), group in matrix.groupby(KEY, sort=False):
peer = named.get((capture_id, txid))
t = labels.loc[(capture_id, txid)]
first = group.iloc[0]
rows.append({
"stem_fired": validity.dandelion_stem(group["delta_vs_first_s"], cfg) is not None,
"coinjoin_fired": validity.coinjoin(validity.tx_of(txid, t), cfg) is not None,
"onion_fired": validity.tor_onion(peer) is not None,
"v2_fired": validity.v2_passive_tap(first.get("capture_source"),
first.get("unreadable_flows")) is not None,
"stem": bool(t["stem_length"] > 0),
"stem_through_observer": bool(t["stem_through_observer"]),
"coinjoin": t["shape"] == "coinjoin",
"onion_origin": bool(t["onion_origin"]),
})
return pd.DataFrame(rows)
def _pr(name: str, against: str, fired: pd.Series, actual: pd.Series, cfg: dict,
test_set: str) -> dict:
from eval.ground_truth.score import _ci
tp, n_fired, n_actual = int((fired & actual).sum()), int(fired.sum()), int(actual.sum())
return {"condition": "simulated", "test set": test_set, "detector": name,
"tier": validity.TIER[name], "ground truth": against,
"transactions": len(fired), "truly present": n_actual, "fired": n_fired,
"true positives": tp,
"precision": round(tp / n_fired, 3) if n_fired else None,
"precision 95% CI": _ci(tp, n_fired, cfg) if n_fired else "n/a",
"recall": round(tp / n_actual, 3) if n_actual else None,
"recall 95% CI": _ci(tp, n_actual, cfg) if n_actual else "n/a"}
def precision_recall(d: pd.DataFrame, cfg: dict, test_set: str) -> list[dict]:
return [
_pr(validity.DANDELION_STEM, "a stem was used", d["stem_fired"], d["stem"], cfg,
test_set),
_pr(validity.DANDELION_STEM, "the stem passed through an observer",
d["stem_fired"], d["stem_through_observer"], cfg, test_set),
_pr(validity.COINJOIN, "CoinJoin shape", d["coinjoin_fired"], d["coinjoin"], cfg,
test_set),
_pr(validity.TOR_ONION, "sender reachable only over Tor", d["onion_fired"],
d["onion_origin"], cfg, test_set),
_pr(validity.V2_PASSIVE_TAP, "a v2 peer is missing from a pcap capture",
d["v2_fired"], pd.Series(False, index=d.index), cfg, test_set),
]
# --- the policies under both metrics ------------------------------------------
def cutoffs(calibration: pd.DataFrame, cfg: dict) -> dict:
return {(p, m): origin_eval.choose_cutoff_for(calibration, policy(cfg, p), m)[0]
for p in POLICIES for m in origin_eval.METRICS}
def policy_rows(frame: pd.DataFrame, cuts: dict, cfg: dict, corpus_name: str,
label: str) -> list[dict]:
from eval.ground_truth.score import _ci
rows = []
for p in POLICIES:
pcfg = policy(cfg, p)
revised = with_cutoff(pcfg, cuts[(p, "revised")])
p6 = with_cutoff(pcfg, cuts[(p, "p6")])
flagged = origin_eval.flagged_at(frame, revised)
kept = frame[~flagged]
right = int(kept["correct"].sum())
r = origin_eval.cost_score(frame, revised, metric="revised")
rows.append({
"condition": "simulated", "corpus": corpus_name, "test set": label,
"policy": p, "n": len(frame),
"cost, revised metric": r["cost_weighted_score"],
"cost, P6 metric": origin_eval.cost_score(frame, p6, metric="p6")
["cost_weighted_score"],
"abstention rate": round(float(flagged.mean()), 3),
"acc if answered": round(right / len(kept), 3) if len(kept) else None,
"acc if answered 95% CI": _ci(right, len(kept), cfg) if len(kept) else "n/a",
"correct_actionable": r["correct_actionable"],
"correct_infrastructure": r["correct_infrastructure"],
"wrong_uninvolved_third_party": r["wrong_uninvolved_third_party"],
"qualified_correct": r["qualified_correct"],
"qualified_wrong": r["qualified_wrong"],
"coinjoin_input_misattribution": r["coinjoin_input_misattribution"],
"cutoff, revised": cuts[(p, "revised")], "cutoff, P6": cuts[(p, "p6")],
})
return rows
def withheld_table(frame: pd.DataFrame, cuts: dict, cfg: dict, corpus_name: str,
label: str) -> pd.DataFrame:
"""P6's per-detector table, again: for each reason, the transactions it
fires on that the layer-off system would have answered, and what those
answers were (P6 metric, off's P6 cutoff — P6's method exactly), beside
what the tiered policy now does with them."""
off = with_cutoff(policy(cfg, "off"), cuts[("off", "p6")])
outcome = origin_eval.outcomes(frame, off, metric="p6")
rows = []
for reason in validity.REASONS:
hit = frame["validity_reasons"].map(lambda r: reason in list(r))
answered = hit & (outcome != "abstained")
counts = outcome[answered].value_counts()
rows.append({
"condition": "simulated", "corpus": corpus_name, "test set": label,
"reason": reason, "tier now": validity.TIER[reason],
"tiered policy": {"ABSTAIN": "withholds", "QUALIFIED": "answers, qualified",
"ANNOTATE": "answers, flagged"}[validity.TIER[reason]],
"fired": int(hit.sum()), "would have been answered": int(answered.sum()),
"…correct_actionable": int(counts.get("correct_actionable", 0)),
"…correct_infrastructure": int(counts.get("correct_infrastructure", 0)),
"…wrong_uninvolved_third_party": int(counts.get("wrong_uninvolved_third_party", 0)),
})
return pd.DataFrame(rows)
#: Every code `eval.origin.abstention_reasons` can return, in its order.
ABSTENTION_CODES = (*validity.REASONS, origin_eval.RELAY_AT_TOP, origin_eval.BELOW_CUTOFF)
def abstention_breakdown(frame: pd.DataFrame, cuts: dict, cfg: dict, corpus_name: str,
label: str) -> list[dict]:
"""Why each policy withheld what it withheld, one count per reason code, at
the same (revised-metric) cutoff as `policy_rows`' abstention rate — so the
codes sum to the abstentions and coverage is explained beside accuracy."""
rows = []
for p in POLICIES:
pcfg = with_cutoff(policy(cfg, p), cuts[(p, "revised")])
reasons = origin_eval.abstention_reasons(frame, pcfg)
kept = frame[reasons.isna()]
counts = reasons.value_counts()
rows.append({
"condition": "simulated", "corpus": corpus_name, "test set": label,
"policy": p, "n": len(frame), "abstained": int(reasons.notna().sum()),
"abstention rate": round(float(reasons.notna().mean()), 3) if len(frame) else None,
"acc if answered": round(float(kept["correct"].mean()), 3) if len(kept) else None,
**{code: int(counts.get(code, 0)) for code in ABSTENTION_CODES},
})
return rows
def tier_counts(frame: pd.DataFrame, corpus_name: str, label: str) -> list[dict]:
reasons = frame["validity_reasons"].explode().dropna().value_counts()
return [{"condition": "simulated", "corpus": corpus_name, "test set": label,
"tier": t, "transactions": int((frame["validity_tier"] == t).sum()),
"reasons fired": ", ".join(f"{r} {int(reasons.get(r, 0))}"
for r in validity.REASONS if validity.TIER[r] == t)
or "—"}
for t in (validity.PASS, validity.ABSTAIN, validity.QUALIFIED, validity.ANNOTATE)]
def score_corpus(result: dict, cfg: dict, name: str) -> dict:
"""Every table for one corpus, from its `origination.evaluate.run` result."""
model, parts, truth = result["model"], result["parts"], result["truth"]
mixes = origination_eval.coinjoins_of({"truth": result["truth_frame"]})
calibration = label_coinjoins(
model.decide(parts["calibration"], truth, cfg, result["shapes"]), mixes)
cuts = cutoffs(calibration, cfg)
systems, withheld, tiers, abstentions = [], [], [], []
for role, label in TEST_SETS.items():
frame = result["decided"][role]
systems += policy_rows(frame, cuts, cfg, name, label)
withheld.append(withheld_table(frame, cuts, cfg, name, label))
tiers += tier_counts(frame, name, label)
abstentions += abstention_breakdown(frame, cuts, cfg, name, label)
return {"systems": pd.DataFrame(systems),
"withheld": pd.concat(withheld, ignore_index=True),
"tiers": pd.DataFrame(tiers), "abstentions": pd.DataFrame(abstentions)}
def run(cfg: dict | None = None, rebuild: bool = False, base: dict | None = None) -> dict:
"""`base` is section 9's `origination.evaluate.run` result on the base
corpus; the variant is run here."""
cfg = cfg or config.load()
manifest = corpus.load_manifest(corpus.VALIDITY_MANIFEST)
variant = origination_eval.run(cfg, rebuild, manifest=manifest)
labels = variant["truth_frame"]
pr = []
for role, label in TEST_SETS.items():
d = detections(variant["parts"][role], variant["decided"][role], labels, cfg)
pr += precision_recall(d, cfg, label)
scored = {"variant": score_corpus(variant, cfg, "variant")}
false_alarms = None
if base is not None:
scored["base"] = score_corpus(base, cfg, "base")
frame = base["decided"]["cross_test"]
fired = frame["validity_reasons"].map(lambda r: validity.DANDELION_STEM in list(r))
false_alarms = {"transactions": len(frame), "fired": int(fired.sum()),
"rate": round(float(fired.mean()), 4) if len(frame) else None,
"corpus": base["corpus"]}
def joined(key):
return pd.concat([s[key] for s in scored.values()], ignore_index=True)
return {"corpus": variant["corpus"], "summary": variant["summary"],
"base_corpus": base["corpus"] if base else None,
"precision_recall": pd.DataFrame(pr), "systems": joined("systems"),
"withheld": joined("withheld"), "tiers": joined("tiers"),
"abstentions": joined("abstentions"),
"base_false_alarms": false_alarms, "omissions": OMISSIONS}
# --- the decision rule ---------------------------------------------------------
def verdict(result: dict) -> str:
"""The pre-registered rule: tiered beats off iff its cross-topology cost on
the revised metric is higher on the variant corpus; an unqualified
improvement also needs it higher on the P6 metric."""
s = result["systems"]
cross = s[(s["corpus"] == "variant") & (s["test set"] == TEST_SETS["cross_test"])]
row = cross.set_index("policy")
off, tiered, binary = row.loc["off"], row.loc["tiered"], row.loc["binary"]
rev = (tiered["cost, revised metric"], off["cost, revised metric"])
p6 = (tiered["cost, P6 metric"], off["cost, P6 metric"])
beats_rev, beats_p6 = rev[0] > rev[1], p6[0] > p6[1]
if beats_rev and beats_p6:
head = ("**Verdict (variant, cross-topology): the revised validity layer beats "
f"no-validity on both metrics** — revised {rev[0]} against {rev[1]}, "
f"P6 metric {p6[0]} against {p6[1]}.")
elif beats_rev:
head = ("**Verdict (variant, cross-topology): the revised layer beats "
f"no-validity on the revised metric only** — {rev[0]} against {rev[1]}; "
f"on the P6 metric it scores {p6[0]} against {p6[1]}. As pre-registered, "
"that is reported as exactly this and not as an unqualified improvement: "
"the difference comes from pricing the CoinJoin ownership claim, not from "
"the policy answering better by P6's measure.")
else:
head = ("**Verdict (variant, cross-topology): the revised validity layer still "
f"does not beat no-validity on the revised metric** — {rev[0]} against "
f"{rev[1]} (P6 metric: {p6[0]} against {p6[1]}). Reported as found; the "
"weights were not changed after this result.")
return (head + f" P6's binary gate, rescored on the same transactions: revised "
f"{binary['cost, revised metric']}, P6 metric {binary['cost, P6 metric']}.")
# --- rendering --------------------------------------------------------------
TAXONOMY = """\
Origin attribution names the peer that announced a transaction first, or most
centrally, to an observer. **Calibration** says how often such a name is right.
The **validity layer** (`analysis/validity.py`) asks two prior questions: can
an attribution be about this transaction at all, and if so, what may it
claim? Every detector runs; every reason that fires is reported; the verdict's
**tier** is the most severe among them.
| tier | effect |
| --- | --- |
| `ABSTAIN` | the answer is withheld, through the existing rule — `low_confidence_origin` in `engines/propagation`, `eval.origin.flagged_at` everywhere else — not a second path |
| `QUALIFIED` | the answer is given with its meaning restricted, and never shaped like an IP attribution |
| `ANNOTATE` | the answer is given unchanged, with the flag and its evidence |
| `PASS` | nothing fired |
`validity.enforce: false` switches the layer off in that one place;
`validity.mode: binary` restores P6's gate, where every non-PASS verdict
abstained.
Every origin the system emits — `engines/propagation`'s estimates, the
origination model's output, the API's `/transactions/{txid}/propagation`, the
red-team origin table, each attribution lead, and the console views of them —
carries a probability, its calibration basis, and a verdict
`{"tier", "reason", "reasons", "confidence", "evidence"}`; where an answer is
given, `answer` says what it may claim: `ip_attribution`, `onion_identity`
(no `ip` field) or `broadcasting_peer` (with `input_ownership: not
attributable`). `tests/test_api.py` asserts the verdict at the API boundary;
`tests/test_validity.py` asserts that no TOR_ONION answer carries an IP and no
COINJOIN answer attributes input ownership.
## Reason codes
`confidence` is confidence that the *condition* is present, not that the
attribution is wrong.
| code | tier | fires when | reads | catches | does not catch |
| --- | --- | --- | --- | --- | --- |
| `DEGENERATE` | ABSTAIN | 0 or 1 peers announced the transaction | `candidate_count` (the matrix's `degenerate`; a single-observation tree) | a capture with nothing to rank | — it is definitional |
| `NOT_REACHABLE` | ABSTAIN | every candidate is a known public relay | the matrix's `scope_out`; the same rule on a tree | the zero-ceiling case: the sender is not among what this observer can see | a sender absent from the candidates while an ordinary node happens to be among them — that is the unobserved-origin case calibration prices |
| `COINJOIN` | QUALIFIED | `graph.clustering.is_coinjoin`: ≥3 inputs, ≥3 outputs, ≥3 equal-value outputs above dust, no more equal outputs than inputs | the transaction's structure | a mix; the answer becomes "broadcasting peer only", input ownership non-attributable | an equal-value **batch payout** from one owner paying no more equal amounts than it spends inputs (a false positive: structure cannot see input ownership); a CoinJoin with unequal denominations or fewer than three equal outputs (PayJoin, some WabiSabi rounds) |
| `TOR_ONION` | QUALIFIED | the named peer is a `.onion` address | the peer address (onion parsing in `p2p/capture_reader.py`) | a hidden-service peer named as origin; the answer becomes an onion identity, never an IP | a Tor sender whose transaction entered through a mixed-transport node: the observer sees that node's clearnet address, and nothing in one capture marks it |
| `DANDELION_STEM` | ANNOTATE | the gap after the first announcement is ≥ `isolation_ratio` × the median gap among the rest, with ≥ `min_candidates` candidates | announcement times | an observer on or next to a stem, which hears one peer, then silence, then the broadcast | a stem that ends away from every observer (the common case); and it false-alarms on sparse captures. A flag only: BIP-156 was never merged into Bitcoin Core, which has no stem phase |
| `V2_PASSIVE_TAP` | ANNOTATE | a pcap capture held port-8333 flows that never decoded | `capture_source`, `unreadable_flows` | a packet capture blind to its BIP-324 v2 peers, whose announcements are then missing from the candidates | which peers those were. It never fires on debug.log or .btcap: the node writes those, and as a session endpoint it decrypts. A pcap on the node's own host is as blind as a span port — tcpdump holds no session keys |
| `NOT_ASSESSED` | — | not a detector: a stored output from before this layer | — | keeps an old lead from being served as PASS | — |
**Taint and leads.** A CoinJoin terminates taint: `graph.entity_graph` counts
the CoinJoin transactions on each entity edge (`mix_count`, from the same
`is_coinjoin`), and `fusion.taint.propagate` does not cross an edge made of
nothing else. A correlation lead says "this IP broadcast for this cluster's
inputs", so none is built from a COINJOIN or TOR_ONION answer or a
NOT_REACHABLE one. DEGENERATE observations are kept — aggregating many
single-candidate observations, each weighted by its low confidence, is what
correlation is for — and ANNOTATE flags ride along; a lead's own verdict is
`PASS` or `ANNOTATE`, naming how many of its observations carried each flag.
The DANDELION_STEM thresholds were fixed in config.yaml before the first P6
result and have not been tuned since.
"""
def _omitted(omissions: str) -> str:
return f"\n*`condition=\"simulated\"`.* {omissions}\n"
def section(result: dict, md_table) -> str:
om = result["omissions"]
s = result["summary"]
prevalence = s["validity_conditions"]
fa = result["base_false_alarms"]
rows = [{"condition": "simulated", "measure": k.replace("_", " "),
"transactions": v["transactions"] if isinstance(v, dict) else v,
"share of broadcast": v["share"] if isinstance(v, dict) else None}
for k, v in prevalence.items()
if k not in ("captures_with_toggle_on", "transactions")]
systems = result["systems"]
out = [
f"`condition=\"simulated\"`. **What the simulator omits:** {om}\n",
f"\nVariant corpus `{result['corpus']}` (`origination/manifest_validity.json`) and "
f"base corpus `{result['base_corpus']}` (`origination/manifest.json`). The model is "
"fitted, calibrated and scored on each exactly as in section 9. Variant captures "
"with each toggle on: "
+ ", ".join(f"{k} {v}" for k, v in prevalence["captures_with_toggle_on"].items())
+ f", of {s['captures']}.\n",
"\n### The policies under both metrics\n\n"
"Same test transactions for every row. `off`: no validity layer. `binary`: P6's "
"gate (every non-PASS verdict abstains), current detectors. `tiered`: the "
"revision. Each metric's cost is at that (policy, metric)'s own cutoff, chosen "
"on the calibration captures; the other columns are at the revised-metric "
"cutoff. **The headline is the cost score and `acc if answered`.**\n\n",
"#### Variant corpus\n\n",
md_table(systems[systems["corpus"] == "variant"]), _omitted(om),
"\n#### Base corpus (no invalidating condition simulated)\n\n",
md_table(systems[systems["corpus"] == "base"]), _omitted(om),
"\n" + verdict(result) + "\n",
("\n### Why each answer was withheld\n\n"
"Every abstention, by the one reason code `eval.origin.abstention_reasons` gives "
"it — the validity verdict's leading reason when its tier is withheld, then "
"`RELAY_AT_TOP` (a listed public relay ranked first), then `BELOW_CUTOFF` "
"(calibrated confidence under the policy's cutoff). Same frames and "
"revised-metric cutoffs as the table above; the codes sum to `abstained`, so "
"coverage (1 − abstention rate) is explained next to `acc if answered`.\n\n"),
md_table(result["abstentions"]), _omitted(om),
"\n### Verdicts by tier\n\n", md_table(result["tiers"]), _omitted(om),
"\n### What each reason fires on, and what the answers were (P6's table, again)\n\n"
"At the layer-off policy's P6-metric cutoff, P6's method exactly: the "
"transactions each reason fires on that the layer-off system would have "
"answered, and whether those answers were right. The `tiered policy` column is "
"what the revision now does with them.\n\n",
md_table(result["withheld"]), _omitted(om),
"\n### Detector precision and recall against the generator's labels\n\n"
"Each detector on its own over every observed test transaction of the variant. "
"TOR_ONION depends on which peer is named; the model's choice is used. "
"V2_PASSIVE_TAP cannot fire here: every simulated observer is a node, which "
"writes its own log, so no simulated capture is a pcap.\n\n",
md_table(result["precision_recall"]), _omitted(om),
]
if fa:
out.append(f"\nOn the base corpus `{fa['corpus']}` (cross-topology test set), where "
"no transaction has a stem, DANDELION_STEM fires on "
f"{fa['fired']} of {fa['transactions']} transactions ({fa['rate']}): "
"every one a false alarm. It now only annotates them.\n")
out += ["\n### Ground-truth prevalence (variant)\n\n", md_table(pd.DataFrame(rows)),
_omitted(om)]
return "".join(out)
def preserved(path: Path | str) -> str:
"""The pre-registration and the frozen P6 results, verbatim from the doc."""
path = Path(path)
if not path.exists():
return ""
out, keep = [], False
for line in path.read_text().splitlines(keepends=True):
if line.startswith("## "):
keep = line.rstrip("\n") in PRESERVED
if keep:
out.append(line)
return "".join(out)
DOC_HEADER = """# Validity layer
**Generated by `python -m eval.report`, except the two sections copied through
verbatim: `Metric revision` (pre-registered before the rerun) and `P6 results
(frozen)`. Do not edit by hand.** The results are also section 10 of
`eval/results.md`, from the same run.
"""
def write_doc(result: dict, md_table, path: Path | str) -> Path:
path = Path(path)
kept = preserved(path)
if not all(h in kept for h in PRESERVED):
raise RuntimeError(f"{path} lost a preserved section; restore it from git "
"before regenerating")
path.write_text(DOC_HEADER + TAXONOMY + "\n" + kept.rstrip("\n") + "\n\n"
+ "## Results — metric revision\n\n" + section(result, md_table))
return path