eval/actor_queue.py

"""The actor queue against the entity queue, on generator ground truth: triage.

Both queues come from one dataset, one fusion bundle and one fitted stacker, so
the only difference is the unit: an entity (address cluster) or an actor
(clusters joined through peer identities, fusion/actors.py). Ground truth is
`eval.actors.actors_of`: the illicit operations pre-registered in
docs/detection_unit_protocol.md.

A queue item *finds* an operation when it holds any of the operation's
wallets. A wallet's true controller, for purity, is its operation if it has
one, else its generator actor.
"""

from __future__ import annotations

import json
import tempfile
from collections import Counter
from pathlib import Path

import pandas as pd

from engines.correlation.profile import build_sources
from features import relay
from fusion import actors as A
from fusion.pipeline import build_alerts, collect_signals
from ingest.ip_intel import load_intel
from p2p import demo_capture

from .actors import actors_of
from .datasets import Dataset

KS = (10, 25, 50)


def dataset_actors(dataset: Dataset, stacker, cfg: dict) -> dict:
    """The fusion bundle, the entity alerts and the actors of one dataset.
    Its relay matrix is built the way the served one is (p2p.demo_capture,
    then features.relay), in a scratch directory. `stacker=None` fits one on
    this dataset's own labels, as eval.fusion_eval does: entities holding an
    illicit operation's wallet."""
    cfg = json.loads(json.dumps(cfg))
    cfg["ingest"]["input_dir"] = str(dataset.raw)       # this dataset's node intel
    df = dataset.frame()
    bundle = collect_signals(df, cfg, dataset.raw)
    if stacker is None:
        from fusion.stacker import train

        from .actors import illicit_entities
        bad = illicit_entities(actors_of(dataset), bundle["features"].entity_of)
        stacker = train(bundle["signals"], bundle["signals"]["entity_id"].isin(bad).astype(int), cfg)
    alerts = build_alerts(bundle, stacker, cfg)
    with tempfile.TemporaryDirectory() as tmp:
        capture = Path(tmp) / "demo_capture"
        demo_capture.write(df, capture)
        matrix, _, _ = relay.build(capture, [demo_capture.COLLECTOR], cfg)
    src = build_sources(cfg, transactions=df, matrix=matrix, features=bundle["features"],
                        intel=load_intel(None, dataset.raw, cfg))
    links = A.peer_links(src, cfg)
    actors = A.build(bundle["signals"], stacker, links, cfg, set(alerts["entity_id"]))
    return {"bundle": bundle, "alerts": alerts, "actors": actors, "links": links}


def _wallets(entity: str, clusters: dict) -> set[str]:
    return set(clusters.get(entity, {entity}))


def triage(dataset: Dataset, built: dict) -> dict:
    gt = dataset.ground_truth()
    operations = actors_of(dataset)
    clusters = built["bundle"]["features"].clustering.clusters
    op_of = {w: op.actor_id for op in operations for w in op.wallets}
    controller = lambda w: op_of.get(w) or gt["wallets"].get(w, w)
    present = {op.actor_id for op in operations
               if any(built["bundle"]["features"].entity_of(w) for w in op.wallets)}

    entity_items = [_wallets(e, clusters) for e in built["alerts"]["entity_id"]]
    queue = A.queue(built["actors"])
    actor_items = [set().union(*(_wallets(m, clusters) for m in members))
                   for members in queue["members"]]

    def found(items):
        return [{op_of[w] for w in wallets if w in op_of} for wallets in items]

    def rows(name, items):
        hits = found(items)
        out = {"queue": name, "items": len(items)}
        for k in KS:
            top = hits[:k]
            seen = set().union(*top) if top else set()
            out[f"precision@{k}"] = round(sum(bool(h) for h in top) / len(top), 3) if top else None
            out[f"recall@{k}"] = round(len(seen) / len(present), 3) if present else None
        # "Items reviewed before finding N", for N = 1, 3, 10, 20 and all of the
        # operations present, whichever of those the dataset holds.
        for n in sorted({k for k in (1, 3, 10, 20) if k <= len(present)} | {len(present)} - {0}):
            reviewed = None
            seen = set()
            for i, h in enumerate(hits, 1):
                seen |= h
                if len(seen) >= n:
                    reviewed = i
                    break
            label = "all" if n == len(present) else str(n)
            out[f"reviewed to find {label}"] = reviewed if reviewed is not None else "not reached"
        return out

    table = [rows("entity queue (before)", entity_items), rows("actor queue", actor_items)]

    def purity(wallets):
        c = Counter(controller(w) for w in wallets)
        return c.most_common(1)[0][1] / sum(c.values())

    multi = queue[queue["members"].map(len) > 1]
    majority = lambda m: Counter(controller(w) for w in _wallets(m, clusters)).most_common(1)[0][0]
    wrong = [len({majority(m) for m in members}) > 1 for members in multi["members"]]
    all_multi = built["actors"][built["actors"]["members"].map(len) > 1]
    wrong_all = [len({majority(m) for m in members}) > 1 for members in all_multi["members"]]
    quality = {
        "entity alerts": len(entity_items), "actor alerts": len(actor_items),
        "alert count reduction": round(1 - len(actor_items) / len(entity_items), 3)
        if entity_items else None,
        "illicit operations present": len(present),
        "actor purity (alerted, mean)": round(sum(map(purity, actor_items)) / len(actor_items), 3)
        if actor_items else None,
        "entity purity (alerted, mean)": round(sum(map(purity, entity_items)) / len(entity_items), 3)
        if entity_items else None,
        "alerted actors joining 2+ clusters": len(multi),
        "wrong-merge rate (alerted multi-cluster actors)": round(sum(wrong) / len(wrong), 3)
        if wrong else None,
        "actors joining 2+ clusters (all)": len(all_multi),
        "wrong-merge rate (all multi-cluster actors)": round(sum(wrong_all) / len(wrong_all), 3)
        if wrong_all else None,
        "links": len(built["links"]),
        "joining links": sum(A.may_join(lk) for lk in built["links"]),
    }
    return {"table": pd.DataFrame(table), "quality": quality}


