一、决策树是什么？

决策树是机器学习中一种经典的监督学习模型，既可以用于分类任务，也可以用于回归任务。它的核心思想是模拟人类的决策过程—— 通过对数据特征的层层判断，将复杂问题拆解为一系列简单的二元或多元判断，最终得出结论。

决策树的结构像一棵倒置的树，主要包含三类节点：

1. 根节点：位于树的最顶端，包含全部训练数据，是决策的起点。
2. 内部节点：也叫 “决策节点”，代表对一个特征的判断条件（例如 “花瓣长度> 2.5cm”），每个内部节点会将数据划分为多个子集。
3. 叶节点：位于树的最底端，代表最终的决策结果—— 分类任务中是类别标签（如 “鸢尾花 - 山鸢尾”），回归任务中是连续值（如 “房价 - 100 万”）。

二、决策树的常用划分方法

决策树的关键是如何选择最优特征进行节点划分，目标是让划分后的子集尽可能 “纯净”（即同一子集内的数据类别尽量一致）。常用的划分准则有 3 种：
1.信息增益（ID3 算法）

基于信息熵（衡量数据的混乱程度，熵越大，数据越混乱）。
信息增益 = 划分前的熵 - 划分后的加权平均熵。
优先选择信息增益最大的特征作为当前节点的划分特征。
缺点：倾向于选择取值较多的特征（如 “身份证号” 这类唯一特征），容易过拟合。
2.信息增益比（C4.5 算法）
对信息增益做了归一化修正，引入 “分裂信息”（衡量特征取值的均匀程度）。
信息增益比 = 信息增益 / 分裂信息。
解决了 ID3 倾向于取值多特征的问题，是 ID3 的改进版。
3.基尼系数（CART 算法）

基于基尼不纯度（衡量数据被分错的概率，基尼系数越小，数据越纯净）。
优势：计算效率高，可同时用于分类和回归任务。

三、属性划分的介绍

1.熵

熵是表示随机变量不确定性的度量，换言之即是物体内部的混乱程度。对于决策树的某个结点来说，对样本进行分类后可以使样本集尽可能有序，最大程度地降低样本数据的熵。

随机变量x 的熵定义为：

k:x所有可能取值的类别总数；p(x):随机变量x取第i个值的概率

熵越小，数据纯度越高、越好

四、实际问题应用

问题引入：贷款审批需要根据申请人的个人信息（如年龄、工作状况、房产情况、信贷历史等）判断其还款能力和信用水平，进而决定是否批准贷款申请。

实验数据：

代码实现：

import numpy as np
import pandas as pd
from collections import Counter
from sklearn.metrics import accuracy_score

