| """Exact NumPy learner and preprocessing from the tested cached replay.""" |
| from __future__ import annotations |
| from typing import Any |
| import numpy as np |
|
|
| class PerformanceError(ValueError): |
| """A fail-closed data or algorithm contract violation.""" |
|
|
| def _impute_and_scale(train_x: np.ndarray, test_x: np.ndarray, *, scale: bool) -> tuple[np.ndarray, np.ndarray, dict[str, np.ndarray]]: |
| train_x = np.asarray(train_x, dtype=np.float64) |
| test_x = np.asarray(test_x, dtype=np.float64) |
| if train_x.ndim != 2 or test_x.ndim != 2 or train_x.shape[1] != test_x.shape[1] or not len(train_x): |
| raise PerformanceError("TRAIN_TEST_FEATURE_SHAPE_INVALID") |
| medians = np.zeros(train_x.shape[1], dtype=np.float64) |
| for column in range(train_x.shape[1]): |
| finite = train_x[np.isfinite(train_x[:, column]), column] |
| medians[column] = float(np.median(finite)) if len(finite) else 0.0 |
| train = np.where(np.isfinite(train_x), train_x, medians[None, :]) |
| test = np.where(np.isfinite(test_x), test_x, medians[None, :]) |
| means = train.mean(axis=0) if scale else np.zeros(train.shape[1], dtype=np.float64) |
| stds = train.std(axis=0) if scale else np.ones(train.shape[1], dtype=np.float64) |
| stds[stds < 1e-12] = 1.0 |
| train = (train - means[None, :]) / stds[None, :] |
| test = (test - means[None, :]) / stds[None, :] |
| if not np.isfinite(train).all() or not np.isfinite(test).all(): |
| raise PerformanceError("TRAIN_ONLY_FEATURE_TRANSFORM_NONFINITE") |
| return train, test, {"medians": medians, "means": means, "scales": stds} |
|
|
| class _HistogramTreeRegressor: |
| """Small deterministic squared-loss histogram gradient booster using NumPy.""" |
|
|
| engine = "d0403ck_b_numpy_histogram_gbr_v1" |
|
|
| def __init__(self, *, max_depth: int, max_iter: int, learning_rate: float, |
| min_samples_leaf: int, random_state: int = 0, max_bins: int = 255): |
| self.max_depth = int(max_depth) |
| self.max_iter = int(max_iter) |
| self.learning_rate = float(learning_rate) |
| self.min_samples_leaf = int(min_samples_leaf) |
| self.random_state = int(random_state) |
| self.max_bins = int(max_bins) |
| self.thresholds: list[np.ndarray] = [] |
| self.trees: list[Any] = [] |
| self.initial_value = 0.0 |
|
|
| def _bin_matrix(self, x: np.ndarray) -> np.ndarray: |
| return np.column_stack([ |
| np.searchsorted(self.thresholds[column], x[:, column], side="right").astype(np.int16) |
| for column in range(x.shape[1]) |
| ]) |
|
|
| def _build_tree(self, bins: np.ndarray, residual: np.ndarray, indices: np.ndarray, depth: int) -> Any: |
| value = float(residual[indices].mean()) if len(indices) else 0.0 |
| if depth >= self.max_depth or len(indices) < 2 * self.min_samples_leaf: |
| return {"value": value} |
| best_gain = 0.0 |
| best: tuple[int, int] | None = None |
| for feature in range(bins.shape[1]): |
| bin_values = bins[indices, feature] |
| bin_count = int(self.thresholds[feature].size + 1) |
| counts = np.bincount(bin_values, minlength=bin_count).astype(np.int64) |
| sums = np.bincount(bin_values, weights=residual[indices], minlength=bin_count) |
| left_count = np.cumsum(counts)[:-1] |
| if not len(left_count): |
| continue |
| right_count = len(indices) - left_count |
| valid = (left_count >= self.min_samples_leaf) & (right_count >= self.min_samples_leaf) |
| if not np.any(valid): |
| continue |
| left_sum = np.cumsum(sums)[:-1] |
| total_sum = float(sums.sum()) |
| gain = np.zeros(len(left_count), dtype=np.float64) |
| gain[valid] = ( |
| (left_sum[valid] ** 2) / left_count[valid] |
| + ((total_sum - left_sum[valid]) ** 2) / right_count[valid] |
| ) |
| threshold_bin = int(np.argmax(gain)) |
| candidate_gain = float(gain[threshold_bin]) |
| if candidate_gain > best_gain + 1e-12: |
| best_gain = candidate_gain |
| best = (feature, threshold_bin) |
| if best is None: |
| return {"value": value} |
| feature, threshold_bin = best |
| left_mask = bins[indices, feature] <= threshold_bin |
| left = indices[left_mask] |
| right = indices[~left_mask] |
| if len(left) < self.min_samples_leaf or len(right) < self.min_samples_leaf: |
| return {"value": value} |
| return { |
| "feature": feature, |
| "threshold_bin": threshold_bin, |
| "left": self._build_tree(bins, residual, left, depth + 1), |
| "right": self._build_tree(bins, residual, right, depth + 1), |
| } |
|
|
| def _predict_tree(self, bins: np.ndarray, tree: Any) -> np.ndarray: |
| output = np.zeros(len(bins), dtype=np.float64) |
|
|
| def apply(node: Any, indices: np.ndarray) -> None: |
| if not len(indices): |
| return |
| if "value" in node: |
| output[indices] = float(node["value"]) |
| return |
| feature = int(node["feature"]) |
| left_mask = bins[indices, feature] <= int(node["threshold_bin"]) |
| apply(node["left"], indices[left_mask]) |
| apply(node["right"], indices[~left_mask]) |
|
|
| apply(tree, np.arange(len(bins), dtype=np.int64)) |
| return output |
|
|
| def fit(self, x: np.ndarray, y: np.ndarray) -> "_HistogramTreeRegressor": |
| x = np.asarray(x, dtype=np.float64) |
| y = np.asarray(y, dtype=np.float64) |
| if x.ndim != 2 or y.shape != (len(x),) or not len(x) or not np.isfinite(x).all() or not np.isfinite(y).all(): |
| raise PerformanceError("GBT_TRAINING_ARRAY_INVALID") |
| self.thresholds = [] |
| for column in range(x.shape[1]): |
| unique = np.unique(x[:, column]) |
| if len(unique) <= self.max_bins: |
| cuts = (unique[:-1] + unique[1:]) / 2.0 |
| else: |
| quantiles = np.linspace(0.0, 1.0, self.max_bins + 1)[1:-1] |
| cuts = np.unique(np.quantile(x[:, column], quantiles)) |
| self.thresholds.append(np.asarray(cuts, dtype=np.float64)) |
| bins = self._bin_matrix(x) |
| self.initial_value = float(y.mean()) |
| prediction = np.full(len(y), self.initial_value, dtype=np.float64) |
| self.trees = [] |
| all_indices = np.arange(len(y), dtype=np.int64) |
| for _ in range(self.max_iter): |
| residual = y - prediction |
| tree = self._build_tree(bins, residual, all_indices, 0) |
| self.trees.append(tree) |
| prediction += self.learning_rate * self._predict_tree(bins, tree) |
| return self |
|
|
| def predict(self, x: np.ndarray) -> np.ndarray: |
| x = np.asarray(x, dtype=np.float64) |
| if x.ndim != 2 or x.shape[1] != len(self.thresholds) or not np.isfinite(x).all(): |
| raise PerformanceError("GBT_PREDICTION_ARRAY_INVALID") |
| bins = self._bin_matrix(x) |
| prediction = np.full(len(x), self.initial_value, dtype=np.float64) |
| for tree in self.trees: |
| prediction += self.learning_rate * self._predict_tree(bins, tree) |
| return prediction |
|
|
