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