从零构建CART决策树：原理详解与Python实战避坑指南

1次阅读

没有评论

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

传统 ID3 决策树在实际应用中存在几个明显缺陷：

只能处理离散型特征，无法直接用于现实场景中大量存在的连续值数据
采用信息增益作为分裂标准，容易偏向取值较多的特征（例如用户 ID 这种无意义的高基数特征）
多叉树结构导致模型复杂度难以控制，容易过拟合

CART(Classification And Regression Tree) 通过以下改进解决了这些问题：

引入二元分裂机制，所有节点最多分裂为两个子节点
使用基尼系数替代信息增益，避免特征取值数量的影响
原生支持连续特征处理（通过寻找最佳分割点）

基尼系数衡量的是数据集中随机抽取两个样本，其类别标签不一致的概率。对于包含 K 个类别的数据集 $D$，其计算公式为：

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

其中 $|C_k|$ 表示第 k 类样本的数量。当我们在特征 $A$ 的取值 $a$ 处将数据集划分为 $D_1$ 和 $D_2$ 两部分时，划分后的基尼指数为：

$$
Gini_{split}(D,A) = \frac{|D_1|}{|D|}Gini(D_1) + \frac{|D_2|}{|D|}Gini(D_2)
$$

与信息增益相比，基尼系数计算不需要耗时的对数运算，且在实际应用中往往能产生更平衡的树结构。

class DecisionNode:
    """决策树节点类"""
    def __init__(self, feature_idx=None, threshold=None,
                 left=None, right=None, value=None):
        # 分裂特征索引（仅内部节点使用）self.feature_idx = feature_idx  
        # 分裂阈值（连续特征）self.threshold = threshold
        # 左右子节点
        self.left = left
        self.right = right
        # 叶节点存储的预测值
        self.value = value

import numpy as np

def gini(y):
    """计算基尼系数"""
    m = y.shape[0]
    if m == 0:
        return 0
    # 计算每个类别的占比平方和
    return 1 - np.sum((np.bincount(y) / m) ** 2)

def gini_split(y_left, y_right):
    """计算分裂后的加权基尼指数"""
    n = len(y_left) + len(y_right)
    return (len(y_left)/n)*gini(y_left) + (len(y_right)/n)*gini(y_right)

class CARTClassifier:
    def __init__(self, max_depth=None, min_samples_split=2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split

    def _best_split(self, X, y):
        """寻找当前数据集的最佳分裂特征和阈值"""
        m, n = X.shape
        best_gini = float('inf')
        best_feature, best_thresh = None, None

        for feature_idx in range(n):
            # 对当前特征的所有唯一值尝试作为分裂点
            thresholds = np.unique(X[:, feature_idx])
            for threshold in thresholds:
                # 根据阈值划分数据集
                left_mask = X[:, feature_idx] <= threshold
                current_gini = gini_split(y[left_mask], y[~left_mask])

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

        return best_feature, best_thresh

    def _build_tree(self, X, y, depth=0):
        """递归构建决策树"""
        # 终止条件检查
        if (len(np.unique(y)) == 1 or 
            len(y) < self.min_samples_split or
            (self.max_depth is not None and depth >= self.max_depth)):
            return DecisionNode(value=np.argmax(np.bincount(y)))

        # 寻找最佳分裂
        feature, threshold = self._best_split(X, y)
        if feature is None:  # 无法找到有效分裂
            return DecisionNode(value=np.argmax(np.bincount(y)))

        # 递归构建子树
        left_mask = X[:, feature] <= threshold
        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=threshold,
                           left=left, right=right)

决策树的预测过程天然适合并行化，我们可以利用 joblib 来加速批量预测：

from joblib import Parallel, delayed

class ParallelPredictor:
    def __init__(self, tree, n_jobs=4):
        self.tree = tree
        self.n_jobs = n_jobs

    def _predict_single(self, x):
        """单个样本的预测逻辑"""
        node = self.tree
        while node.value is None:
            if x[node.feature_idx] <= node.threshold:
                node = node.left
            else:
                node = node.right
        return node.value

    def predict(self, X):
        """并行预测接口"""
        return Parallel(n_jobs=self.n_jobs)(delayed(self._predict_single)(x) for x in X)

为了让我们的实现能够无缝接入现有机器学习流水线，可以模仿 sklearn 的 API 风格：

from sklearn.base import BaseEstimator, ClassifierMixin

class SKCompatCART(BaseEstimator, ClassifierMixin):
    """sklearn 兼容的 CART 实现"""
    def __init__(self, max_depth=None, min_samples_split=2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split

    def fit(self, X, y):
        self.tree_ = CARTClassifier(
            max_depth=self.max_depth,
            min_samples_split=self.min_samples_split
        )._build_tree(X, y)
        return self

    def predict(self, X):
        predictor = ParallelPredictor(self.tree_)
        return predictor.predict(X)

当处理连续特征时，直接使用浮点数比较可能存在精度问题。改进方案：

# 在_best_split 方法中增加容差处理
left_mask = np.abs(X[:, feature_idx] - threshold) < 1e-8
left_mask |= (X[:, feature_idx] < threshold)

对于类别型特征，建议使用目标编码（Target Encoding）而非 One-Hot：

from category_encoders import TargetEncoder

# 在训练前预处理
encoder = TargetEncoder()
X_train_encoded = encoder.fit_transform(X_train, y_train)
X_test_encoded = encoder.transform(X_test)

使用 graphviz 可视化时，中文可能出现乱码，解决方案：

import graphviz

def plot_tree(tree, feature_names):
    dot = graphviz.Digraph(engine='dot', 
                          graph_attr={'dpi': '300'},
                          node_attr={'fontname': 'Microsoft YaHei'})
    # 构建可视化节点的递归逻辑...
    return dot

在 UCI 的葡萄酒数据集上进行测试：

from sklearn.datasets import load_wine
from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import train_test_split

# 数据准备
data = load_wine()
X_train, X_test, y_train, y_test = train_test_split(data.data, data.target, test_size=0.2, random_state=42)

# sklearn 版本
sk_tree = DecisionTreeClassifier(criterion='gini', max_depth=3)
sk_tree.fit(X_train, y_train)
print("sklearn 准确率:", sk_tree.score(X_test, y_test))

# 自实现版本
my_tree = SKCompatCART(max_depth=3)
my_tree.fit(X_train, y_train)
print("自实现准确率:", np.mean(my_tree.predict(X_test) == y_test))

典型测试结果：