C4.5决策树算法在数据挖掘中的核心原理与实战优化

1次阅读

没有评论

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

决策树算法是数据挖掘中的经典方法，而 C4.5 算法作为 ID3 算法的改进版，主要解决了两个关键问题：

ID3 无法直接处理连续值属性
ID3 对缺失值敏感且容易过拟合

在工业场景中，我们经常会遇到包含连续特征（如温度、价格）和缺失值的数据集，这正是 C4.5 大显身手的地方。

C4.5 使用信息增益比而非 ID3 的信息增益来选择划分属性，计算公式为：

$$\text{GainRatio}(S,A) = \frac{\text{Gain}(S,A)}{\text{SplitInfo}(S,A)}$$

其中 SplitInfo 是分裂信息量，用于惩罚取值较多的属性：

$$\text{SplitInfo}(S,A) = -\sum_{i=1}^n \frac{|S_i|}{|S|} \log_2 \frac{|S_i|}{|S|}$$

这个改进有效避免了 ID3 倾向于选择取值多的属性的问题。

对于连续属性，C4.5 采用二分法处理：

将属性值排序
计算相邻值的中点作为候选切分点
选择使信息增益最大的切分点
将连续值转换为二元离散值

C4.5 采用自底向上的剪枝策略，核心思想是：

计算子树在训练集上的错误率 $e$
加上连续性修正项 $(0.5)$：$E = e + 0.5$
计算标准误差：$SE = \sqrt{E(N-E)/N}$
比较剪枝前后的估计错误率

import numpy as np
from collections import Counter

class C45Node:
    def __init__(self, feature=None, threshold=None, value=None):
        self.feature = feature  # 划分特征
        self.threshold = threshold  # 连续特征划分阈值
        self.children = {}  # 子节点
        self.value = value  # 叶节点类别

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

    def fit(self, X, y):
        self.n_classes = len(set(y))
        self.n_features = X.shape[1]
        self.tree = self._grow_tree(X, y)

    def _grow_tree(self, X, y, depth=0):
        n_samples = X.shape[0]

        # 终止条件
        if (n_samples < self.min_samples_split or 
            (self.max_depth is not None and depth >= self.max_depth)):
            leaf_value = self._most_common_label(y)
            return C45Node(value=leaf_value)

        # 选择最佳划分特征
        best_feature, best_threshold = self._best_split(X, y)

        # 创建节点
        node = C45Node(feature=best_feature, threshold=best_threshold)

        # 递归构建子树
        if best_threshold is None:  # 离散特征
            for value in set(X[:, best_feature]):
                mask = X[:, best_feature] == value
                node.children[value] = self._grow_tree(X[mask], y[mask], depth + 1)
        else:  # 连续特征
            left_mask = X[:, best_feature] <= best_threshold
            right_mask = ~left_mask
            node.children['left'] = self._grow_tree(X[left_mask], y[left_mask], depth + 1)
            node.children['right'] = self._grow_tree(X[right_mask], y[right_mask], depth + 1)

        return node

    def _best_split(self, X, y):
        best_gain_ratio = -1
        best_feature = None
        best_threshold = None

        for feature_idx in range(self.n_features):
            # 处理连续特征
            thresholds = self._get_thresholds(X[:, feature_idx])
            for threshold in thresholds:
                gain_ratio = self._information_gain_ratio(X[:, feature_idx], y, threshold)
                if gain_ratio > best_gain_ratio:
                    best_gain_ratio = gain_ratio
                    best_feature = feature_idx
                    best_threshold = threshold

        return best_feature, best_threshold

    def _information_gain_ratio(self, feature, y, threshold=None):
        # 计算信息增益比
        parent_entropy = self._entropy(y)

        if threshold is None:  # 离散特征
            split_info = 0
            child_entropy = 0
            for value in set(feature):
                mask = feature == value
                p = sum(mask) / len(feature)
                child_entropy += p * self._entropy(y[mask])
                split_info -= p * np.log2(p) if p > 0 else 0
        else:  # 连续特征
            left_mask = feature <= threshold
            right_mask = ~left_mask

            p_left = sum(left_mask) / len(feature)
            p_right = 1 - p_left

            child_entropy = p_left * self._entropy(y[left_mask])
            child_entropy += p_right * self._entropy(y[right_mask])

            split_info = -p_left * np.log2(p_left) if p_left > 0 else 0
            split_info -= p_right * np.log2(p_right) if p_right > 0 else 0

        gain = parent_entropy - child_entropy
        gain_ratio = gain / split_info if split_info > 0 else 0

        return gain_ratio

    def _entropy(self, y):
        # 计算熵
        counts = np.bincount(y)
        ps = counts / len(y)
        return -np.sum([p * np.log2(p) for p in ps if p > 0])

    def _get_thresholds(self, feature):
        # 获取连续特征的候选切分点
        unique_values = np.unique(feature)
        if len(unique_values) <= 1:
            return [None]

        thresholds = []
        for i in range(1, len(unique_values)):
            thresholds.append((unique_values[i] + unique_values[i-1]) / 2)

        return thresholds

    def _most_common_label(self, y):
        # 返回出现次数最多的类别
        counter = Counter(y)
        return counter.most_common(1)[0][0]

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

    def _predict_one(self, x):
        node = self.tree
        while node.value is None:
            if node.threshold is None:  # 离散特征
                node = node.children.get(x[node.feature], None)
            else:  # 连续特征
                if x[node.feature] <= node.threshold:
                    node = node.children['left']
                else:
                    node = node.children['right']

            if node is None:  # 处理缺失值
                return None

        return node.value