CART决策树代码实现：从算法原理到工程实践

1次阅读

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

决策树是机器学习中最直观的可解释模型之一，它通过树形结构模拟人类决策过程。在众多决策树算法中，ID3、C4.5 和 CART 形成了经典的三代演进：

ID3：采用信息增益选择特征，仅支持离散特征且容易过拟合
C4.5：引入信息增益率改进特征选择，支持连续特征和缺失值处理
CART（本文重点）：使用基尼系数作为分裂标准，支持回归和分类任务，生成二叉树结构便于工程优化

基尼系数衡量数据集的纯度，值越小表示纯度越高。其计算公式为：

Gini(D) = 1 - \sum_{k=1}^{K} (\frac{|C_k|}{|D|})^2

计算特征 A 在某划分点下的基尼指数：

def gini_impurity(y: np.ndarray) -> float:
    """计算基尼不纯度"""
    _, counts = np.unique(y, return_counts=True)
    return 1 - np.sum((counts / len(y)) ** 2)

递归构建决策树需要明确的停止条件：

当前节点样本数小于预设阈值（min_samples_split）
所有特征都已使用或特征带来的增益小于阈值（min_impurity_decrease）
达到最大树深度（max_depth）
节点样本完全属于同一类别

采用代价复杂度剪枝（CCP），平衡子树复杂度与误差率：

def prune_tree(node, alpha):
    """基于 MDL 原则的剪枝"""
    if node.is_leaf:
        return

    # 计算子树误差与叶节点误差
    subtree_error = calculate_error(node)
    leaf_error = calculate_error_if_pruned(node)

    if leaf_error - subtree_error < alpha * count_leaves(node):
        node.left = node.right = None
        node.is_leaf = True

from typing import Union, Optional
import numpy as np

class DecisionNode:
    """决策树节点类"""
    def __init__(self, 
                 feature_idx: Optional[int] = None,
                 threshold: Optional[float] = None,
                 left: Optional['DecisionNode'] = None,
                 right: Optional['DecisionNode'] = None,
                 value: Optional[Union[int, float]] = None):
        self.feature_idx = feature_idx  # 分裂特征索引
        self.threshold = threshold      # 分裂阈值
        self.left = left                # 左子树
        self.right = right              # 右子树
        self.value = value              # 叶节点预测值

class CARTClassifier:
    """CART 分类树实现"""
    def __init__(self, 
                 max_depth: int = 3,
                 min_samples_split: int = 2,
                 min_impurity_decrease: float = 0.0):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split
        self.min_impurity_decrease = min_impurity_decrease
        self.root = None

    def _find_best_split(self, X: np.ndarray, y: np.ndarray) -> tuple:
        """寻找最优分裂特征和阈值"""
        best_gini = float('inf')
        best_feature, best_thresh = None, None

        for feature_idx in range(X.shape[1]):
            thresholds = np.unique(X[:, feature_idx])
            for threshold in thresholds:
                left_mask = X[:, feature_idx] <= threshold
                gini = self._weighted_gini(y[left_mask], y[~left_mask])

                if gini < best_gini:
                    best_gini = gini
                    best_feature = feature_idx
                    best_thresh = threshold

        return best_feature, best_thresh, best_gini

    def _weighted_gini(self, *y_groups: np.ndarray) -> float:
        """计算加权基尼系数"""
        total_samples = sum(len(y) for y in y_groups)
        return sum(len(y)/total_samples * gini_impurity(y) for y in y_groups)

    def fit(self, X: np.ndarray, y: np.ndarray):
        """构建决策树"""
        self.root = self._build_tree(X, y, depth=0)

    def _build_tree(self, X: np.ndarray, y: np.ndarray, depth: int) -> DecisionNode:
        """递归构建决策树"""
        # 终止条件判断
        if (depth >= self.max_depth 
            or len(y) < self.min_samples_split 
            or gini_impurity(y) <= self.min_impurity_decrease):
            return DecisionNode(value=self._leaf_value(y))

        # 寻找最优分裂
        feature, thresh, gini = self._find_best_split(X, y)
        if gini == float('inf'):
            return DecisionNode(value=self._leaf_value(y))

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

        return DecisionNode(feature_idx=feature, 
                           threshold=thresh,
                           left=left,
                           right=right)

在 UCI 乳腺癌数据集上的测试结果：