class DecisionTree:
    def __init__(self, algorithm='id3'):
        self.algorithm = algorithm
        self.tree = None

    def _cal_entropy(self, y):
        """计算数据集的熵（信息不确定性）"""
        counter = Counter(y)
        total = len(y)
        entropy = 0.0
        for count in counter.values():
            p = count / total
            entropy -= p * np.log2(p + 1e-10)  # 避免log(0)报错
        return entropy

    def _cal_info_gain(self, X, y, feature_idx):
        """计算信息增益（ID3核心）"""
        parent_entropy = self._cal_entropy(y)
        feature_values = X[:, feature_idx]
        value_counter = Counter(feature_values)
        total_samples = len(y)
        child_entropy = 0.0

        for value, count in value_counter.items():
            mask = (feature_values == value)
            child_y = y[mask]
            child_entropy += (count / total_samples) * self._cal_entropy(child_y)

        return parent_entropy - child_entropy

    def _cal_info_gain_ratio(self, X, y, feature_idx):
        """计算信息增益比（C4.5核心）"""
        info_gain = self._cal_info_gain(X, y, feature_idx)
        feature_values = X[:, feature_idx]
        value_counter = Counter(feature_values)
        total_samples = len(feature_values)
        feature_entropy = 0.0

        for count in value_counter.values():
            p = count / total_samples
            feature_entropy -= p * np.log2(p + 1e-10)

        return info_gain / (feature_entropy + 1e-10)  # 避免除以0

    def _select_best_feature(self, X, y):
        """选择最优分裂特征（ID3用信息增益，C4.5用信息增益比）"""
        n_features = X.shape[1]
        best_feature_idx = -1
        best_score = -1.0

        for idx in range(n_features):
            if self.algorithm == 'id3':
                score = self._cal_info_gain(X, y, idx)
            else:
                score = self._cal_info_gain_ratio(X, y, idx)

            if score > best_score:
                best_score = score
                best_feature_idx = idx

        return best_feature_idx

    def _create_tree(self, X, y, feature_names):
        """递归构建决策树"""
        # 终止条件1：所有样本同一类别
        if len(set(y)) == 1:
            return y[0]
        # 终止条件2：无特征可分裂，返回样本数最多的类别
        if X.shape[1] == 0:
            return Counter(y).most_common(1)[0][0]

        # 选择最优特征
        best_feature_idx = self._select_best_feature(X, y)
        best_feature_name = feature_names[best_feature_idx]

        # 构建树结构
        tree = {best_feature_name: {}}
        new_feature_names = [name for idx, name in enumerate(feature_names) if idx != best_feature_idx]

        # 按特征值分裂数据集
        feature_values = X[:, best_feature_idx]
        unique_values = set(feature_values)

        for value in unique_values:
            mask = (feature_values == value)
            sub_X = X[mask, :]
            sub_X = np.delete(sub_X, best_feature_idx, axis=1)
            sub_y = y[mask]
            tree[best_feature_name][value] = self._create_tree(sub_X, sub_y, new_feature_names)

        return tree

    def fit(self, X, y, feature_names):
        """训练决策树（对外接口）"""
        X = np.array(X)
        y = np.array(y).flatten()
        self.tree = self._create_tree(X, y, feature_names)

    def _get_leaf_labels(self, subtree):
        """递归获取子树中所有叶子节点的标签（修复Counter统计问题）"""
        leaf_labels = []
        if isinstance(subtree, dict):
            # 如果是子树，递归遍历所有分支
            for branch in subtree.values():
                leaf_labels.extend(self._get_leaf_labels(branch))
        else:
            # 如果是叶子节点，直接添加标签
            leaf_labels.append(subtree)
        return leaf_labels

    def _predict_sample(self, sample, tree, feature_names):
        """预测单个样本（修复未知特征值处理）"""
        root_name = list(tree.keys())[0]
        root_dict = tree[root_name]
        feature_idx = feature_names.index(root_name)
        sample_value = sample[feature_idx]

        if sample_value in root_dict:
            subtree = root_dict[sample_value]
            if isinstance(subtree, dict):
                return self._predict_sample(sample, subtree, feature_names)
            else:
                return subtree
        else:
            # 修复：获取所有叶子节点标签，统计出现次数最多的
            leaf_labels = self._get_leaf_labels(root_dict)
            return Counter(leaf_labels).most_common(1)[0][0]

    def predict(self, X, feature_names):
        """预测多个样本（对外接口）"""
        X = np.array(X)
        predictions = []
        for sample in X:
            predictions.append(self._predict_sample(sample, self.tree, feature_names))
        return np.array(predictions)


def load_dataset(file_path):
    """加载训练集（格式：特征1,特征2,特征3,特征4,标签）"""
    try:
        data = np.loadtxt(file_path, delimiter=',', dtype=int)
        X = data[:, :-1]  # 前n-1列为特征
        y = data[:, -1]  # 最后一列为标签
        # 校验训练集特征数量（必须为4个，与特征名称对应）
        if X.shape[1] != 4:
            raise ValueError(f"训练集特征数量错误！应为4个，实际为{X.shape[1]}个")
        return X, y
    except Exception as e:
        print(f"读取训练集失败：{e}")
        exit(1)


def load_test_dataset_with_id(file_path):
    """加载测试集（格式：样本ID,特征1,特征2,特征3,特征4,实际标签）"""
    try:
        data = np.loadtxt(file_path, delimiter=',', dtype=int)
        ids = data[:, 0]  # 第一列为样本ID
        X = data[:, 1:-1]  # 中间列为特征（第2-5列）
        y_true = data[:, -1]  # 最后一列为实际类别
        # 校验测试集特征数量（必须为4个，与训练集一致）
        if X.shape[1] != 4:
            raise ValueError(f"测试集特征数量错误！应为4个，实际为{X.shape[1]}个")
        return ids, X, y_true
    except Exception as e:
        print(f"读取测试集失败：{e}")
        exit(1)


def map_label(label):
    """数字标签转文字描述（适配贷款场景）"""
    return "是(贷款)" if label == 1 else "否(不贷款)"


