CART决策树分类结构实战指南：从原理到Python实现

1次阅读

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

在机器学习中，我们经常会遇到分类问题。线性模型（如逻辑回归）虽然简单易懂，但在处理非线性可分数据时表现往往不尽如人意。比如下面这样的数据分布：

import matplotlib.pyplot as plt
from sklearn.datasets import make_moons

X, y = make_moons(n_samples=100, noise=0.15, random_state=42)
plt.scatter(X[:, 0], X[:, 1], c=y, cmap='coolwarm')
plt.show()

这种情况下，决策树就能大显身手了。它通过一系列 if-then 规则将数据空间划分为矩形区域，特别适合处理这类非线性边界的问题。

CART（Classification and Regression Trees）使用基尼系数作为分裂标准。基尼系数反映了数据的不纯度，定义为：

$$Gini(p) = 1 – \sum_{k=1}^K p_k^2$$

其中 $p_k$ 是第 k 类样本在节点中的比例。与 ID3/C4.5 使用信息增益不同，CART 的计算更简单高效。

from dataclasses import dataclass
from typing import Any, Optional

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

import numpy as np
from collections import Counter

class DecisionTreeClassifier:
    def __init__(self, max_depth=5):
        self.max_depth = max_depth

    def _gini(self, y: np.ndarray) -> float:
        """计算基尼不纯度"""
        counts = np.bincount(y)
        return 1 - np.sum((counts / len(y)) ** 2)

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

        for idx in range(X.shape[1]):
            thresholds = np.unique(X[:, idx])
            for thresh in thresholds:
                left_mask = X[:, idx] <= thresh
                gini = (left_mask.sum() * self._gini(y[left_mask]) + 
                       (~left_mask).sum() * self._gini(y[~left_mask])) / len(y)
                if gini < best_gini:
                    best_gini = gini
                    best_idx = idx
                    best_thresh = thresh
        return best_idx, best_thresh

    def fit(self, X: np.ndarray, y: np.ndarray, depth=0) -> TreeNode:
        """递归构建决策树"""
        # 终止条件：纯度达标或达到最大深度
        if depth >= self.max_depth or len(np.unique(y)) == 1:
            return TreeNode(value=Counter(y).most_common(1)[0][0])

        # 寻找最佳分裂
        idx, thresh = self._best_split(X, y)
        if idx is None:  # 无法继续分裂
            return TreeNode(value=Counter(y).most_common(1)[0][0])

        # 递归构建子树
        left_mask = X[:, idx] <= thresh
        node = TreeNode(
            feature_idx=idx,
            threshold=thresh,
            left=self.fit(X[left_mask], y[left_mask], depth+1),
            right=self.fit(X[~left_mask], y[~left_mask], depth+1)
        )
        return node

def prune(node: TreeNode, X_val: np.ndarray, y_val: np.ndarray) -> bool:
    """后剪枝：验证集准确率提升则剪枝"""
    if node.left is None or node.right is None:
        return False

    # 如果是中间节点，尝试合并为叶节点
    if prune(node.left, X_val, y_val) or prune(node.right, X_val, y_val):
        return True

    # 计算当前节点和合并后的准确率
    original_acc = evaluate(node, X_val, y_val)
    merged_value = most_common_class([node.left.value, node.right.value])
    merged_acc = evaluate(TreeNode(value=merged_value), X_val, y_val)

    if merged_acc >= original_acc:
        node.left = node.right = None
        node.value = merged_value
        return True
    return False

类别不平衡处理

在计算基尼系数时加入类别权重

def _gini(self, y: np.ndarray, class_weight=None) -> float:
    if class_weight is None:
        class_weight = np.ones(len(np.unique(y)))
    counts = np.bincount(y)
    return 1 - np.sum((counts / len(y) * class_weight) ** 2)

连续值分箱陷阱
避免等宽分箱导致的信息丢失
优先考虑基于百分位数的分箱方法

可视化技巧

from sklearn.tree import export_graphviz

dot_data = export_graphviz(
    tree_model,
    out_file=None, 
    feature_names=feature_names,
    class_names=target_names,
    filled=True
)
graph = graphviz.Source(dot_data)
graph.render("decision_tree")  # 保存为 PDF