C4.5决策树算法实战：从原理到Python实现与调优指南

1次阅读

没有评论

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

决策树算法最早源于 ID3（Iterative Dichotomiser 3），它使用信息增益作为特征选择标准。但 ID3 存在明显缺陷：

倾向于选择取值较多的特征（如 ID 类特征）
无法处理连续型特征
没有剪枝机制容易过拟合

C4.5 作为 ID3 的改进版，主要做了三大升级：

用信息增益比替代信息增益，解决特征偏好问题
新增连续值离散化处理能力
引入后剪枝 (pre-pruning) 策略

信息增益比 = 信息增益 / 特征固有值(intrinsic value)

计算步骤：

计算数据集 D 的经验熵 H(D)
计算特征 A 对 D 的经验条件熵 H(D|A)
信息增益 = H(D) – H(D|A)
特征固有值 = -Σ(|Di|/|D|)*log2(|Di|/|D|)
信息增益比 = 信息增益 / 特征固有值

其中特征固有值相当于对特征取值数目做了归一化。

C4.5 处理连续特征的步骤：

将特征值排序
计算相邻值的中间点作为候选划分点
选择使信息增益比最大的划分点
将该特征转换为二值特征（≤划分点 or >划分点）

import numpy as np
from collections import Counter

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

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

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

    def _split_continuous(self, X, feature_idx):
        """连续值最佳划分点选择"""
        values = sorted(set(X[:, feature_idx]))
        thresholds = [(values[i] + values[i+1])/2 for i in range(len(values)-1)]

        best_gain_ratio = 0
        best_threshold = None

        for threshold in thresholds:
            left_idx = X[:, feature_idx] <= threshold
            right_idx = X[:, feature_idx] > threshold

            if sum(left_idx) == 0 or sum(right_idx) == 0:
                continue

            gain_ratio = self._gain_ratio(X, feature_idx, threshold)

            if gain_ratio > best_gain_ratio:
                best_gain_ratio = gain_ratio
                best_threshold = threshold

        return best_threshold, best_gain_ratio

    def _gain_ratio(self, X, feature_idx, threshold=None):
        """计算信息增益比"""
        parent_entropy = self._entropy(X[:, -1])

        if threshold is not None:  # 连续特征
            left_idx = X[:, feature_idx] <= threshold
            right_idx = X[:, feature_idx] > threshold

            n = len(X)
            n_left, n_right = sum(left_idx), sum(right_idx)

            if n_left == 0 or n_right == 0:
                return 0

            child_entropy = (n_left/n)*self._entropy(X[left_idx, -1]) + \
                            (n_right/n)*self._entropy(X[right_idx, -1])

            gain = parent_entropy - child_entropy
            iv = - (n_left/n)*np.log2(n_left/n) - (n_right/n)*np.log2(n_right/n)

        else:  # 离散特征
            feature_values = X[:, feature_idx]
            unique_values = set(feature_values)

            child_entropy = 0
            iv = 0

            for value in unique_values:
                idx = feature_values == value
                p = sum(idx) / len(X)
                child_entropy += p * self._entropy(X[idx, -1])
                iv += -p * np.log2(p)

            gain = parent_entropy - child_entropy

        return gain / iv if iv != 0 else 0

    def _best_split(self, X):
        """寻找最佳分裂特征"""
        best_gain_ratio = 0
        best_feature = None
        best_threshold = None

        for feature_idx in range(X.shape[1]-1):
            if len(set(X[:, feature_idx])) > 10:  # 假设连续特征
                threshold, gain_ratio = self._split_continuous(X, feature_idx)
            else:  # 离散特征
                threshold = None
                gain_ratio = self._gain_ratio(X, feature_idx)

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

        return best_feature, best_threshold

    def _build_tree(self, X, depth=0):
        """递归构建决策树"""
        n_samples, n_features = X.shape

        # 终止条件
        if (len(set(X[:, -1])) == 1 or 
            n_samples < self.min_samples_split or
            (self.max_depth is not None and depth >= self.max_depth)):
            most_common = Counter(X[:, -1]).most_common(1)[0][0]
            return C45Node(value=most_common)

        # 寻找最佳分裂
        feature_idx, threshold = self._best_split(X)

        if feature_idx is None:  # 无法继续分裂
            most_common = Counter(X[:, -1]).most_common(1)[0][0]
            return C45Node(value=most_common)

        # 创建节点
        node = C45Node(feature_idx=feature_idx, threshold=threshold)

        # 连续特征处理
        if threshold is not None:
            left_idx = X[:, feature_idx] <= threshold
            right_idx = X[:, feature_idx] > threshold

            node.children['left'] = self._build_tree(X[left_idx], depth+1)
            node.children['right'] = self._build_tree(X[right_idx], depth+1)
        else:  # 离散特征
            feature_values = X[:, feature_idx]
            unique_values = set(feature_values)

            for value in unique_values:
                idx = feature_values == value
                node.children[value] = self._build_tree(X[idx], depth+1)

        return node

    def fit(self, X, y):
        """训练模型"""
        dataset = np.column_stack([X, y])
        self.root = self._build_tree(dataset)

    def _predict(self, x, node):
        """递归预测单个样本"""
        if node.value is not None:
            return node.value

        feature_value = x[node.feature_idx]

        if node.threshold is not None:  # 连续特征
            if feature_value <= node.threshold:
                return self._predict(x, node.children['left'])
            else:
                return self._predict(x, node.children['right'])
        else:  # 离散特征
            if feature_value in node.children:
                return self._predict(x, node.children[feature_value])
            else:  # 处理未见过的特征值
                return Counter([n.value for n in node.children.values()]).most_common(1)[0][0]

    def predict(self, X):
        """批量预测"""
        return np.array([self._predict(x, self.root) for x in X])

max_depth: 树的最大深度
值越小模型越简单，可能欠拟合
值越大模型越复杂，可能过拟合
建议从 3 -10 开始尝试
min_samples_split: 节点分裂最小样本数
值越大分裂条件越严格
典型值 2 -20

C4.5 采用悲观剪枝(Pessimistic Pruning)：

def prune(self, node=None, parent=None, branch_name=None):
    if node is None:
        node = self.root

    # 后序遍历
    if node.children:
        for branch, child in list(node.children.items()):
            self.prune(child, node, branch)

    # 尝试剪枝
    if node.children:
        # 计算剪枝前后的错误率
        original_error = self._calculate_error(node)

        # 临时保存子节点
        temp_children = node.children

        # 尝试变成叶节点
        node.children = {}
        pruned_error = self._calculate_error(node)

        # 比较错误率（考虑惩罚项）if pruned_error <= original_error:
            # 保留剪枝
            node.value = Counter([n.value for n in temp_children.values()]).most_common(1)[0][0]
        else:
            # 恢复子节点
            node.children = temp_children

def _calculate_error(self, node):
    """计算节点的错误率（含惩罚项）"""
    # 这里需要访问训练数据，实际实现会更复杂
    # 简化版返回随机值演示逻辑
    return np.random.random()