From 56235f220aa70cd3e50fbb8f1b64a33f0988fa87 Mon Sep 17 00:00:00 2001 From: Carlos Ortiz Date: Sun, 27 Sep 2026 10:08:37 -0600 Subject: [PATCH] Select the target and the configuration on data the report never sees compile_selected implements the protocol of #75. Rows are split into Fit, Select and Report: stratified, out-of-time by date, or by group so one applicant lands in one partition. The ceiling (the teacher) is tuned once by the user's factory, frozen, and refit K times for cross-fitted soft targets; one search plus K fits, never K searches. Every (alpha, trees, depth, monotone) configuration is compiled on Fit and scored on Select as an integer artifact, so quantization and banding are priced too. The simplest configuration statistically tied with the best wins, preferring alpha = 1 as a governance prior recorded as the tie rule; the floor gates; the winner is refit on Fit union Select; result.report() reads Report once and writes the figures into a hash-covered provenance block, so what ships carries its own reported cost. Below a configurable event count in Select, nested cross-validation takes over with a warning. Whether the whitebox learns from the ceiling's probabilities or from the labels is now a selected parameter. Every step is also a plain function. compileml.batch.score_batch is the runtime's score path over a matrix in NumPy, asserted bit-identical to decide() on every row across calibration modes, input precision and missing values; 200,000 rows in under a second. Selection ranks compiled artifacts with it and never computes attribution. The trees axis is free: extract_trees passes an ExtractedModel through, so one fit at the largest tree count is extracted once and compiled from prefixes, verified equal to a separate fit for both backends. train_whitebox(backend="hist") selects the histogram backend explicitly, measured at 1-2 s where the classic backend took 18 s at 30k rows and 293 s at 300k; compile_selected uses it throughout, and the default is unchanged. Validation check 11, selection hygiene, reads the provenance block and lists broken protocol: soft targets not cross-fitted, partitions that overlap, Report read more than once. Advisory unless required, following check 10. sweep_whitebox warns when alpha_grid is not passed and when one holdout is used both to choose and to report. Part of #75; the documentation and the benchmark rerun follow separately. --- CHANGELOG.md | 35 + src/compileml/__init__.py | 13 +- src/compileml/batch.py | 122 ++++ src/compileml/cli.py | 8 +- src/compileml/compile/distill.py | 15 +- src/compileml/compile/extract.py | 10 +- src/compileml/select.py | 978 ++++++++++++++++++++++++++++ src/compileml/tune/sweeps.py | 20 +- src/compileml/validate/framework.py | 71 ++ tests/test_batch.py | 64 ++ tests/test_select.py | 447 +++++++++++++ tests/test_validate.py | 4 +- 12 files changed, 1781 insertions(+), 6 deletions(-) create mode 100644 src/compileml/batch.py create mode 100644 src/compileml/select.py create mode 100644 tests/test_batch.py create mode 100644 tests/test_select.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a434b2..7207a1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,41 @@ inside each artifact (`schema_version`). ## [Unreleased] +### Added +- `compile_selected` ([#75](https://github.com/orgoca/CompileML/issues/75)): + the target and the configuration are chosen on data the report never sees. + Rows are split into Fit, Select and Report; the ceiling (the teacher) is + tuned once, frozen and refit K times for cross-fitted soft targets; every + (α, trees, depth, monotone) configuration is compiled on Fit and scored on + Select as an integer artifact; the simplest configuration statistically + tied with the best wins, preferring α = 1 as a governance prior; the floor + gates; the winner is refit on Fit ∪ Select; and `result.report()` reads + Report once, writing the figures into a hash-covered provenance block. Below + a configurable event count in Select, nested cross-validation takes over. + Whether the whitebox learns from the ceiling's probabilities or from the + labels is now a selected parameter, not an assumption. +- `compileml.batch.score_batch`: the runtime's score path over a matrix, in + NumPy, asserted bit-identical to `decide()` on every row — 200,000 rows in + under a second. Selection ranks compiled artifacts with it; deployments + still use the runtime or an export. +- Validation check 11, selection hygiene: reads the provenance block and + lists broken protocol — soft targets not cross-fitted, partitions that + overlap, Report read more than once. Advisory unless + `require_selection_hygiene=True`, following check 10. +- `train_whitebox(backend="hist")` selects the histogram backend explicitly. + It was measured at 1–2 s where the classic backend took 18 s at 30k rows and + 293 s at 300k; `compile_selected` uses it throughout. The default is + unchanged, since changing it would change every artifact built with the + defaults. +- `extract_trees` passes an `ExtractedModel` through, so a boosted fit can be + extracted once and compiled from prefixes of its trees: the first k trees of + a fit are the k-tree model, verified for both backends. + +### Changed +- `sweep_whitebox` warns when `alpha_grid` is not passed — its default of + pure distillation is the assumption real data falsified — and when `X_val` + is used both to choose and to report. The defaults switch in 1.0. + ## [0.8.0] - 2026-09-13 What compilation cost where decisions are made — and what to do about it. diff --git a/src/compileml/__init__.py b/src/compileml/__init__.py index 52b5774..4673c6d 100644 --- a/src/compileml/__init__.py +++ b/src/compileml/__init__.py @@ -18,4 +18,15 @@ # `compileml_version` recorded inside every artifact can never disagree. __version__ = "0.8.0" -__all__ = ["decide", "load_artifact", "verify_artifact", "__version__"] +__all__ = ["compile_selected", "decide", "load_artifact", "verify_artifact", "__version__"] + + +def __getattr__(name): + # compile_selected is learning-side (NumPy, scikit-learn). Importing it + # lazily keeps `import compileml` free of those dependencies, so the + # runtime story - copy compileml/runtime anywhere - stays true of the root. + if name == "compile_selected": + from compileml.select import compile_selected + + return compile_selected + raise AttributeError(f"module 'compileml' has no attribute {name!r}") diff --git a/src/compileml/batch.py b/src/compileml/batch.py new file mode 100644 index 0000000..0a3fa09 --- /dev/null +++ b/src/compileml/batch.py @@ -0,0 +1,122 @@ +"""Score many rows through an artifact at once, with NumPy. + +The runtime scores one row at a time in the standard library, which is what +lets it run anywhere. Selection scores forty compiled candidates on hundreds +of thousands of rows, where a Python loop costs minutes per candidate. This +module does the same arithmetic vectorised — the same integer sums, the same +float64 comparison, the same rounding — and the test suite asserts it agrees +with :func:`compileml.runtime.decide` on every row. + +Learning side: it reads an artifact and never makes a decision anyone acts +on. Deployments use the runtime or an export. +""" + +from __future__ import annotations + +import numpy as np + +from compileml.runtime.calibrate import PD_SCALE +from compileml.runtime.io import ArtifactError +from compileml.runtime.score import LEAF + + +def _div_rha_nonneg(num: np.ndarray, den) -> np.ndarray: + """Spec §2.2 for non-negative numerators, in int64.""" + den = np.asarray(den, dtype=np.int64) + return (2 * num + den) // (2 * den) + + +def _prepare(artifact: dict, X) -> np.ndarray: + """The runtime's ``_prepare_row`` over a matrix (spec §4, §8).""" + names = artifact["features"]["names"] + X_arr = np.array(X, dtype=np.float64, copy=True) + if X_arr.ndim != 2 or X_arr.shape[1] != len(names): + raise ArtifactError(f"expected {len(names)} feature columns, got shape {X_arr.shape}") + missing = np.isnan(X_arr) + if missing.any(): + if artifact["features"].get("missing_policy", "baseline") == "reject": + rows = int(missing.any(axis=1).sum()) + raise ArtifactError(f"{rows} row(s) have missing values and missing_policy is 'reject'") + baseline = np.asarray(artifact["features"]["baseline"], dtype=np.float64) + X_arr = np.where(missing, baseline[None, :], X_arr) + if artifact["model"].get("input_precision", "float64") == "float32": + X_arr = X_arr.astype(np.float32).astype(np.float64) + return X_arr + + +def raw_micro_batch(model: dict, X: np.ndarray) -> np.ndarray: + """``score_micro`` for every row: integer sums of the leaves each row reaches.""" + n = X.shape[0] + acc = np.full(n, int(model["base_micro"]), dtype=np.int64) + rows = np.arange(n) + for tree in model["trees"]: + feature = np.asarray(tree["feature"], dtype=np.int64) + threshold = np.asarray(tree["threshold"], dtype=np.float64) + left = np.asarray(tree["left"], dtype=np.int64) + right = np.asarray(tree["right"], dtype=np.int64) + value = np.asarray(tree["value_micro"], dtype=np.int64) + node = np.zeros(n, dtype=np.int64) + active = feature[node] != LEAF + while active.any(): + idx = rows[active] + at = node[idx] + go_left = X[idx, feature[at]] <= threshold[at] + node[idx] = np.where(go_left, left[at], right[at]) + active = feature[node] != LEAF + acc += value[node] + return acc + + +def calibrate_ppm_batch( + latent_micro: np.ndarray, calibration: dict | None, micro_scale: int +) -> np.ndarray: + """``calibrate_ppm`` for every row (spec §6).""" + if not calibration: + return _div_rha_nonneg(latent_micro * PD_SCALE, micro_scale) + f = np.asarray(calibration["f_micro"], dtype=np.int64) + pd = np.asarray(calibration["pd_ppm"], dtype=np.int64) + k = len(f) + out = np.empty_like(latent_micro) + low = latent_micro <= f[0] + high = latent_micro >= f[k - 1] + out[low] = pd[0] + out[high] = pd[k - 1] + mid = ~(low | high) + if mid.any(): + lm = latent_micro[mid] + hi = np.searchsorted(f, lm, side="right") + lo = hi - 1 + if calibration.get("mode", "linear_int") == "step": + out[mid] = pd[lo] + else: + num = (pd[hi] - pd[lo]) * (lm - f[lo]) + out[mid] = pd[lo] + _div_rha_nonneg(num, f[hi] - f[lo]) + return out + + +def score_batch(artifact: dict, X) -> dict[str, np.ndarray]: + """Score a matrix of rows: the runtime's score path, one array per output. + + Returns ``raw_micro``, ``latent_micro``, ``latent_int``, ``band_idx`` and + ``pd_ppm`` as int64 arrays, plus ``pd`` as float64. Every integer equals + what ``decide(artifact, row, explain=False)`` returns for that row. + """ + model = artifact["model"] + micro_scale = int(model["micro_scale"]) + ratio = micro_scale // int(artifact["scale"]) + X_arr = _prepare(artifact, X) + + raw = raw_micro_batch(model, X_arr) + latent_micro = np.clip(raw, 0, micro_scale) + latent_int = _div_rha_nonneg(latent_micro, ratio) + edges = np.asarray(artifact["bands"]["edges_int"], dtype=np.int64)[1:-1] + band_idx = np.searchsorted(edges, latent_int, side="right") + pd_ppm = calibrate_ppm_batch(latent_micro, artifact.get("calibration"), micro_scale) + return { + "raw_micro": raw, + "latent_micro": latent_micro, + "latent_int": latent_int, + "band_idx": band_idx, + "pd_ppm": pd_ppm, + "pd": pd_ppm / PD_SCALE, + } diff --git a/src/compileml/cli.py b/src/compileml/cli.py index da9ebf9..b2d12ae 100644 --- a/src/compileml/cli.py +++ b/src/compileml/cli.py @@ -175,6 +175,7 @@ def cmd_validate(args) -> int: X_val=X_val, y_val=y_val, require_full_reason_coverage=args.require_reasons, + require_selection_hygiene=args.require_selection_hygiene, ) print(json.dumps(report, indent=2)) return 0 if report["all_pass"] else 1 @@ -275,7 +276,12 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument("--out", help="output file (default: stdout)") p.set_defaults(func=cmd_scorecard) - p = sub.add_parser("validate", help="run the 10-check validation framework") + p = sub.add_parser("validate", help="run the validation framework") + p.add_argument( + "--require-selection-hygiene", + action="store_true", + help="fail check 11 when the provenance block records a broken selection protocol", + ) p.add_argument("artifact") p.add_argument("--csv", help="validation CSV (features + outcome)") p.add_argument("--y-col", help="outcome column name in --csv") diff --git a/src/compileml/compile/distill.py b/src/compileml/compile/distill.py index 2e886d5..bca7a60 100644 --- a/src/compileml/compile/distill.py +++ b/src/compileml/compile/distill.py @@ -35,6 +35,7 @@ def train_whitebox( monotone_constraints=None, teacher_latent=None, sample_weight=None, + backend: str | None = None, ): """Fit a whitebox GBM to a target. @@ -54,6 +55,14 @@ def train_whitebox( the backend to ``HistGradientBoostingRegressor``; ``None`` (or all zeros) keeps the classic ``GradientBoostingRegressor``. + ``backend`` overrides that choice: ``"hist"`` for + ``HistGradientBoostingRegressor`` and ``"gbr"`` for the classic + ``GradientBoostingRegressor``. The histogram backend is the one to use + at scale — it bins features and runs multithreaded, and was measured at + 1–2 s where the classic backend took 18 s at 30k rows and 293 s at 300k + — but the default stays ``None`` (classic unless constrained) because + changing it would change every artifact built with the defaults. + ``sample_weight`` — one non-negative weight per row — tells the fit where fidelity matters. A whitebox has a fixed budget and, unweighted, spends it where most of the squared error is: the largest segment and the busiest @@ -97,7 +106,11 @@ def train_whitebox( raise ValueError("sample_weight must be finite, non-negative, and not all zero") feature_names = list(X.columns) if hasattr(X, "columns") else None cst = normalize_constraints(monotone_constraints, X_arr.shape[1], feature_names=feature_names) - if cst is not None: + if backend not in (None, "hist", "gbr"): + raise ValueError("backend must be None, 'hist' or 'gbr'") + if backend == "gbr" and cst is not None: + raise ValueError("monotone constraints need the 'hist' backend") + if cst is not None or backend == "hist": model = HistGradientBoostingRegressor( max_iter=n_estimators, max_depth=max_depth, diff --git a/src/compileml/compile/extract.py b/src/compileml/compile/extract.py index f0c937c..bcc8b66 100644 --- a/src/compileml/compile/extract.py +++ b/src/compileml/compile/extract.py @@ -369,7 +369,15 @@ def _extract_lightgbm(model) -> ExtractedModel: def extract_trees(model) -> ExtractedModel: - """Extract a fitted model into float tree arrays; dispatches on family.""" + """Extract a fitted model into float tree arrays; dispatches on family. + + An :class:`ExtractedModel` passes through unchanged. That is what lets a + caller extract a boosted model once and build artifacts from prefixes of + its trees — the first ``k`` trees of a boosted fit *are* the ``k``-tree + model — without refitting. + """ + if isinstance(model, ExtractedModel): + return model if _is_sklearn_hist(model): extracted = _extract_sklearn_hist(model) validate_extraction(extracted, model) diff --git a/src/compileml/select.py b/src/compileml/select.py new file mode 100644 index 0000000..5c4bbce --- /dev/null +++ b/src/compileml/select.py @@ -0,0 +1,978 @@ +"""Select the target and the configuration on data the report never sees. + +Three models with fixed jobs. The **ceiling** (the teacher) is the strongest +model the user can train under a declared budget; it prices compilation and +may supply soft targets, and it never ships. The **floor** — a WoE logistic +reference or a champion scorecard's Gini — says whether adopting the artifact +is worth it. The **candidate** is the depth-≤2 whitebox, one per +configuration, trained on labels, ceiling predictions, or a blend. + +Three disjoint partitions. **Fit** trains everything; **Select** chooses the +target and the configuration; **Report** is read once, for the number that +gets published. The selected configuration is refit on Fit ∪ Select before +Report is read, so the selection curve ranks configurations, not the artifact +that ships. + +Whether the whitebox should learn from the ceiling's probabilities or from +the labels is a parameter this module selects, not an assumption: on real +data, the whitebox trained on labels has beaten the distilled one. Every +step here is also a plain function, so the protocol can be followed by hand. + +Learning side: NumPy and scikit-learn throughout. Nothing here runs at +decision time. +""" + +from __future__ import annotations + +import hashlib +import warnings +from dataclasses import dataclass, field, replace + +import numpy as np +from sklearn.base import clone +from sklearn.metrics import roc_auc_score +from sklearn.model_selection import GroupShuffleSplit, StratifiedKFold, train_test_split + +from compileml.artifact import build_artifact +from compileml.bands import monotone_quantile_bands +from compileml.batch import score_batch +from compileml.compile import extract_trees, train_whitebox +from compileml.reference import fit_reference +from compileml.runtime.io import canonical_hash + +DEFAULT_ALPHAS = (0.0, 0.25, 0.5, 0.75, 1.0) + + +def default_grid() -> dict: + """The default search space: five targets, four tree counts, two depths. + + ``monotone`` holds ``None`` and/or ``"declared"``; the declared set comes + from ``compile_selected(monotone_constraints=...)`` and is never inferred. + Depth 3 is not in the grid because it forfeits exact attribution; add it + explicitly if that trade is acceptable. + """ + return { + "alpha": DEFAULT_ALPHAS, + "trees": (20, 40, 80, 160), + "depth": (1, 2), + "monotone": (None,), + } + + +# ------------------------------------------------------------------ metrics +def gini(y, score) -> float: + y = np.asarray(y, dtype=int) + if len(np.unique(y)) < 2: + return float("nan") + return 2.0 * float(roc_auc_score(y, np.asarray(score, dtype=float))) - 1.0 + + +def ks_statistic(y, score) -> float: + """Maximum separation between the score distributions of the two classes.""" + y = np.asarray(y, dtype=int) + score = np.asarray(score, dtype=float) + if y.sum() == 0 or y.sum() == len(y): + return float("nan") + order = np.argsort(score, kind="stable") + pos = np.cumsum(y[order]) / y.sum() + neg = np.cumsum(1 - y[order]) / (len(y) - y.sum()) + return float(np.max(np.abs(pos - neg))) + + +def brier(y, prob) -> float: + y = np.asarray(y, dtype=float) + return float(np.mean((np.asarray(prob, dtype=float) - y) ** 2)) + + +# --------------------------------------------------------------- partitions +@dataclass +class Partitions: + """Row indices of the three partitions, and how they were made.""" + + fit: np.ndarray + select: np.ndarray + report: np.ndarray + method: str + + @property + def dev(self) -> np.ndarray: + """Fit ∪ Select — what the selected configuration is refit on.""" + return np.concatenate([self.fit, self.select]) + + +def make_partitions( + y, + *, + method: str = "stratified", + fractions=(0.6, 0.2, 0.2), + date=None, + group=None, + explicit: dict | None = None, + seed: int = 42, +) -> Partitions: + """Split rows into Fit, Select and Report. + + - ``method="stratified"`` (default): stratified on the label. + - ``date``: Report is the latest slice by date; Fit and Select are split + within the earlier period. + - ``group``: split by group so one applicant lands in one partition. Not + stratified, which the method name records. + - ``explicit``: ``{"fit": idx, "select": idx, "report": idx}``, checked + for disjointness. + """ + y = np.asarray(y).reshape(-1) + n = len(y) + if explicit is not None: + parts = {k: np.asarray(explicit[k], dtype=int) for k in ("fit", "select", "report")} + for k, idx in parts.items(): + if idx.size == 0 or idx.min() < 0 or idx.max() >= n or len(np.unique(idx)) != len(idx): + raise ValueError(f"explicit {k} indices must be unique and within range") + for a, b in (("fit", "select"), ("fit", "report"), ("select", "report")): + if np.intersect1d(parts[a], parts[b]).size: + raise ValueError(f"explicit partitions overlap: {a} and {b}") + return Partitions(parts["fit"], parts["select"], parts["report"], "explicit") + + f_fit, f_select, f_report = (float(v) for v in fractions) + if min(f_fit, f_select, f_report) <= 0 or abs(f_fit + f_select + f_report - 1.0) > 1e-9: + raise ValueError("fractions must be positive and sum to 1") + idx = np.arange(n) + + if date is not None: + order = idx[np.argsort(np.asarray(date), kind="stable")] + n_report = max(1, int(round(n * f_report))) + report, rest = order[-n_report:], order[:-n_report] + method = "out_of_time" + else: + if group is not None: + splitter = GroupShuffleSplit(n_splits=1, test_size=f_report, random_state=seed) + rest_pos, report_pos = next(splitter.split(idx, groups=np.asarray(group))) + rest, report = idx[rest_pos], idx[report_pos] + method = "group" + else: + rest, report = train_test_split(idx, test_size=f_report, stratify=y, random_state=seed) + method = "stratified" + + select_share = f_select / (f_fit + f_select) + if group is not None: + splitter = GroupShuffleSplit(n_splits=1, test_size=select_share, random_state=seed + 1) + fit_pos, select_pos = next(splitter.split(rest, groups=np.asarray(group)[rest])) + fit, select = rest[fit_pos], rest[select_pos] + method = method if method != "stratified" else "group" + if date is not None: + method = "out_of_time_group" + else: + fit, select = train_test_split( + rest, test_size=select_share, stratify=y[rest], random_state=seed + 1 + ) + return Partitions(np.sort(fit), np.sort(select), np.sort(report), method) + + +def row_content_hashes(X, y) -> np.ndarray: + """A 128-bit digest per row of its features and label, for disjointness checks.""" + X = np.ascontiguousarray(np.asarray(X, dtype=np.float64)) + y = np.asarray(y).reshape(-1) + out = np.empty(len(y), dtype="S16") + for i in range(len(y)): + out[i] = hashlib.sha256(X[i].tobytes() + bytes([int(y[i])])).digest()[:16] + return out + + +def partition_hashes(hashes: np.ndarray, parts: Partitions) -> dict: + """One hex digest per partition, over its sorted row digests.""" + result = {} + for name in ("fit", "select", "report"): + rows = np.sort(hashes[getattr(parts, name)]) + result[name] = hashlib.sha256(b"".join(rows.tolist())).hexdigest() + return result + + +def duplicate_rows(hashes: np.ndarray, parts: Partitions) -> int: + """Rows in Select whose content appears in Fit, plus rows in Report whose + content appears in Fit ∪ Select. Legitimate in credit data (coarse bins, + repeat applicants), so a count rather than a failure.""" + fit_set = set(hashes[parts.fit].tolist()) + dup = sum(1 for h in hashes[parts.select].tolist() if h in fit_set) + dev_set = fit_set | set(hashes[parts.select].tolist()) + dup += sum(1 for h in hashes[parts.report].tolist() if h in dev_set) + return int(dup) + + +# ------------------------------------------------------------------ ceiling +def ceiling_predict(model, X) -> np.ndarray: + """A ceiling's score: positive-class probability, or the prediction.""" + if hasattr(model, "predict_proba"): + return np.asarray(model.predict_proba(X))[:, 1].astype(float) + return np.asarray(model.predict(X), dtype=float).reshape(-1) + + +def freeze_ceiling(model): + """The configuration to refit: a search's best estimator, or the model itself.""" + frozen = getattr(model, "best_estimator_", model) + try: + clone(frozen) + except Exception as exc: # pragma: no cover - depends on the user's object + raise TypeError( + "the ceiling factory must return a scikit-learn-compatible estimator " + "(or a fitted search exposing best_estimator_) so its frozen " + "configuration can be refit for out-of-fold predictions; otherwise pass " + "ceiling_oof= and ceiling_scores= instead" + ) from exc + return frozen + + +def params_hash(estimator) -> str: + """A digest of the frozen hyperparameters, so provenance can name what was refit.""" + try: + text = repr(sorted(estimator.get_params(deep=False).items())) + except Exception: # pragma: no cover + text = repr(estimator) + return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] + + +@dataclass +class _Ceiling: + kind: str # "factory" | "predictions" + factory: object = None + frozen: object = None + oof: np.ndarray | None = None # full-length, for kind == "predictions" + scores: np.ndarray | None = None + budget: dict | None = None + factory_calls: int = 0 + fits: int = 0 + + def tune(self, X, y, idx): + """Call the factory once on ``idx`` and freeze what it returns.""" + self.frozen = freeze_ceiling(self.factory(X[idx], y[idx])) + self.factory_calls += 1 + self.fits += 1 + + def out_of_fold(self, X, y, idx, k: int, seed: int) -> np.ndarray: + """Cross-fitted predictions on ``idx``: k refits of the frozen configuration.""" + if self.kind == "predictions": + return self.oof[idx] + out = np.empty(len(idx), dtype=float) + folds = StratifiedKFold(n_splits=k, shuffle=True, random_state=seed) + for train_pos, held_pos in folds.split(idx, y[idx]): + model = clone(self.frozen).fit(X[idx[train_pos]], y[idx[train_pos]]) + self.fits += 1 + out[held_pos] = ceiling_predict(model, X[idx[held_pos]]) + return out + + def score(self, X, y, train_idx, eval_idx) -> np.ndarray: + """The frozen configuration refit on ``train_idx``, scored on ``eval_idx``.""" + if self.kind == "predictions": + return self.scores[eval_idx] + model = clone(self.frozen).fit(X[train_idx], y[train_idx]) + self.fits += 1 + return ceiling_predict(model, X[eval_idx]) + + @property + def family(self) -> str: + if self.kind == "predictions": + return "user_predictions" + cls = type(self.frozen) + return f"{cls.__module__}.{cls.__name__}" + + +# --------------------------------------------------------------------- grid +def _resolve_grid(grid, monotone_constraints, has_ceiling) -> tuple[dict, list[str]]: + g = dict(default_grid()) + g.update(grid or {}) + notes = [] + alphas = tuple(float(a) for a in g["alpha"]) + if not has_ceiling: + if any(a < 1.0 for a in alphas): + notes.append("no ceiling predictions: alpha forced to (1.0,) — labels only") + alphas = (1.0,) + else: + for endpoint in (0.0, 1.0): + if endpoint not in alphas: + alphas = tuple(sorted({*alphas, endpoint})) + notes.append(f"alpha endpoint {endpoint} added: endpoints are always searched") + if any(not 0.0 <= a <= 1.0 for a in alphas): + raise ValueError("alpha values must lie in [0, 1]") + trees = tuple(sorted({int(t) for t in g["trees"]})) + depths = tuple(sorted({int(d) for d in g["depth"]})) + if any(d > 2 for d in depths): + notes.append("depth > 2 in the grid: those candidates forfeit exact attribution") + monos = [] + for m in g["monotone"]: + if m is None: + monos.append(None) + elif m == "declared": + if monotone_constraints is None: + raise ValueError("grid asks for the declared monotone set but none was given") + monos.append(monotone_constraints) + else: + raise ValueError("grid['monotone'] entries must be None or 'declared'") + return {"alpha": alphas, "trees": trees, "depth": depths, "monotone": tuple(monos)}, notes + + +def _config_key(alpha, trees, depth, mono_index) -> tuple: + return (float(alpha), int(trees), int(depth), int(mono_index)) + + +def evaluate_grid( + X, + y, + train_idx, + eval_idx, + *, + grid: dict, + oof_train: np.ndarray | None, + feature_names, + n_bands: int, + backend: str, + learning_rate: float, + sample_weight=None, + seed: int = 42, +) -> dict[tuple, dict]: + """Every configuration in the grid, compiled on ``train_idx`` and scored on ``eval_idx``. + + One fit per (alpha, depth, monotone) at the largest tree count; the + smaller counts are prefixes of that fit, extracted once and truncated. + Each prefix is quantized, calibrated and banded on the training rows and + scored on the evaluation rows through the batch scorer — the compiled + artifact is what is ranked, not the float whitebox. + """ + X_tr, y_tr = X[train_idx], y[train_idx] + X_ev = X[eval_idx] + w_tr = None if sample_weight is None else np.asarray(sample_weight, dtype=float)[train_idx] + baseline = np.median(X_tr, axis=0) + positives = int(y_tr.sum()) + trees_wanted = set(grid["trees"]) + max_trees = max(grid["trees"]) + rows: dict[tuple, dict] = {} + + for alpha in grid["alpha"]: + target = y_tr.astype(float) if alpha >= 1.0 else alpha * y_tr + (1.0 - alpha) * oof_train + for depth in grid["depth"]: + for m_index, mono in enumerate(grid["monotone"]): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + model, _ = train_whitebox( + X_tr, + target, + n_estimators=max_trees, + max_depth=depth, + learning_rate=learning_rate, + random_state=seed, + monotone_constraints=mono, + sample_weight=w_tr, + backend=backend, + ) + extracted = extract_trees(model) + stages = {} + for i, pred in enumerate(model.staged_predict(X_tr), start=1): + if i in trees_wanted: + stages[i] = np.clip(pred, 0.0, 1.0) + for k in grid["trees"]: + prefix = replace(extracted, trees=extracted.trees[:k]) + latent = stages[k] + artifact = build_artifact( + prefix, + feature_names, + baseline, + monotone_quantile_bands(latent, y_tr, n_bands=n_bands), + calibration_latent=latent, + calibration_y=y_tr, + monotone_constraints=mono, + ) + scored = score_batch(artifact, X_ev) + rows[_config_key(alpha, k, depth, m_index)] = { + "alpha": float(alpha), + "n_estimators": int(k), + "max_depth": int(depth), + "monotone": mono is not None, + "eps": positives / (k * (2**depth - 1)), + "_raw": scored["raw_micro"], + "_pd": scored["pd"], + } + return rows + + +# ---------------------------------------------------------------- selection +def _metric(name: str, y, raw, pd) -> float: + if name == "gini": + return gini(y, raw) + if name == "brier": + return -brier(y, pd) # higher is better throughout selection + raise ValueError("selection_metric must be 'gini' or 'brier'") + + +def choose( + rows: dict[tuple, dict], y_eval, *, metric: str, tie_se: float, n_boot: int, seed: int +) -> dict: + """One-SE-simplest: the simplest configuration statistically tied with the best. + + Bootstrap the evaluation rows to get the standard error of the best + configuration's metric; everything within ``tie_se`` standard errors is + tied. Among the tied, prefer α = 1 (no training dependency on the + ceiling), then fewer trees, then lower depth. The α preference is a + governance prior, recorded as the tie rule, not an empirical claim. + """ + y_eval = np.asarray(y_eval, dtype=int) + for row in rows.values(): + row["gini"] = gini(y_eval, row["_raw"]) + row["brier"] = brier(y_eval, row["_pd"]) + row["score"] = _metric(metric, y_eval, row["_raw"], row["_pd"]) + best_key = max(rows, key=lambda k: rows[k]["score"]) + best = rows[best_key] + + rng = np.random.default_rng(seed) + n = len(y_eval) + samples = [] + for _ in range(n_boot): + pick = rng.integers(0, n, n) + samples.append(_metric(metric, y_eval[pick], best["_raw"][pick], best["_pd"][pick])) + se = float(np.std(samples, ddof=1)) if n_boot > 1 else 0.0 + + tied = [k for k, r in rows.items() if r["score"] >= best["score"] - tie_se * se] + chosen_key = min(tied, key=lambda k: (-k[0], k[1], k[2], k[3])) + for k, r in rows.items(): + r["tied"] = k in tied + r["selected"] = k == chosen_key + return { + "key": chosen_key, + "best_key": best_key, + "se": se, + "n_tied": len(tied), + "metric": metric, + "tie_se": tie_se, + "n_boot": n_boot, + } + + +# ------------------------------------------------------------------- result +@dataclass +class SelectionResult: + """What ``compile_selected`` returns: the artifact, the curve, and one report.""" + + artifact: dict | None + selection_curve: list[dict] + selected: dict + partitions: Partitions + floor_gate: dict + notes: list[str] + report_evaluations: int = 0 + _state: dict = field(default_factory=dict, repr=False) + + def report(self) -> dict: + """Evaluate the shipped artifact on Report, once. + + Gini, KS and Brier for artifact, ceiling and floor; retention against + the ceiling and the ratio against the floor at equal weight, each with + a bootstrap interval. The figures are written into the artifact's + provenance and the artifact is rehashed, so what ships carries its own + reported cost. A second call raises: nothing may be chosen after + Report is read, and this counter is hygiene rather than proof. + """ + if self.report_evaluations >= 1: + raise RuntimeError( + "Report has already been evaluated once; a second evaluation after " + "changing anything would invalidate the published figure. Rerun the " + "whole pipeline if the configuration changes." + ) + if self.artifact is None: + raise RuntimeError("no artifact: the selected configuration did not clear the floor") + s = self._state + y_rep = s["y"][self.partitions.report] + scored = score_batch(self.artifact, s["X"][self.partitions.report]) + art_raw, art_pd = scored["raw_micro"], scored["pd"] + ceil = s["ceiling_report"] + floor_scores, floor_gini = s["floor_report"] + + def block(raw, prob): + return { + "gini": gini(y_rep, raw), + "ks": ks_statistic(y_rep, raw), + "brier": brier(y_rep, prob) if prob is not None else None, + } + + art_m = block(art_raw, art_pd) + ceil_m = ( + block(ceil, ceil) if ceil is not None else {"gini": None, "ks": None, "brier": None} + ) + floor_m = ( + block(floor_scores, floor_scores) + if floor_scores is not None + else {"gini": floor_gini, "ks": None, "brier": None} + ) + + rng = np.random.default_rng(s["seed"] + 7) + n = len(y_rep) + ret, ratio, art_g = [], [], [] + for _ in range(s["n_boot"]): + pick = rng.integers(0, n, n) + g_art = gini(y_rep[pick], art_raw[pick]) + art_g.append(g_art) + if ceil is not None: + g_ceil = gini(y_rep[pick], ceil[pick]) + ret.append(100 * g_art / g_ceil if g_ceil > 0 else float("nan")) + if floor_scores is not None: + g_floor = gini(y_rep[pick], floor_scores[pick]) + ratio.append(100 * g_art / g_floor if g_floor > 0 else float("nan")) + elif floor_gini: + ratio.append(100 * g_art / floor_gini) + + def interval(values): + values = [v for v in values if v == v] + if not values: + return None + lo, hi = np.percentile(values, [2.5, 97.5]) + return [float(lo), float(hi)] + + report = { + "rows": int(n), + "artifact": art_m, + "ceiling": ceil_m, + "floor": floor_m, + "retention_pct": ( + 100 * art_m["gini"] / ceil_m["gini"] + if ceil_m["gini"] and ceil_m["gini"] > 0 + else None + ), + "retention_ci": interval(ret), + "floor_ratio_pct": ( + 100 * art_m["gini"] / floor_m["gini"] + if floor_m["gini"] and floor_m["gini"] > 0 + else None + ), + "floor_ratio_ci": interval(ratio), + "artifact_gini_ci": interval(art_g), + "n_boot": int(s["n_boot"]), + "evaluations": 1, + } + self.report_evaluations = 1 + prov = self.artifact["metadata"]["provenance"] + prov["report"] = _plain(report) + self.artifact["artifact_hash"] = canonical_hash(self.artifact) + report["artifact_hash"] = self.artifact["artifact_hash"] + return report + + +def _plain(value): + if isinstance(value, dict): + return {str(k): _plain(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_plain(v) for v in value] + if isinstance(value, np.generic): + return value.item() + return value + + +def _column(X_raw, spec, name): + if spec is None: + return None + if isinstance(spec, str): + if not hasattr(X_raw, "columns"): + raise ValueError(f"{name} is a column name but X has no columns") + return np.asarray(X_raw[spec]) + return np.asarray(spec).reshape(-1) + + +# ---------------------------------------------------------- the entry point +def compile_selected( + X, + y, + *, + ceiling=None, + ceiling_oof=None, + ceiling_scores=None, + ceiling_budget: dict | None = None, + reference="woe", + partitions="stratified", + date=None, + group=None, + row_id=None, + grid: dict | None = None, + monotone_constraints=None, + selection_metric: str = "gini", + tie_se: float = 1.0, + n_boot: int = 200, + k_folds: int = 5, + min_select_events: int = 300, + n_bands: int = 10, + feature_names=None, + reasons: dict | None = None, + sample_weight=None, + allow_below_floor: bool = False, + eps_warn: float = 10.0, + duplicate_threshold: float = 0.01, + backend: str = "hist", + learning_rate: float = 0.2, + seed: int = 42, + build_kwargs: dict | None = None, +) -> SelectionResult: + """Compile the whitebox whose target and configuration were chosen on Select. + + Args: + X, y: All rows and binary outcomes; the function partitions them. + ceiling: A callable ``(X_fit, y_fit) -> fitted model``. It is called + once, its configuration is frozen, and that configuration is + refit ``k_folds`` times for cross-fitted soft targets. One search + plus K fits — never K searches. + ceiling_oof, ceiling_scores: For users who can only supply + predictions: out-of-fold predictions used as soft targets, and + the ceiling's scores used for retention, both full-length. + Provenance then records ``cross_fitted: "user_asserted"``. + ceiling_budget: Free-form dict recorded in provenance — trials, CV + scheme, search space — so retention is comparable across artifacts. + reference: ``"woe"`` fits the WoE logistic floor inside Fit; a float + is a champion scorecard's Gini; ``None`` skips the floor gate, + with a warning. + partitions: ``"stratified"``, or an explicit dict of index arrays. + date, group, row_id: Arrays (or column names, when X has columns). + ``date`` makes Report the latest slice; ``group`` keeps one + applicant in one partition; ``row_id`` turns partition overlap + into a hard failure. + grid: See :func:`default_grid`. + selection_metric: ``"gini"`` (default) or ``"brier"``. + tie_se: Width of the tie band in standard errors. + min_select_events: Below this many events in Select, selection uses + nested cross-validation over Fit ∪ Select instead, with a warning. + allow_below_floor: Build the artifact even if the selected + configuration does not beat the floor on Select. + eps_warn: Provisional events-per-split threshold below which the + grid should hold interior α values; a diagnostic, never an input. + duplicate_threshold: Maximum share of Select and Report rows whose + content also appears in an earlier partition. + backend: ``"hist"`` (the fast histogram backend) or ``"gbr"``. + + Returns: + A :class:`SelectionResult`. Call ``.report()`` once for the figure. + """ + X_raw = X + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2-D array of rows") + y_arr = np.asarray(y).reshape(-1) + if len(y_arr) != X_arr.shape[0]: + raise ValueError("X and y must have the same number of rows") + if not np.isin(y_arr, [0, 1]).all(): + raise ValueError("y must contain only 0 and 1") + y_arr = y_arr.astype(int) + names = ( + [str(n) for n in feature_names] + if feature_names is not None + else ( + [str(c) for c in X_raw.columns] + if hasattr(X_raw, "columns") + else [f"f{i}" for i in range(X_arr.shape[1])] + ) + ) + if len(names) != X_arr.shape[1]: + raise ValueError("feature_names length does not match X") + notes: list[str] = [] + + # ---------------------------------------------------------- partitions + date_arr, group_arr, row_ids = ( + _column(X_raw, v, n) for v, n in ((date, "date"), (group, "group"), (row_id, "row_id")) + ) + if isinstance(partitions, dict): + parts = make_partitions(y_arr, explicit=partitions, seed=seed) + elif partitions == "stratified": + parts = make_partitions(y_arr, date=date_arr, group=group_arr, seed=seed) + else: + raise ValueError("partitions must be 'stratified' or a dict of index arrays") + if row_ids is not None: + overlap = 0 + for a, b in ( + (parts.fit, parts.select), + (parts.fit, parts.report), + (parts.select, parts.report), + ): + overlap += int(np.intersect1d(row_ids[a], row_ids[b]).size) + if overlap: + raise ValueError(f"{overlap} row id(s) appear in more than one partition") + hashes = row_content_hashes(X_arr, y_arr) + duplicates = duplicate_rows(hashes, parts) + dup_fraction = duplicates / (len(parts.select) + len(parts.report)) + if dup_fraction > duplicate_threshold: + raise ValueError( + f"{duplicates} rows in Select or Report ({dup_fraction:.1%}) have content " + "identical to a " + f"row in an earlier partition, above duplicate_threshold={duplicate_threshold}. " + "Repeat applicants? Pass group= so each lands in one partition." + ) + + # ------------------------------------------------------------- ceiling + if ceiling is not None and (ceiling_oof is not None or ceiling_scores is not None): + raise ValueError( + "pass either ceiling= (a factory) or ceiling_oof=/ceiling_scores=, not both" + ) + if ceiling is not None: + ceil = _Ceiling(kind="factory", factory=ceiling, budget=ceiling_budget) + elif ceiling_oof is not None: + oof = np.asarray(ceiling_oof, dtype=float).reshape(-1) + scores = np.asarray( + ceiling_scores if ceiling_scores is not None else ceiling_oof, dtype=float + ).reshape(-1) + if len(oof) != len(y_arr) or len(scores) != len(y_arr): + raise ValueError("ceiling_oof and ceiling_scores must be full-length") + ceil = _Ceiling(kind="predictions", oof=oof, scores=scores, budget=ceiling_budget) + else: + ceil = None + notes.append("no ceiling: retention cannot be reported; the floor is the only yardstick") + + resolved, grid_notes = _resolve_grid(grid, monotone_constraints, ceil is not None) + notes.extend(grid_notes) + + # --------------------------------------------------------------- floor + if reference is None: + warnings.warn( + "no reference floor: the artifact cannot be compared with what a risk team " + "builds in an afternoon. Pass reference='woe' or a champion scorecard's Gini.", + stacklevel=2, + ) + floor_is_gini = isinstance(reference, (int, float)) and not isinstance(reference, bool) + + # ----------------------------------------------------------- selection + select_events = int(y_arr[parts.select].sum()) + nested = select_events < min_select_events + if nested: + warnings.warn( + f"Select holds {select_events} events, below min_select_events={min_select_events}: " + "selecting with nested cross-validation over Fit ∪ Select instead. Report stays " + "held out.", + stacklevel=2, + ) + sw = None if sample_weight is None else np.asarray(sample_weight, dtype=float).reshape(-1) + + def run_split(train_idx, eval_idx): + oof_train = None + if ceil is not None and any(a < 1.0 for a in resolved["alpha"]): + if ceil.kind == "factory" and (ceil.frozen is None or nested): + ceil.tune(X_arr, y_arr, train_idx) + oof_train = ceil.out_of_fold(X_arr, y_arr, train_idx, k_folds, seed) + elif ceil is not None and ceil.kind == "factory" and ceil.frozen is None: + ceil.tune(X_arr, y_arr, train_idx) + rows = evaluate_grid( + X_arr, + y_arr, + train_idx, + eval_idx, + grid=resolved, + oof_train=oof_train, + feature_names=names, + n_bands=n_bands, + backend=backend, + learning_rate=learning_rate, + sample_weight=sw, + seed=seed, + ) + floor_scores = None + if reference is not None and not floor_is_gini: + ref = fit_reference( + X_arr[train_idx], y_arr[train_idx], feature_names=names, random_state=seed + ) + floor_scores = ref.score(X_arr[eval_idx]) + return rows, floor_scores + + if not nested: + rows, floor_select = run_split(parts.fit, parts.select) + y_eval = y_arr[parts.select] + else: + dev = parts.dev + folds = StratifiedKFold(n_splits=k_folds, shuffle=True, random_state=seed) + pooled: dict[tuple, dict] = {} + floor_parts, y_parts = [], [] + for train_pos, held_pos in folds.split(dev, y_arr[dev]): + fold_rows, fold_floor = run_split(dev[train_pos], dev[held_pos]) + y_parts.append(y_arr[dev[held_pos]]) + if fold_floor is not None: + floor_parts.append(fold_floor) + for key, row in fold_rows.items(): + agg = pooled.setdefault( + key, + { + **{k: v for k, v in row.items() if not k.startswith("_")}, + "_raw": [], + "_pd": [], + }, + ) + agg["_raw"].append(row["_raw"]) + agg["_pd"].append(row["_pd"]) + rows = { + k: {**r, "_raw": np.concatenate(r["_raw"]), "_pd": np.concatenate(r["_pd"])} + for k, r in pooled.items() + } + y_eval = np.concatenate(y_parts) + floor_select = np.concatenate(floor_parts) if floor_parts else None + + selected = choose( + rows, y_eval, metric=selection_metric, tie_se=tie_se, n_boot=n_boot, seed=seed + ) + chosen = rows[selected["key"]] + eps_min = min(r["eps"] for r in rows.values()) + if eps_min < eps_warn and set(resolved["alpha"]) <= {0.0, 1.0} and ceil is not None: + warnings.warn( + f"minimum events per split in the grid is {eps_min:.1f}, below eps_warn={eps_warn}, " + "and the " + "alpha grid holds only endpoints; interior values (a blend of labels and ceiling " + "predictions) are worth searching at this event count. The threshold is provisional.", + stacklevel=2, + ) + + # ---------------------------------------------------------- floor gate + if reference is None: + floor_gini_select = None + elif floor_is_gini: + floor_gini_select = float(reference) + else: + floor_gini_select = gini(y_eval, floor_select) + clears = floor_gini_select is None or chosen["gini"] >= floor_gini_select + floor_gate = { + "passed": bool(clears), + "select_gini": chosen["gini"], + "floor_gini": floor_gini_select, + "enforced": bool(reference is not None and not allow_below_floor), + } + + curve = sorted( + ({k: v for k, v in r.items() if not k.startswith("_")} for r in rows.values()), + key=lambda r: (-r["score"], r["n_estimators"], r["max_depth"]), + ) + for r in curve: + r.pop("score", None) + + result_common = dict( + selection_curve=curve, + selected={ + "alpha": chosen["alpha"], + "n_estimators": chosen["n_estimators"], + "max_depth": chosen["max_depth"], + "monotone": chosen["monotone"], + "select_gini": chosen["gini"], + "select_brier": chosen["brier"], + "eps": chosen["eps"], + **{k: v for k, v in selected.items() if k not in ("key", "best_key")}, + "nested": nested, + }, + partitions=parts, + floor_gate=floor_gate, + notes=notes, + ) + if not clears and not allow_below_floor: + return SelectionResult(artifact=None, **result_common) + + # --------------------------------------------------------- final refit + dev = parts.dev + alpha = chosen["alpha"] + mono = monotone_constraints if chosen["monotone"] else None + cross_fitted: object = None + if alpha < 1.0: + if ceil.kind == "factory" and (ceil.frozen is None or nested): + ceil.tune(X_arr, y_arr, dev) + oof_dev = ceil.out_of_fold(X_arr, y_arr, dev, k_folds, seed) + target = alpha * y_arr[dev] + (1.0 - alpha) * oof_dev + cross_fitted = True if ceil.kind == "factory" else "user_asserted" + else: + target = y_arr[dev].astype(float) + if ceil is not None and ceil.kind == "factory" and (ceil.frozen is None or nested): + ceil.tune(X_arr, y_arr, dev) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + model, _ = train_whitebox( + X_arr[dev], + target, + n_estimators=chosen["n_estimators"], + max_depth=chosen["max_depth"], + learning_rate=learning_rate, + random_state=seed, + monotone_constraints=mono, + sample_weight=None if sw is None else sw[dev], + backend=backend, + ) + latent = np.clip(model.predict(X_arr[dev]), 0.0, 1.0) + part_hashes = partition_hashes(hashes, parts) + provenance = { + "target": {"alpha": alpha, "uses_ceiling": alpha < 1.0, "loss": "squared_error"}, + "selection": { + "metric": selection_metric, + "rule": "one_se_simplest", + "tie_se": tie_se, + "se": selected["se"], + "n_tied": selected["n_tied"], + "n_configs": len(rows), + "n_boot": n_boot, + "nested": nested, + "k_folds": k_folds if nested else None, + "tie_rule": "prefer alpha=1, then fewer trees, then lower depth — a governance prior", + }, + "partitions": { + "method": parts.method + ("_nested" if nested else ""), + "fit_rows": int(len(parts.fit)), + "select_rows": int(len(parts.select)), + "report_rows": int(len(parts.report)), + "content_hashes": part_hashes, + "duplicate_rows": duplicates, + "duplicate_fraction": dup_fraction, + "duplicate_threshold": duplicate_threshold, + "row_id_overlap": 0 if row_ids is not None else None, + }, + "soft_targets": {"cross_fitted": cross_fitted, "k": k_folds if alpha < 1.0 else None}, + "ceiling": ( + None + if ceil is None + else { + "kind": ceil.kind, + "family": ceil.family, + "params_hash": params_hash(ceil.frozen) if ceil.kind == "factory" else None, + "budget": ceil.budget, + "factory_calls": ceil.factory_calls, + "fits": ceil.fits, + } + ), + "floor": ( + None + if reference is None + else { + "kind": "gini" if floor_is_gini else "woe_logit", + "select_gini": floor_gini_select, + } + ), + "eps_min": eps_min, + "eps_warn": eps_warn, + "report": None, + } + build_kwargs = dict(build_kwargs or {}) + metadata = {**build_kwargs.pop("metadata", {}), "provenance": _plain(provenance)} + artifact = build_artifact( + model, + names, + np.median(X_arr[dev], axis=0), + monotone_quantile_bands(latent, y_arr[dev], n_bands=n_bands), + calibration_latent=latent, + calibration_y=y_arr[dev], + monotone_constraints=mono, + reasons=reasons, + metadata=metadata, + X_sample=X_arr[dev][:500], + **build_kwargs, + ) + + ceiling_report = ceil.score(X_arr, y_arr, dev, parts.report) if ceil is not None else None + if reference is None: + floor_report = (None, None) + elif floor_is_gini: + floor_report = (None, float(reference)) + else: + ref = fit_reference(X_arr[dev], y_arr[dev], feature_names=names, random_state=seed) + floor_report = (ref.score(X_arr[parts.report]), None) + if ceil is not None: + artifact["metadata"]["provenance"]["ceiling"]["fits"] = ceil.fits + artifact["artifact_hash"] = canonical_hash(artifact) + + return SelectionResult( + artifact=artifact, + _state={ + "X": X_arr, + "y": y_arr, + "ceiling_report": ceiling_report, + "floor_report": floor_report, + "n_boot": n_boot, + "seed": seed, + }, + **result_common, + ) diff --git a/src/compileml/tune/sweeps.py b/src/compileml/tune/sweeps.py index caecb08..1fcb72c 100644 --- a/src/compileml/tune/sweeps.py +++ b/src/compileml/tune/sweeps.py @@ -48,7 +48,7 @@ def sweep_whitebox( *, trees_grid=(10, 20, 40, 80, 160), depth_grid=(1, 2, 3), - alpha_grid=(0.0,), + alpha_grid=None, learning_rate: float = 0.2, random_state: int = 42, X_val=None, @@ -102,6 +102,24 @@ def sweep_whitebox( weighted sweep beside an unweighted one shows what moving the budget buys one segment and costs the others. """ + if alpha_grid is None: + warnings.warn( + "sweep_whitebox defaulted to pure distillation, alpha_grid=(0.0,), which real " + "data has falsified: a whitebox trained on labels has beaten the distilled one. " + "Pass alpha_grid explicitly (at least the endpoints (0.0, 1.0)), or use " + "compile_selected, which selects alpha on data it does not report on. The " + "default becomes (0.0, 0.5, 1.0) in 1.0.", + FutureWarning, + stacklevel=2, + ) + alpha_grid = (0.0,) + if X_val is not None: + warnings.warn( + "one holdout used both to choose a configuration and to report its retention " + "leaks: the figure chosen on it is optimistic. Keep it for choosing, and report " + "on rows never used to choose. compile_selected keeps Select and Report apart.", + stacklevel=2, + ) X_arr = np.asarray(X, dtype=float) y_arr = np.asarray(y, dtype=int).reshape(-1) has_teacher = teacher_latent is not None diff --git a/src/compileml/validate/framework.py b/src/compileml/validate/framework.py index 280a84a..94a9c83 100644 --- a/src/compileml/validate/framework.py +++ b/src/compileml/validate/framework.py @@ -25,6 +25,9 @@ 9. monotone_constraints declared directions re-verified against the shipped integer trees (skipped when the artifact declares none) + 11. selection_hygiene the provenance block, when present, records a + protocol that was followed: cross-fitted soft + targets, disjoint partitions, one Report read 10. reference_floor the artifact out-scores a reference model on the same data — the floor that teacher retention alone can never supply @@ -75,6 +78,7 @@ def validate_artifact( require_full_reason_coverage: bool = False, reference=None, require_reference_floor: bool = False, + require_selection_hygiene: bool = False, seed: int = 42, ) -> dict: """Run the ten-check framework. Checks lacking inputs skip, not fail. @@ -97,6 +101,12 @@ def validate_artifact( require_reference_floor: Make check 10 *fail* when the artifact does not clear the reference, rather than recording it as evidence (the ``max_within_band_auc`` precedent). + require_selection_hygiene: Make check 11 *fail* on a provenance + block that records a broken protocol — soft targets that were not + cross-fitted, partitions that overlap, Report read more than once + — rather than listing the issues as evidence. Check 11 skips when + the artifact carries no provenance block; it can verify what the + block records, not that the block is honest. Returns: {"all_pass": bool, "checks": {name: {"pass", "skipped", ...evidence}}} @@ -371,4 +381,65 @@ def bootstrap_cutoffs(): else: checks["10_reference_floor"] = {"pass": True, "skipped": True} + # ------------------------------------------------ 11 selection hygiene + provenance = (artifact.get("metadata") or {}).get("provenance") + if provenance: + checks["11_selection_hygiene"] = _selection_hygiene(provenance, require_selection_hygiene) + else: + checks["11_selection_hygiene"] = {"pass": True, "skipped": True} + return {"all_pass": all(c["pass"] for c in checks.values()), "checks": checks} + + +def _selection_hygiene(provenance: dict, required: bool) -> dict: + """Check 11: does the provenance block record a protocol that was followed? + + Issues fail the check when it is required; notes never do. The check reads + what ``compile_selected`` wrote and cannot verify it against data it does + not have, which is why it is advisory by default. + """ + issues: list[str] = [] + notes: list[str] = [] + target = provenance.get("target") or {} + alpha = float(target.get("alpha", 1.0)) + cross_fitted = (provenance.get("soft_targets") or {}).get("cross_fitted") + if alpha < 1.0: + if cross_fitted is True: + pass + elif cross_fitted == "user_asserted": + notes.append("soft-target cross-fitting is asserted by the user, not verified") + else: + issues.append( + "soft targets were not cross-fitted (alpha < 1 with in-sample ceiling predictions)" + ) + + parts = provenance.get("partitions") or {} + hashes = parts.get("content_hashes") or {} + if hashes and len(set(hashes.values())) != len(hashes): + issues.append("partition content hashes are not distinct") + if parts.get("row_id_overlap"): + issues.append(f"{parts['row_id_overlap']} row id(s) appear in more than one partition") + dup, threshold = parts.get("duplicate_fraction"), parts.get("duplicate_threshold") + if dup is not None and threshold is not None and dup > threshold: + issues.append(f"duplicate fraction {dup:.4f} exceeds threshold {threshold}") + + report = provenance.get("report") + evaluations = report.get("evaluations") if report else None + if report is None: + issues.append("Report has not been evaluated: the artifact carries no reported figure") + elif evaluations != 1: + issues.append(f"Report was evaluated {evaluations} times; the protocol allows one") + elif parts.get("report_rows") is not None and report.get("rows") != parts.get("report_rows"): + issues.append("the reported row count does not match the Report partition") + + return { + "pass": bool(not issues or not required), + "skipped": False, + "required": bool(required), + "alpha": alpha, + "cross_fitted": cross_fitted, + "report_evaluations": evaluations, + "duplicate_fraction": dup, + "issues": issues, + "notes": notes, + } diff --git a/tests/test_batch.py b/tests/test_batch.py new file mode 100644 index 0000000..cfe63c8 --- /dev/null +++ b/tests/test_batch.py @@ -0,0 +1,64 @@ +"""The batch scorer is the runtime's score path, vectorised: same integers, every row.""" + +import copy + +import numpy as np +import pytest + +from compileml.batch import score_batch +from compileml.runtime import ArtifactError, decide, load_artifact + +FIELDS = ("raw_micro", "latent_micro", "latent_int", "band_idx", "pd_ppm") + + +@pytest.fixture(scope="module") +def artifact(): + return load_artifact("tests/data/reference_artifact.json") + + +@pytest.fixture(scope="module") +def rows(artifact): + """Ordinary rows, missing values, and values far enough out to hit both clamps.""" + p = len(artifact["features"]["names"]) + rng = np.random.default_rng(0) + X = rng.standard_normal((2000, p)) * 2 + X[::7, 0] = np.nan + X[::11, 3] = 1e6 + X[::13, 4] = -1e6 + return X + + +def _agrees(art, X): + batch = score_batch(art, X) + for i, row in enumerate(X): + d = decide(art, [None if v != v else float(v) for v in row], explain=False) + assert tuple(d[f] for f in FIELDS) == tuple(int(batch[f][i]) for f in FIELDS), i + + +def test_bit_identical_to_decide(artifact, rows): + _agrees(artifact, rows) + + +@pytest.mark.parametrize("variant", ["step", "none", "float32"]) +def test_bit_identical_across_calibration_modes_and_precision(artifact, rows, variant): + art = copy.deepcopy(artifact) + if variant == "step": + art["calibration"]["mode"] = "step" + elif variant == "none": + art["calibration"] = None + else: + art["model"]["input_precision"] = "float32" + _agrees(art, rows) + + +def test_reject_policy_refuses_missing_values(artifact, rows): + art = copy.deepcopy(artifact) + art["features"]["missing_policy"] = "reject" + with pytest.raises(ArtifactError, match="reject"): + score_batch(art, rows) + score_batch(art, np.nan_to_num(rows)) # no missing values: fine + + +def test_wrong_width_is_refused(artifact, rows): + with pytest.raises(ArtifactError, match="feature columns"): + score_batch(artifact, rows[:, :-1]) diff --git a/tests/test_select.py b/tests/test_select.py new file mode 100644 index 0000000..68ad2c6 --- /dev/null +++ b/tests/test_select.py @@ -0,0 +1,447 @@ +"""compile_selected: the protocol, and the properties #75 promises. + +The fixture data is small and the grids are short so the suite stays fast; +the properties do not depend on scale. +""" + +import copy +import warnings + +import numpy as np +import pytest +from sklearn.base import BaseEstimator, ClassifierMixin +from sklearn.ensemble import HistGradientBoostingClassifier + +import compileml +from compileml.artifact import build_artifact +from compileml.bands import monotone_quantile_bands +from compileml.batch import score_batch +from compileml.compile import extract_trees, train_whitebox +from compileml.runtime import canonical_hash, verify_artifact +from compileml.select import ( + choose, + compile_selected, + duplicate_rows, + make_partitions, + row_content_hashes, +) +from compileml.tune import sweep_whitebox +from compileml.validate import validate_artifact + +N, P = 4000, 6 +FEATURES = [f"f{i}" for i in range(P)] +SMALL_GRID = {"alpha": (0.0, 1.0), "trees": (20, 40), "depth": (1, 2)} + + +class CountingCeiling(ClassifierMixin, BaseEstimator): + """A ceiling that counts its fits, so the protocol's fit budget is testable.""" + + fits = 0 + + def __init__(self, max_iter=60): + self.max_iter = max_iter + + def fit(self, X, y): + CountingCeiling.fits += 1 + self.model_ = HistGradientBoostingClassifier( + max_iter=self.max_iter, max_depth=3, learning_rate=0.1, random_state=0 + ).fit(X, y) + self.classes_ = self.model_.classes_ + return self + + def predict_proba(self, X): + return self.model_.predict_proba(X) + + def predict(self, X): + return self.model_.predict(X) + + +@pytest.fixture(scope="module") +def data(): + rng = np.random.default_rng(1) + X = rng.standard_normal((N, P)) + logit = 1.2 * X[:, 0] - 0.8 * X[:, 1] + 0.7 * X[:, 2] * X[:, 3] - 1.0 + y = (rng.random(N) < 1 / (1 + np.exp(-logit))).astype(int) + return X, y + + +@pytest.fixture(scope="module") +def run(data): + """One flat run with a counting factory, shared by the tests that read it.""" + X, y = data + calls = {"factory": 0} + CountingCeiling.fits = 0 + + def factory(Xf, yf): + calls["factory"] += 1 + return CountingCeiling().fit(Xf, yf) + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + result = compile_selected( + X, + y, + ceiling=factory, + reference="woe", + grid=SMALL_GRID, + n_boot=30, + k_folds=3, + min_select_events=50, + seed=3, + ) + return result, calls + + +# ---------------------------------------------------------------- partitions +def test_partitions_are_disjoint_and_cover_every_row(data): + _, y = data + parts = make_partitions(y, seed=0) + all_idx = np.sort(np.concatenate([parts.fit, parts.select, parts.report])) + assert np.array_equal(all_idx, np.arange(N)) + assert abs(len(parts.report) / N - 0.2) < 0.01 and abs(len(parts.fit) / N - 0.6) < 0.01 + # stratified: event rates agree + rates = [y[getattr(parts, k)].mean() for k in ("fit", "select", "report")] + assert max(rates) - min(rates) < 0.02 + + +def test_out_of_time_report_is_the_latest_slice(data): + _, y = data + date = np.arange(N) # row i happened on day i + parts = make_partitions(y, date=date, seed=0) + assert parts.report.min() > max(parts.fit.max(), parts.select.max()) + assert parts.method == "out_of_time" + + +def test_group_split_keeps_each_group_in_one_partition(data): + _, y = data + group = np.arange(N) // 10 # ten rows per applicant + parts = make_partitions(y, group=group, seed=0) + seen = {} + for name in ("fit", "select", "report"): + for g in np.unique(group[getattr(parts, name)]): + assert seen.setdefault(g, name) == name, f"group {g} straddles partitions" + assert parts.method == "group" + + +def test_explicit_partitions_must_not_overlap(data): + _, y = data + with pytest.raises(ValueError, match="overlap"): + make_partitions(y, explicit={"fit": [0, 1, 2], "select": [2, 3], "report": [4]}) + with pytest.raises(ValueError, match="sum to 1"): + make_partitions(y, fractions=(0.5, 0.3, 0.3)) + + +def test_content_duplicates_are_counted_and_thresholded(data): + X, y = data + hashes = row_content_hashes(X, y) + parts = make_partitions(y, seed=0) + assert duplicate_rows(hashes, parts) == 0 + X_dup = X.copy() + X_dup[parts.report[:100]] = X[parts.fit[:100]] # report rows copied from fit + y_dup = y.copy() + y_dup[parts.report[:100]] = y[parts.fit[:100]] + assert duplicate_rows(row_content_hashes(X_dup, y_dup), parts) == 100 + with pytest.raises(ValueError, match="identical"): + compile_selected( + X_dup, + y_dup, + partitions={"fit": parts.fit, "select": parts.select, "report": parts.report}, + reference=0.1, + grid={"alpha": (1.0,), "trees": (20,), "depth": (1,)}, + n_boot=5, + min_select_events=10, + duplicate_threshold=0.01, + ) + + +def test_row_id_overlap_is_a_hard_failure(data): + X, y = data + parts = make_partitions(y, seed=0) + row_id = np.arange(N) + row_id[parts.report[0]] = row_id[parts.fit[0]] # one id in two partitions + with pytest.raises(ValueError, match="row id"): + compile_selected( + X, + y, + row_id=row_id, + partitions={"fit": parts.fit, "select": parts.select, "report": parts.report}, + reference=0.1, + grid={"alpha": (1.0,), "trees": (20,), "depth": (1,)}, + n_boot=5, + ) + + +# ------------------------------------------------------------ the prefix trick +def test_prefix_of_a_fit_is_the_smaller_fit(data): + """The first 20 trees of an 80-tree fit compile to the same artifact as a 20-tree fit.""" + X, y = data + target = y.astype(float) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + big, _ = train_whitebox(X, target, n_estimators=80, backend="hist", random_state=0) + small, _ = train_whitebox(X, target, n_estimators=20, backend="hist", random_state=0) + extracted = extract_trees(big) + prefix = copy.copy(extracted) + prefix.trees = extracted.trees[:20] + latent = np.clip(small.predict(X), 0, 1) + bands = monotone_quantile_bands(latent, y, n_bands=6) + art_prefix = build_artifact(prefix, FEATURES, np.median(X, axis=0), bands) + art_small = build_artifact(small, FEATURES, np.median(X, axis=0), bands) + assert np.array_equal( + score_batch(art_prefix, X)["raw_micro"], score_batch(art_small, X)["raw_micro"] + ) + + +# ------------------------------------------------------------------ ceiling +def test_ceiling_is_tuned_once_and_refit_frozen(run): + """One search, K fits for out-of-fold targets, one refit for Report scoring.""" + result, calls = run + k = 3 + assert calls["factory"] == 1 + prov = result.artifact["metadata"]["provenance"]["ceiling"] + assert prov["factory_calls"] == 1 + expected = 1 + k + 1 # tune, out-of-fold on Fit, refit on Fit ∪ Select for Report + if result.selected["alpha"] < 1.0: + expected += k # out-of-fold regenerated on Fit ∪ Select for the final refit + assert prov["fits"] == expected == CountingCeiling.fits + + +# ---------------------------------------------------------------- selection +def _rows(y, preds): + return { + key: { + "alpha": key[0], + "n_estimators": key[1], + "max_depth": key[2], + "monotone": False, + "eps": 50.0, + "_raw": raw, + "_pd": np.clip(raw / raw.max(), 0, 1), + } + for key, raw in preds.items() + } + + +def test_tie_rule_prefers_alpha_one_then_fewer_trees_then_lower_depth(): + rng = np.random.default_rng(0) + y = (rng.random(500) < 0.3).astype(int) + same = (y * 1000 + rng.integers(0, 200, 500)).astype(np.int64) + rows = _rows( + y, + { + (0.0, 20, 1, 0): same, + (1.0, 80, 2, 0): same, + (1.0, 20, 2, 0): same, + (1.0, 20, 1, 0): same, + }, + ) + chosen = choose(rows, y, metric="gini", tie_se=1.0, n_boot=20, seed=0) + assert chosen["key"] == (1.0, 20, 1, 0) + assert chosen["n_tied"] == 4 + + +def test_a_clearly_better_configuration_wins_regardless_of_the_prior(): + rng = np.random.default_rng(0) + y = (rng.random(500) < 0.3).astype(int) + sharp = (y * 1000 + rng.integers(0, 100, 500)).astype(np.int64) + noise = rng.integers(0, 1000, 500).astype(np.int64) + rows = _rows(y, {(0.0, 80, 2, 0): sharp, (1.0, 20, 1, 0): noise}) + chosen = choose(rows, y, metric="gini", tie_se=1.0, n_boot=20, seed=0) + assert chosen["key"] == (0.0, 80, 2, 0) + assert chosen["n_tied"] == 1 + + +def test_selection_curve_and_selected_agree(run): + result, _ = run + curve = result.selection_curve + assert len(curve) == 2 * 2 * 2 # alpha x trees x depth + assert sum(r["selected"] for r in curve) == 1 + top = next(r for r in curve if r["selected"]) + assert (top["alpha"], top["n_estimators"], top["max_depth"]) == ( + result.selected["alpha"], + result.selected["n_estimators"], + result.selected["max_depth"], + ) + assert all("_raw" not in r for r in curve) + assert all(r["eps"] > 0 for r in curve) + + +# ------------------------------------------------------ floor and no ceiling +def test_floor_gate_stops_without_an_artifact(data): + X, y = data + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + stopped = compile_selected( + X[:1500], + y[:1500], + reference=0.99, + grid={"alpha": (1.0,), "trees": (20,), "depth": (2,)}, + n_boot=5, + min_select_events=10, + seed=3, + ) + forced = compile_selected( + X[:1500], + y[:1500], + reference=0.99, + grid={"alpha": (1.0,), "trees": (20,), "depth": (2,)}, + n_boot=5, + min_select_events=10, + seed=3, + allow_below_floor=True, + ) + assert stopped.artifact is None and stopped.floor_gate["passed"] is False + with pytest.raises(RuntimeError, match="floor"): + stopped.report() + assert forced.artifact is not None + + +def test_no_ceiling_forces_labels_and_reports_no_retention(data): + X, y = data + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + result = compile_selected( + X[:1500], + y[:1500], + reference=0.2, + grid={"trees": (20,), "depth": (2,)}, + n_boot=5, + min_select_events=10, + seed=3, + ) + assert result.selected["alpha"] == 1.0 + assert any("forced" in n for n in result.notes) + report = result.report() + assert report["retention_pct"] is None and report["floor_ratio_pct"] > 0 + + +# --------------------------------------------------------- report and hash +def test_report_is_read_once_and_written_into_the_hash(run): + result, _ = run + before = result.artifact["artifact_hash"] + assert verify_artifact(result.artifact) + report = result.report() + prov = result.artifact["metadata"]["provenance"] + assert prov["report"]["evaluations"] == 1 + assert report["artifact_hash"] == result.artifact["artifact_hash"] != before + assert verify_artifact(result.artifact) + assert report["retention_ci"][0] <= report["retention_pct"] <= report["retention_ci"][1] + assert report["floor_ratio_ci"][0] <= report["floor_ratio_pct"] <= report["floor_ratio_ci"][1] + with pytest.raises(RuntimeError, match="already been evaluated"): + result.report() + + +def test_selection_never_computes_attribution(data, monkeypatch): + import compileml.runtime.explain as explain + + def boom(*args, **kwargs): + raise AssertionError("attribution was computed during selection") + + monkeypatch.setattr(explain, "contributions_half_micro", boom) + X, y = data + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + result = compile_selected( + X[:1500], + y[:1500], + reference=0.1, + grid={"alpha": (1.0,), "trees": (20,), "depth": (1,)}, + n_boot=5, + min_select_events=10, + seed=3, + ) + assert result.artifact is not None + + +# ------------------------------------------------------------------ nested +def test_small_select_falls_back_to_nested_cv(data): + X, y = data + with pytest.warns(UserWarning, match="nested"): + result = compile_selected( + X[:1200], + y[:1200], + reference=0.1, + grid={"alpha": (1.0,), "trees": (20,), "depth": (1,)}, + n_boot=5, + k_folds=3, + min_select_events=10_000, + seed=3, + ) + assert result.selected["nested"] is True + assert result.artifact["metadata"]["provenance"]["partitions"]["method"].endswith("_nested") + assert result.report()["rows"] == len(result.partitions.report) + + +# ---------------------------------------------------------------- check 11 +def test_check_11_passes_on_a_reported_artifact_and_catches_tampering(run, data): + result, _ = run + X, y = data + if result.report_evaluations == 0: + result.report() + idx = result.partitions.report + v = validate_artifact(result.artifact, X[idx], y[idx], require_selection_hygiene=True) + c11 = v["checks"]["11_selection_hygiene"] + assert c11["pass"] and not c11["skipped"] and c11["issues"] == [] + + tampered = copy.deepcopy(result.artifact) + prov = tampered["metadata"]["provenance"] + prov["report"]["evaluations"] = 2 + prov["target"]["alpha"] = 0.5 + prov["soft_targets"]["cross_fitted"] = False + tampered["artifact_hash"] = canonical_hash(tampered) + advisory = validate_artifact(tampered, X[idx], y[idx])["checks"]["11_selection_hygiene"] + assert advisory["pass"] and len(advisory["issues"]) == 2 + required = validate_artifact(tampered, X[idx], y[idx], require_selection_hygiene=True) + assert required["checks"]["11_selection_hygiene"]["pass"] is False + assert required["all_pass"] is False + + plain = copy.deepcopy(result.artifact) + del plain["metadata"]["provenance"] + plain["artifact_hash"] = canonical_hash(plain) + assert validate_artifact(plain)["checks"]["11_selection_hygiene"]["skipped"] + + +# --------------------------------------------------------- warnings and API +def test_sweep_whitebox_warns_about_its_old_defaults(data): + X, y = data + teacher = 1 / (1 + np.exp(-(X[:, 0] - 1))) + with pytest.warns(FutureWarning, match="alpha_grid"): + sweep_whitebox( + X[:800], teacher[:800], y[:800], trees_grid=(10,), depth_grid=(1,), reference=0.1 + ) + with pytest.warns(UserWarning, match="choose"): + sweep_whitebox( + X[:800], + teacher[:800], + y[:800], + trees_grid=(10,), + depth_grid=(1,), + alpha_grid=(0.0,), + X_val=X[800:1000], + y_val=y[800:1000], + teacher_latent_val=teacher[800:1000], + reference=0.1, + ) + + +def test_eps_warning_when_endpoints_only_and_events_are_scarce(data): + X, y = data + with pytest.warns(UserWarning, match="events per split"): + compile_selected( + X[:600], + y[:600], + ceiling=lambda Xf, yf: CountingCeiling(max_iter=20).fit(Xf, yf), + reference=0.1, + grid={"alpha": (0.0, 1.0), "trees": (160,), "depth": (2,)}, + n_boot=5, + k_folds=2, + min_select_events=1, + seed=3, + ) + + +def test_compile_selected_is_importable_from_the_root_lazily(): + assert compileml.compile_selected is compile_selected + missing = "no_such_thing" + with pytest.raises(AttributeError): + getattr(compileml, missing) diff --git a/tests/test_validate.py b/tests/test_validate.py index a07e423..1890ec8 100644 --- a/tests/test_validate.py +++ b/tests/test_validate.py @@ -68,7 +68,9 @@ def test_all_checks_pass(fitted, reference): reference=reference, # check 10 skips without one ) assert report["all_pass"], {k: v for k, v in report["checks"].items() if not v["pass"]} - assert not any(c["skipped"] for c in report["checks"].values()) + # Check 11 reads a provenance block that only compile_selected writes. + skipped = [k for k, c in report["checks"].items() if c["skipped"]] + assert skipped == ["11_selection_hygiene"] assert report["checks"]["8_reason_coverage"]["coverage"] == 1.0 assert report["checks"]["9_monotone_constraints"]["constrained_features"] == ["x0", "x1", "x2"]