Python实战：CART决策树算法原理与高效实现

1次阅读

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

决策树是机器学习中的一种树形结构模型，通过递归地对数据进行划分来完成分类或回归任务。CART（Classification and Regression Trees）是其中最经典的算法之一，与其他决策树算法相比有显著特点：

二叉树结构 ：每个节点只分裂为两个子节点，简化决策路径
基尼系数 ：采用基尼不纯度（Gini Impurity）作为分类任务的分裂标准
多功能性 ：既可处理分类问题，也可处理回归问题（使用方差最小化）

基尼系数衡量数据集的不纯度，值越小表示纯度越高。对于包含 K 个类别的数据集 D：

def gini_index(y):
    _, counts = np.unique(y, return_counts=True)
    probabilities = counts / len(y)
    return 1 - np.sum(probabilities**2)

CART 通过穷举所有可能的分裂方式（特征 + 阈值）来寻找最优分裂点：

对每个特征的所有可能取值进行排序
取相邻值的中间点作为候选阈值
计算每个分裂点的加权基尼系数
选择使基尼系数降低最多的分裂方案

算法	分裂标准	树结构	连续值处理	缺失值处理
ID3	信息增益	多叉树	不支持	不支持
C4.5	信息增益比	多叉树	支持	支持
CART	基尼系数 / 方差	二叉树	支持	支持

主要优势：
– 二叉树结构计算效率更高
– 基尼系数计算比信息熵更简单
– 天然支持回归任务

from dataclasses import dataclass
from typing import Union

@dataclass
class Node:
    feature_idx: int = None    # 分裂特征索引
    threshold: float = None    # 分裂阈值
    left: 'Node' = None        # 左子树
    right: 'Node' = None       # 右子树
    value: Union[int, float] = None  # 叶节点预测值

class CARTDecisionTree:
    def __init__(self, max_depth=5, min_samples_split=2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split

    def _best_split(self, X, y):
        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

    def _weighted_gini(self, left_y, right_y):
        n = len(left_y) + len(right_y)
        return (len(left_y)/n)*gini_index(left_y) + \
               (len(right_y)/n)*gini_index(right_y)

    def fit(self, X, y, node=None, depth=0):
        if node is None:
            self.root = Node()
            node = self.root

        # 终止条件检查
        if (depth >= self.max_depth or 
            len(y) < self.min_samples_split or 
            len(np.unique(y)) == 1):
            node.value = np.mean(y) if self._is_regression else np.argmax(np.bincount(y))
            return

        # 寻找最佳分裂
        feature_idx, threshold = self._best_split(X, y)
        if feature_idx is None:
            return

        # 递归构建子树
        left_mask = X[:, feature_idx] <= threshold
        node.feature_idx = feature_idx
        node.threshold = threshold
        node.left = Node()
        node.right = Node()

        self.fit(X[left_mask], y[left_mask], node.left, depth+1)
        self.fit(X[~left_mask], y[~left_mask], node.right, depth+1)

最大深度限制 ：通过 max_depth 参数控制树生长
最小样本分裂 ：节点样本数少于 min_samples_split 时停止分裂
最小纯度提升 ：分裂必须带来显著的基尼系数降低

def prune(self, node, X_val, y_val):
    if node.value is not None:
        return node.value

    # 在验证集上测试剪枝效果
    left_mask = X_val[:, node.feature_idx] <= node.threshold

    # 计算剪枝前后的错误率
    original_error = self._calculate_error(X_val, y_val)

    # 临时转换为叶节点
    original_value = node.value
    node.value = np.mean(y_val) if self._is_regression else np.argmax(np.bincount(y_val))
    pruned_error = self._calculate_error(X_val, y_val)

    # 决策是否剪枝
    if pruned_error <= original_error:
        node.left = node.right = None
    else:
        node.value = original_value
        if node.left:
            self.prune(node.left, X_val[left_mask], y_val[left_mask])
        if node.right:
            self.prune(node.right, X_val[~left_mask], y_val[~left_mask])

特征预处理 ：
对连续特征进行离散化（等宽 / 等频分箱）
对高基数类别特征采用目标编码
内存优化 ：
使用稀疏矩阵存储处理 one-hot 编码
限制树的深度防止内存爆炸
计算加速 ：
对排序等耗时操作使用 numpy 向量化
对大数据集采用随机子采样
API 设计规范 ：
实现 sklearn 的 fit/predict 接口
提供 get_params/set_params 方法

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

# 数据准备
iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(iris.data, iris.target, test_size=0.2)

# 模型训练
tree = CARTDecisionTree(max_depth=3)
tree.fit(X_train, y_train)

# 评估
preds = tree.predict(X_test)
accuracy = np.mean(preds == y_test)
print(f"Test accuracy: {accuracy:.2%}")