从零实现 CART 回归树
实验目标与环境
把上一节的阈值表变成一个能训练、能预测的对象。待验证命题是:同一套平方误差规则应当同时解释手算答案、递归代码和成熟库在受控样本上的结果。
核心实验使用 Python 3.9+ 标准库,无网络、无输出文件,四行输入来自上一节。第三方对照固定 Python 3.12 和 scikit-learn 1.6.1,使用后文的独立运行命令。2026-09-07 的实际结果见实验记录。
1. 先规定接口和状态
fit(X,y) 生成树,predict(X) 只沿树查询。节点保存:
| 字段 | 含义 |
|---|---|
value、count | 当前区域的均值、样本数 |
feature、threshold | 内部节点采用的特征列和阈值 |
gain | 这次切分减少的 SSE |
left、right | 两个子节点 |
feature is None 表示叶子。所有节点先按叶子建立,只有找到合法且严格改善的切分,才增加孩子。这样停止分支自然保留一个合法预测。
根节点深度为 0:max_depth=0 只有根叶,max_depth=1 最多问一次问题。深度是路径长度,不是叶子数量。
2. 自己写之前,先描述算法
build(当前样本索引, 当前深度):
计算均值和父节点 SSE
如果深度或样本数不允许切分,返回叶子
遍历每一列:
对当前样本的该列数值排序并去重
遍历相邻值中点:
按 <= 和 > 分成左右索引
任一边不足 min_samples_leaf,跳过
计算父 SSE - 左 SSE - 右 SSE
如果严格超过当前最好增益,记住候选
如果没有候选,返回叶子
否则递归建立两个孩子,返回内部节点同增益时保留最先遇到的候选,便于复现。这里直接重算每个候选的误差,一节点最坏约为
3. 阅读并运行完整源码
以下是仓库中实际运行的源码;本节重点读到 CARTRegressor.predict 为止。SquaredGBDT 留到后两节。
"""Readable numeric CART and squared-loss GBDT; no third-party dependencies.
This is an exact-search teaching implementation, not a LightGBM replacement.
Run from the repository root: python3 examples/python/tree_boosting.py
"""
from dataclasses import dataclass
from math import isfinite
X_TOY = [[1.0], [2.0], [3.0], [4.0]]
Y_TOY = [1.0, 1.0, 3.0, 3.0]
def mean(values):
return sum(values) / len(values)
def sse(values):
center = mean(values)
return sum((value - center) ** 2 for value in values)
def mse(actual, predicted):
if not actual or len(actual) != len(predicted):
raise ValueError("MSE requires equal, nonempty sequences")
return mean([(a - p) ** 2 for a, p in zip(actual, predicted)])
def validate(X, y):
if not X or len(X) != len(y) or not X[0]:
raise ValueError("X and y must have matching nonzero sample counts")
width = len(X[0])
if any(len(row) != width for row in X):
raise ValueError("X must be rectangular")
if any(not isfinite(v) for row in X for v in row) or any(not isfinite(v) for v in y):
raise ValueError("This teaching tree accepts only finite numeric values")
@dataclass
class Node:
value: float
count: int
feature: object = None
threshold: object = None
gain: float = 0.0
left: object = None
right: object = None
class CARTRegressor:
def __init__(self, max_depth=2, min_samples_leaf=1, min_gain=0.0):
if not isinstance(max_depth, int) or max_depth < 0:
raise ValueError("max_depth must be a nonnegative integer")
if not isinstance(min_samples_leaf, int) or min_samples_leaf < 1:
raise ValueError("min_samples_leaf must be a positive integer")
if not isfinite(min_gain) or min_gain < 0:
raise ValueError("min_gain must be finite and nonnegative")
self.max_depth = max_depth
self.min_samples_leaf = min_samples_leaf
self.min_gain = min_gain
self.root = None
def fit(self, X, y):
validate(X, y)
self.n_features = len(X[0])
def build(indices, depth):
labels = [y[i] for i in indices]
node = Node(mean(labels), len(indices))
parent_sse = sse(labels)
if depth >= self.max_depth or len(indices) < 2 * self.min_samples_leaf:
return node
best = None
best_gain = self.min_gain
for feature in range(self.n_features):
values = sorted({X[i][feature] for i in indices})
for low, high in zip(values, values[1:]):
threshold = low / 2 + high / 2
left = [i for i in indices if X[i][feature] <= threshold]
right = [i for i in indices if X[i][feature] > threshold]
if min(len(left), len(right)) < self.min_samples_leaf:
continue
gain = parent_sse - sse([y[i] for i in left]) - sse([y[i] for i in right])
# Strict improvement: deterministic first candidate wins ties.
if gain > best_gain:
best_gain = gain
best = feature, threshold, left, right
if best is not None:
feature, threshold, left, right = best
node.feature, node.threshold, node.gain = feature, threshold, best_gain
node.left, node.right = build(left, depth + 1), build(right, depth + 1)
return node
self.root = build(list(range(len(y))), 0)
return self
def predict(self, X):
if self.root is None:
raise ValueError("Call fit before predict")
if any(len(row) != self.n_features or any(not isfinite(v) for v in row) for row in X):
raise ValueError("Prediction features must match training width and be finite")
result = []
for row in X:
node = self.root
while node.feature is not None:
node = node.left if row[node.feature] <= node.threshold else node.right
result.append(node.value)
return result
class SquaredGBDT:
def __init__(self, n_estimators=2, learning_rate=0.5, max_depth=1, min_samples_leaf=1):
if not isinstance(n_estimators, int) or n_estimators < 1:
raise ValueError("n_estimators must be positive")
if not isfinite(learning_rate) or not 0 < learning_rate <= 1:
raise ValueError("Teaching implementation requires 0 < learning_rate <= 1")
self.n_estimators = n_estimators
self.learning_rate = learning_rate
self.max_depth = max_depth
self.min_samples_leaf = min_samples_leaf
self.trees = []
self.base = None
def fit(self, X, y):
validate(X, y)
self.base, self.trees = mean(y), []
prediction = [self.base] * len(y)
self.train_mse = [mse(y, prediction)]
self.trace = []
for _ in range(self.n_estimators):
residual = [a - p for a, p in zip(y, prediction)]
tree = CARTRegressor(self.max_depth, self.min_samples_leaf).fit(X, residual)
correction = tree.predict(X)
prediction = [p + self.learning_rate * c for p, c in zip(prediction, correction)]
self.trees.append(tree)
self.train_mse.append(mse(y, prediction))
self.trace.append({"residual": residual, "correction": correction, "prediction": prediction[:]})
return self
def predict(self, X):
if self.base is None:
raise ValueError("Call fit before predict")
prediction = [self.base] * len(X)
for tree in self.trees:
prediction = [p + self.learning_rate * c for p, c in zip(prediction, tree.predict(X))]
return prediction
def self_check():
tree = CARTRegressor(max_depth=1).fit(X_TOY, Y_TOY)
assert tree.root.threshold == 2.5 and tree.root.gain == 4.0
assert tree.predict(X_TOY) == Y_TOY
assert tree.predict([[0.0], [2.5], [5.0]]) == [1.0, 1.0, 3.0]
assert CARTRegressor(max_depth=0).fit(X_TOY, Y_TOY).predict(X_TOY) == [2.0] * 4
assert CARTRegressor(min_samples_leaf=3).fit(X_TOY, Y_TOY).root.feature is None
assert CARTRegressor().fit(X_TOY, [7.0] * 4).root.feature is None
assert CARTRegressor().fit([[1.0]] * 4, Y_TOY).root.feature is None
# Changed labels must produce changed learned values, not hard-coded toy answers.
changed = [2 * y + 5 for y in Y_TOY]
assert CARTRegressor(1).fit(X_TOY, changed).predict(X_TOY) == changed
duplicates = CARTRegressor(1).fit([[1.0], [1.0], [2.0], [2.0]], Y_TOY)
assert duplicates.root.threshold == 1.5
multi = CARTRegressor(1).fit([[0.0, x[0]] for x in X_TOY], Y_TOY)
assert multi.root.feature == 1
for bad_X, bad_y in [([], []), ([[1.0]], []), ([[float('nan')]], [1.0]), ([[1.0], []], [1.0, 2.0])]:
try:
CARTRegressor().fit(bad_X, bad_y)
except ValueError:
pass
else:
raise AssertionError("Invalid input should be rejected")
model = SquaredGBDT().fit(X_TOY, Y_TOY)
assert model.train_mse == [1.0, 0.25, 0.0625]
assert model.predict(X_TOY) == [1.25, 1.25, 2.75, 2.75]
assert model.trace[1]['residual'] == [-0.5, -0.5, 0.5, 0.5]
slower = SquaredGBDT(learning_rate=0.1).fit(X_TOY, Y_TOY)
assert slower.train_mse[-1] > model.train_mse[-1]
assert abs(slower.train_mse[-1] - 0.6561) < 1e-12
assert SquaredGBDT(n_estimators=3).fit(X_TOY, Y_TOY).train_mse[-1] == 0.015625
model.fit(X_TOY, changed)
assert len(model.trees) == 2 # Refit must reset previous trees.
assert model.predict(X_TOY) == [7.5, 7.5, 10.5, 10.5]
# Newton statistics for the same toy residuals, lambda_l2=1.
g_left, h_left, g_right, h_right = 2.0, 2.0, -2.0, 2.0
improvement = 0.5 * (g_left ** 2 / (h_left + 1) + g_right ** 2 / (h_right + 1))
assert abs(improvement - 4 / 3) < 1e-12
print('CART split=2.5, gain(SSE)=4.0, prediction=[1.0, 1.0, 3.0, 3.0]')
original = SquaredGBDT().fit(X_TOY, Y_TOY)
for index, row in enumerate(original.trace, 1):
print(f'round {index}: {row}')
print(f'train MSE={original.train_mse}')
print('tree and boosting variants passed')
if __name__ == '__main__':
self_check()在仓库根目录运行:
python3 examples/python/tree_boosting.py应看到:
CART split=2.5, gain(SSE)=4.0, prediction=[1.0, 1.0, 3.0, 3.0]程序还输出后面 GBDT 单元的两轮结果。先只解释本行:根节点切分列 0、阈值 2.5、左叶为 1、右叶为 3。
4. 沿一个样本走完预测
对 [[2.5]],循环检查 2.5 <= 2.5,进入左叶,返回 1。对 [[5]],进入右叶,返回 3。预测不会重新计算阈值,也不需要标签;训练结束后,原训练样本没有被保存在模型中。
先自行实现 predict,再用上述两个输入调试。常见错误是训练用 <=、预测却用 <,使边界点走到另一边。
5. 必做变式
| 变化 | 验收条件 | 解释 |
|---|---|---|
max_depth=0 | 所有预测为 2 | 容量限制确实生效 |
min_samples_leaf=3 | 四行数据无法分成两个合法叶子 | 限制作用于两个孩子 |
| 特征全相同 | 返回一个叶子 | 不能凭标签强拆相同输入 |
标签变为 2*y+5 | 叶子值为 7、11 | 预测由数据学出 |
| 增加一列全零特征放在第一列 | 学到的切分列为 1 | 搜索覆盖全部特征 |
| 输入 NaN 或不等长行 | 明确抛出 ValueError | 本实现未实现缺失值路由 |
这些变式已放在同一源码的 self_check 中。独立实现验收时先不看答案,写完自己的类,再用同样的输入和不变量检查;运行作者代码只证明环境和示例正常。
6. 与 scikit-learn 对照
安装好 uv 后,在仓库根目录运行:
uv run --extra trees --frozen python examples/python/lightgbm_lab.py此命令使用 uv 管理的隔离环境,不修改仓库已有的 PyTorch 依赖环境。首次下载需要网络;后续执行使用缓存。macOS 如果报告缺少 libomp,按 LightGBM 安装说明安装 OpenMP 运行库后重试。
对照程序比较四行数据上的阈值与预测,还比较固定 seed=13 的 80 行、三特征数据上的深度 3 回归树预测。它也运行后续 GBDT 和 LightGBM 实验。
不要要求任意数据上的树结构都逐字相同。并列候选的选择、阈值精度、缺失处理和实现约束都可能不同。本次受控对照的结论仅是这些输入下预测一致。
实验记录、失败与清理
2026-09-07 核心脚本实测:阈值 2.5,SSE 下降 4,所有变式通过。完整第三方对照记录见 LightGBM 训练实验。
如果出现无穷递归,先检查是否允许空孩子,或是否把同一组索引原样传回自己;如果均值异常,检查左右样本是否丢失或重复。不要通过忽略异常让错误输入继续训练。
核心脚本不产生文件;第三方实验的临时模型自动删除,依赖缓存可保留。记录自己的 Python 版本、修改内容与输出,即可复现。
自测与下一步
为什么模型在预测时不用保存全部训练数据?
参考答案
分组规则已经编码为特征和阈值,区域内最优常数已经编码为叶子值。常规预测只需要这些学习结果。重新训练和审计可能需要原始数据,但它们不属于这份最小预测对象的必要状态。
下一步:梯度提升为什么逐轮拟合修正量。