CART决策树算法原理深度解析与工程实践优化

1次阅读

共计 2928 个字符，预计需要花费 8 分钟才能阅读完成。

决策树分裂质量的核心指标，计算公式为：
$$ Gini(D) = 1 – \sum_{k=1}^{K} (\frac{|C_k|}{|D|})^2 $$
其中 $C_k$ 表示第 k 类样本的数量。分裂后的 Gini 指数：
$$ Gini_{split}(D) = \sum_{v=1}^{V} \frac{|D_v|}{|D|} Gini(D_v) $$

分裂标准 ：
ID3 使用信息增益（偏向多值特征）
C4.5 使用增益率（解决 ID3 偏置）
CART 使用 Gini 系数（计算效率更高）
输出类型 ：
CART 可同时处理分类（二叉树）和回归（MSE）任务
ID3/C4.5 仅支持分类

class TreeNode:
    def __init__(self, feature_idx=None, threshold=None, value=None, left=None, right=None):
        self.feature_idx = feature_idx  # 分裂特征索引
        self.threshold = threshold      # 分裂阈值
        self.value = value              # 叶节点预测值
        self.left = left                # 左子树
        self.right = right              # 右子树

def _build_tree(self, X, y, depth=0):
    # 终止条件：达到最大深度或样本纯净
    if depth >= self.max_depth or self._is_pure(y):
        return TreeNode(value=self._leaf_value(y))

    # 寻找最佳分裂
    best_gini = float('inf')
    best_feature, best_thresh = None, None

    # 特征预排序加速（见 3.2 节）for feature_idx in self._get_candidate_features():
        thresholds = np.unique(X[:, feature_idx])
        for thresh in thresholds:
            left_idx = X[:, feature_idx] <= thresh
            gini = self._calc_split_gini(y, left_idx)
            if gini < best_gini:
                best_gini = gini
                best_feature, best_thresh = feature_idx, thresh

    # 递归构建子树
    left_idx = X[:, best_feature] <= best_thresh
    left = self._build_tree(X[left_idx], y[left_idx], depth+1)
    right = self._build_tree(X[~left_idx], y[~left_idx], depth+1)

    return TreeNode(best_feature, best_thresh, left=left, right=right)

def prune(self, tree, X_val, y_val):
    if tree.is_leaf:
        return tree

    # 先剪枝子树
    tree.left = self.prune(tree.left, X_val, y_val)
    tree.right = self.prune(tree.right, X_val, y_val)

    # 计算剪枝前后验证集误差
    original_loss = self._calculate_loss(y_val, self._predict(X_val, tree))
    merged_loss = self._calculate_loss(y_val, np.full(len(y_val), self._leaf_value(y_val)))

    # 判断是否剪枝
    if merged_loss <= original_loss + self.ccp_alpha:
        return TreeNode(value=self._leaf_value(y_val))
    return tree

训练阶段 ：
常规实现：$O(mn^2)$（m 特征数，n 样本数）
优化后：$O(mn\log n)$（特征预排序）
预测阶段 ：$O(\text{ 树深度})$

# 在初始化时对所有特征排序
self._sorted_indices = [np.argsort(X[:, i]) for i in range(X.shape[1])]
self._sorted_values = [np.sort(X[:, i]) for i in range(X.shape[1])]

# 分裂时直接使用排序结果
def _find_best_split(self, feature_idx, y):
    sorted_y = y[self._sorted_indices[feature_idx]]
    # ... 使用二分查找优化阈值搜索...

# 特征并行化方案
def train_distributed(data_rdd):
    # 各 worker 计算局部最佳分裂
    splits = data_rdd.mapPartitions(lambda part: [find_local_best_split(part)]
    ).collect()

    # Driver 聚合全局最佳分裂
    global_best = select_global_best(splits)

    # 广播分裂结果并递归处理
    if not stopping_criteria(global_best):
        left_rdd, right_rdd = split_data(data_rdd, global_best)
        left_tree = train_distributed(left_rdd)
        right_tree = train_distributed(right_rdd)
        return combine_trees(global_best, left_tree, right_tree)

加权 Gini 系数 ：
$$ Gini_{weighted} = \sum_{k=1}^{K} w_k \cdot \frac{|C_k|}{|D|} (1 – \frac{|C_k|}{|D|}) $$
过采样 / 欠采样 ：SMOTE 等算法调整样本分布

# 等频分桶示例
def bin_continuous(feature, n_bins=10):
    percentiles = np.linspace(0, 100, n_bins+1)
    bins = np.percentile(feature, percentiles)
    return np.digitize(feature, bins[:-1])