决策树实战:从零解析CART算法构造流程与实现细节

1次阅读
没有评论

共计 3955 个字符,预计需要花费 10 分钟才能阅读完成。

image.webp

一、CART 算法基础认知

决策树作为最直观的机器学习算法之一,其核心是通过一系列规则对数据进行分割。CART(Classification and Regression Trees)算法由 Breiman 等人于 1984 年提出,特点是:

决策树实战:从零解析 CART 算法构造流程与实现细节

  • 二叉树结构 :每个节点仅有两个子节点
  • 万能算法 :既可处理分类任务(使用基尼系数),也可处理回归任务(使用平方误差)
  • 递归分裂 :采用贪婪算法自顶向下构建

1.1 分类树 vs 回归树

分类树核心指标

基尼系数计算公式:

Gini(D) = 1 - Σ(p_i)^2  # p_i 是第 i 类样本占比 

分裂时选择使基尼指数下降最大的特征:

ΔGini = Gini(D) - (|D1|/|D|)Gini(D1) - (|D2|/|D|)Gini(D2)

回归树核心指标

平方误差最小化:

min Σ(y_i - c1)^2 + Σ(y_j - c2)^2  # c1/c2 为左右子节点均值 

二、算法完整实现流程

2.1 特征选择阶段

连续值处理(以年龄特征为例)

  1. 排序所有取值:[22,25,28,30,32]
  2. 计算候选分割点:(22+25)/2=23.5, (25+28)/2=26.5,…
  3. 对每个分割点计算指标增益
def find_best_split(X, y):
    best_gain = -1
    best_feature, best_value = None, None

    for feature in range(X.shape[1]):
        unique_values = np.unique(X[:, feature])
        for val in unique_values:
            left_mask = X[:, feature] <= val
            # 计算当前分割的基尼增益或平方误差减少
            current_gain = calculate_gain(y, left_mask)
            if current_gain > best_gain:
                best_gain = current_gain
                best_feature = feature
                best_value = val
    return best_feature, best_value

2.2 节点分裂实现

关键递归逻辑:

class TreeNode:
    def __init__(self, depth=0):
        self.left = None
        self.right = None
        self.feature = None
        self.value = None
        self.depth = depth
        self.leaf_value = None  # 叶节点预测值

def build_tree(X, y, max_depth=5):
    node = TreeNode()

    # 终止条件检查
    if max_depth<=0 or len(y)<2:
        node.leaf_value = np.mean(y) if is_regression else np.bincount(y).argmax()
        return node

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

    # 分裂数据集
    left_mask = X[:, feature] <= value
    node.feature = feature
    node.value = value

    # 递归构建子树
    node.left = build_tree(X[left_mask], y[left_mask], max_depth-1)
    node.right = build_tree(X[~left_mask], y[~left_mask], max_depth-1)

    return node

三、工程实践关键技巧

3.1 过拟合预防方案

预剪枝策略
– 最大深度限制(max_depth)
– 最小样本分裂数(min_samples_split)
– 叶节点最小样本数(min_samples_leaf)

后剪枝实现 (CCP 代价复杂度剪枝):

def prune_tree(node, alpha):
    if node.left is None: 
        return 0, node.leaf_value

    # 计算子树误差
    left_cost, left_val = prune_tree(node.left, alpha)
    right_cost, right_val = prune_tree(node.right, alpha)

    # 计算当前节点误差
    current_cost = calculate_node_cost(node)

    # 比较剪枝前后代价
    if current_cost + alpha < left_cost + right_cost:
        node.left = node.right = None
        node.leaf_value = calculate_leaf_value(node)
        return current_cost + alpha, node.leaf_value
    else:
        return left_cost + right_cost, None

3.2 类别不平衡处理

通过样本权重调整:

class_weight = {0:1, 1:5}  # 少数类权重增大
sample_weight = np.array([class_weight[y_i] for y_i in y])

# 修改基尼系数计算:weighted_gini = 1 - sum((np.sum(sample_weight[y==k])/total_weight)**2 
                      for k in classes)

四、完整代码实现

import numpy as np
from graphviz import Digraph

class CART:
    def __init__(self, task='classification', max_depth=3):
        self.task = task
        self.max_depth = max_depth

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

    def _build_tree(self, X, y, depth=0):
        node = {'depth': depth}

        # 终止条件
        if depth >= self.max_depth or len(np.unique(y)) == 1:
            node['leaf'] = True
            node['value'] = np.mean(y) if self.task=='regression' \
                           else np.bincount(y).argmax()
            return node

        # 寻找最佳分裂
        feature, value = self._find_split(X, y)
        if feature is None:  # 无法继续分裂
            node['leaf'] = True
            node['value'] = np.mean(y) if self.task=='regression' \
                           else np.bincount(y).argmax()
            return node

        # 递归构建子树
        left_mask = X[:, feature] <= value
        node['feature'] = feature
        node['value'] = value
        node['left'] = self._build_tree(X[left_mask], y[left_mask], depth+1)
        node['right'] = self._build_tree(X[~left_mask], y[~left_mask], depth+1)

        return node

    def _find_split(self, X, y):
        # 实际实现参考前文 find_best_split
        pass

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

    def _predict_one(self, x, node=None):
        if node is None:
            node = self.tree_

        if node.get('leaf', False):
            return node['value']

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

    def visualize(self):
        dot = Digraph()
        self._add_node(dot, self.tree_)
        return dot

    def _add_node(self, dot, node, parent=None, edge_label=''):
        node_id = str(id(node))

        if node.get('leaf', False):
            dot.node(node_id, f"leaf={node['value']:.2f}", shape='box')
        else:
            dot.node(node_id, f"X[{node['feature']}] <= {node['value']:.2f}")

        if parent is not None:
            dot.edge(parent, node_id, label=edge_label)

        if 'left' in node:
            self._add_node(dot, node['left'], node_id, 'True')
        if 'right' in node:
            self._add_node(dot, node['right'], node_id, 'False')

五、延伸思考与实践

  1. 多分类任务适配 :当前实现主要针对二分类,如何扩展基尼系数计算使其支持多分类?
  2. 缺失值处理 :当特征存在缺失值时,CART 算法常用的替代分裂(Surrogate Splits)机制如何实现?
  3. 并行优化 :在特征选择阶段,各个特征的计算相互独立,如何使用多进程加速训练过程?

通过本文的详细拆解,相信读者已经掌握 CART 算法的核心思想与实现要点。建议在理解基本原理后,尝试用不同数据集进行实验,观察参数变化对模型效果的影响,这是掌握算法最有效的方式。

正文完
 0
评论(没有评论)