| """Evaluate a fine-tuned classifier on the unseen test sets (or the val split). |
| |
| This is the ONLY script allowed to read data/processed/test_holdout.parquet (TREC-06) and |
| data/processed/test_holdout3.parquet (the three-way holdout). Everything upstream |
| (data_prep, tokenizer, pretrain, finetune) is forbidden from touching them, so the numbers |
| written here are the project's honest generalisation estimate. |
| |
| Subcommand ``run --checkpoint <finetune best dir> [--split test|test3|val|test,test3]``: |
| |
| 1. load ``EncoderForClassification`` from ``config.json`` + ``model.safetensors`` (strict; |
| ``num_labels`` comes from ``config.json["model"]``), batched inference under bf16 |
| autocast with length-bucketed dynamic padding and bool key-padding masks -> the full |
| softmax ``P(class)`` for every row. |
| 2. **Binary roll-up, identical to the binary-era report.** ``P(unwanted) = P(phishing) + |
| P(spam)`` (for a 2-class checkpoint this is exactly ``P(class 1)``), so accuracy, |
| precision, recall, F1, specificity, ROC-AUC, PR-AUC, recall at 1% / 0.1% FPR and the |
| best-F1 threshold stay comparable with every earlier run. **Recall on the unwanted |
| class is reported first**: in a defensive triage tool a missed phish (false negative) |
| is the costly error, a false positive costs a human a glance. Plus: a credential-lure |
| slice (``CREDENTIAL_LURE_PHRASES``), a per-source table, and the token-length confound |
| check that RUN_LOG entry 2 asks for. |
| 3. **Three-way metrics** (RUN_LOG entry 6) on the rows whose ``label3`` is in {0, 1, 2}: |
| 3x3 confusion matrix (plot + table), per-class precision / recall / F1, macro-F1, and a |
| phishing-vs-spam confusion restricted to the true unwanted rows. Rows labelled |
| ``-1`` / ``"unwanted"`` (TREC-05 / TREC-07 / CEAS-08 positives, whose corpora never |
| separated spam from phishing) only enter the binary roll-up. The three-way decision |
| rule is **roll-up first**: a row is unwanted iff ``P(unwanted) >= threshold``, and then |
| the kind is ``argmax(P(phishing), P(spam))``, so the 3x3 matrix collapses exactly onto |
| the binary confusion matrix above; plain argmax-of-3 is reported next to it. On the |
| TREC-06 split there is no phishing/spam ground truth, so only an *informational* |
| breakdown of how the flagged rows split into predicted phishing vs spam is written, |
| and it says so. A 2-class checkpoint skips every three-way section with a log line. |
| 4. single-email latency like the instructor's step 5: tokenize + forward + softmax at |
| batch 1 timed with ``timeit.default_timer``, GPU bf16 and CPU fp32 (16 threads), |
| printed next to the 74 ms benchmark -> ``latency.json`` (measured once per invocation). |
| 5. plots (confusion matrices, ROC, PR), ``error_analysis.md`` (top false negatives / |
| positives, phishing->spam and spam->phishing confusions with previews, lexical |
| patterns, written summary), ``metrics.json`` (with a ``threeway`` block), ONE line in |
| ``experiments.jsonl`` and ONE RUN_LOG.md entry per invocation covering every split |
| that was run (skipped for ``--limit-rows`` dev runs, which are not evidence). |
| ``reports/pretrain_curves.png`` and ``finetune_curves.png`` are written by |
| ``pretrain.py plot`` / ``finetune.py plot`` and are never touched here; |
| ``--finetune-run-dir`` / ``--pretrain-run-dir`` opt into extra ``eval_*_curves.png``. |
| |
| Outputs for ``--split val`` get a ``_val`` suffix and for ``--split test3`` a ``_test3`` |
| suffix so no split's report overwrites another's. ``--out-dir`` redirects every file, so a |
| self-test never writes into reports/; ``--data-dir`` points at fixture parquet files. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import logging |
| import math |
| import re |
| import sys |
| import timeit |
| import unicodedata |
| from collections import Counter |
| from dataclasses import dataclass |
| from datetime import datetime |
| from pathlib import Path |
| from typing import Any, Sequence |
|
|
| sys.path.insert(0, str(Path(__file__).parent)) |
|
|
| import matplotlib |
|
|
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import pyarrow.parquet as pq |
| import torch |
| from safetensors.torch import load_file |
| from sklearn.metrics import ( |
| ConfusionMatrixDisplay, |
| average_precision_score, |
| confusion_matrix, |
| precision_recall_curve, |
| precision_recall_fscore_support, |
| roc_auc_score, |
| roc_curve, |
| ) |
|
|
| import common |
| from model import EncoderConfig, EncoderForClassification, count_parameters |
| from tokenizer import encode_batch, load_tokenizer |
|
|
| LOG = logging.getLogger("evaluate") |
|
|
| SPLIT_FILES: dict[str, str] = {"test": "test_holdout.parquet", "val": "finetune_val.parquet", "test3": "test_holdout3.parquet"} |
| SPLIT_SUFFIX: dict[str, str] = {"test": "", "val": "_val", "test3": "_test3"} |
| SPLIT_TITLE: dict[str, str] = {"test": "TREC-06 binary holdout", "val": "validation split", "test3": "three-way holdout"} |
| |
| |
| LABEL_NAMES: dict[int, str] = {0: "Legitimate", 1: "Phishing", 2: "Spam"} |
| UNWANTED_NAME = "Unwanted (phishing or spam)" |
| BINARY_NAMES: dict[int, str] = {0: "Legitimate", 1: UNWANTED_NAME} |
| LABEL_SET_OF: dict[int, str] = {0: "legit", 1: "phish", 2: "spam", -1: "unwanted"} |
| DECISION_RULE = "rollup_then_argmax" |
| DECISION_RULE_DOC = ("unwanted iff P(phishing)+P(spam) >= threshold, then kind = argmax(P(phishing), P(spam)); " |
| "the 3x3 matrix therefore collapses exactly onto the binary confusion matrix. argmax-of-3 reported alongside.") |
| INSTRUCTOR_BENCHMARK_MS = 74.0 |
| DEFAULT_THRESHOLD = 0.5 |
| CPU_LATENCY_BUDGET_S = 60.0 |
| CPU_LATENCY_FALLBACK_N = 10 |
| TOP_K_ERRORS = 25 |
| TOP_K_CONFUSIONS = 25 |
| PREVIEW_CHARS = 160 |
| ENCODE_BATCH_ROWS = 2048 |
| LENGTH_BUCKETS: tuple[tuple[int, int, str], ...] = ( |
| (0, 64, "<=64"), (65, 128, "65-128"), (129, 256, "129-256"), (257, 512, "257-512"), (513, 10**9, ">512"), |
| ) |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| CREDENTIAL_LURE_PHRASES: tuple[str, ...] = ( |
| "verify your account", "verify your identity", "confirm your identity", "confirm your account", |
| "enter your password", "update your account", "update your information", "update your billing", |
| "update your payment", "click here to verify", "click the link below to verify", |
| "account has been suspended", "account has been limited", "account will be suspended", |
| "account has been locked", "unusual sign-in", "unusual activity", "suspicious activity", |
| "paypal account", "ebay account", "password expire", "password will expire", "reset your password", |
| "invoice attached", "wire transfer", "gift card", "urgent", "urgently", "security alert", "security notice", |
| "login to your account", "log in to your account", "sign in to your account", |
| "bank account", "failure to verify", "within 24 hours", "within 48 hours", |
| ) |
| LURE_MIN_POSITIVES = 20 |
| _LURE_NON_ALNUM = re.compile(r"[^a-z0-9]+") |
|
|
| |
| C_BLUE, C_ORANGE, C_AQUA, C_VIOLET = "#2a78d6", "#eb6834", "#1baf7a", "#4a3aa7" |
| C_INK, C_INK2, C_GRID, C_SURFACE = "#0b0b0b", "#52514e", "#dddcd7", "#fcfcfb" |
| PLOT_DPI = 150 |
|
|
|
|
| |
| @dataclass |
| class Split: |
| name: str |
| path: Path |
| doc_id: list[str] |
| text: list[str] |
| subject: list[str] |
| body: list[str] |
| label: np.ndarray |
| label3: np.ndarray |
| label_set: list[str] |
| source: list[str] |
| limit_rows: int | None |
|
|
| @property |
| def n(self) -> int: |
| return len(self.text) |
|
|
| @property |
| def sfx(self) -> str: |
| return SPLIT_SUFFIX[self.name] |
|
|
|
|
| def check_eval_config(cfg: dict) -> int: |
| """Reconcile the documented ``eval:`` dials with this module's constants and return ``top_k_confusions``. |
| |
| ``label_names``, ``unwanted_name`` and ``threeway_decision_rule`` are the three-way label |
| contract shared with finetune.py, export.py and the app, so they are fixed in code and |
| config.yaml may only restate them: a drift raises here, loudly, instead of producing a card |
| whose names differ from the metrics. ``top_k_confusions`` is a real dial (size of the |
| phishing->spam / spam->phishing preview tables). ``eval.metrics_threeway`` is documentation |
| only: it lists the sections the three-way block always contains.""" |
| ev = cfg.get("eval") or {} |
| names = ev.get("label_names") |
| if names is not None: |
| got = {int(c): str(v) for c, v in dict(names).items()} |
| if got != LABEL_NAMES: |
| raise ValueError(f"config eval.label_names {got} != evaluate.LABEL_NAMES {LABEL_NAMES}: the label contract is fixed in code") |
| if ev.get("unwanted_name") is not None and str(ev["unwanted_name"]) != UNWANTED_NAME: |
| raise ValueError(f"config eval.unwanted_name {ev['unwanted_name']!r} != {UNWANTED_NAME!r}: fixed in code") |
| if ev.get("threeway_decision_rule") is not None and str(ev["threeway_decision_rule"]) != DECISION_RULE: |
| raise ValueError(f"config eval.threeway_decision_rule {ev['threeway_decision_rule']!r} != {DECISION_RULE!r}: only this rule is implemented") |
| top_k = int(ev.get("top_k_confusions", TOP_K_CONFUSIONS)) |
| if top_k < 1: |
| raise ValueError(f"config eval.top_k_confusions must be >= 1, got {top_k}") |
| return top_k |
|
|
|
|
| def split_path(data_dir: Path, name: str, test3_file: str) -> Path: |
| return data_dir / (test3_file if name == "test3" else SPLIT_FILES[name]) |
|
|
|
|
| def load_split(data_dir: Path, name: str, limit_rows: int | None, seed: int, test3_file: str = SPLIT_FILES["test3"]) -> Split: |
| """Read one labelled split. ``limit_rows`` takes a seeded random sample (dev only); |
| the full run reads every row so the report is the whole holdout. |
| |
| ``label3`` / ``label_set`` are read when the file carries them (data_prep writes them |
| for every split since RUN_LOG entry 6). A binary-era file without them is still |
| evaluable: every positive is an unsplit "unwanted" row (-1) and every negative is |
| legitimate (0), which is precisely the TREC-06 contract, so the binary roll-up is |
| unaffected and the three-way section simply has no rows.""" |
| path = split_path(data_dir, name, test3_file) |
| if not path.exists(): |
| raise FileNotFoundError(path) |
| have = set(pq.read_schema(path).names) |
| cols = ["doc_id", "text", "subject", "body", "source"] + [c for c in ("label", "label3", "label_set") if c in have] |
| if "label" not in have and "label3" not in have: |
| raise ValueError(f"{path}: neither 'label' nor 'label3' column present") |
| table = pq.read_table(path, columns=cols) |
| if limit_rows is not None and table.num_rows > limit_rows: |
| idx = np.sort(np.random.default_rng(seed).choice(table.num_rows, size=limit_rows, replace=False)) |
| table = table.take(idx) |
| d = table.to_pydict() |
| n = table.num_rows |
| if "label3" in have: |
| label3 = np.asarray(d["label3"], dtype=np.int64) |
| bad = sorted(set(label3.tolist()) - set(LABEL_SET_OF)) |
| if bad: |
| raise ValueError(f"{path}: label3 values outside {{-1,0,1,2}}: {bad}") |
| else: |
| label3 = None |
| if "label" in have: |
| label = np.asarray(d["label"], dtype=np.int64) |
| else: |
| label = (label3 != 0).astype(np.int64) |
| if label3 is None: |
| label3 = np.where(label == 1, -1, 0).astype(np.int64) |
| LOG.info("%s has no label3 column: positives treated as unwanted-but-unsplit (-1), negatives as legitimate (0)", path.name) |
| if "label_set" in have: |
| label_set = [str(s) for s in d["label_set"]] |
| |
| |
| |
| mismatch = [i for i in range(n) if label_set[i] != LABEL_SET_OF[int(label3[i])]] |
| if mismatch: |
| i = mismatch[0] |
| raise ValueError(f"{path}: 'label_set' disagrees with 'label3' on {len(mismatch)} row(s) (first: row {i} " |
| f"label_set={label_set[i]!r} but label3={int(label3[i])} implies {LABEL_SET_OF[int(label3[i])]!r}); " |
| f"contract: {LABEL_SET_OF}") |
| else: |
| label_set = [LABEL_SET_OF[int(v)] for v in label3] |
| if bool(((label3 != 0).astype(np.int64) != label).any()): |
| raise ValueError(f"{path}: 'label' disagrees with 'label3 != 0' on some rows (contract: label = 1 for phish/spam/unwanted)") |
| return Split( |
| name=name, path=path, |
| doc_id=[str(x) for x in d["doc_id"]], |
| text=[t or "" for t in d["text"]], |
| subject=[s or "" for s in d["subject"]], |
| body=[b or "" for b in d["body"]], |
| label=label, label3=label3, label_set=label_set, |
| source=[str(s) for s in d["source"]], |
| limit_rows=limit_rows, |
| ) |
|
|
|
|
| @dataclass |
| class Encoded: |
| """Per-row token ids as the model sees them, plus the untruncated length for the confound check.""" |
|
|
| ids: list[list[int]] |
| n_tokens: np.ndarray |
| n_tokens_full: np.ndarray |
| truncated: np.ndarray |
|
|
|
|
| def encode_split(tok, texts: Sequence[str], max_len: int) -> Encoded: |
| """Encode once without truncation (so the full length is known), then truncate the |
| content to ``max_len - 2`` and keep ``[CLS] ... [SEP]``, which is exactly what |
| ``tokenizers`` truncation + the BERT template produce.""" |
| tok.no_truncation() |
| tok.no_padding() |
| sep_id = tok.token_to_id("[SEP]") |
| ids: list[list[int]] = [] |
| full = np.empty(len(texts), dtype=np.int64) |
| for i in range(0, len(texts), ENCODE_BATCH_ROWS): |
| for j, enc in enumerate(tok.encode_batch_fast(list(texts[i:i + ENCODE_BATCH_ROWS]))): |
| seq = enc.ids |
| full[i + j] = len(seq) |
| if len(seq) > max_len: |
| seq = seq[: max_len - 1] + [sep_id] |
| ids.append(seq) |
| n_tok = np.minimum(full, max_len) |
| return Encoded(ids, n_tok, full, full > max_len) |
|
|
|
|
| |
| def load_classifier(ckpt_dir: Path, device: torch.device) -> tuple[EncoderForClassification, EncoderConfig, dict]: |
| """Rebuild the classifier from ``config.json["model"]`` (including ``num_labels``) and |
| load ``model.safetensors`` strictly. |
| |
| Strict loading is non-negotiable: a half-loaded encoder still produces confident |
| probabilities, and nothing downstream could tell the difference. A head whose shape |
| disagrees with ``num_labels`` fails here, loudly. Weights may be fp32 (training |
| checkpoint) or bf16 (export); ``load_state_dict`` casts into the fp32 module, and the |
| GPU path runs under bf16 autocast either way. |
| """ |
| cfg_json = common.read_json(ckpt_dir / "config.json") |
| mcfg = EncoderConfig.from_dict(cfg_json["model"]) |
| mcfg.grad_checkpointing = False |
| if mcfg.num_labels not in (2, 3): |
| raise ValueError(f"num_labels={mcfg.num_labels}: this evaluator understands 2 (binary) or 3 (legitimate/phishing/spam)") |
| with torch.device(device): |
| model = EncoderForClassification(mcfg) |
| sd = load_file(str(ckpt_dir / "model.safetensors"), device="cpu") |
| model.load_state_dict(sd, strict=True) |
| model.eval() |
| dtypes = sorted({str(v.dtype).replace("torch.", "") for v in sd.values()}) |
| LOG.info("loaded %s: %d tensors (%s), %s params, num_labels=%d", ckpt_dir / "model.safetensors", len(sd), dtypes, |
| f"{count_parameters(model)['total']:,}", mcfg.num_labels) |
| return model, mcfg, cfg_json |
|
|
|
|
| def _pad_batch(enc: Encoded, idx: np.ndarray, multiple: int) -> tuple[torch.Tensor, torch.Tensor]: |
| L = int(enc.n_tokens[idx].max()) |
| L = int(math.ceil(L / multiple) * multiple) |
| arr = np.zeros((len(idx), L), dtype=np.int64) |
| mask = np.zeros((len(idx), L), dtype=np.bool_) |
| for r, i in enumerate(idx): |
| seq = enc.ids[i] |
| arr[r, : len(seq)] = seq |
| mask[r, : len(seq)] = True |
| return torch.from_numpy(arr), torch.from_numpy(mask) |
|
|
|
|
| @torch.inference_mode() |
| def predict_probs(model: EncoderForClassification, enc: Encoded, device: torch.device, |
| batch_size: int, pad_multiple: int) -> tuple[np.ndarray, dict[str, Any]]: |
| """Full softmax ``[n, num_labels]`` per row. Rows are sorted longest-first so each batch |
| pads to its own longest member (dynamic padding; ~2x fewer pad tokens than padding to |
| 512) and an OOM would surface on the first batch rather than after ten minutes.""" |
| n = len(enc.ids) |
| k = int(model.config.num_labels) |
| order = np.argsort(-enc.n_tokens, kind="stable") |
| probs = np.empty((n, k), dtype=np.float32) |
| use_amp = device.type == "cuda" |
| padded_tokens = 0 |
| n_batches = math.ceil(n / batch_size) |
| t0 = common.timer() |
| for b, start in enumerate(range(0, n, batch_size)): |
| idx = order[start: start + batch_size] |
| ids, mask = _pad_batch(enc, idx, pad_multiple) |
| padded_tokens += ids.numel() |
| ids, mask = ids.to(device, non_blocking=True), mask.to(device, non_blocking=True) |
| with torch.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=use_amp): |
| logits = model(ids, mask) |
| probs[idx] = logits.float().softmax(-1).cpu().numpy() |
| if b % 20 == 0 or b == n_batches - 1: |
| LOG.info("inference batch %d/%d (L=%d) %.1fs", b + 1, n_batches, ids.shape[1], common.timer() - t0) |
| if use_amp: |
| torch.cuda.synchronize() |
| secs = common.timer() - t0 |
| info = { |
| "device": str(device), "autocast_bf16": use_amp, "batch_size": batch_size, "pad_to_multiple_of": pad_multiple, |
| "num_labels": k, "rows": n, "seconds": round(secs, 2), "rows_per_s": round(n / secs, 1), |
| "padded_tokens": int(padded_tokens), "real_tokens": int(enc.n_tokens.sum()), |
| "padded_tokens_per_s": round(padded_tokens / secs), |
| "peak_vram_gb": round(common.vram_stats()["max_allocated_gb"], 2), |
| } |
| LOG.info("inference: %s", info) |
| return probs, info |
|
|
|
|
| def p_unwanted(probs: np.ndarray) -> np.ndarray: |
| """Binary roll-up: P(unwanted) = sum of every non-legitimate class. For a 2-class head |
| this is exactly P(class 1), so the binary report is byte-for-byte the old definition.""" |
| return np.clip(probs[:, 1:].sum(axis=1), 0.0, 1.0).astype(np.float32) |
|
|
|
|
| |
| def _safe_div(a: float, b: float) -> float: |
| return float(a) / float(b) if b else 0.0 |
|
|
|
|
| def threshold_metrics(y: np.ndarray, p: np.ndarray, thr: float) -> dict[str, Any]: |
| """Confusion-matrix metrics at ``p >= thr``. An empty slice (e.g. no credential-lure |
| rows in a small ``--limit-rows`` sample) returns all-zero counts instead of hitting |
| sklearn's minimum-one-sample check.""" |
| if len(y) == 0: |
| tn = fp = fn = tp = 0 |
| else: |
| pred = (p >= thr).astype(np.int64) |
| cm = confusion_matrix(y, pred, labels=[0, 1]) |
| tn, fp, fn, tp = (int(v) for v in cm.ravel()) |
| precision = _safe_div(tp, tp + fp) |
| recall = _safe_div(tp, tp + fn) |
| return { |
| "threshold": thr, |
| "accuracy": _safe_div(tp + tn, len(y)), |
| "precision": precision, |
| "recall": recall, |
| "f1": _safe_div(2 * precision * recall, precision + recall), |
| "specificity": _safe_div(tn, tn + fp), |
| "false_positive_rate": _safe_div(fp, tn + fp), |
| "false_negative_rate": _safe_div(fn, tp + fn), |
| "confusion_matrix": {"tn": tn, "fp": fp, "fn": fn, "tp": tp}, |
| "n": int(len(y)), "n_positive": int(y.sum()), "n_negative": int(len(y) - y.sum()), |
| } |
|
|
|
|
| def _both_classes(y: np.ndarray) -> bool: |
| return bool(y.min() == 0 and y.max() == 1) if len(y) else False |
|
|
|
|
| def threshold_free_metrics(y: np.ndarray, p: np.ndarray) -> dict[str, Any]: |
| """ROC-AUC, PR-AUC (average precision), recall at a fixed FPR budget read off the ROC |
| curve, and the threshold that maximises F1. The FPR-budget numbers are the ones an |
| inbox operator actually asks ("if I may bother users with 1 false alarm in 100 |
| legitimate emails, how many phish do I catch?"). Thresholds are to be applied as |
| ``p >= threshold``; when the only point inside the budget is the ROC origin, sklearn's |
| sentinel threshold is +inf and ``threshold`` is reported as None (no finite score |
| cut reaches that budget) rather than clamped to 1.0, which bf16 probabilities that |
| round to exactly 1.0 would still pass.""" |
| if not _both_classes(y): |
| LOG.warning("only one class present: ROC/PR metrics undefined") |
| return {"roc_auc": None, "pr_auc": None, "recall_at_fpr_1pct": None, "recall_at_fpr_0_1pct": None, |
| "best_f1": None, "curves": None} |
| fpr, tpr, roc_thr = roc_curve(y, p) |
| prec, rec, pr_thr = precision_recall_curve(y, p) |
|
|
| def recall_at_fpr(budget: float) -> dict[str, float]: |
| ok = np.nonzero(fpr <= budget)[0] |
| i = ok[np.argmax(tpr[ok])] |
| t = float(roc_thr[i]) |
| return {"fpr_budget": budget, "recall": float(tpr[i]), "fpr": float(fpr[i]), |
| "threshold": t if math.isfinite(t) else None, "threshold_rule": "flag when p >= threshold"} |
|
|
| f1 = 2 * prec[:-1] * rec[:-1] / np.clip(prec[:-1] + rec[:-1], 1e-12, None) |
| j = int(np.argmax(f1)) |
| return { |
| "roc_auc": float(roc_auc_score(y, p)), |
| "pr_auc": float(average_precision_score(y, p)), |
| "recall_at_fpr_1pct": recall_at_fpr(0.01), |
| "recall_at_fpr_0_1pct": recall_at_fpr(0.001), |
| "best_f1": {"threshold": float(pr_thr[j]), "f1": float(f1[j]), "precision": float(prec[j]), "recall": float(rec[j])}, |
| "curves": {"fpr": fpr, "tpr": tpr, "precision": prec, "recall": rec}, |
| } |
|
|
|
|
| def lure_normalize(s: str) -> str: |
| """Word-boundary-preserving sibling of ``common.normalize_text``: HTML tags removed, |
| NFKC -> lowercase -> every non-alphanumeric run collapses to one space, padded with a |
| space on both ends so a phrase match ``" ph "`` is a whole-word match |
| ("Verify your<br>account" -> " verify your account ").""" |
| s = _HTML_TAG.sub(" ", s or "") |
| return " " + _LURE_NON_ALNUM.sub(" ", unicodedata.normalize("NFKC", s).lower()).strip() + " " |
|
|
|
|
| def lure_mask(texts: Sequence[str], labels: np.ndarray | None = None) -> tuple[np.ndarray, Counter, dict[str, dict[str, int]]]: |
| """Rows containing any lure phrase as whole words, per-phrase hit counts, and (when |
| labels are given) per-phrase positive/negative hits so the slice's purity is visible.""" |
| norm_phrases = [(ph, lure_normalize(ph)) for ph in CREDENTIAL_LURE_PHRASES] |
| hits = np.zeros(len(texts), dtype=np.bool_) |
| per_phrase: Counter = Counter() |
| by_label: dict[str, dict[str, int]] = {ph: {"positive": 0, "negative": 0} for ph in CREDENTIAL_LURE_PHRASES} |
| for i, t in enumerate(texts): |
| n = lure_normalize(t) |
| for ph, nph in norm_phrases: |
| if nph in n: |
| hits[i] = True |
| per_phrase[ph] += 1 |
| if labels is not None: |
| by_label[ph]["positive" if int(labels[i]) == 1 else "negative"] += 1 |
| return hits, per_phrase, {ph: c for ph, c in by_label.items() if per_phrase[ph]} |
|
|
|
|
| def slice_metrics(y: np.ndarray, p: np.ndarray, thr: float) -> dict[str, Any]: |
| m = threshold_metrics(y, p, thr) |
| m["roc_auc"] = float(roc_auc_score(y, p)) if _both_classes(y) else None |
| return m |
|
|
|
|
| def per_source_table(split: Split, p: np.ndarray, thr: float) -> dict[str, dict[str, Any]]: |
| out: dict[str, dict[str, Any]] = {} |
| src = np.asarray(split.source) |
| for s in sorted(set(split.source)): |
| sel = src == s |
| m = slice_metrics(split.label[sel], p[sel], thr) |
| m["label3_counts"] = _label3_counts(split.label3[sel]) |
| out[s] = m |
| return out |
|
|
|
|
| def _label3_counts(label3: np.ndarray) -> dict[str, int]: |
| return {LABEL_SET_OF[k]: int((label3 == k).sum()) for k in (0, 1, 2, -1)} |
|
|
|
|
| def length_confound(split: Split, enc: Encoded, p: np.ndarray, thr: float) -> dict[str, Any]: |
| """Mean token length of FP/FN vs correct rows, by label, and recall/FPR per length |
| bucket. If errors concentrate in the length range where the classes' length |
| distributions overlap (long positives, short legitimate), the model is at least |
| partly reading length rather than content. ``by_class3`` adds the phishing / spam / |
| unsplit split of the positive class so a phishing-vs-spam length gap is visible too.""" |
| y, pred = split.label, (p >= thr).astype(np.int64) |
| groups = {"tp": (y == 1) & (pred == 1), "tn": (y == 0) & (pred == 0), |
| "fp": (y == 0) & (pred == 1), "fn": (y == 1) & (pred == 0)} |
|
|
| def stats(sel: np.ndarray) -> dict[str, Any]: |
| if not sel.any(): |
| return {"n": 0, "mean_tokens_full": None, "mean_tokens_fed": None, "median_tokens_full": None, "pct_truncated": None} |
| return {"n": int(sel.sum()), "mean_tokens_full": round(float(enc.n_tokens_full[sel].mean()), 1), |
| "mean_tokens_fed": round(float(enc.n_tokens[sel].mean()), 1), |
| "median_tokens_full": float(np.median(enc.n_tokens_full[sel])), |
| "pct_truncated": round(100 * float(enc.truncated[sel].mean()), 1)} |
|
|
| by_outcome = {k: stats(v) for k, v in groups.items()} |
| by_outcome["correct"] = stats(groups["tp"] | groups["tn"]) |
| by_outcome["wrong"] = stats(groups["fp"] | groups["fn"]) |
| by_label = {"legitimate": stats(y == 0), "unwanted": stats(y == 1)} |
| by_class3 = {LABEL_SET_OF[k]: stats(split.label3 == k) for k in (0, 1, 2, -1) if (split.label3 == k).any()} |
| buckets = [] |
| for lo, hi, name in LENGTH_BUCKETS: |
| sel = (enc.n_tokens_full >= lo) & (enc.n_tokens_full <= hi) |
| if not sel.any(): |
| continue |
| m = threshold_metrics(y[sel], p[sel], thr) |
| buckets.append({"bucket": name, "n": m["n"], "n_positive": m["n_positive"], "positive_rate": _safe_div(m["n_positive"], m["n"]), |
| "recall": m["recall"] if m["n_positive"] else None, |
| "fpr": m["false_positive_rate"] if m["n_negative"] else None, "accuracy": m["accuracy"]}) |
| return {"by_outcome": by_outcome, "by_label": by_label, "by_class3": by_class3, "by_length_bucket": buckets} |
|
|
|
|
| |
| @dataclass |
| class ThreeWay: |
| """Three-way result for one split. ``metrics`` is the JSON block; ``pred3`` / ``kind`` |
| are per-row arrays (full length) that the error analysis uses, None for a 2-class head.""" |
|
|
| evaluated: bool |
| metrics: dict[str, Any] |
| pred3: np.ndarray | None |
| kind: np.ndarray | None |
|
|
|
|
| def predict3(probs: np.ndarray, p_unw: np.ndarray, thr: float) -> tuple[np.ndarray, np.ndarray]: |
| """Roll-up-first decision: ``kind`` = argmax over the unwanted classes (1 phishing / |
| 2 spam) for every row; ``pred3`` = kind when P(unwanted) >= thr else 0 (legitimate). |
| Choosing the kind *conditionally* on being unwanted (rather than argmax-of-3) keeps the |
| three-way prediction consistent with the binary verdict the app shows: a row the app |
| flags can never be "legitimate" in the 3x3 table, and vice versa.""" |
| kind = 1 + probs[:, 1:3].argmax(axis=1) |
| pred3 = np.where(p_unw >= thr, kind, 0).astype(np.int64) |
| return pred3, kind.astype(np.int64) |
|
|
|
|
| def _cm_dict(mat: np.ndarray, labels: Sequence[int]) -> dict[str, dict[str, int]]: |
| return {f"true_{LABEL_NAMES[a].lower()}": {f"pred_{LABEL_NAMES[b].lower()}": int(mat[i, j]) for j, b in enumerate(labels)} |
| for i, a in enumerate(labels)} |
|
|
|
|
| def _macro_block(y: np.ndarray, pred: np.ndarray) -> dict[str, Any]: |
| p, r, f, s = precision_recall_fscore_support(y, pred, labels=[0, 1, 2], zero_division=0) |
| per_class = {LABEL_NAMES[k]: {"precision": float(p[k]), "recall": float(r[k]), "f1": float(f[k]), "support": int(s[k])} for k in range(3)} |
| present = [k for k in range(3) if s[k] > 0] |
| return {"accuracy": float((y == pred).mean()) if len(y) else 0.0, |
| "macro_f1": float(np.mean([f[k] for k in range(3)])), |
| "macro_f1_present_classes": float(np.mean([f[k] for k in present])) if present else 0.0, |
| "classes_present": [LABEL_NAMES[k] for k in present], |
| "per_class": per_class} |
|
|
|
|
| def threeway_metrics(split: Split, probs: np.ndarray, p_unw: np.ndarray, thr: float) -> ThreeWay: |
| """Three-way block for one split. |
| |
| * 2-class head: nothing to evaluate (the checkpoint has no phishing-vs-spam output). |
| * 3-class head: ``predicted_kind_breakdown`` is ALWAYS written (how the flagged rows |
| split into predicted phishing vs spam); it is informational on TREC-06, where every |
| positive is an unsplit "unwanted" row and no ground truth exists, and it says so. |
| * rows with ``label3`` in {0, 1, 2}: 3x3 confusion matrix, per-class P/R/F1, macro-F1 |
| (over all three classes, zero F1 for an absent class, plus the mean over the classes |
| actually present), argmax-of-3 as a secondary rule, a phishing-vs-spam confusion on |
| the true unwanted rows (the kind decision alone, i.e. "given it is unwanted, which |
| kind", plus the roll-up view that also shows how many were missed as legitimate), |
| and per-source class counts / accuracy / macro-F1. |
| """ |
| k = probs.shape[1] |
| base: dict[str, Any] = {"num_labels": k, "label_names": {str(c): LABEL_NAMES[c] for c in range(k) if c in LABEL_NAMES}, |
| "unwanted_name": UNWANTED_NAME, "decision_rule": DECISION_RULE, "decision_rule_doc": DECISION_RULE_DOC, |
| "threshold": thr, "label3_counts": _label3_counts(split.label3)} |
| if k != 3: |
| base["label_names"] = {"0": BINARY_NAMES[0], "1": BINARY_NAMES[1]} |
| return ThreeWay(False, {**base, "evaluated": False, "reason": f"{k}-class checkpoint: no phishing-vs-spam head, three-way sections skipped"}, None, None) |
| pred3, kind = predict3(probs, p_unw, thr) |
| argmax3 = probs.argmax(axis=1) |
| y, y3 = split.label, split.label3 |
| flagged = p_unw >= thr |
|
|
| def kinds(sel: np.ndarray) -> dict[str, int]: |
| return {"n": int(sel.sum()), "predicted_phishing": int((kind[sel] == 1).sum()), "predicted_spam": int((kind[sel] == 2).sum())} |
|
|
| base["predicted_kind_breakdown"] = { |
| "note": "How the model's flagged rows split into predicted phishing vs spam (kind = argmax(P(phishing), P(spam))). " |
| "Informational: for rows labelled -1/unwanted (all TREC-06 positives) there is NO phishing-vs-spam ground truth.", |
| "true_positives_flagged": kinds((y == 1) & flagged), "false_positives_flagged": kinds((y == 0) & flagged), |
| "missed_positives": kinds((y == 1) & ~flagged), "all_flagged": kinds(flagged), |
| "unsplit_unwanted_rows_flagged": kinds((y3 == -1) & flagged), |
| } |
| full = (y3 >= 0) & (y3 <= 2) |
| base["rows_with_full_label"] = int(full.sum()) |
| base["rows_unwanted_only"] = int((y3 == -1).sum()) |
| |
| |
| |
| if not (full & (y3 >= 1)).any(): |
| return ThreeWay(False, {**base, "evaluated": False, |
| "reason": "no rows labelled phishing or spam: every positive is unwanted-but-unsplit (no phishing-vs-spam ground truth)"}, |
| pred3, kind) |
| yf, pf, af = y3[full], pred3[full], argmax3[full] |
| mat = confusion_matrix(yf, pf, labels=[0, 1, 2]) |
| main = _macro_block(yf, pf) |
| alt = _macro_block(yf, af) |
| tu = full & (y3 >= 1) |
| if tu.any(): |
| kmat = confusion_matrix(y3[tu], kind[tu], labels=[1, 2]) |
| rmat = confusion_matrix(y3[tu], pred3[tu], labels=[0, 1, 2])[1:, :] |
| pvs = {"n_true_unwanted": int(tu.sum()), |
| "kind_only": {"doc": "given the row is unwanted, is argmax(P(phishing), P(spam)) right? (independent of the roll-up threshold)", |
| "labels": ["Phishing", "Spam"], "matrix": kmat.tolist(), "cells": _cm_dict(kmat, [1, 2]), |
| "accuracy": float(np.trace(kmat) / max(kmat.sum(), 1)), |
| "phishing_as_spam": int(kmat[0, 1]), "spam_as_phishing": int(kmat[1, 0])}, |
| "rollup_rule": {"doc": "rows = true Phishing / Spam; columns = predicted Legitimate (missed) / Phishing / Spam under the roll-up rule", |
| "rows": ["Phishing", "Spam"], "columns": ["Legitimate", "Phishing", "Spam"], "matrix": rmat.tolist(), |
| "missed_as_legitimate": {"phishing": int(rmat[0, 0]), "spam": int(rmat[1, 0])}}} |
| else: |
| pvs = {"n_true_unwanted": 0, "kind_only": None, "rollup_rule": None} |
| src = np.asarray(split.source) |
| per_source: dict[str, Any] = {} |
| for s in sorted(set(split.source)): |
| sel = full & (src == s) |
| if not sel.any(): |
| continue |
| b = _macro_block(y3[sel], pred3[sel]) |
| per_source[s] = {"n": int(sel.sum()), "counts": {LABEL_NAMES[c]: int((y3[sel] == c).sum()) for c in range(3)}, |
| "accuracy": b["accuracy"], "macro_f1": b["macro_f1"], "macro_f1_present_classes": b["macro_f1_present_classes"], |
| "per_class_recall": {c: v["recall"] for c, v in b["per_class"].items() if v["support"]}} |
| metrics = {**base, "evaluated": True, "reason": None, |
| "confusion_matrix": {"labels": [LABEL_NAMES[c] for c in range(3)], "matrix": mat.tolist(), "cells": _cm_dict(mat, [0, 1, 2])}, |
| "accuracy": main["accuracy"], "macro_f1": main["macro_f1"], "macro_f1_present_classes": main["macro_f1_present_classes"], |
| "classes_present": main["classes_present"], "per_class": main["per_class"], |
| "argmax_rule": {"doc": "plain argmax over the three probabilities (secondary; can disagree with the binary verdict near the threshold)", |
| "accuracy": alt["accuracy"], "macro_f1": alt["macro_f1"], "per_class": alt["per_class"], |
| "rows_where_rules_differ": int((pf != af).sum())}, |
| "phish_vs_spam_on_true_unwanted": pvs, "per_source": per_source} |
| return ThreeWay(True, metrics, pred3, kind) |
|
|
|
|
| |
| def _time_single(model, tok, texts: Sequence[str], device: torch.device, max_len: int, |
| use_amp: bool, warmup: int = 3) -> list[float]: |
| """Wall-clock ms per email for tokenize + forward + softmax at batch 1, measured with |
| ``timeit.default_timer`` exactly like the instructor's step-5 inference timer. |
| ``.item()`` forces the GPU sync so the stop time includes the kernel work (the |
| roll-up sum is part of what the app computes per request, so it is timed too).""" |
| times: list[float] = [] |
| with torch.inference_mode(): |
| for k, t in enumerate(texts): |
| t0 = timeit.default_timer() |
| ids, mask = encode_batch(tok, [t], max_len) |
| ids, mask = ids.to(device), mask.to(device) |
| with torch.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=use_amp): |
| logits = model(ids, mask) |
| _ = logits.float().softmax(-1)[0, 1:].sum().item() |
| if k >= warmup: |
| times.append((timeit.default_timer() - t0) * 1000.0) |
| return times |
|
|
|
|
| def _summ(times: list[float], n_tokens: Sequence[int]) -> dict[str, Any]: |
| arr = np.asarray(times) |
| return {"n": int(len(arr)), "avg_ms": round(float(arr.mean()), 2), "p50_ms": round(float(np.median(arr)), 2), |
| "p95_ms": round(float(np.percentile(arr, 95)), 2), "max_ms": round(float(arr.max()), 2), |
| "avg_s": round(float(arr.mean()) / 1000.0, 4), "mean_tokens": round(float(np.mean(n_tokens)), 1), |
| "ratio_to_benchmark": round(float(arr.mean()) / INSTRUCTOR_BENCHMARK_MS, 3)} |
|
|
|
|
| def measure_latency(model, tok, split: Split, enc: Encoded, device: torch.device, max_len: int, |
| n_samples: int, cpu_threads: int, seed: int) -> dict[str, Any]: |
| """GPU bf16 then CPU fp32 on the same seeded sample of real emails. The CPU pass is |
| cut to 10 samples when 3 warmups predict more than 60 s for the full count (the |
| smoke test measured ~0.7 s per 512-token email on 16 threads). Moves the model to |
| the CPU, so it is the last thing an invocation does with the model.""" |
| rng = np.random.default_rng(seed) |
| n = min(n_samples, split.n) |
| idx = rng.choice(split.n, size=n, replace=False) |
| texts = [split.text[i] for i in idx] |
| toks = [int(enc.n_tokens[i]) for i in idx] |
| out: dict[str, Any] = {"instructor_benchmark_ms": INSTRUCTOR_BENCHMARK_MS, "samples_requested": n_samples, |
| "sample_split": split.name, "max_len": max_len} |
| warm = texts[:3] + texts |
| if device.type == "cuda": |
| torch.cuda.synchronize() |
| gpu_times = _time_single(model, tok, warm, device, max_len, use_amp=True) |
| out["gpu_bf16"] = _summ(gpu_times, toks) |
| out["gpu_bf16"]["device"] = torch.cuda.get_device_name(device) |
| LOG.info("GPU bf16 latency: %s", out["gpu_bf16"]) |
| else: |
| out["gpu_bf16"] = None |
| model.cpu().float() |
| torch.set_num_threads(cpu_threads) |
| cpu = torch.device("cpu") |
| probe = _time_single(model, tok, texts[:3] + texts[:3], cpu, max_len, use_amp=False, warmup=3) |
| est_s = float(np.mean(probe)) / 1000.0 * n |
| n_cpu = n if est_s <= CPU_LATENCY_BUDGET_S or n <= CPU_LATENCY_FALLBACK_N else CPU_LATENCY_FALLBACK_N |
| if n_cpu != n: |
| LOG.info("CPU latency: %d samples would take ~%.0f s (> %.0f s budget); using %d", n, est_s, CPU_LATENCY_BUDGET_S, n_cpu) |
| cpu_times = _time_single(model, tok, texts[:3] + texts[:n_cpu], cpu, max_len, use_amp=False) |
| out["cpu_fp32"] = _summ(cpu_times, toks[:n_cpu]) |
| out["cpu_fp32"]["threads"] = torch.get_num_threads() |
| out["cpu_fp32"]["estimated_full_run_s"] = round(est_s, 1) |
| LOG.info("CPU fp32 latency: %s", out["cpu_fp32"]) |
| return out |
|
|
|
|
| |
| def _style(ax, title: str, xlabel: str, ylabel: str) -> None: |
| ax.set_title(title, color=C_INK, fontsize=11, loc="left") |
| ax.set_xlabel(xlabel, color=C_INK2) |
| ax.set_ylabel(ylabel, color=C_INK2) |
| ax.grid(True, color=C_GRID, linewidth=0.6) |
| for s in ("top", "right"): |
| ax.spines[s].set_visible(False) |
| for s in ("left", "bottom"): |
| ax.spines[s].set_color(C_GRID) |
| ax.tick_params(colors=C_INK2) |
| ax.set_facecolor(C_SURFACE) |
|
|
|
|
| def _save(fig, path: Path) -> None: |
| fig.patch.set_facecolor(C_SURFACE) |
| fig.tight_layout() |
| fig.savefig(path, dpi=PLOT_DPI) |
| plt.close(fig) |
| LOG.info("wrote %s", path) |
|
|
|
|
| def plot_confusion(cm: dict[str, int], path: Path, title: str) -> None: |
| mat = np.array([[cm["tn"], cm["fp"]], [cm["fn"], cm["tp"]]]) |
| fig, ax = plt.subplots(figsize=(5.6, 4.8)) |
| ConfusionMatrixDisplay(mat, display_labels=[BINARY_NAMES[0], BINARY_NAMES[1]]).plot( |
| ax=ax, cmap="Blues", values_format="d", colorbar=False) |
| ax.set_title(title, color=C_INK, fontsize=11, loc="left") |
| ax.set_xlabel("Predicted label", color=C_INK2) |
| ax.set_ylabel("True label", color=C_INK2) |
| ax.tick_params(colors=C_INK2) |
| plt.setp(ax.get_xticklabels(), rotation=15, ha="right", fontsize=8) |
| plt.setp(ax.get_yticklabels(), fontsize=8) |
| _save(fig, path) |
|
|
|
|
| def plot_confusion3(mat: Sequence[Sequence[int]], path: Path, title: str) -> None: |
| fig, ax = plt.subplots(figsize=(5.6, 4.8)) |
| ConfusionMatrixDisplay(np.asarray(mat), display_labels=[LABEL_NAMES[c] for c in range(3)]).plot( |
| ax=ax, cmap="Blues", values_format="d", colorbar=False) |
| ax.set_title(title, color=C_INK, fontsize=11, loc="left") |
| ax.set_xlabel("Predicted label (roll-up first, then argmax phishing/spam)", color=C_INK2, fontsize=9) |
| ax.set_ylabel("True label (corpus provenance)", color=C_INK2) |
| ax.tick_params(colors=C_INK2) |
| _save(fig, path) |
|
|
|
|
| def plot_roc(tf: dict[str, Any], path: Path, title: str) -> None: |
| c = tf["curves"] |
| fig, axes = plt.subplots(1, 2, figsize=(10, 4.4)) |
| for ax, logx in zip(axes, (False, True)): |
| ax.plot(c["fpr"], c["tpr"], color=C_BLUE, linewidth=2, label=f"ROC (AUC {tf['roc_auc']:.4f})") |
| if not logx: |
| ax.plot([0, 1], [0, 1], color=C_GRID, linewidth=1, linestyle="--") |
| for key, col, lab, dy in (("recall_at_fpr_1pct", C_ORANGE, "1% FPR", 16), ("recall_at_fpr_0_1pct", C_AQUA, "0.1% FPR", 4)): |
| r = tf[key] |
| x = max(r["fpr"], 1e-4 if logx else 0) |
| ax.scatter([x], [r["recall"]], color=col, s=48, zorder=3) |
| ax.annotate(f"recall {r['recall']:.3f} @ {lab}", (x, r["recall"]), textcoords="offset points", |
| xytext=(8, dy), color=C_INK2, fontsize=8, annotation_clip=False) |
| if logx: |
| ax.set_xscale("log") |
| ax.set_xlim(1e-4, 1) |
| _style(ax, "ROC, log false-positive rate", "False positive rate (log)", "Recall (true positive rate)") |
| else: |
| ax.set_xlim(0, 1) |
| _style(ax, title, "False positive rate", "Recall (true positive rate)") |
| ax.set_ylim(0, 1.02) |
| ax.legend(loc="lower right", frameon=False, labelcolor=C_INK2) |
| _save(fig, path) |
|
|
|
|
| def plot_pr(tf: dict[str, Any], base_rate: float, path: Path, title: str) -> None: |
| c = tf["curves"] |
| fig, ax = plt.subplots(figsize=(5.6, 4.4)) |
| ax.plot(c["recall"], c["precision"], color=C_BLUE, linewidth=2, label=f"PR (AP {tf['pr_auc']:.4f})") |
| ax.axhline(base_rate, color=C_GRID, linewidth=1, linestyle="--") |
| ax.text(0.01, base_rate + 0.01, f"positive rate {base_rate:.3f}", color=C_INK2, fontsize=8) |
| b = tf["best_f1"] |
| ax.scatter([b["recall"]], [b["precision"]], color=C_ORANGE, s=48, zorder=3) |
| ax.annotate(f"best F1 {b['f1']:.3f} @ thr {b['threshold']:.3f}", (b["recall"], b["precision"]), |
| textcoords="offset points", xytext=(-8, -14), ha="right", color=C_INK2, fontsize=8) |
| ax.set_xlim(0, 1) |
| ax.set_ylim(0, 1.02) |
| _style(ax, title, "Recall (unwanted)", "Precision (unwanted)") |
| ax.legend(loc="lower left", frameon=False, labelcolor=C_INK2) |
| _save(fig, path) |
|
|
|
|
| def _read_csv(path: Path): |
| import pandas as pd |
|
|
| try: |
| df = pd.read_csv(path) |
| except Exception as e: |
| LOG.warning("could not read %s: %s", path, e) |
| return None |
| return df if len(df) else None |
|
|
|
|
| def plot_pretrain_curves(run_dir: Path, path: Path) -> bool: |
| """Loss (train + val) and learning rate from the pretraining run's metrics.csv; two |
| panels, one axis each (never a twin axis).""" |
| df = _read_csv(run_dir / "metrics.csv") |
| if df is None or "step" not in df or "loss" not in df: |
| return False |
| fig, axes = plt.subplots(1, 2, figsize=(10, 4.2)) |
| ax = axes[0] |
| ax.plot(df["step"], df["loss"], color=C_BLUE, linewidth=1.6, label="train MLM loss") |
| if "val_mlm_loss" in df and df["val_mlm_loss"].notna().any(): |
| v = df[df["val_mlm_loss"].notna()] |
| ax.plot(v["step"], v["val_mlm_loss"], color=C_ORANGE, linewidth=2, marker="o", markersize=4, label="val MLM loss") |
| ax.axhline(math.log(32768), color=C_GRID, linewidth=1, linestyle="--") |
| ax.text(float(df["step"].min()), math.log(32768) + 0.05, "ln(V) = uniform guess", color=C_INK2, fontsize=8) |
| _style(ax, f"MLM pretraining loss ({run_dir.name})", "optimizer step", "cross-entropy") |
| ax.legend(frameon=False, labelcolor=C_INK2) |
| ax = axes[1] |
| if "lr" in df: |
| ax.plot(df["step"], df["lr"], color=C_AQUA, linewidth=2) |
| _style(ax, "Learning rate schedule", "optimizer step", "lr") |
| ax.ticklabel_format(axis="y", style="sci", scilimits=(0, 0)) |
| else: |
| ax.axis("off") |
| _save(fig, path) |
| return True |
|
|
|
|
| def plot_finetune_curves(run_dir: Path, path: Path) -> bool: |
| """Training loss per logging step (metrics.csv) and val metrics per evaluation |
| (epochs.csv). finetune.py appends an epochs.csv row every ``eval_every`` steps, not |
| only at epoch ends, so both panels use the optimizer ``step`` as the x-axis (plotting |
| against ``epoch`` stacks every mid-epoch evaluation on one vertical line); rows with |
| ``is_epoch_end == 1`` get a hollow marker.""" |
| m = _read_csv(run_dir / "metrics.csv") |
| e = _read_csv(run_dir / "epochs.csv") |
| if m is None and e is None: |
| return False |
| fig, axes = plt.subplots(1, 2, figsize=(10, 4.2)) |
| ax = axes[0] |
| if m is not None and "loss" in m: |
| x = m["step"] if "step" in m else np.arange(len(m)) |
| ax.plot(x, m["loss"], color=C_BLUE, linewidth=1.4, label="train loss") |
| for col, colr in (("val_loss", C_ORANGE),): |
| if col in m and m[col].notna().any(): |
| v = m[m[col].notna()] |
| ax.plot(v["step"] if "step" in v else v.index, v[col], color=colr, linewidth=2, marker="o", markersize=4, label=col) |
| _style(ax, f"Fine-tuning loss ({run_dir.name})", "optimizer step", "cross-entropy") |
| ax.legend(frameon=False, labelcolor=C_INK2) |
| else: |
| ax.axis("off") |
| ax = axes[1] |
| if e is not None: |
| x = e["step"] if "step" in e else (e["epoch"] if "epoch" in e else np.arange(1, len(e) + 1)) |
| series = [c for c in ("f1", "recall", "precision", "accuracy", "roc_auc", "pr_auc", "macro_f1_3way") if c in e] |
| series = series or [c for c in e.columns if c.startswith("val_") and c != "val_loss"][:6] |
| colors = [C_BLUE, C_ORANGE, C_AQUA, "#eda100", "#e87ba4", "#008300", C_VIOLET] |
| for col, colr in zip(series, colors): |
| ax.plot(x, e[col], color=colr, linewidth=2, marker="o", markersize=4, label=col) |
| ax.annotate(f"{float(e[col].iloc[-1]):.3f}", (float(x.iloc[-1]), float(e[col].iloc[-1])), |
| textcoords="offset points", xytext=(6, 0), color=C_INK2, fontsize=8) |
| if "is_epoch_end" in e and (e["is_epoch_end"] == 1).any(): |
| ends = e[e["is_epoch_end"] == 1] |
| for xe in (ends["step"] if "step" in ends else ends.index): |
| ax.axvline(float(xe), color=C_GRID, linewidth=1, linestyle="--") |
| ax.text(float((ends["step"] if "step" in ends else ends.index).iloc[0]), 0.03, "epoch end", color=C_INK2, |
| fontsize=8, rotation=90, va="bottom", ha="right", transform=ax.get_xaxis_transform()) |
| _style(ax, "Validation metrics per evaluation", "optimizer step", "score") |
| if series: |
| ax.legend(frameon=False, labelcolor=C_INK2, fontsize=8) |
| else: |
| ax.axis("off") |
| _save(fig, path) |
| return True |
|
|
|
|
| |
| _WORD = re.compile(r"[a-z][a-z0-9'\-]{2,}") |
| _HTML_TAG = re.compile(r"<\s*/?\s*[a-z!][^>]{0,80}>", re.I) |
| _LIST_BODY = re.compile(r"unsubscribe|mailing list|listserv|majordomo|listinfo|to be removed from this list|digest", re.I) |
| _LIST_SUBJ = re.compile(r"^\s*(re:\s*)?\[[^\]]{1,40}\]") |
|
|
|
|
| def preview(subject: str, body: str, n: int = PREVIEW_CHARS) -> str: |
| """Subject and body on one line, whitespace collapsed, pipes escaped for the markdown table.""" |
| s = re.sub(r"\s+", " ", f"{subject.strip()} │ {body.strip()}").replace("|", "\\|") |
| return (s[:n] + "...") if len(s) > n else s |
|
|
|
|
| def is_mailing_list(subject: str, body: str) -> bool: |
| return bool(_LIST_SUBJ.search(subject) or _LIST_BODY.search(body)) |
|
|
|
|
| def is_html_heavy(body: str) -> bool: |
| return len(_HTML_TAG.findall(body)) >= 10 or bool(re.search(r"<\s*(html|table|td|tr)\b", body, re.I)) |
|
|
|
|
| def over_represented(a_texts: Sequence[str], b_texts: Sequence[str], top: int = 20, alpha: float = 0.5) -> list[dict[str, Any]]: |
| """Tokens with the highest document-frequency ratio in group A vs group B (smoothed). |
| Simple lexical statistics, not a model explanation: it says what the missed emails |
| tend to contain that caught ones do not.""" |
| if not a_texts or not b_texts: |
| return [] |
| df_a: Counter = Counter() |
| df_b: Counter = Counter() |
| for t in a_texts: |
| df_a.update(set(_WORD.findall(t.lower()))) |
| for t in b_texts: |
| df_b.update(set(_WORD.findall(t.lower()))) |
| min_df = max(3, int(0.05 * len(a_texts))) |
| rows = [] |
| for w, c in df_a.items(): |
| if c < min_df: |
| continue |
| ra, rb = (c + alpha) / (len(a_texts) + alpha), (df_b.get(w, 0) + alpha) / (len(b_texts) + alpha) |
| rows.append({"token": w, "share_a": round(c / len(a_texts), 3), "share_b": round(df_b.get(w, 0) / len(b_texts), 3), |
| "ratio": round(ra / rb, 2)}) |
| rows.sort(key=lambda r: (-r["ratio"], -r["share_a"])) |
| return rows[:top] |
|
|
|
|
| def error_analysis(split: Split, enc: Encoded, probs: np.ndarray, p: np.ndarray, thr: float, at: dict[str, Any], tf: dict[str, Any], |
| lure: dict[str, Any], conf: dict[str, Any], tw: ThreeWay, run_name: str, ts: str) -> tuple[str, dict[str, Any]]: |
| y = split.label |
| pred = (p >= thr).astype(np.int64) |
| fn_idx = np.nonzero((y == 1) & (pred == 0))[0] |
| fp_idx = np.nonzero((y == 0) & (pred == 1))[0] |
| tp_idx = np.nonzero((y == 1) & (pred == 1))[0] |
| tn_idx = np.nonzero((y == 0) & (pred == 0))[0] |
| fn_top = fn_idx[np.argsort(p[fn_idx])][:TOP_K_ERRORS] |
| fp_top = fp_idx[np.argsort(-p[fp_idx])][:TOP_K_ERRORS] |
|
|
| def share(idx: np.ndarray, fn) -> float | None: |
| return round(float(np.mean([fn(i) for i in idx])), 3) if len(idx) else None |
|
|
| comp = { |
| grp: { |
| "n": int(len(idx)), |
| "mailing_list_or_digest": share(idx, lambda i: is_mailing_list(split.subject[i], split.body[i])), |
| "html_heavy": share(idx, lambda i: is_html_heavy(split.body[i])), |
| "very_short_lt20_tokens": share(idx, lambda i: enc.n_tokens_full[i] - 2 < 20), |
| "truncated_gt512": share(idx, lambda i: bool(enc.truncated[i])), |
| "lure_phrase": share(idx, lambda i: bool(lure["mask"][i])), |
| } |
| for grp, idx in (("fp", fp_idx), ("tn", tn_idx), ("fn", fn_idx), ("tp", tp_idx)) |
| } |
| fn_vs_tp = over_represented([split.text[i] for i in fn_idx], [split.text[i] for i in tp_idx]) |
| fp_vs_tn = over_represented([split.text[i] for i in fp_idx], [split.text[i] for i in tn_idx]) |
|
|
| cm = at["confusion_matrix"] |
| k = probs.shape[1] |
| L = [f"# Error analysis — {split.name} split ({SPLIT_TITLE[split.name]}), run `{run_name}` ({ts})", "", |
| "All quoted emails are rows of public corpora " |
| f"(source {', '.join(sorted(set(split.source)))}); nothing here is private mail. " |
| f"Threshold {thr}. Binary positive class = {UNWANTED_NAME}, P(unwanted) = " |
| + ("P(phishing) + P(spam)." if k == 3 else "P(class 1) of the 2-class head."), "", |
| "## Counts (binary roll-up)", "", |
| f"- rows {at['n']:,} (positives {at['n_positive']:,}, {100 * at['n_positive'] / max(at['n'], 1):.1f}%); " |
| f"label3 counts {lure_free_counts(split)}", |
| f"- **recall (unwanted) {at['recall']:.4f}** — {cm['fn']:,} missed unwanted emails (false negatives) out of {at['n_positive']:,}", |
| f"- precision {at['precision']:.4f} — {cm['fp']:,} false alarms out of {cm['fp'] + cm['tp']:,} flagged", |
| f"- F1 {at['f1']:.4f}, accuracy {at['accuracy']:.4f}, specificity {at['specificity']:.4f}", |
| f"- ROC-AUC {_fmt(tf['roc_auc'])}, PR-AUC {_fmt(tf['pr_auc'])}; " |
| f"recall at 1% FPR {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])}, " |
| f"at 0.1% FPR {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])}", |
| f"- credential-lure slice: {lure['metrics']['n']:,} rows, {lure['metrics']['n_positive']:,} positive " |
| f"({100 * lure['label_purity']:.1f}% label purity: slice precision cannot exceed what the phrase list selects), " |
| f"recall {_fmt(lure['metrics']['recall'])}, precision {_fmt(lure['metrics']['precision'])}, " |
| f"{lure['metrics']['confusion_matrix']['fn']} missed", ""] |
|
|
| def table(title: str, idx: np.ndarray) -> list[str]: |
| rows = [f"## {title}", "", "| # | doc_id | P(unwanted) | tokens (full) | source | label_set | preview (subject │ body, 160 chars) |", |
| "|---:|---|---:|---:|---|---|---|"] |
| for kk, i in enumerate(idx, 1): |
| rows.append(f"| {kk} | `{split.doc_id[i]}` | {p[i]:.4f} | {enc.n_tokens_full[i]} | {split.source[i]} | {split.label_set[i]} | " |
| f"{preview(split.subject[i], split.body[i])} |") |
| if not len(idx): |
| rows.append("| - | none | | | | | |") |
| return rows + [""] |
|
|
| L += table(f"Top-{TOP_K_ERRORS} false negatives (missed unwanted mail, most confident first)", fn_top) |
| L += table(f"Top-{TOP_K_ERRORS} false positives (legitimate flagged, most confident first)", fp_top) |
|
|
| |
| stats3: dict[str, Any] = {} |
| L += ["## Three-way (legitimate / phishing / spam)", ""] |
| if k != 3: |
| L += [f"Skipped: {tw.metrics.get('reason')}.", ""] |
| else: |
| m3 = tw.metrics |
| bd = m3["predicted_kind_breakdown"] |
| if tw.evaluated: |
| cmx = m3["confusion_matrix"] |
| L += [f"Decision rule: {DECISION_RULE_DOC}", "", |
| f"Rows with a full label: {m3['rows_with_full_label']:,}; unwanted-but-unsplit rows (binary roll-up only): {m3['rows_unwanted_only']:,}.", "", |
| "| true \\ predicted | " + " | ".join(cmx["labels"]) + " |", "|---|" + "---:|" * 3] |
| for i, lab in enumerate(cmx["labels"]): |
| L.append(f"| **{lab}** | " + " | ".join(f"{v:,}" for v in cmx["matrix"][i]) + " |") |
| L += ["", f"- accuracy {m3['accuracy']:.4f}, **macro-F1 {m3['macro_f1']:.4f}** (over 3 classes; over the " |
| f"{len(m3['classes_present'])} present: {m3['macro_f1_present_classes']:.4f}); argmax-of-3 rule: macro-F1 " |
| f"{m3['argmax_rule']['macro_f1']:.4f}, {m3['argmax_rule']['rows_where_rules_differ']:,} rows differ between the rules", "", |
| "| class | precision | recall | F1 | support |", "|---|---:|---:|---:|---:|"] |
| for lab, v in m3["per_class"].items(): |
| L.append(f"| {lab} | {v['precision']:.4f} | {v['recall']:.4f} | {v['f1']:.4f} | {v['support']:,} |") |
| pvs = m3["phish_vs_spam_on_true_unwanted"] |
| if pvs["kind_only"]: |
| ko, rr = pvs["kind_only"], pvs["rollup_rule"] |
| L += ["", f"### Phishing vs spam on the {pvs['n_true_unwanted']:,} true unwanted rows", "", |
| f"Kind decision alone (given the row is unwanted, argmax(P(phishing), P(spam))): accuracy {ko['accuracy']:.4f}; " |
| f"phishing read as spam {ko['phishing_as_spam']:,}, spam read as phishing {ko['spam_as_phishing']:,}.", "", |
| "| true \\ predicted (roll-up rule) | Legitimate (missed) | Phishing | Spam |", "|---|---:|---:|---:|"] |
| for i, lab in enumerate(rr["rows"]): |
| L.append(f"| **{lab}** | " + " | ".join(f"{v:,}" for v in rr["matrix"][i]) + " |") |
| L += ["", "### Per source (three-way rows)", "", "| source | n | Legitimate | Phishing | Spam | accuracy | macro-F1 (present classes) |", |
| "|---|---:|---:|---:|---:|---:|---:|"] |
| for s, v in m3["per_source"].items(): |
| c3 = v["counts"] |
| L.append(f"| {s} | {v['n']:,} | {c3['Legitimate']:,} | {c3['Phishing']:,} | {c3['Spam']:,} | {v['accuracy']:.4f} | {v['macro_f1_present_classes']:.4f} |") |
| |
| y3, pred3 = split.label3, tw.pred3 |
| ps_idx = np.nonzero((y3 == 1) & (pred3 == 2))[0] |
| sp_idx = np.nonzero((y3 == 2) & (pred3 == 1))[0] |
| ps_top = ps_idx[np.argsort(-probs[ps_idx, 2])][:TOP_K_CONFUSIONS] |
| sp_top = sp_idx[np.argsort(-probs[sp_idx, 1])][:TOP_K_CONFUSIONS] |
|
|
| def table3(title: str, idx: np.ndarray) -> list[str]: |
| rows = [f"### {title}", "", "| # | doc_id | P(legit) | P(phish) | P(spam) | tokens (full) | source | preview (subject │ body, 160 chars) |", |
| "|---:|---|---:|---:|---:|---:|---|---|"] |
| for kk, i in enumerate(idx, 1): |
| rows.append(f"| {kk} | `{split.doc_id[i]}` | {probs[i, 0]:.4f} | {probs[i, 1]:.4f} | {probs[i, 2]:.4f} | {enc.n_tokens_full[i]} | " |
| f"{split.source[i]} | {preview(split.subject[i], split.body[i])} |") |
| if not len(idx): |
| rows.append("| - | none | | | | | | |") |
| return rows + [""] |
|
|
| L += [""] |
| L += table3(f"Top-{TOP_K_CONFUSIONS} phishing -> spam confusions ({len(ps_idx):,} total, most confident first)", ps_top) |
| L += table3(f"Top-{TOP_K_CONFUSIONS} spam -> phishing confusions ({len(sp_idx):,} total, most confident first)", sp_top) |
| ps_vs_pp = over_represented([split.text[i] for i in ps_idx], [split.text[i] for i in np.nonzero((y3 == 1) & (pred3 == 1))[0]]) |
| sp_vs_ss = over_represented([split.text[i] for i in sp_idx], [split.text[i] for i in np.nonzero((y3 == 2) & (pred3 == 2))[0]]) |
| L += ["### Tokens over-represented in phishing read as spam vs phishing read correctly", ""] + _ratio_table(ps_vs_pp, "phish->spam", "phish->phish") |
| L += ["", "### Tokens over-represented in spam read as phishing vs spam read correctly", ""] + _ratio_table(sp_vs_ss, "spam->phish", "spam->spam") + [""] |
| stats3 = {"n_phish_as_spam": int(len(ps_idx)), "n_spam_as_phish": int(len(sp_idx)), |
| "top_phish_as_spam_doc_ids": [split.doc_id[i] for i in ps_top], "top_spam_as_phish_doc_ids": [split.doc_id[i] for i in sp_top], |
| "phish_as_spam_tokens": ps_vs_pp, "spam_as_phish_tokens": sp_vs_ss} |
| else: |
| L += [f"No three-way ground truth on this split: {m3['reason']}.", ""] |
| tpf, fpf, mis = bd["true_positives_flagged"], bd["false_positives_flagged"], bd["missed_positives"] |
| L += ["### Predicted kind of the flagged rows (informational — " |
| + ("no phishing-vs-spam ground truth on this split)" if not tw.evaluated else "includes the unsplit rows, which have no kind label)"), "", |
| "| group | n | predicted Phishing | predicted Spam |", "|---|---:|---:|---:|", |
| f"| true positives flagged | {tpf['n']:,} | {tpf['predicted_phishing']:,} | {tpf['predicted_spam']:,} |", |
| f"| false positives flagged | {fpf['n']:,} | {fpf['predicted_phishing']:,} | {fpf['predicted_spam']:,} |", |
| f"| missed positives (kind the model would have given) | {mis['n']:,} | {mis['predicted_phishing']:,} | {mis['predicted_spam']:,} |", ""] |
|
|
| L += ["## Patterns (simple lexical statistics, not a model explanation)", "", |
| "### Tokens over-represented in false negatives vs true positives (document frequency, smoothed ratio)", ""] |
| L += _ratio_table(fn_vs_tp, "FN", "TP") |
| L += ["", "### Tokens over-represented in false positives vs true negatives", ""] |
| L += _ratio_table(fp_vs_tn, "FP", "TN") |
| L += ["", "### Composition of the error groups (share of rows)", "", |
| "| group | n | mailing-list / digest | HTML-heavy | very short (<20 content tokens) | truncated (>512) | lure phrase |", |
| "|---|---:|---:|---:|---:|---:|---:|"] |
| for grp in ("fp", "tn", "fn", "tp"): |
| c = comp[grp] |
| L.append(f"| {grp.upper()} | {c['n']:,} | {_pct(c['mailing_list_or_digest'])} | {_pct(c['html_heavy'])} | " |
| f"{_pct(c['very_short_lt20_tokens'])} | {_pct(c['truncated_gt512'])} | {_pct(c['lure_phrase'])} |") |
| L += ["", "### Length confound (RUN_LOG entry 2)", "", |
| "| group | n | mean tokens (full) | mean tokens fed | median (full) | % truncated |", "|---|---:|---:|---:|---:|---:|"] |
| for grp in ("tp", "tn", "fp", "fn", "correct", "wrong"): |
| s = conf["by_outcome"][grp] |
| L.append(f"| {grp.upper()} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | " |
| f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |") |
| for lab in ("legitimate", "unwanted"): |
| s = conf["by_label"][lab] |
| L.append(f"| all {lab} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | " |
| f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |") |
| for lab, s in conf["by_class3"].items(): |
| if lab in ("phish", "spam", "unwanted"): |
| L.append(f"| all label3={lab} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | " |
| f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |") |
| L += ["", "| length bucket (full tokens) | n | positive rate | recall | FPR | accuracy |", "|---|---:|---:|---:|---:|---:|"] |
| for b in conf["by_length_bucket"]: |
| L.append(f"| {b['bucket']} | {b['n']:,} | {b['positive_rate']:.3f} | {_fmt(b['recall'])} | {_fmt(b['fpr'])} | {b['accuracy']:.4f} |") |
|
|
| L += ["", "## Summary", ""] + summary_sentences(at, tf, lure, conf, comp, fn_vs_tp, fp_vs_tn, tw) |
| stats = {"n_fn": int(len(fn_idx)), "n_fp": int(len(fp_idx)), "composition": comp, |
| "fn_vs_tp_tokens": fn_vs_tp, "fp_vs_tn_tokens": fp_vs_tn, |
| "top_fn_doc_ids": [split.doc_id[i] for i in fn_top], "top_fp_doc_ids": [split.doc_id[i] for i in fp_top], |
| "threeway": stats3 or None} |
| return "\n".join(L) + "\n", stats |
|
|
|
|
| def lure_free_counts(split: Split) -> str: |
| c = _label3_counts(split.label3) |
| return ", ".join(f"{k} {v:,}" for k, v in c.items() if v) |
|
|
|
|
| def summary_sentences(at, tf, lure, conf, comp, fn_vs_tp, fp_vs_tn, tw: ThreeWay | None = None) -> list[str]: |
| cm = at["confusion_matrix"] |
| out = [f"- The classifier catches **{100 * at['recall']:.2f}% of unwanted mail (phishing or spam)** at threshold {at['threshold']} " |
| f"({cm['fn']:,} of {at['n_positive']:,} missed) while flagging {100 * at['false_positive_rate']:.2f}% of legitimate mail " |
| f"({cm['fp']:,} false alarms). Recall is the headline because a missed phish is the costly error for a small-business inbox."] |
| if tf["recall_at_fpr_1pct"]: |
| r1 = tf["recall_at_fpr_1pct"] |
| thr_txt = f"threshold {r1['threshold']:.3f}, flag when p >= threshold" if r1["threshold"] is not None else "no finite threshold reaches this budget" |
| out.append(f"- Holding false alarms to 1 in 100 legitimate emails keeps recall at {100 * r1['recall']:.2f}% " |
| f"({thr_txt}); at 1 in 1,000 it is " |
| f"{100 * tf['recall_at_fpr_0_1pct']['recall']:.2f}%. The best-F1 threshold is {tf['best_f1']['threshold']:.3f} " |
| f"(F1 {tf['best_f1']['f1']:.4f}).") |
| if tw is not None and tw.evaluated: |
| m3 = tw.metrics |
| pc = m3["per_class"] |
| pvs = m3["phish_vs_spam_on_true_unwanted"] |
| s = (f"- Three-way: macro-F1 **{m3['macro_f1']:.4f}** over {m3['rows_with_full_label']:,} rows with a full label " |
| f"(recall legitimate {100 * pc['Legitimate']['recall']:.1f}%, phishing {100 * pc['Phishing']['recall']:.1f}%, spam {100 * pc['Spam']['recall']:.1f}%).") |
| if pvs["kind_only"]: |
| ko = pvs["kind_only"] |
| s += (f" Given that an email is unwanted, the model tells phishing from spam {100 * ko['accuracy']:.1f}% of the time " |
| f"({ko['phishing_as_spam']:,} phishing read as spam, {ko['spam_as_phishing']:,} spam read as phishing). " |
| "The phishing/spam labels follow corpus provenance (Nazario + Nigerian fraud + synthetic lures = phishing; " |
| "SpamAssassin + Enron-Spam spam folders = spam), so part of this confusion is corpus style, not lure content.") |
| out.append(s) |
| elif tw is not None and tw.metrics.get("num_labels") == 3: |
| bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"] |
| out.append(f"- Three-way: this split has no phishing-vs-spam ground truth; of the {bd['n']:,} true positives the model flagged, " |
| f"it called {bd['predicted_phishing']:,} phishing and {bd['predicted_spam']:,} spam (informational only).") |
| lm = lure["metrics"] |
| if lm["n_positive"] >= LURE_MIN_POSITIVES: |
| out.append(f"- On the {lm['n']:,}-row credential-lure slice ({len(CREDENTIAL_LURE_PHRASES)} documented phrases, " |
| f"{100 * lure['label_purity']:.1f}% of its rows are labelled positive) recall is " |
| f"{100 * lm['recall']:.2f}% with precision {100 * lm['precision']:.2f}%: " |
| f"{'the highest-cost phish are caught at least as well as the average' if lm['recall'] >= at['recall'] else 'the highest-cost phish are caught LESS often than the average positive, which is the first thing to fix'}.") |
| elif lm["n"]: |
| out.append(f"- The credential-lure slice has only {lm['n']:,} rows ({lm['n_positive']:,} positive, {lm['confusion_matrix']['fn']} missed): " |
| f"too few rows to compare its recall with the overall recall (needs >= {LURE_MIN_POSITIVES} positives).") |
| else: |
| out.append("- No row of this sample contains a credential-lure phrase; the slice is empty.") |
| bo, bl = conf["by_outcome"], conf["by_label"] |
| mean = {g: bo[g]["mean_tokens_full"] for g in ("tp", "tn", "fp", "fn")} |
| out.append(f"- Length confound: missed unwanted mail averages {_fmt(mean['fn'], 0)} tokens vs {_fmt(mean['tp'], 0)} for caught; " |
| f"false positives average {_fmt(mean['fp'], 0)} tokens vs {_fmt(mean['tn'], 0)} for correct legitimate mail " |
| f"(all legitimate {_fmt(bl['legitimate']['mean_tokens_full'], 0)}, all unwanted {_fmt(bl['unwanted']['mean_tokens_full'], 0)}).") |
| if all(mean[g] is not None for g in mean): |
| follows_length = mean["fn"] > mean["tp"] and mean["fp"] < mean["tn"] |
| out.append(f" Legitimate mail is ~{bl['legitimate']['mean_tokens_full'] / max(bl['unwanted']['mean_tokens_full'] or 1, 1):.1f}x longer than unwanted mail on average; " |
| + ("the errors sit on long unwanted mail and short legitimate mail, so the model partly reads length rather than content." |
| if follows_length else "the error groups do not follow the length gap, so length is not the main shortcut.")) |
| c3 = conf.get("by_class3") or {} |
| if "phish" in c3 and "spam" in c3: |
| out.append(f" Phishing rows average {_fmt(c3['phish']['mean_tokens_full'], 0)} tokens vs spam {_fmt(c3['spam']['mean_tokens_full'], 0)}: " |
| "a length gap between the two unwanted kinds is a second shortcut to watch in the phishing-vs-spam confusion.") |
| buckets = [b for b in conf["by_length_bucket"] if b["recall"] is not None and b["n_positive"] >= 20] |
| if buckets: |
| worst = min(buckets, key=lambda b: b["recall"]) |
| out.append(f"- Recall by length is lowest in the {worst['bucket']}-token bucket ({100 * worst['recall']:.1f}%, {worst['n_positive']:,} positives).") |
| fp_c = comp["fp"] |
| if fp_c["n"]: |
| out.append(f"- Of the {fp_c['n']:,} false positives, {_pct(fp_c['mailing_list_or_digest'])} look like mailing-list or digest traffic, " |
| f"{_pct(fp_c['html_heavy'])} are HTML-heavy and {_pct(fp_c['very_short_lt20_tokens'])} are very short " |
| f"(TN reference: {_pct(comp['tn']['mailing_list_or_digest'])} / {_pct(comp['tn']['html_heavy'])} / {_pct(comp['tn']['very_short_lt20_tokens'])}).") |
| if fn_vs_tp: |
| out.append("- Tokens most over-represented in missed unwanted mail vs caught: " + ", ".join(f"`{r['token']}`" for r in fn_vs_tp[:10]) + ".") |
| if fp_vs_tn: |
| out.append("- Tokens most over-represented in false positives vs correct legitimate mail: " + ", ".join(f"`{r['token']}`" for r in fp_vs_tn[:10]) + ".") |
| return out |
|
|
|
|
| def _ratio_table(rows: list[dict[str, Any]], a: str, b: str) -> list[str]: |
| if not rows: |
| return ["(too few rows in one of the groups)"] |
| out = [f"| token | share of {a} docs | share of {b} docs | ratio |", "|---|---:|---:|---:|"] |
| out += [f"| `{r['token']}` | {r['share_a']:.3f} | {r['share_b']:.3f} | {r['ratio']:.2f} |" for r in rows] |
| return out |
|
|
|
|
| def _fmt(v: Any, nd: int = 4) -> str: |
| if v is None or v is False: |
| return "n/a" |
| return f"{v:.{nd}f}" |
|
|
|
|
| def _pct(v: float | None) -> str: |
| return "n/a" if v is None else f"{100 * v:.1f}%" |
|
|
|
|
| |
| def next_run_log_number(path: Path) -> int: |
| if not path.exists(): |
| return 0 |
| nums = [int(m.group(1)) for m in re.finditer(r"^## (\d+)\.", path.read_text(encoding="utf-8"), re.M)] |
| return (max(nums) + 1) if nums else 0 |
|
|
|
|
| @dataclass |
| class SplitResult: |
| """Everything one split produced, kept until the invocation-level artifacts are written.""" |
|
|
| split: Split |
| enc: Encoded |
| probs: np.ndarray |
| p_unw: np.ndarray |
| inf: dict[str, Any] |
| at: dict[str, Any] |
| tf: dict[str, Any] |
| lure: dict[str, Any] |
| per_source: dict[str, Any] |
| conf: dict[str, Any] |
| tw: ThreeWay |
| plots: dict[str, str] |
| err_path: Path |
| err_stats: dict[str, Any] |
|
|
|
|
| def _split_table(r: SplitResult) -> list[str]: |
| at, tf, lure, tw = r.at, r.tf, r.lure, r.tw |
| cm = at["confusion_matrix"] |
| lines = [f"### {r.split.name} split ({SPLIT_TITLE[r.split.name]}): `{r.split.path.name}`, {r.split.n:,} rows " |
| f"({at['n_positive']:,} unwanted, {100 * at['n_positive'] / max(r.split.n, 1):.1f}%; label3 {lure_free_counts(r.split)}). " |
| f"Inference {r.inf['rows_per_s']} rows/s on {r.inf['device']} (bf16 autocast, batch {r.inf['batch_size']}, length-bucketed), " |
| f"peak VRAM {r.inf['peak_vram_gb']} GB.", "", |
| "| binary roll-up metric (P(unwanted) = P(phishing) + P(spam)) | value |", "|---|---:|", |
| f"| **recall (unwanted)** | **{at['recall']:.4f}** ({cm['fn']:,} missed of {at['n_positive']:,}) |", |
| f"| precision | {at['precision']:.4f} ({cm['fp']:,} false alarms) |", |
| f"| F1 | {at['f1']:.4f} |", f"| accuracy | {at['accuracy']:.4f} |", f"| specificity | {at['specificity']:.4f} |", |
| f"| ROC-AUC | {_fmt(tf['roc_auc'])} |", f"| PR-AUC | {_fmt(tf['pr_auc'])} |", |
| f"| recall @ 1% FPR | {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])} |", |
| f"| recall @ 0.1% FPR | {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])} |", |
| f"| best-F1 threshold | {_fmt(tf['best_f1'] and tf['best_f1']['threshold'], 3)} (F1 {_fmt(tf['best_f1'] and tf['best_f1']['f1'])}) |", |
| f"| credential-lure slice | {lure['metrics']['n']:,} rows ({100 * lure['label_purity']:.1f}% labelled positive), " |
| f"recall {_fmt(lure['metrics']['recall'])}, precision {_fmt(lure['metrics']['precision'])} |", ""] |
| if tw.evaluated: |
| m3 = tw.metrics |
| cmx = m3["confusion_matrix"] |
| pc = m3["per_class"] |
| lines += [f"**Three-way** on the {m3['rows_with_full_label']:,} rows with label3 in {{0,1,2}} ({m3['rows_unwanted_only']:,} unsplit rows " |
| f"enter the roll-up only; rule: {DECISION_RULE}): accuracy {m3['accuracy']:.4f}, **macro-F1 {m3['macro_f1']:.4f}**; " |
| + "; ".join(f"{lab} P {v['precision']:.3f} / R {v['recall']:.3f} / F1 {v['f1']:.3f} (n={v['support']:,})" for lab, v in pc.items()) |
| + ". 3x3 (rows true, cols predicted Legitimate/Phishing/Spam): " |
| + "; ".join(f"{lab} {'/'.join(f'{v:,}' for v in cmx['matrix'][i])}" for i, lab in enumerate(cmx["labels"])) + "."] |
| pvs = m3["phish_vs_spam_on_true_unwanted"] |
| if pvs["kind_only"]: |
| ko = pvs["kind_only"] |
| lines[-1] += (f" Phishing-vs-spam on the {pvs['n_true_unwanted']:,} true unwanted rows: kind accuracy {ko['accuracy']:.4f} " |
| f"({ko['phishing_as_spam']:,} phishing->spam, {ko['spam_as_phishing']:,} spam->phishing).") |
| lines += ["", "| source | n | Legitimate | Phishing | Spam | accuracy | macro-F1 (present classes) |", "|---|---:|---:|---:|---:|---:|---:|"] |
| for s, v in m3["per_source"].items(): |
| c3 = v["counts"] |
| lines.append(f"| {s} | {v['n']:,} | {c3['Legitimate']:,} | {c3['Phishing']:,} | {c3['Spam']:,} | {v['accuracy']:.4f} | {v['macro_f1_present_classes']:.4f} |") |
| lines.append("") |
| elif tw.metrics.get("num_labels") == 3: |
| bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"] |
| lines += [f"**Three-way:** no phishing-vs-spam ground truth on this split ({tw.metrics['reason']}); informational: of the " |
| f"{bd['n']:,} flagged true positives the model called {bd['predicted_phishing']:,} phishing and {bd['predicted_spam']:,} spam.", ""] |
| else: |
| lines += [f"**Three-way:** skipped ({tw.metrics.get('reason')}).", ""] |
| return lines |
|
|
|
|
| def run_log_entry(num: int, ts: str, run_name: str, results: list[SplitResult], ckpt: Path, lat: dict, out_dir: Path, dev: bool, k: int) -> str: |
| g, c = lat.get("gpu_bf16"), lat.get("cpu_fp32") |
| names = " + ".join(r.split.name for r in results) |
| lines = [f"## {num}. {ts} — Evaluation on the {names} split{'s' if len(results) > 1 else ''} (evaluate.py run, `{run_name}`)" |
| f"{' [DEV RUN, --limit-rows]' if dev else ''}", "", |
| f"**Checkpoint.** `{ckpt}` ({k}-class head{'; three-way label contract of RUN_LOG entry 6' if k == 3 else ''}). " |
| "Binary roll-up metrics use the same definitions as the binary-era runs so they stay comparable.", ""] |
| for r in results: |
| lines += _split_table(r) |
| lines += ["**Latency (batch 1, tokenize + forward + softmax, timeit.default_timer) vs the instructor's 74 ms.** " |
| + (f"GPU bf16: {g['avg_ms']} ms avg / {g['p50_ms']} ms p50 over {g['n']} emails ({g['ratio_to_benchmark']:.2f}x). " if g else "GPU: not available. ") |
| + (f"CPU fp32 ({c['threads']} threads): {c['avg_ms']} ms avg / {c['p50_ms']} ms p50 over {c['n']} emails ({c['ratio_to_benchmark']:.2f}x)." if c else ""), |
| ""] |
| r0 = results[0] |
| conf = r0.conf |
| lines += [f"**Length confound ({r0.split.name}).** FN mean {_fmt(conf['by_outcome']['fn']['mean_tokens_full'], 0)} tokens vs TP {_fmt(conf['by_outcome']['tp']['mean_tokens_full'], 0)}; " |
| f"FP {_fmt(conf['by_outcome']['fp']['mean_tokens_full'], 0)} vs TN {_fmt(conf['by_outcome']['tn']['mean_tokens_full'], 0)} " |
| f"(see error_analysis{r0.split.sfx}.md for the per-bucket table).", "", |
| f"**Artifacts.** `{out_dir}`: " + "; ".join( |
| f"metrics{r.split.sfx}.json, latency{r.split.sfx}.json, confusion_matrix{r.split.sfx}.png" |
| + (f", confusion_matrix3{r.split.sfx}.png" if r.tw.evaluated else "") |
| + f", roc_curve{r.split.sfx}.png, pr_curve{r.split.sfx}.png, error_analysis{r.split.sfx}.md" for r in results) |
| + " (training curves: reports/pretrain_curves.png and finetune_curves.png from pretrain.py / finetune.py `plot`)."] |
| return "\n".join(lines) |
|
|
|
|
| |
| def _summary(r: SplitResult) -> dict[str, Any]: |
| """The experiments.jsonl summary for one split: the binary keys of the binary-era line |
| plus a ``threeway`` sub-dict (None when not evaluated).""" |
| at, tf, lure, tw = r.at, r.tf, r.lure, r.tw |
| s = {"accuracy": at["accuracy"], "precision": at["precision"], "recall": at["recall"], "f1": at["f1"], |
| "specificity": at["specificity"], "roc_auc": tf["roc_auc"], "pr_auc": tf["pr_auc"], |
| "recall_at_fpr_1pct": tf["recall_at_fpr_1pct"] and tf["recall_at_fpr_1pct"]["recall"], |
| "recall_at_fpr_0_1pct": tf["recall_at_fpr_0_1pct"] and tf["recall_at_fpr_0_1pct"]["recall"], |
| "confusion_matrix": at["confusion_matrix"], "n": r.split.n, "threshold": at["threshold"], |
| "lure_recall": lure["metrics"]["recall"] if lure["metrics"]["n_positive"] else None, |
| "f1_unwanted": at["f1"], "threeway": None} |
| if tw.evaluated: |
| m3 = tw.metrics |
| pvs = m3["phish_vs_spam_on_true_unwanted"] |
| s["threeway"] = {"n": m3["rows_with_full_label"], "accuracy": m3["accuracy"], "macro_f1": m3["macro_f1"], |
| "macro_f1_present_classes": m3["macro_f1_present_classes"], |
| "per_class": {lab: {"precision": v["precision"], "recall": v["recall"], "f1": v["f1"], "support": v["support"]} |
| for lab, v in m3["per_class"].items()}, |
| "confusion_matrix_3x3": m3["confusion_matrix"]["matrix"], |
| "kind_accuracy_on_true_unwanted": pvs["kind_only"]["accuracy"] if pvs["kind_only"] else None} |
| s["macro_f1_3way"] = m3["macro_f1"] |
| return s |
|
|
|
|
| def evaluate_split(name: str, model, mcfg: EncoderConfig, tok, cfg: dict, args: argparse.Namespace, device: torch.device, |
| max_len: int, out_dir: Path, run_name: str, ts: str, data_dir: Path) -> SplitResult: |
| dev = args.limit_rows is not None |
| test3_file = str((cfg.get("eval") or {}).get("test3_file") or SPLIT_FILES["test3"]) |
| split = load_split(data_dir, name, args.limit_rows, int(cfg["seed"]), test3_file) |
| LOG.info("%s split: %d rows, %d unwanted (%.1f%%), label3 %s, sources %s%s", split.name, split.n, int(split.label.sum()), |
| 100 * split.label.mean(), _label3_counts(split.label3), sorted(set(split.source)), f" [limit_rows={args.limit_rows}]" if dev else "") |
| t0 = common.timer() |
| enc = encode_split(tok, split.text, max_len) |
| LOG.info("encoded %d rows in %.1fs: mean %.0f tokens (full), %.1f%% truncated at %d", |
| split.n, common.timer() - t0, enc.n_tokens_full.mean(), 100 * enc.truncated.mean(), max_len) |
|
|
| probs, inf = predict_probs(model, enc, device, args.batch_size, int(cfg["finetune"].get("dynamic_padding_multiple", 8))) |
| p_unw = p_unwanted(probs) |
| thr = args.threshold |
| at = threshold_metrics(split.label, p_unw, thr) |
| tf = threshold_free_metrics(split.label, p_unw) |
| lmask, per_phrase, phrase_by_label = lure_mask(split.text, split.label) |
| lure = {"mask": lmask, "phrases": list(CREDENTIAL_LURE_PHRASES), "phrase_hits": dict(per_phrase.most_common()), |
| "phrase_hits_by_label": phrase_by_label, |
| "metrics": slice_metrics(split.label[lmask], p_unw[lmask], thr), |
| "label3_counts": _label3_counts(split.label3[lmask])} |
| |
| |
| lure["label_purity"] = _safe_div(lure["metrics"]["n_positive"], lure["metrics"]["n"]) |
| per_source = per_source_table(split, p_unw, thr) |
| conf = length_confound(split, enc, p_unw, thr) |
| tw = threeway_metrics(split, probs, p_unw, thr) |
| if tw.evaluated and lmask.any(): |
| full = lmask & (split.label3 >= 0) |
| lure["threeway_accuracy"] = float((tw.pred3[full] == split.label3[full]).mean()) if full.any() else None |
| if not tw.evaluated: |
| LOG.info("three-way sections skipped on %s: %s", split.name, tw.metrics.get("reason")) |
|
|
| cm = at["confusion_matrix"] |
| print(f"\n=== {split.name} split / {SPLIT_TITLE[split.name]} ({split.n:,} rows, {at['n_positive']:,} unwanted) — run {run_name} ===") |
| print(f"RECALL (unwanted = phishing or spam, the metric that matters): {at['recall']:.4f} missed: {cm['fn']:,} / {at['n_positive']:,}") |
| print(f"precision {at['precision']:.4f} F1 {at['f1']:.4f} accuracy {at['accuracy']:.4f} specificity {at['specificity']:.4f}") |
| print(f"confusion matrix [[TN {cm['tn']:,}, FP {cm['fp']:,}], [FN {cm['fn']:,}, TP {cm['tp']:,}]]") |
| print(f"ROC-AUC {_fmt(tf['roc_auc'])} PR-AUC {_fmt(tf['pr_auc'])} " |
| f"recall@1%FPR {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])} " |
| f"recall@0.1%FPR {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])} " |
| f"best-F1 thr {_fmt(tf['best_f1'] and tf['best_f1']['threshold'], 3)}") |
| lm = lure["metrics"] |
| print(f"credential-lure slice: {lm['n']:,} rows ({lm['n_positive']:,} positive = {100 * lure['label_purity']:.1f}% label purity) " |
| f"recall {_fmt(lm['recall'])} precision {_fmt(lm['precision'])}" |
| + (f" [< {LURE_MIN_POSITIVES} positives: not compared to overall recall]" if lm['n_positive'] < LURE_MIN_POSITIVES else "")) |
| print("per source: " + "; ".join(f"{s}: n={m['n']:,} recall={_fmt(m['recall'])} prec={_fmt(m['precision'])}" for s, m in per_source.items())) |
| bo = conf["by_outcome"] |
| print(f"mean tokens (full): TP {_fmt(bo['tp']['mean_tokens_full'], 0)} FN {_fmt(bo['fn']['mean_tokens_full'], 0)} " |
| f"TN {_fmt(bo['tn']['mean_tokens_full'], 0)} FP {_fmt(bo['fp']['mean_tokens_full'], 0)}") |
| if tw.evaluated: |
| m3 = tw.metrics |
| print(f"THREE-WAY ({m3['rows_with_full_label']:,} rows with label3 in {{0,1,2}}): accuracy {m3['accuracy']:.4f} MACRO-F1 {m3['macro_f1']:.4f} " |
| + " ".join(f"{lab[:5]} P/R/F1 {v['precision']:.3f}/{v['recall']:.3f}/{v['f1']:.3f} (n={v['support']:,})" for lab, v in m3["per_class"].items())) |
| print("3x3 [rows true L/P/S, cols pred L/P/S]: " + str(m3["confusion_matrix"]["matrix"])) |
| pvs = m3["phish_vs_spam_on_true_unwanted"] |
| if pvs["kind_only"]: |
| ko = pvs["kind_only"] |
| print(f"phishing vs spam on {pvs['n_true_unwanted']:,} true unwanted rows: kind accuracy {ko['accuracy']:.4f} " |
| f"phish->spam {ko['phishing_as_spam']:,} spam->phish {ko['spam_as_phishing']:,}") |
| print("per source (3-way): " + "; ".join(f"{s}: n={v['n']:,} L/P/S={v['counts']['Legitimate']}/{v['counts']['Phishing']}/{v['counts']['Spam']} " |
| f"acc={v['accuracy']:.3f}" for s, v in m3["per_source"].items())) |
| elif mcfg.num_labels == 3: |
| bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"] |
| print(f"three-way: no ground truth on this split ({tw.metrics['reason']}); flagged true positives split " |
| f"{bd['predicted_phishing']:,} phishing / {bd['predicted_spam']:,} spam (informational)") |
| else: |
| print(f"three-way: skipped ({tw.metrics.get('reason')})") |
|
|
| sfx = split.sfx |
| plots: dict[str, str] = {} |
| p_cm = out_dir / f"confusion_matrix{sfx}.png" |
| plot_confusion(cm, p_cm, f"Confusion matrix (binary roll-up) — {split.name} split, threshold {thr}") |
| plots["confusion_matrix"] = str(p_cm) |
| if tw.evaluated: |
| p_cm3 = out_dir / f"confusion_matrix3{sfx}.png" |
| plot_confusion3(tw.metrics["confusion_matrix"]["matrix"], p_cm3, f"Three-way confusion matrix — {split.name} split, threshold {thr}") |
| plots["confusion_matrix3"] = str(p_cm3) |
| if tf["curves"] is not None: |
| p_roc, p_pr = out_dir / f"roc_curve{sfx}.png", out_dir / f"pr_curve{sfx}.png" |
| plot_roc(tf, p_roc, f"ROC (binary roll-up) — {split.name} split") |
| plot_pr(tf, split.label.mean(), p_pr, f"Precision-recall (binary roll-up) — {split.name} split") |
| plots["roc_curve"], plots["pr_curve"] = str(p_roc), str(p_pr) |
| |
| |
| |
| |
| for flag, plotter, key in ((args.finetune_run_dir, plot_finetune_curves, "finetune"), (args.pretrain_run_dir, plot_pretrain_curves, "pretrain")): |
| if not flag: |
| continue |
| p_curve = out_dir / f"eval_{key}_curves{sfx}.png" |
| if plotter(Path(flag), p_curve): |
| plots[f"eval_{key}_curves"] = str(p_curve) |
| else: |
| LOG.warning("no usable metrics.csv under %s: %s skipped", flag, p_curve.name) |
|
|
| md, err_stats = error_analysis(split, enc, probs, p_unw, thr, at, tf, lure, conf, tw, run_name, ts) |
| p_err = out_dir / f"error_analysis{sfx}.md" |
| p_err.write_text(md, encoding="utf-8") |
| LOG.info("wrote %s", p_err) |
| return SplitResult(split, enc, probs, p_unw, inf, at, tf, lure, per_source, conf, tw, plots, p_err, err_stats) |
|
|
|
|
| def run(args: argparse.Namespace) -> int: |
| cfg = common.load_config(args.config) |
| global LOG, TOP_K_CONFUSIONS |
| LOG = common.setup_logging("evaluate", cfg["paths"]["logs"]) |
| TOP_K_CONFUSIONS = check_eval_config(cfg) |
| common.set_seed(int(cfg["seed"])) |
| common.print_device_info() |
| device = torch.device("cuda" if torch.cuda.is_available() and not args.cpu else "cpu") |
| ts = datetime.now().strftime("%Y-%m-%d %H:%M") |
| dev = args.limit_rows is not None |
| splits = [s.strip() for s in args.split.split(",") if s.strip()] |
| bad = [s for s in splits if s not in SPLIT_FILES] |
| if bad or not splits: |
| raise ValueError(f"--split must be a comma list of {sorted(SPLIT_FILES)}, got {args.split!r}") |
| if len(set(splits)) != len(splits): |
| raise ValueError(f"--split lists a split twice: {args.split!r}") |
|
|
| ckpt = Path(args.checkpoint).resolve() |
| run_name = args.run_name or (ckpt.parent.name if ckpt.name in ("best", "last") else ckpt.name) |
| out_dir = Path(args.out_dir) if args.out_dir else Path(cfg["paths"]["reports"]) |
| out_dir.mkdir(parents=True, exist_ok=True) |
| data_dir = Path(args.data_dir) if args.data_dir else Path(cfg["paths"]["data_processed"]) |
| LOG.info("run_name=%s splits=%s checkpoint=%s out_dir=%s data_dir=%s device=%s", run_name, splits, ckpt, out_dir, data_dir, device) |
|
|
| model, mcfg, ckpt_cfg = load_classifier(ckpt, device) |
| k = int(mcfg.num_labels) |
| max_len = int(args.max_len or mcfg.max_seq_len) |
| tok = load_tokenizer(cfg["paths"]["tokenizer"]) |
| if tok.get_vocab_size() > mcfg.vocab_size: |
| raise RuntimeError(f"tokenizer vocab {tok.get_vocab_size()} > model vocab {mcfg.vocab_size}") |
| if k == 2: |
| LOG.info("2-class checkpoint: binary roll-up only; every three-way section is skipped") |
| else: |
| LOG.info("3-class checkpoint: labels %s; binary roll-up P(unwanted) = P(phishing) + P(spam); rule %s", LABEL_NAMES, DECISION_RULE) |
|
|
| |
| |
| results = [evaluate_split(name, model, mcfg, tok, cfg, args, device, max_len, out_dir, run_name, ts, data_dir) for name in splits] |
| r0 = results[0] |
| lat = measure_latency(model, tok, r0.split, r0.enc, device, max_len, args.latency_samples, args.cpu_threads, int(cfg["seed"])) |
| lat.update({"checkpoint": str(ckpt), "run_name": run_name, "params": count_parameters(model)["total"], "timestamp": ts, |
| "num_labels": k, "measured_once_for_splits": splits}) |
| g, c = lat.get("gpu_bf16"), lat["cpu_fp32"] |
| print(f"\nlatency per email (batch 1, tokenize+forward+softmax) vs instructor benchmark {INSTRUCTOR_BENCHMARK_MS:.0f} ms:") |
| if g: |
| print(f" GPU bf16 : avg {g['avg_ms']:.2f} ms p50 {g['p50_ms']:.2f} ms (n={g['n']}, {g['ratio_to_benchmark']:.2f}x benchmark)") |
| print(f" CPU fp32 : avg {c['avg_ms']:.1f} ms p50 {c['p50_ms']:.1f} ms (n={c['n']}, {c['threads']} threads, {c['ratio_to_benchmark']:.2f}x benchmark)") |
|
|
| for r in results: |
| split, sfx = r.split, r.split.sfx |
| common.write_json(lat, out_dir / f"latency{sfx}.json") |
| tf_out = {kk: v for kk, v in r.tf.items() if kk != "curves"} |
| metrics = { |
| "timestamp": ts, "run_name": run_name, "split": split.name, "split_file": str(split.path), "split_title": SPLIT_TITLE[split.name], |
| "limit_rows": args.limit_rows, "dev_run": dev, "checkpoint": str(ckpt), "splits_in_invocation": splits, |
| "model_config": mcfg.to_dict(), "num_labels": k, "params": count_parameters(model)["total"], |
| "checkpoint_config": {kk: v for kk, v in ckpt_cfg.items() if kk != "model"}, |
| "tokenizer": str(cfg["paths"]["tokenizer"]), "max_len": max_len, |
| "label_names": {str(c3): LABEL_NAMES[c3] for c3 in range(k)} if k == 3 else {str(c2): BINARY_NAMES[c2] for c2 in range(2)}, |
| "binary_rollup": {"unwanted_name": UNWANTED_NAME, "classes": list(range(1, k)), "rule": "P(unwanted) = sum of P(class) for class >= 1; flag when >= threshold"}, |
| "split_sizes": {"rows": split.n, "positive": r.at["n_positive"], "negative": r.at["n_negative"], |
| "positive_rate": round(_safe_div(r.at["n_positive"], split.n), 4), |
| "label3": _label3_counts(split.label3), |
| "by_source": {s: {"rows": m["n"], "positive": m["n_positive"], "label3": m["label3_counts"]} for s, m in r.per_source.items()}, |
| "mean_tokens_full": round(float(r.enc.n_tokens_full.mean()), 1), |
| "pct_truncated": round(100 * float(r.enc.truncated.mean()), 2)}, |
| "threshold": args.threshold, "at_threshold": r.at, "threshold_free": tf_out, |
| "threeway": r.tw.metrics, |
| "credential_lure_slice": {"n_phrases": len(CREDENTIAL_LURE_PHRASES), "phrases": r.lure["phrases"], |
| "match_rule": "whole-word match on NFKC-lowercased text with non-alphanumerics collapsed to spaces", |
| "phrase_hits": r.lure["phrase_hits"], "phrase_hits_by_label": r.lure["phrase_hits_by_label"], |
| "label_purity": round(r.lure["label_purity"], 4), "min_positives_for_comparison": LURE_MIN_POSITIVES, |
| "label3_counts": r.lure["label3_counts"], "threeway_accuracy": r.lure.get("threeway_accuracy"), |
| "metrics": r.lure["metrics"]}, |
| "per_source": r.per_source, "length_confound": r.conf, "inference": r.inf, "latency": lat, |
| "error_analysis": {"path": str(r.err_path), **r.err_stats}, "plots": r.plots, |
| } |
| p_metrics = out_dir / f"metrics{sfx}.json" |
| common.write_json(metrics, p_metrics) |
| LOG.info("wrote %s", p_metrics) |
|
|
| |
| by_name = {r.split.name: _summary(r) for r in results} |
| line = {"run_name": run_name, "timestamp": ts, "stage": "evaluate", "checkpoint": str(ckpt), "splits": splits, "num_labels": k, |
| "dev_run": dev, "limit_rows": args.limit_rows, |
| "val": by_name.get("val"), "test": by_name.get("test"), "test3": by_name.get("test3"), |
| "latency_ms": {"gpu_bf16_avg": g and g["avg_ms"], "cpu_fp32_avg": c["avg_ms"]}} |
| p_exp = out_dir / "experiments.jsonl" |
| with p_exp.open("a", encoding="utf-8") as fh: |
| fh.write(json.dumps(line, default=common._json_default) + "\n") |
| LOG.info("appended %s", p_exp) |
|
|
| |
| num = next_run_log_number(common.project_root() / "RUN_LOG.md") |
| entry = run_log_entry(num, ts, run_name, results, ckpt, lat, out_dir, dev, k) |
| p_entry = out_dir / f"run_log_entry_{'_'.join(splits)}.md" |
| p_entry.write_text(entry + "\n", encoding="utf-8") |
| if dev or args.no_run_log: |
| LOG.info("RUN_LOG.md not appended (%s); entry saved to %s", "dev run" if dev else "--no-run-log", p_entry) |
| else: |
| common.append_run_log(entry) |
| LOG.info("appended entry %d to RUN_LOG.md", num) |
| return 0 |
|
|
|
|
| def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) |
| ap.add_argument("--config", default="config.yaml") |
| sub = ap.add_subparsers(dest="cmd", required=True) |
| r = sub.add_parser("run", help="evaluate a fine-tuned checkpoint on the test / test3 / val split(s)") |
| r.add_argument("--checkpoint", required=True, help="fine-tune best/ dir with model.safetensors + config.json") |
| r.add_argument("--split", default="test", |
| help=f"one of {sorted(SPLIT_FILES)} or a comma list (e.g. test,test3): one invocation, one experiments line, one RUN_LOG entry") |
| r.add_argument("--limit-rows", type=int, default=None, help="dev only: seeded random subset per split; skips the RUN_LOG append") |
| r.add_argument("--batch-size", type=int, default=64) |
| r.add_argument("--latency-samples", type=int, default=None, help="default config.eval.latency_samples (100)") |
| r.add_argument("--out-dir", default=None, help="default config.paths.reports") |
| r.add_argument("--data-dir", default=None, help="default config.paths.data_processed (fixture dir for self-tests)") |
| r.add_argument("--run-name", default=None, help="default: the run dir name (parent of best/)") |
| r.add_argument("--threshold", type=float, default=DEFAULT_THRESHOLD) |
| r.add_argument("--max-len", type=int, default=None, help="default model max_seq_len") |
| r.add_argument("--cpu-threads", type=int, default=16) |
| r.add_argument("--cpu", action="store_true", help="force CPU inference (dev)") |
| r.add_argument("--finetune-run-dir", default=None, |
| help="opt-in: also render <out-dir>/eval_finetune_curves.png from this run's metrics.csv/epochs.csv " |
| "(reports/finetune_curves.png itself belongs to finetune.py plot)") |
| r.add_argument("--pretrain-run-dir", default=None, |
| help="opt-in: also render <out-dir>/eval_pretrain_curves.png from this run's metrics.csv " |
| "(reports/pretrain_curves.png itself belongs to pretrain.py plot)") |
| r.add_argument("--no-run-log", action="store_true", help="do not append to RUN_LOG.md") |
| args = ap.parse_args(argv) |
| if args.latency_samples is None: |
| args.latency_samples = int(common.load_config(args.config)["eval"]["latency_samples"]) |
| return args |
|
|
|
|
| def main(argv: Sequence[str] | None = None) -> int: |
| args = parse_args(argv) |
| return run(args) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
