"""Metrics and threshold selection for multi-label ATT&CK classification. Design notes that matter for reading the numbers: * **macro-F1 is the headline, not micro-F1.** 79% of sentences carry no label and the technique distribution is heavily long-tailed (T1027 has 678 examples, T1072 has ~20). Micro-F1 is dominated by a handful of frequent techniques and flatters every model. Macro-F1 is what tells you whether the tail works. * **Two threshold regimes are reported.** A single global threshold, and per-class thresholds tuned on dev. Per-class tuning fits 49 free parameters on a dev set with very few positives per class, so it can overfit; reporting both makes the size of that effect visible instead of hiding it. """ from __future__ import annotations import json from dataclasses import dataclass import numpy as np from . import config def to_matrix(records: list[dict], labels: list[str]) -> np.ndarray: index = {l: i for i, l in enumerate(labels)} Y = np.zeros((len(records), len(labels)), dtype=np.int8) for i, rec in enumerate(records): for l in rec["labels"]: j = index.get(l) if j is not None: Y[i, j] = 1 return Y def _prf(tp: float, fp: float, fn: float) -> tuple[float, float, float]: p = tp / (tp + fp) if tp + fp else 0.0 r = tp / (tp + fn) if tp + fn else 0.0 f = 2 * p * r / (p + r) if p + r else 0.0 return p, r, f @dataclass class Report: micro_f1: float macro_f1: float micro_precision: float micro_recall: float macro_precision: float macro_recall: float per_class: dict n_examples: int n_positive_labels: int def as_dict(self) -> dict: return { "micro_f1": round(self.micro_f1, 4), "macro_f1": round(self.macro_f1, 4), "micro_precision": round(self.micro_precision, 4), "micro_recall": round(self.micro_recall, 4), "macro_precision": round(self.macro_precision, 4), "macro_recall": round(self.macro_recall, 4), "n_examples": self.n_examples, "n_positive_labels": self.n_positive_labels, "per_class": self.per_class, } def evaluate(Y_true: np.ndarray, Y_pred: np.ndarray, labels: list[str]) -> Report: tp = (Y_true & Y_pred).sum(axis=0).astype(float) fp = ((1 - Y_true) & Y_pred).sum(axis=0).astype(float) fn = (Y_true & (1 - Y_pred)).sum(axis=0).astype(float) mip, mir, mif = _prf(tp.sum(), fp.sum(), fn.sum()) per_class, ps, rs, fs = {}, [], [], [] for j, name in enumerate(labels): p, r, f = _prf(tp[j], fp[j], fn[j]) per_class[name] = { "precision": round(p, 4), "recall": round(r, 4), "f1": round(f, 4), "support": int(Y_true[:, j].sum()), "predicted": int(Y_pred[:, j].sum()), } ps.append(p) rs.append(r) fs.append(f) return Report( micro_f1=mif, macro_f1=float(np.mean(fs)), micro_precision=mip, micro_recall=mir, macro_precision=float(np.mean(ps)), macro_recall=float(np.mean(rs)), per_class=per_class, n_examples=int(Y_true.shape[0]), n_positive_labels=int(Y_true.sum()), ) def tune_global_threshold( Y_true: np.ndarray, scores: np.ndarray, grid=None ) -> tuple[float, float]: """Pick the single threshold maximising macro-F1 on the given (dev) set.""" grid = grid or config.THRESHOLD_GRID best_t, best_f = 0.5, -1.0 for t in grid: Y_pred = (scores >= t).astype(np.int8) tp = (Y_true & Y_pred).sum(axis=0).astype(float) fp = ((1 - Y_true) & Y_pred).sum(axis=0).astype(float) fn = (Y_true & (1 - Y_pred)).sum(axis=0).astype(float) f = float(np.mean([_prf(tp[j], fp[j], fn[j])[2] for j in range(Y_true.shape[1])])) if f > best_f: best_t, best_f = t, f return best_t, best_f def tune_per_class_thresholds( Y_true: np.ndarray, scores: np.ndarray, grid=None ) -> np.ndarray: """One threshold per technique, each maximising that technique's dev F1.""" grid = grid or config.THRESHOLD_GRID out = np.full(Y_true.shape[1], 0.5) for j in range(Y_true.shape[1]): best_t, best_f = 0.5, -1.0 yt = Y_true[:, j] for t in grid: yp = (scores[:, j] >= t).astype(np.int8) _, _, f = _prf( float((yt & yp).sum()), float(((1 - yt) & yp).sum()), float((yt & (1 - yp)).sum()), ) if f > best_f: best_t, best_f = t, f out[j] = best_t return out def apply_thresholds(scores: np.ndarray, thresholds) -> np.ndarray: return (scores >= np.asarray(thresholds)).astype(np.int8) def save_report(name: str, scheme: str, reports: dict) -> None: path = config.RESULTS_DIR / f"{name}__{scheme}.json" path.write_text(json.dumps(reports, indent=2), encoding="utf-8") print(f" -> {path.relative_to(config.REPO_ROOT)}")