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"]