File size: 7,178 Bytes
08bd580 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """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
|