fix(ml): train on few players without crashing on empty splits

docs/how-to/evaluate-models.md

@@ -62,6 +62,13 @@ overwhelmingly legit; it biases recall at low FPR optimistically if cheaters hid
 **Split.** Players (pseudonyms) are split into train / val / test, never windows. The split is
 stratified by player label (cheat, legit, unlabelled), shuffled with `run.seed`, and asserted to
 have no player in two parts. The pipeline runs once per `game_context`, with its own split.
+A label stratum needs at least 2 players to put one in the test split and at least 3 to fill all
+three parts. With fewer players the val / test split is empty: the candidate is not scored on it,
+the report lists it under "No test metrics", and the ONNX parity check and benchmark use the
+windows of the first non-empty split (test, val, train).
+For smoke tests with few players, `run.split_by = "session"` splits by session instead. The same
+player can then be in train and test (leakage), so the report carries a warning banner and the
+numbers must not be compared with player-split results. It cannot be combined with `[tune]`.
 
 ## 2. Config reference (`eval.toml`)
 
@@ -76,6 +83,7 @@ unlabelled = "exclude"   # or "assume_legit"
 calibration = "auto"     # auto | isotonic | platt | ecdf | none, see "Calibration" below
 contexts = []            # empty = every game_context in the data
 split = [0.6, 0.2, 0.2]  # train / val / test share of players
+# split_by = "player"    # or "session" (leakage-prone, smoke tests with few players only)
 fpr_targets = [0.01, 0.001]
 min_confidence = 0.0
 learning_curve = [0.1, 0.25, 0.5, 1.0]  # empty list disables it

docs/reference/ml-jobs.md

@@ -125,12 +125,26 @@ it uploads the reports and registers nothing. **`train`** does the same, then fo
 takes the best usable candidate (highest `select_by` on the test split, ties by lower p99 latency;
 the ONNX parity check must pass), uploads its artefacts and registers the model.
 
+**Few players.** The split is by player and stratified by label, so a stratum needs at least 3
+players to fill train, val and test, and at least 2 to get a test split. With fewer, the val
+and/or test split is empty. The job does not fail for that: nothing is scored on an empty split,
+the candidate's `test` / `val` metrics are absent with `test_skipped` / `val_skipped` (for example
+`test split empty: needs at least 2 players per label stratum`), the ONNX parity check and the
+benchmark run on the windows of the first non-empty split (test, val, train; reported as
+`onnx.parity_on` in the report and `parity_on` in the result), and `train` still registers the
+model, without test metrics (the selection treats a missing metric as worst). With very few players, `split_by: "session"` gives every split windows (see the table).
+The result lists
+what the caller must know under `warnings`. A candidate without usable training windows is
+skipped with its `reason`; the job only fails if no candidate in any context is usable, and the
+message names each reason.
+
 | Param | Default | Meaning |
 |---|---|---|
 | `candidates` | `["isolation_forest", "gradient_boosting"]` | list of names or `{name: {hyperparameters}}`; `raw_hier_classifier` / `raw_hier_anomaly` need the raw build and cannot be mixed with the others; the torch candidates take `lr` / `pretrain_lr` as a number or `"auto"` (range test, see [learning-rate-finder.md](../explanation/learning-rate-finder.md)); the result then lists the found rate, rule, schedule and curve under `training` |
 | `game_context` / `game_contexts` | all | contexts to evaluate |
 | `unlabelled` | `exclude` | or `assume_legit` |
 | `label_sources`, `min_confidence` | all, 0 | which labels count (for example `["honeypot", "staff_verdict"]`) |
+| `split_by` | `player` | or `session`: split by session instead of player. **Leakage-prone** (the same player can be in train and test), so the metrics are not comparable with player-split ones; only for smoke tests with few players. Recorded as `split_by` in the report and the result, which then carries a warning. Not combinable with `tune` |
 | `seed`, `threads`, `seq_len`, `split`, `fpr_targets`, `learning_curve`, `benchmark` | as `eval.toml` | `threads` at most 8 |
 | `calibration` | `auto` | `isotonic`, `platt`, `ecdf`; `none` is refused |
 | `select_by` | `recall@fpr0.1pct` | metric that picks the candidate |

