Skip to content

从零实现 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. 自己写之前,先描述算法 ​

text
build(当前样本索引, 当前深度):
    计算均值和父节点 SSE
    如果深度或样本数不允许切分,返回叶子
    遍历每一列:
        对当前样本的该列数值排序并去重
        遍历相邻值中点:
            按 <= 和 > 分成左右索引
            任一边不足 min_samples_leaf,跳过
            计算父 SSE - 左 SSE - 右 SSE
            如果严格超过当前最好增益,记住候选
    如果没有候选,返回叶子
    否则递归建立两个孩子,返回内部节点

同增益时保留最先遇到的候选,便于复现。这里直接重算每个候选的误差,一节点最坏约为 O(dn2):每列最多 n−1 个阈值,每次扫描 n 行。这不是高性能实现。排序加前缀统计可以把每列的候选扫描降到线性,之后再学习 LightGBM 的分箱。

3. 阅读并运行完整源码 ​

以下是仓库中实际运行的源码;本节重点读到 CARTRegressor.predict 为止。SquaredGBDT 留到后两节。

python
"""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()

在仓库根目录运行:

bash
python3 examples/python/tree_boosting.py

应看到:

text
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 后,在仓库根目录运行:

bash
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 版本、修改内容与输出,即可复现。

自测与下一步 ​

为什么模型在预测时不用保存全部训练数据?

参考答案

分组规则已经编码为特征和阈值,区域内最优常数已经编码为叶子值。常规预测只需要这些学习结果。重新训练和审计可能需要原始数据,但它们不属于这份最小预测对象的必要状态。

下一步:梯度提升为什么逐轮拟合修正量。

最近更新