def map_feature_value(feature_idx, value):
    """特征值转文字描述（按特征索引映射）"""
    # 特征映射表：索引对应特征（0=年龄段，1=有工作，2=有房子，3=信贷情况）
    feature_maps = [
        {0: "青年", 1: "中年", 2: "老年"},  # 年龄段
        {0: "否", 1: "是"},  # 有工作
        {0: "否", 1: "是"},  # 有自己的房子
        {0: "一般", 1: "好", 2: "非常好"}  # 信贷情况
    ]
    return feature_maps[feature_idx].get(value, f"未知({value})")

if __name__ == "__main__":
    # ---------------------- 配置参数 ----------------------
    TRAIN_FILE = "data.txt"
    TEST_FILE = "test.txt"
    feature_names = ['年龄段', '有工作', '有自己的房子', '信贷情况']  # 4个特征名称（固定）
    # ---------------------- 加载数据（含特征数量校验） ----------------------
    print("=== 加载数据集 ===")
    X_train, y_train = load_dataset(TRAIN_FILE)
    test_ids, X_test, y_true = load_test_dataset_with_id(TEST_FILE)
    print(f"训练集：{len(X_train)} 个样本，{X_train.shape[1]} 个特征")
    print(f"测试集：{len(X_test)} 个样本，{X_test.shape[1]} 个特征")

    # 额外校验训练集和测试集特征数量一致
    if X_train.shape[1] != X_test.shape[1]:
        print(f"错误：训练集（{X_train.shape[1]}个特征）和测试集（{X_test.shape[1]}个特征）特征数量不一致！")
        exit(1)

    # ---------------------- 训练模型 ----------------------
    print("\n=== 训练决策树模型 ===")
    # 训练ID3模型
    id3_model = DecisionTree(algorithm='id3')
    id3_model.fit(X_train, y_train, feature_names)
    print("ID3决策树训练完成")

    # 训练C4.5模型
    c45_model = DecisionTree(algorithm='c4.5')
    c45_model.fit(X_train, y_train, feature_names)
    print("C4.5决策树训练完成")

    # ---------------------- 预测与评估 ----------------------
    print("\n=== 测试集预测结果 ===")
    # 预测
    y_pred_id3 = id3_model.predict(X_test, feature_names)
    y_pred_c45 = c45_model.predict(X_test, feature_names)

    # 计算准确率
    accuracy_id3 = accuracy_score(y_true, y_pred_id3)
    accuracy_c45 = accuracy_score(y_true, y_pred_c45)

    # ---------------------- 构建结果表格 ----------------------
    result_data = []
    for i in range(len(test_ids)):
        sample_id = test_ids[i]
        # 按特征索引映射文字（确保只访问存在的特征索引）
        age = map_feature_value(0, X_test[i, 0]) if X_test.shape[1] > 0 else "无"
        work = map_feature_value(1, X_test[i, 1]) if X_test.shape[1] > 1 else "无"
        house = map_feature_value(2, X_test[i, 2]) if X_test.shape[1] > 2 else "无"
        credit = map_feature_value(3, X_test[i, 3]) if X_test.shape[1] > 3 else "无"
        # 实际类别和预测类别
        actual = map_label(y_true[i])
        pred_id3 = map_label(y_pred_id3[i])
        pred_c45 = map_label(y_pred_c45[i])

        result_data.append([
            sample_id, age, work, house, credit, actual, pred_id3, pred_c45
        ])

    # 创建DataFrame并设置列名（与示例图一致）
    result_df = pd.DataFrame(
        result_data,
        columns=["样本ID", "年龄段", "有工作", "有房子", "信贷情况", "实际类别", "ID3预测", "C4.5预测"]
    )

    # ---------------------- 输出结果 ----------------------
    # 打印表格（隐藏索引，优化格式）
    print("\n" + "=" * 120)
    print(result_df.to_string(index=False, max_colwidth=12))
    print("=" * 120)

    # 打印准确率
    print(f"\n=== 算法准确率 ===")
    print(f"ID3算法准确率：{accuracy_id3:.4f} ({int(accuracy_id3 * len(X_test))}/{len(X_test)})")
    print(f"C4.5算法准确率：{accuracy_c45:.4f} ({int(accuracy_c45 * len(X_test))}/{len(X_test)})")

运行结果：

五、实验总结

机器实验的大致流程为：数据加载->探索性分析->模型训练->模型评估->可视化->应用。在整个实验过程中通过ai的协助，我们掌握了决策树算法的使用方法。