def evaluate(datasets: dict[str, Dataset], stackers: dict, cfg: dict) -> dict:
    out = {}
    for name, dataset in datasets.items():
        built = dataset_actors(dataset, stackers[name], cfg)
        out[name] = triage(dataset, built)
    return out


# --- power: many seeds, larger datasets ----------------------------------------------
#: Sizes chosen so each dataset holds about thirty illicit operations (shifted
#: instances are larger, so it needs more transactions). Actors scale with size.
POWER_SIZES = {"standard": 9600, "shifted": 18000}
POWER_SEEDS = (101, 102, 103, 104, 105)
T_975 = {2: 12.706, 3: 4.303, 4: 3.182, 5: 2.776, 6: 2.571, 7: 2.447, 8: 2.365, 9: 2.306}


def power_datasets(cfg: dict, seeds=POWER_SEEDS, sizes=None) -> dict[tuple[str, int], Dataset]:
    from .datasets import build
    out = {}
    for condition, n in (sizes or POWER_SIZES).items():
        c = json.loads(json.dumps(cfg))
        c["eval"]["n_actors"], c["eval"]["n_transactions"] = n // 6, n
        for seed in seeds:
            out[(condition, seed)] = build(cfg["eval"]["default_rate"], condition == "shifted",
                                           seed, c, name_prefix=f"power-{condition}-n{n}")
    return out


def _interval(values: list[float]) -> str:
    """Mean and a 95% t-interval across seeds."""
    import statistics
    v = [float(x) for x in values if x is not None and not isinstance(x, str)]
    if not v:
        return "—"
    if len(v) == 1:
        return f"{v[0]:.3f} (one seed)"
    half = T_975.get(len(v), 2.0) * statistics.stdev(v) / len(v) ** 0.5
    return f"{statistics.mean(v):.3f} [{statistics.mean(v) - half:.3f}, {statistics.mean(v) + half:.3f}]"


def power(cfg: dict, seeds=POWER_SEEDS, sizes=None) -> dict:
    """The P9 comparison on every (condition, seed) dataset, each with its own
    fitted stacker, and the spread across seeds. The join rule is fusion/actors.py's,
    unchanged."""
    per = []
    for (condition, seed), dataset in power_datasets(cfg, seeds, sizes).items():
        result = triage(dataset, dataset_actors(dataset, None, cfg))
        table = result["table"].set_index("queue")
        per.append({"condition": condition, "seed": seed, "table": table,
                    "quality": result["quality"]})
    return {"per_seed": per, "summary": summarise_power(per)}


def summarise_power(per: list[dict]) -> dict:
    rows, diffs, merges, sizes = [], [], [], []
    for condition in sorted({p["condition"] for p in per}):
        runs = [p for p in per if p["condition"] == condition]
        metrics = [c for c in runs[0]["table"].columns if c != "items"]
        common = [m for m in metrics if all(m in r["table"].columns for r in runs)]
        for queue in runs[0]["table"].index:
            rows.append({"condition": condition, "queue": queue, "seeds": len(runs),
                         "items": _interval([r["table"].loc[queue, "items"] for r in runs]),
                         **{m: _interval([r["table"].loc[queue, m] for r in runs]) for m in common}})
        before, after = runs[0]["table"].index
        for m in ["items", *common]:
            pairs = [(r["table"].loc[before, m], r["table"].loc[after, m]) for r in runs]
            pairs = [(a, b) for a, b in pairs if not isinstance(a, str) and not isinstance(b, str)]
            diffs.append({"condition": condition, "measure": m,
                          "seeds with both": len(pairs),
                          "actor − entity": _interval([b - a for a, b in pairs])})
        q = [r["quality"] for r in runs]
        merges.append({"condition": condition,
                       **{k: _interval([x[k] for x in q]) for k in (
                           "illicit operations present", "alert count reduction",
                           "actor purity (alerted, mean)",
                           "wrong-merge rate (alerted multi-cluster actors)",
                           "wrong-merge rate (all multi-cluster actors)")},
                       "alerted multi-cluster actors (total)":
                           sum(x["alerted actors joining 2+ clusters"] for x in q)})
        sizes += [{"condition": condition, "seed": r["seed"],
                   "operations": r["quality"]["illicit operations present"],
                   "entity alerts": r["quality"]["entity alerts"],
                   "actor alerts": r["quality"]["actor alerts"],
                   **{f"reviewed to find all ({q.split()[0]})": r["table"].loc[q].get(
                       "reviewed to find all", "—") for q in r["table"].index}}
                  for r in runs]
    return {"queues": pd.DataFrame(rows), "paired": pd.DataFrame(diffs),
            "quality": pd.DataFrame(merges), "datasets": pd.DataFrame(sizes)}