Download analysis/evaluate.py from Anish-g/btc-intel: direct link, hf CLI and curl.
- Browser
- Download file 25.3 kB
-
https://huggingface.co/spaces/Anish-g/btc-intel/resolve/main/analysis/evaluate.py
- Command line
-
hf download hf://spaces/Anish-g/btc-intel/analysis/evaluate.py
-
curl -L -o evaluate.py https://huggingface.co/spaces/Anish-g/btc-intel/resolve/main/analysis/evaluate.py
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 | |