C4.5决策树算法Python实现：从数学推导到工程实践

1次阅读

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

决策树算法在特征选择时面临三个核心挑战：如何量化特征对分类的贡献度，如何处理连续型特征的分割问题，以及当特征取值分布不均时如何避免选择偏差。这些挑战直接影响了模型的准确性和泛化能力。

ID3 算法使用信息增益作为特征选择标准，计算公式为：
$$Gain(S,A) = Entropy(S) – \sum_{v\in Values(A)} \frac{|S_v|}{|S|}Entropy(S_v)$$

而 C4.5 引入了信息增益率来解决 ID3 对多值特征的偏好问题：
$$GainRatio(S,A) = \frac{Gain(S,A)}{SplitInfo(S,A)}$$
其中分裂信息：
$$SplitInfo(S,A) = -\sum_{v\in Values(A)} \frac{|S_v|}{|S|}log_2\frac{|S_v|}{|S|}$$

import numpy as np
from typing import List, Dict, Union, Optional

class C45DecisionTree:
    def __init__(self, max_depth: int = 5, min_samples_split: int = 2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split
        self.tree = None

    def _entropy(self, y: np.ndarray) -> float:
        """计算熵值"""
        _, counts = np.unique(y, return_counts=True)
        probs = counts / len(y)
        return -np.sum(probs * np.log2(probs))

    def _split_info(self, X_col: np.ndarray) -> float:
        """计算分裂信息"""
        _, counts = np.unique(X_col, return_counts=True)
        probs = counts / len(X_col)
        return -np.sum(probs * np.log2(probs))

    def _find_best_split(self, X: np.ndarray, y: np.ndarray) -> Dict:
        """寻找最佳分割特征和分割点"""
        best_gain_ratio = -1
        best_feature = None
        best_threshold = None

        for feature_idx in range(X.shape[1]):
            # 处理连续特征
            if len(np.unique(X[:, feature_idx])) > 10:
                thresholds = np.unique(X[:, feature_idx])
                for threshold in thresholds:
                    mask = X[:, feature_idx] <= threshold
                    gain = self._information_gain(y, mask)
                    split_info = self._split_info(mask.astype(int))
                    gain_ratio = gain / split_info if split_info != 0 else 0

                    if gain_ratio > best_gain_ratio:
                        best_gain_ratio = gain_ratio
                        best_feature = feature_idx
                        best_threshold = threshold
            else:  # 处理离散特征
                gain = self._information_gain(y, X[:, feature_idx])
                split_info = self._split_info(X[:, feature_idx])
                gain_ratio = gain / split_info if split_info != 0 else 0

                if gain_ratio > best_gain_ratio:
                    best_gain_ratio = gain_ratio
                    best_feature = feature_idx
                    best_threshold = None

        return {'feature': best_feature, 'threshold': best_threshold}

    def fit(self, X: np.ndarray, y: np.ndarray, depth: int = 0) -> None:
        """训练决策树"""
        # 预剪枝条件
        if (depth >= self.max_depth or 
            len(y) < self.min_samples_split or 
            len(np.unique(y)) == 1):
            self.tree = {'leaf': True, 'class': np.argmax(np.bincount(y))}
            return

        split = self._find_best_split(X, y)
        if split['feature'] is None:
            self.tree = {'leaf': True, 'class': np.argmax(np.bincount(y))}
            return

        # 构建子树
        self.tree = {'feature': split['feature'],
            'threshold': split['threshold'],
            'left': None,
            'right': None,
            'leaf': False
        }

        if split['threshold'] is not None:  # 连续特征分割
            mask = X[:, split['feature']] <= split['threshold']
            self.tree['left'] = C45DecisionTree(self.max_depth, self.min_samples_split)
            self.tree['left'].fit(X[mask], y[mask], depth + 1)

            self.tree['right'] = C45DecisionTree(self.max_depth, self.min_samples_split)
            self.tree['right'].fit(X[~mask], y[~mask], depth + 1)
        else:  # 离散特征分割
            unique_values = np.unique(X[:, split['feature']])
            self.tree['branches'] = {}

            for value in unique_values:
                mask = X[:, split['feature']] == value
                subtree = C45DecisionTree(self.max_depth, self.min_samples_split)
                subtree.fit(X[mask], y[mask], depth + 1)
                self.tree['branches'][value] = subtree

    def predict(self, X: np.ndarray) -> np.ndarray:
        """预测方法"""
        return np.array([self._predict_single(x) for x in X])

    def _predict_single(self, x: np.ndarray) -> int:
        """单个样本预测"""
        node = self.tree
        while not node['leaf']:
            if node['threshold'] is not None:  # 连续特征
                if x[node['feature']] <= node['threshold']:
                    node = node['left'].tree
                else:
                    node = node['right'].tree
            else:  # 离散特征
                value = x[node['feature']]
                if value in node['branches']:
                    node = node['branches'][value].tree
                else:
                    break
        return node['class']

# 单元测试示例
def test_c45():
    # 测试 1：纯分类数据
    X = np.array([[0,1], [1,0], [1,1], [0,0]])
    y = np.array([0,1,1,0])
    model = C45DecisionTree(max_depth=2)
    model.fit(X, y)
    assert np.all(model.predict(X) == y)

    # 测试 2：连续特征处理
    X = np.random.rand(100, 3)
    y = (X[:,0] + X[:,1] > 1).astype(int)
    model = C45DecisionTree(max_depth=3)
    model.fit(X, y)
    assert model.predict([[0.1,0.1,0.5]])[0] == 0

if __name__ == '__main__':
    test_c45()