ml/src/corvus_ml/eval/config.py

@@ -15,6 +15,7 @@
 from corvus_ml.raw.tokenizer import TokenizerConfig
 
 UNLABELLED_MODES = ("exclude", "assume_legit")
+SPLIT_BY = ("player", "session")
 
 
 @dataclass
@@ -70,6 +71,9 @@ class EvalConfig:
     calibration: str = "auto"  # auto | isotonic | platt | ecdf | none (see corvus_ml.calibration)
     contexts: list[str] = field(default_factory=list)  # empty = every context in the data
     split: tuple[float, float, float] = (0.6, 0.2, 0.2)
+    # "player" (default) or "session": session splits leak (a player is in train and test) and
+    # are only meant for smoke tests with few players.
+    split_by: str = "player"
     fpr_targets: tuple[float, ...] = (0.01, 0.001)
     min_confidence: float = 0.0
     learning_curve: tuple[float, ...] = ()
@@ -97,6 +101,7 @@ def from_dict(cls, raw: dict, base: Path = Path()) -> EvalConfig:
             "calibration",
             "contexts",
             "min_confidence",
+            "split_by",
         ):
             if key in run:
                 setattr(cfg, key, run.pop(key))
@@ -110,6 +115,10 @@ def from_dict(cls, raw: dict, base: Path = Path()) -> EvalConfig:
             raise ValueError(f"unknown [run] keys: {sorted(run)}")
         if cfg.unlabelled not in UNLABELLED_MODES:
             raise ValueError(f"run.unlabelled must be one of {UNLABELLED_MODES}")
+        if cfg.split_by not in SPLIT_BY:
+            raise ValueError(f"run.split_by must be one of {SPLIT_BY}")
+        if cfg.split_by == "session" and "tune" in raw:
+            raise ValueError("run.split_by = 'session' cannot be combined with [tune]")
         if cfg.calibration not in CALIBRATION_METHODS:
             raise ValueError(f"run.calibration must be one of {CALIBRATION_METHODS}")
         for key in ("windows", "tokens", "sessions", "labels"):

ml/src/corvus_ml/eval/data.py

@@ -8,6 +8,7 @@
 
 from corvus_ml.features.spec import DEFAULT_AGG_INPUTS, DEFAULT_SEQ_INPUTS
 
+SPLIT_BY = ("player", "session")
 LABEL_UNLABELLED = -1
 LABEL_LEGIT = 0
 LABEL_CHEAT = 1
@@ -58,11 +59,15 @@ def take(self, idx: np.ndarray) -> EvalData:
             None if self.raw_time is None else self.raw_time[idx],
         )
 
-    def group_codes(self) -> np.ndarray:
-        """Integer id per pseudonym, assigned in order of first appearance."""
+    def group_codes(self, by: str = "player") -> np.ndarray:
+        """Integer id per pseudonym (``by="player"``) or per session (``by="session"``),
+        assigned in order of first appearance."""
+        if by not in SPLIT_BY:
+            raise ValueError(f"split_by must be one of {SPLIT_BY}, not {by!r}")
+        keys = self.pseudonym if by == "player" else self.session_id
         codes: dict[bytes, int] = {}
         return np.fromiter(
-            (codes.setdefault(p, len(codes)) for p in self.pseudonym),
+            (codes.setdefault(p, len(codes)) for p in keys),
             dtype=np.int64,
             count=len(self),
         )

ml/src/corvus_ml/eval/report.py

@@ -41,15 +41,26 @@ def _table(header: list[str], rows: list[list[str]]) -> list[str]:
     return out
 
 
+SESSION_SPLIT_BANNER = (
+    "> **Split by session, not by player.** The same player can be in train and in test, so these "
+    "numbers are leakage-prone and not comparable with player-split results. Use this for smoke "
+    "tests with few players only."
+)
+
+
 def to_markdown(report: dict) -> str:
     fprs = report["fpr_targets"]
     lines = [f"# Evaluation report: {report['name']}", ""]
     if report["synthetic"]:
         lines += [SYNTHETIC_BANNER, ""]
