Python实战：从零构建CART决策树模型及避坑指南

1次阅读

没有评论

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

决策树是机器学习中最直观的算法之一，特别适合分类任务。它的优势在于：

模型可解释性强，决策过程像流程图一样清晰可见
对数据预处理要求较低，能自动处理缺失值和异常值
不需要特征缩放，数值型和类别型特征可以混合使用

但新手在手动实现时，经常会遇到这些问题：

忽略连续特征的分割点选择，导致信息利用不充分
忘记设置递归终止条件，造成无限递归或过拟合
缺乏剪枝逻辑，模型在训练集表现好但泛化能力差

常见的决策树算法有 ID3、C4.5 和 CART，它们的主要区别在于：

ID3
使用信息增益作为分裂标准
只能处理离散特征
倾向于选择取值多的特征
C4.5
改进 ID3，使用信息增益比
可以处理连续特征
通过剪枝防止过拟合
CART
使用基尼系数或平方误差
总是生成二叉树
可用于分类和回归

CART 算法的优势在于二叉树结构更简单，计算效率高，且基尼系数计算比信息增益更快速。

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

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

Python 实现示例：

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

对于每个特征，我们需要找到最佳分割点：

对连续特征：排序后取相邻值中点作为候选分割点
对离散特征：尝试所有可能的二分组合

选择使基尼系数下降最多的分割点：

def find_best_split(X, y):
    best_gini = float('inf')
    best_feature = None
    best_value = None

    for feature in range(X.shape[1]):
        values = np.unique(X[:, feature])
        for val in values:
            left_mask = X[:, feature] <= val
            right_mask = ~left_mask

            if np.sum(left_mask) == 0 or np.sum(right_mask) == 0:
                continue

            gini_left = gini(y[left_mask])
            gini_right = gini(y[right_mask])

            weighted_gini = (len(y[left_mask]) * gini_left + 
                            len(y[right_mask]) * gini_right) / len(y)

            if weighted_gini < best_gini:
                best_gini = weighted_gini
                best_feature = feature
                best_value = val

    return best_feature, best_value, best_gini

递归终止条件需要考虑：

当前节点样本数小于 min_samples_split
所有样本属于同一类别
基尼系数下降小于阈值

class Node:
    def __init__(self, feature=None, value=None, left=None, right=None, label=None):
        self.feature = feature  # 分裂特征
        self.value = value      # 分裂值
        self.left = left        # 左子树
        self.right = right      # 右子树
        self.label = label      # 叶节点类别

def build_tree(X, y, max_depth, min_samples_split, depth=0):
    # 终止条件 1: 所有样本属于同一类
    if len(np.unique(y)) == 1:
        return Node(label=y[0])

    # 终止条件 2: 样本数太少
    if len(y) < min_samples_split or depth >= max_depth:
        return Node(label=np.bincount(y).argmax())

    # 寻找最佳分裂
    feature, value, gini_gain = find_best_split(X, y)

    # 终止条件 3: 基尼系数不再下降
    if gini_gain < 1e-6:
        return Node(label=np.bincount(y).argmax())

    # 分裂数据
    left_mask = X[:, feature] <= value
    right_mask = ~left_mask

    # 递归构建子树
    left = build_tree(X[left_mask], y[left_mask], max_depth, min_samples_split, depth+1)
    right = build_tree(X[right_mask], y[right_mask], max_depth, min_samples_split, depth+1)

    return Node(feature=feature, value=value, left=left, right=right)

import numpy as np
from collections import Counter

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

    def fit(self, X, y):
        self.tree = self._build_tree(X, y)

    def predict(self, X):
        return np.array([self._predict(x, self.tree) for x in X])

    def _predict(self, x, node):
        if node.label is not None:
            return node.label

        if x[node.feature] <= node.value:
            return self._predict(x, node.left)
        else:
            return self._predict(x, node.right)

    def _build_tree(self, X, y, depth=0):
        n_samples = len(y)

        # 终止条件
        if (self.max_depth is not None and depth >= self.max_depth) or \
           n_samples < self.min_samples_split or len(np.unique(y)) == 1:
            return Node(label=Counter(y).most_common(1)[0][0])

        # 寻找最佳分裂
        feature, value, _ = find_best_split(X, y)

        # 分裂数据
        left_mask = X[:, feature] <= value
        right_mask = ~left_mask

        # 递归构建子树
        left = self._build_tree(X[left_mask], y[left_mask], depth+1)
        right = self._build_tree(X[right_mask], y[right_mask], depth+1)

        return Node(feature=feature, value=value, left=left, right=right)