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

1次阅读

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

决策树通过一系列规则对数据进行分类或回归，而 CART（Classification and Regression Trees）是其中最经典的算法之一。与 ID3 和 C4.5 不同，CART 有两大特点：

始终采用二叉树结构（即使特征有多个取值也通过二分法处理）
分类任务使用基尼系数，回归任务使用方差作为分裂标准

基尼系数衡量数据的不纯度，计算公式为：

def gini_index(groups, classes):
    n_instances = sum(len(group) for group in groups)
    gini = 0.0
    for group in groups:
        size = len(group)
        if size == 0:
            continue
        score = 0.0
        for class_val in classes:
            p = [row[-1] for row in group].count(class_val) / size
            score += p * p
        gini += (1.0 - score) * (size / n_instances)
    return gini

对于连续特征，CART 的二分过程：
1. 对特征值排序
2. 计算相邻值的中间点作为候选分割阈值
3. 选择基尼系数最小的分割方案

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

from dataclasses import dataclass
from typing import Any, Union

@dataclass
class DecisionNode:
    feature_idx: int = None    # 用于分裂的特征索引
    threshold: float = None    # 分裂阈值
    left: Any = None           # 左子树
    right: Any = None          # 右子树
    value: Any = None          # 叶节点存储的预测值

def build_tree(X, y, max_depth, min_samples_split, depth=0):
    # 终止条件检查
    if len(set(y)) == 1 or len(y) <= min_samples_split or depth >= max_depth:
        return DecisionNode(value=most_common_label(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_idx = X[:, feature_idx] <= threshold
            groups = [y[left_idx], y[~left_idx]]
            gini = gini_index(groups, np.unique(y))

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

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

    return DecisionNode(best_feature, best_thresh, left, right)

最大深度限制（max_depth）
节点最小样本数（min_samples_split）
叶节点最小样本数（min_samples_leaf）

def prune_tree(node, alpha, X_val, y_val):
    if node.value is not None:
        return node

    # 先剪枝子树
    node.left = prune_tree(node.left, alpha, X_val, y_val)
    node.right = prune_tree(node.right, alpha, X_val, y_val)

    # 如果两个子节点都是叶节点，考虑合并
    if node.left.value is not None and node.right.value is not None:
        # 计算合并前后的误差
        merged_error = compute_error(merge_nodes(node), X_val, y_val)
        original_error = compute_error(node, X_val, y_val)

        if merged_error + alpha <= original_error:
            return DecisionNode(value=node.left.value if len(node.left.value) > len(node.right.value) else node.right.value)

    return node

对于连续特征，等频分箱可提升分裂效率：

from sklearn.preprocessing import KBinsDiscretizer

discretizer = KBinsDiscretizer(n_bins=10, encode='ordinal', strategy='quantile')
X_discrete = discretizer.fit_transform(X_continuous)

对大型数据集，可并行计算不同特征的分裂质量：

from joblib import Parallel, delayed

results = Parallel(n_jobs=-1)(delayed(calc_best_split)(X, y, feature_idx) 
    for feature_idx in range(X.shape[1])
)

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 = build_tree(X_train, y_train, max_depth=3, min_samples_split=5)

# 评估
predictions = [predict_tree(tree, x) for x in X_test]
accuracy = np.mean(predictions == y_test)
print(f"Test Accuracy: {accuracy:.2f}")