+    by_session = report.get("split_by", "player") == "session"
+    if by_session:
+        lines += [SESSION_SPLIT_BANNER, ""]
     lines += [
         f"Seed {report['seed']}, threads {report['threads']}, unlabelled windows: "
         f"{report['unlabelled']} ({report['assumed_legit_windows']} assumed legit). "
-        "Metrics are on the test split, which is split by player. `R@x%` is recall at a "
+        "Metrics are on the test split, which is split by "
+        f"{'session' if by_session else 'player'}. `R@x%` is recall at a "
         "false-positive rate of x% on the same score set; `player` aggregates the max score "
         "per player.",
         "",
@@ -96,12 +107,16 @@ def to_markdown(report: dict) -> str:
             if c["status"] != "ok":
                 rows.append([name, *["skipped"] + ["-"] * (len(head) - 2)])
                 continue
-            t, b, o = c["test"], c.get("bench", {}), c.get("onnx", {})
-            row = [name, str(c["n_params"]), _f(t["roc_auc"]), _f(t["pr_auc"])]
+            b, o = c.get("bench", {}), c.get("onnx", {})
+            t = c["test"] or {}  # empty test split: the metric cells stay empty
+            row = [name, str(c["n_params"]), _f(t.get("roc_auc")), _f(t.get("pr_auc"))]
             for f in fprs:
                 tag = _tag(f)
-                low = "" if t[f"resolved@fpr{tag}"] else "*"
-                row += [_f(t[f"recall@fpr{tag}"]) + low, _f(t[f"recall@fpr{tag}_player_max"]) + low]
+                low = "" if t.get(f"resolved@fpr{tag}", True) else "*"
+                row += [
+                    _f(t.get(f"recall@fpr{tag}")) + low,
+                    _f(t.get(f"recall@fpr{tag}_player_max")) + low,
+                ]
             row += [
                 _f(t.get("ece")),
                 _f(b.get("latency_p50_ms"), 3),
@@ -125,6 +140,13 @@ def to_markdown(report: dict) -> str:
         ]
         if skipped:
             lines += ["Skipped:", *skipped, ""]
+        no_test = [
+            f"- `{n}`: {c['test_skipped']}"
+            for n, c in info["candidates"].items()
+            if c["status"] == "ok" and c.get("test_skipped")
+        ]
+        if no_test:
+            lines += ["No test metrics:", *no_test, ""]
         lines += _tune_section(info)
         lines += _lr_section(info)
         lines += _sizes_section(info, budget)
@@ -268,7 +290,7 @@ def _confusion_section(info: dict, fprs: list[float]) -> list[str]:
         if c["status"] != "ok":
             continue
         for f in fprs:
-            k = c["confusion_at_val_threshold"][_tag(f)]
+            k = c["confusion_at_val_threshold"].get(_tag(f), {})
             if "tp" not in k:
                 continue
             rows.append(

ml/src/corvus_ml/eval/runner.py

@@ -113,8 +113,21 @@ def pick(idx: np.ndarray) -> EvalData:
     return pick(idx_train), pick(idx_val)
 
 
-def _eval_metrics(cand: Candidate, data: EvalData, idx: np.ndarray, fprs) -> dict:
+EMPTY_SPLIT_HINT = "needs at least 2 players per label stratum"
+
+
+def _empty_reason(part: str, idx: np.ndarray) -> str:
+    """Why a split cannot be evaluated: it has no windows, or none that carry a label."""
+    if len(idx) == 0:
+        return f"{part} split empty: {EMPTY_SPLIT_HINT}"
+    return f"{part} split has no labelled windows"
+
+
+def _eval_metrics(cand: Candidate, data: EvalData, idx: np.ndarray, fprs) -> dict | None:
+    """Metrics on the labelled windows of ``idx``; None (candidate not called) if there are none."""
     sub = data.take(idx[data.label[idx] != LABEL_UNLABELLED])
+    if len(sub) == 0:
+        return None
     y = (sub.label == LABEL_CHEAT).astype(np.int8)
     s = cand.calibrated_score(sub)
     out = M.window_metrics(y, s, sub.group_codes(), fprs)
@@ -186,11 +199,22 @@ def resolve_spec(cfg: EvalConfig) -> FeatureSpec:
     return FeatureSpec.current(seq_len=cfg.seq_len)
 
 
+def _parity_windows(data: EvalData, split) -> tuple[EvalData, str]:
+    """Real windows for the ONNX parity check and the benchmark: the labelled windows of the
+    first non-empty split in the order test, val, train (then any train window, labelled or
+    not). A tiny dataset may leave the held-out splits empty; the graph is still checked."""
+    for part, idx in (("test", split.test), ("val", split.val), ("train", split.train)):
+        labelled = idx[data.label[idx] != LABEL_UNLABELLED]
+        if len(labelled):
+            return data.take(labelled), part
+    return data.take(split.train), "train"
+
+
 def _export_and_bench(
     cand: Candidate,
     cfg: EvalConfig,
     data: EvalData,
-    test_idx: np.ndarray,
+    split,
     out: Path,
     fprs,
     spec: FeatureSpec,
@@ -200,11 +224,11 @@ def _export_and_bench(
     spec.write(out)  # the exported graphs are only meaningful together with this layout
     if spec.raw_bytes is not None and data.meta.get("tokenizer_json"):
         (out / spec.raw_bytes.tokenizer_file).write_bytes(data.meta["tokenizer_json"])
-    test = data.take(test_idx[data.label[test_idx] != LABEL_UNLABELLED])
+    test, parity_on = _parity_windows(data, split)
     if cand.parity_rows is not None:
         test = test.take(np.arange(min(len(test), cand.parity_rows)))
     if len(test) == 0:
-        return {"error": "no labelled test windows"}
+        return {"error": "no windows to check the ONNX graph on"}
     x = cand.inputs(test)
     shipped = out / f"{cand.name}.onnx"
     native_raw = cand.score(test)
@@ -238,6 +262,7 @@ def _export_and_bench(
         "calibrated": cand.calibration is not None,
         "max_abs_diff": float(np.abs(onnx_scores - native).max()),
         "parity_ok": bool(np.allclose(onnx_scores, native, rtol=PARITY_RTOL, atol=atol)),
+        "parity_on": parity_on,
     }
     if not cfg.bench.enabled:
         return res
@@ -291,14 +316,16 @@ def _quantized(cand, cfg, x, test, path, native, fprs, bench, kw) -> dict:
 
 
 def _learning_curve(name, cfg, data, split, fprs) -> list[dict]:
-    groups = data.group_codes()
+    groups = data.group_codes(cfg.split_by)
 
     def run(frac: float) -> dict[str, float]:
         sub = subsample_groups(data.label, groups, split.train, frac, cfg.seed)
         cand = _fit(name, cfg, data, sub, split.val)
         if cand is None:
             return {}
         m = _eval_metrics(cand, data, split.test, fprs)
+        if m is None:
+            return {}
         keys = ["roc_auc", "pr_auc", *[f"recall@fpr{_tag(f)}" for f in fprs]]
         return {
             "n_train_players": float(np.unique(groups[sub]).size),
@@ -335,13 +362,18 @@ def run_candidate(
     if cand.calibration is not None:
         c = cand.calibration
         res["calibration"] = {"method": c.method, "n_fit": c.n_fit, "fit_on": c.fit_on}
-    res["test"] = _eval_metrics(cand, data, split.test, fprs)
-    res["val"] = _eval_metrics(cand, data, split.val, fprs)
-    yv, sv, _ = _scores_and_labels(cand, data, split.val)
-    yt, st, _ = _scores_and_labels(cand, data, split.test)
-    res["confusion_at_val_threshold"] = {
-        _tag(f): M.confusion_at_val_threshold(yv, sv, yt, st, f) for f in fprs
-    }
+    for part, idx in (("test", split.test), ("val", split.val)):
+        res[part] = _eval_metrics(cand, data, idx, fprs)
+        if res[part] is None:
+            res[f"{part}_skipped"] = _empty_reason(part, idx)
+            log.warning("%s: %s; no %s metrics", name, res[f"{part}_skipped"], part)
+    res["confusion_at_val_threshold"] = {}
+    if res["test"] is not None and res["val"] is not None:
+        yv, sv, _ = _scores_and_labels(cand, data, split.val)
+        yt, st, _ = _scores_and_labels(cand, data, split.test)
+        res["confusion_at_val_threshold"] = {
+            _tag(f): M.confusion_at_val_threshold(yv, sv, yt, st, f) for f in fprs
+        }
     if hasattr(cand, "history"):
         res["training"] = {
             "best_epoch": cand.history.best_epoch,
@@ -354,8 +386,8 @@ def run_candidate(
         if pre is not None and pre.lr_plan is not None:
             res["training"]["pretrain_lr_plan"] = pre.lr_plan
     spec = spec or FeatureSpec.current(seq_len=cfg.seq_len)
-    res.update(_export_and_bench(cand, cfg, data, split.test, out_dir / "onnx" / ctx, fprs, spec))
-    if cfg.learning_curve:
+    res.update(_export_and_bench(cand, cfg, data, split, out_dir / "onnx" / ctx, fprs, spec))
+    if cfg.learning_curve and res["test"] is not None:
         res["learning_curve"] = _learning_curve(name, cfg, data, split, fprs)
     return res
 
@@ -403,7 +435,7 @@ def pretrain_checkpoints(cfg: EvalConfig, out_dir: Path) -> dict:
         sub = data.take(np.flatnonzero(data.game_context == ctx))
         if len(sub) == 0:
             continue
-        split = group_split(sub.label, sub.group_codes(), cfg.seed, cfg.split)
+        split = group_split(sub.label, sub.group_codes(cfg.split_by), cfg.seed, cfg.split)
         pool = sub.take(split.train)
         cand = cand_pkg.create(name, cfg.seed, cfg.threads, params)
         model_cfg = cand.model_config(sub)  # type: ignore[attr-defined]
@@ -449,7 +481,7 @@ def find_learning_rates(cfg: EvalConfig, out_dir: Path, *, steps: int | None = N
         sub = data.take(np.flatnonzero(data.game_context == ctx))
         if len(sub) == 0:
             continue
-        split = group_split(sub.label, sub.group_codes(), cfg.seed, cfg.split)
+        split = group_split(sub.label, sub.group_codes(cfg.split_by), cfg.seed, cfg.split)
         for name, params in cfg.candidates.items():
             cls = cand_pkg.REGISTRY[name]
             if not cls.tunable:  # torch candidates only: sklearn models have no learning rate
@@ -548,6 +580,8 @@ def run(cfg: EvalConfig, out_dir: Path, tune: bool = False) -> dict:
     from corvus_ml.eval.training import seed_everything
 
     seed_everything(cfg.seed, cfg.threads)
+    if tune and cfg.split_by != "player":
+        raise ValueError("tuning needs the split by player (split_by = 'player')")
     unknown = [n for n in cfg.candidates if n not in cand_pkg.REGISTRY]
     if unknown:
         raise ValueError(f"unknown candidates {unknown}; known: {sorted(cand_pkg.REGISTRY)}")
@@ -570,6 +604,7 @@ def run(cfg: EvalConfig, out_dir: Path, tune: bool = False) -> dict:
         "assumed_legit_windows": assumed,
         "fpr_targets": list(cfg.fpr_targets),
         "split": list(cfg.split),
+        "split_by": cfg.split_by,
         "latency_budget_ms": cfg.bench.latency_budget_ms,
         "contexts": {},
     }
@@ -586,7 +621,7 @@ def run(cfg: EvalConfig, out_dir: Path, tune: bool = False) -> dict:
         if len(sub) == 0:
             continue
         groups = sub.group_codes()
-        split = group_split(sub.label, groups, cfg.seed, cfg.split)
+        split = group_split(sub.label, sub.group_codes(cfg.split_by), cfg.seed, cfg.split)
         info = {
             "windows": len(sub),
             "players": int(np.unique(groups).size),