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

1次阅读

没有评论

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

决策树通过一系列规则对数据进行分类或回归。CART（Classification and Regression Trees）与 ID3/C4.5 的主要差异在于：

分裂标准 ：CART 使用基尼系数（Gini Index），而 ID3/C4.5 使用信息增益（Information Gain）
二叉树结构 ：CART 每个节点只产生两个分支（二元分裂），而 ID3/C4.5 可产生多分支
功能完备性 ：CART 同时支持分类和回归任务

基尼系数计算公式如下：

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

其中 $p_k$ 是第 $k$ 类样本在数据集 $D$ 中的比例。基尼系数越小，数据纯度越高。

import numpy as np
import pandas as pd

# 示例：用中位数填充数值型缺失值
def fill_missing_values(df):
    for col in df.select_dtypes(include=np.number):
        median_val = df[col].median()
        df[col].fillna(median_val, inplace=True)
    return df

from sklearn.preprocessing import LabelEncoder

# 类别变量转为数值
def encode_categorical(df, columns):
    encoders = {}
    for col in columns:
        le = LabelEncoder()
        df[col] = le.fit_transform(df[col])
        encoders[col] = le
    return df, encoders

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

from dataclasses import dataclass
from typing import Union, Optional

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

class DecisionTreeClassifier:
    def __init__(self, max_depth=5, min_samples_split=2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split
        self.root = None

    def _best_split(self, X, y):
        best_gini = float('inf')
        best_feature, best_threshold = 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_threshold = threshold
        return best_feature, best_threshold

    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 = TreeNode()
            node = self.root

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

        # 寻找最佳分裂
        feature_idx, threshold = self._best_split(X, y)
        if feature_idx is None:  # 无法继续分裂
            node.value = np.argmax(np.bincount(y))
            return

        # 递归构建子树
        node.feature_idx = feature_idx
        node.threshold = threshold
        left_mask = X[:, feature_idx] <= threshold

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

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

# 等频分箱示例
def equal_freq_binning(series, n_bins=5):
    percentiles = np.linspace(0, 100, n_bins+1)
    bin_edges = np.percentile(series.dropna(), percentiles)
    return pd.cut(series, bins=bin_edges, include_lowest=True)

def prune_tree(node, alpha):
    if node.left is None and node.right is None:
        return node

    # 递归剪枝子树
    if node.left:
        node.left = prune_tree(node.left, alpha)
    if node.right:
        node.right = prune_tree(node.right, alpha)

    # 计算剪枝前后的损失
    if node.left and node.right:
        # 这里简化计算，实际应使用验证集误差
        before_prune_loss = gini_index(np.concatenate([
            node.left.value_samples,
            node.right.value_samples
        ])) + alpha * 2  # 2 个叶节点

        after_prune_loss = gini_index(np.concatenate([node.left.value_samples, node.right.value_samples])
        ) + alpha * 1  # 合并为 1 个节点

        if after_prune_loss <= before_prune_loss:
            node.left = None
            node.right = None
            node.value = np.argmax(np.bincount(np.concatenate([node.left.value_samples, node.right.value_samples])
            ))
    return node

在 MNIST 数据集上的性能测试：

from sklearn.datasets import load_digits
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier as SklearnTree

# 加载数据
digits = load_digits()
X, y = digits.data, digits.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# 自定义模型
my_tree = DecisionTreeClassifier(max_depth=10)
my_tree.fit(X_train, y_train)
my_score = np.mean(my_tree.predict(X_test) == y_test)

# sklearn 模型
sk_tree = SklearnTree(max_depth=10, criterion='gini')
sk_tree.fit(X_train, y_train)
sk_score = sk_tree.score(X_test, y_test)

print(f'自定义模型准确率: {my_score:.4f}')
print(f'sklearn 模型准确率: {sk_score:.4f}')

随机森林扩展 ：通过 bootstrap 采样构建多棵 CART 树，最终投票决策
高维数据优化 ：
特征预筛选（如方差过滤）
使用近似算法加速最佳分裂点查找
对稀疏数据采用特殊存储格式（如 CSR）

使用 graphviz 可视化（需安装 graphviz 软件）：

import graphviz
from sklearn.tree import export_graphviz

dot_data = export_graphviz(
    sk_tree,
    out_file=None, 
    feature_names=[f'pixel_{i}' for i in range(64)],
    class_names=[str(i) for i in range(10)],
    filled=True,
    rounded=True
)
graph = graphviz.Source(dot_data)
graph.render('digits_tree')  # 保存为 PDF

本文从理论到实践完整实现了 CART 决策树模型，关键收获包括：