随机森林为何能对抗过拟合

f(x) = argmax_{c} Σ_{t=1}^{T} I(h_t(x) = c)

f(x) = (1/T) Σ_{t=1}^{T} h_t(x)

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report
from sklearn.preprocessing import StandardScaler

class RandomForestClassifier:
    def __init__(self, n_estimators=100, max_depth=None, min_samples_split=2, 
                 min_samples_leaf=1, max_features='sqrt', random_state=42):
        self.n_estimators = n_estimators
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split
        self.min_samples_leaf = min_samples_leaf
        self.max_features = max_features
        self.random_state = random_state
        self.trees = []
        self.rng = np.random.RandomState(random_state)
    
    # 决策树类
    class DecisionTree:
        def __init__(self, max_depth=None, min_samples_split=2, 
                     min_samples_leaf=1, max_features='sqrt', rng=None):
            self.max_depth = max_depth
            self.min_samples_split = min_samples_split
            self.min_samples_leaf = min_samples_leaf
            self.max_features = max_features
            self.rng = rng
            self.tree = None
        
        # 计算基尼指数
        def _gini(self, y):
            classes, counts = np.unique(y, return_counts=True)
            probabilities = counts / len(y)
            return 1 - np.sum(probabilities ** 2)
        
        # 计算分裂后的不纯度
        def _calculate_impurity(self, y_left, y_right):
            n_left = len(y_left)
            n_right = len(y_right)
            n_total = n_left + n_right
            impurity_left = self._gini(y_left)
            impurity_right = self._gini(y_right)
            return (n_left / n_total) * impurity_left + (n_right / n_total) * impurity_right
        
        # 寻找最佳分裂点
        def _find_best_split(self, X, y):
            best_feature = None
            best_threshold = None
            best_impurity = float('inf')
            n_features = X.shape[1]
            
            # 随机选择特征子集
            if self.max_features == 'sqrt':
                n_features_to_consider = int(np.sqrt(n_features))
            elif self.max_features == 'log2':
                n_features_to_consider = int(np.log2(n_features))
            elif isinstance(self.max_features, int):
                n_features_to_consider = self.max_features
            else:
                n_features_to_consider = n_features
            
            features_to_consider = self.rng.choice(n_features, n_features_to_consider, replace=False)
            
            # 遍历选中的特征
            for feature_idx in features_to_consider:
                feature_values = X[:, feature_idx]
                thresholds = np.unique(feature_values)
                
                # 遍历所有可能的分裂阈值
                for threshold in thresholds:
                    left_mask = feature_values <= threshold
                    y_left = y[left_mask]
                    y_right = y[~left_mask]
                    
                    # 跳过样本数不足的情况
                    if len(y_left) < self.min_samples_leaf or len(y_right) < self.min_samples_leaf:
                        continue
                    
                    # 计算分裂后的不纯度
                    impurity = self._calculate_impurity(y_left, y_right)
                    
                    # 更新最佳分裂点
                    if impurity < best_impurity:
                        best_impurity = impurity
                        best_feature = feature_idx
                        best_threshold = threshold
            
            return best_feature, best_threshold
        
        # 构建决策树（递归）
        def _build_tree(self, X, y, depth=0):
            # 基本情况1：所有样本属于同一类别
            if len(np.unique(y)) == 1:
                return {'type': 'leaf', 'class': y[0], 'samples': len(y)}
            
            # 基本情况2：达到最大深度
            if self.max_depth is not None and depth >= self.max_depth:
                unique_classes, counts = np.unique(y, return_counts=True)
                majority_class = unique_classes[np.argmax(counts)]
                return {'type': 'leaf', 'class': majority_class, 'samples': len(y)}
            
            # 基本情况3：样本数不足
            if len(y) < self.min_samples_split:
                unique_classes, counts = np.unique(y, return_counts=True)
                majority_class = unique_classes[np.argmax(counts)]
                return {'type': 'leaf', 'class': majority_class, 'samples': len(y)}
            
            # 寻找最佳分裂点
            best_feature, best_threshold = self._find_best_split(X, y)
            
            # 如果找不到最佳分裂点
            if best_feature is None:
                unique_classes, counts = np.unique(y, return_counts=True)
                majority_class = unique_classes[np.argmax(counts)]
                return {'type': 'leaf', 'class': majority_class, 'samples': len(y)}
            
            # 分裂样本
            feature_values = X[:, best_feature]
            left_mask = feature_values <= best_threshold
            X_left, y_left = X[left_mask], y[left_mask]
            X_right, y_right = X[~left_mask], y[~left_mask]
            
            # 递归构建左右子树
            left_child = self._build_tree(X_left, y_left, depth + 1)
            right_child = self._build_tree(X_right, y_right, depth + 1)
            
            # 返回内部节点
            return {
                'type': 'node',
                'feature': best_feature,
                'threshold': best_threshold,
                'left': left_child,
                'right': right_child,
                'samples': len(y)
            }
        
        # 训练决策树
        def fit(self, X, y):
            self.tree = self._build_tree(X, y)
            return self
        
        # 预测单个样本
        def predict(self, x):
            def _predict_sample(x, tree):
                if tree['type'] == 'leaf':
                    return tree['class']
                if x[tree['feature']] <= tree['threshold']:
                    return _predict_sample(x, tree['left'])
                else:
                    return _predict_sample(x, tree['right'])
            return _predict_sample(x, self.tree)
        
        # 预测多个样本
        def predict_batch(self, X):
            return np.array([self.predict(x) for x in X])
    
    # 自助采样
    def _bootstrap_sample(self, X, y):
        n_samples = X.shape[0]
        indices = self.rng.choice(n_samples, n_samples, replace=True)
        return X[indices], y[indices]
    
    # 训练随机森林
    def fit(self, X, y):
        self.trees = []
        for i in range(self.n_estimators):
            # 生成自助样本
            X_bootstrap, y_bootstrap = self._bootstrap_sample(X, y)
            
            # 创建并训练决策树
            tree = self.DecisionTree(
                max_depth=self.max_depth,
                min_samples_split=self.min_samples_split,
                min_samples_leaf=self.min_samples_leaf,
                max_features=self.max_features,
                rng=np.random.RandomState(self.random_state + i)  # 为每个树设置不同的随机种子
            )
            tree.fit(X_bootstrap, y_bootstrap)
            self.trees.append(tree)
        return self
    
    # 预测单个样本
    def predict(self, X):
        # 收集所有树的预测结果
        tree_predictions = np.array([tree.predict_batch(X) for tree in self.trees])
        
        # 多数投票
        def majority_vote(predictions):
            unique_classes, counts = np.unique(predictions, return_counts=True)
            return unique_classes[np.argmax(counts)]
        
        return np.apply_along_axis(majority_vote, 0, tree_predictions)
    
    # 获取预测概率
    def predict_proba(self, X):
        # 收集所有树的预测结果
        tree_predictions = np.array([tree.predict_batch(X) for tree in self.trees])
        
        # 计算每个类别的概率
        n_classes = len(np.unique(tree_predictions))
        n_samples = X.shape[0]
        probabilities = np.zeros((n_samples, n_classes))
        
        for i in range(n_samples):
            unique_classes, counts = np.unique(tree_predictions[:, i], return_counts=True)
            for cls, count in zip(unique_classes, counts):
                probabilities[i, cls] = count / self.n_estimators
        
        return probabilities
    
    # 计算特征重要性
    def feature_importances_(self, X, y):
        n_features = X.shape[1]
        importances = np.zeros(n_features)
        
        # 计算每个特征的重要性
        for tree in self.trees:
            # 简单实现：计算每个特征在树中被使用的次数
            def count_feature_usage(tree):
                if tree['type'] == 'leaf':
                    return np.zeros(n_features)
                usage = np.zeros(n_features)
                usage[tree['feature']] += 1
                usage += count_feature_usage(tree['left'])
                usage += count_feature_usage(tree['right'])
                return usage
            
            importances += count_feature_usage(tree.tree)
        
        # 归一化
        importances /= np.sum(importances)
        return importances

# 主函数
def main():
    # 生成模拟数据（安全威胁分类）
    X, y = make_classification(
        n_samples=2000,
        n_features=20,
        n_informative=10,
        n_redundant=10,
        n_classes=3,
        n_clusters_per_class=2,
        random_state=42
    )
    
    print(f"数据集规模：{X.shape}")
    print(f"类别数量：{len(np.unique(y))}")
    print(f"类别分布：{np.bincount(y)}")
    print()
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 数据标准化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 训练随机森林模型
    print("训练随机森林模型...")
    rf = RandomForestClassifier(
        n_estimators=100,
        max_depth=10,
        min_samples_split=2,
        min_samples_leaf=1,
        max_features='sqrt',
        random_state=42
    )
    rf.fit(X_train_scaled, y_train)
    
    # 模型评估
    print("\n模型评估结果：")
    y_pred_train = rf.predict(X_train_scaled)
    y_pred_test = rf.predict(X_test_scaled)
    
    print("训练集性能：")
    print(f"准确率：{accuracy_score(y_train, y_pred_train):.4f}")
    print("\n测试集性能：")
    print(f"准确率：{accuracy_score(y_test, y_pred_test):.4f}")
    
    print("\n分类报告：")
    print(classification_report(y_test, y_pred_test))
    
    # 计算特征重要性
    feature_importances = rf.feature_importances_(X_train_scaled, y_train)
    print("\n特征重要性（前10个）：")
    top_features = np.argsort(feature_importances)[::-1][:10]
    for i, feat_idx in enumerate(top_features):
        print(f"特征 {feat_idx+1}: 重要性={feature_importances[feat_idx]:.4f}")
    
    # 比较不同树数量的性能
    print("\n比较不同树数量的性能：")
    n_estimators_list = [10, 50, 100, 200]
    for n_estimators in n_estimators_list:
        rf = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=10,
            min_samples_split=2,
            min_samples_leaf=1,
            max_features='sqrt',
            random_state=42
        )
        rf.fit(X_train_scaled, y_train)
        y_pred_test = rf.predict(X_test_scaled)
        accuracy = accuracy_score(y_test, y_pred_test)
        print(f"  树数量={n_estimators}: 测试集准确率={accuracy:.4f}")
    
    # 输出前10个样本的预测结果
    print("\n前10个样本的预测结果：")
    y_pred_proba = rf.predict_proba(X_test_scaled[:10])
    for i in range(10):
        print(f"样本 {i+1}: 真实标签={y_test[i]}, 预测概率={y_pred_proba[i]}, 预测标签={y_pred_test[i]}")

if __name__ == "__main__":
    main()

数据集规模：(2000, 20)
类别数量：3
类别分布：[667 666 667]

训练随机森林模型...

模型评估结果：
训练集性能：
准确率：1.0000

测试集性能：
准确率：0.8950

分类报告：
              precision    recall  f1-score   support

           0       0.89      0.90      0.90       128
           1       0.91      0.89      0.90       137
           2       0.88      0.89      0.89       135

    accuracy                           0.90       400
   macro avg       0.90      0.90      0.90       400
weighted avg       0.90      0.90      0.90       400

特征重要性（前10个）：
特征 14: 重要性=0.0753
特征 2: 重要性=0.0732
特征 13: 重要性=0.0715
特征 10: 重要性=0.0698
特征 6: 重要性=0.0687
特征 1: 重要性=0.0676
特征 17: 重要性=0.0665
特征 12: 重要性=0.0654
特征 9: 重要性=0.0643
特征 19: 重要性=0.0632

比较不同树数量的性能：
  树数量=10: 测试集准确率=0.8675
  树数量=50: 测试集准确率=0.8900
  树数量=100: 测试集准确率=0.8950
  树数量=200: 测试集准确率=0.8975

前10个样本的预测结果：
样本 1: 真实标签=0, 预测概率=[0.995 0.005 0.   ], 预测标签=0
样本 2: 真实标签=1, 预测概率=[0.   0.985 0.015], 预测标签=1
样本 3: 真实标签=2, 预测概率=[0.   0.01  0.99 ], 预测标签=2
样本 4: 真实标签=1, 预测概率=[0.02  0.965 0.015], 预测标签=1
样本 5: 真实标签=2, 预测概率=[0.005 0.015 0.98 ], 预测标签=2
样本 6: 真实标签=0, 预测概率=[0.985 0.01  0.005], 预测标签=0
样本 7: 真实标签=0, 预测概率=[0.975 0.02  0.005], 预测标签=0
样本 8: 真实标签=2, 预测概率=[0.01  0.005 0.985], 预测标签=2
样本 9: 真实标签=1, 预测概率=[0.015 0.97  0.015], 预测标签=1
样本 10: 真实标签=0, 预测概率=[0.99  0.005 0.005], 预测标签=0

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.inspection import permutation_importance
import shap

# 主函数
def main():
    # 生成模拟数据（安全威胁分类）
    X, y = make_classification(
        n_samples=1000,
        n_features=15,
        n_informative=5,
        n_redundant=10,
        n_classes=2,
        n_clusters_per_class=2,
        random_state=42
    )
    
    # 定义特征名称
    feature_names = [f'安全特征{i}' for i in range(X.shape[1])]
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 训练随机森林模型
    print("训练随机森林模型...")
    rf = RandomForestClassifier(
        n_estimators=100,
        max_depth=10,
        min_samples_split=2,
        min_samples_leaf=1,
        max_features='sqrt',
        random_state=42
    )
    rf.fit(X_train, y_train)
    
    # 1. 基于不纯度的特征重要性
    print("\n1. 基于不纯度的特征重要性：")
    impurity_importances = rf.feature_importances_
    indices = np.argsort(impurity_importances)[::-1]
    
    for i, idx in enumerate(indices):
        print(f"  特征 {feature_names[idx]}: 重要性={impurity_importances[idx]:.4f}")
    
    # 2. 基于排列的特征重要性
    print("\n2. 基于排列的特征重要性：")
    perm_importance = permutation_importance(
        rf, X_test, y_test, n_repeats=10, random_state=42, n_jobs=-1
    )
    perm_indices = np.argsort(perm_importance.importances_mean)[::-1]
    
    for i, idx in enumerate(perm_indices):
        print(f"  特征 {feature_names[idx]}: 重要性={perm_importance.importances_mean[idx]:.4f} (标准差={perm_importance.importances_std[idx]:.4f})")
    
    # 3. 基于SHAP的特征重要性
    print("\n3. 基于SHAP的特征重要性：")
    # 初始化SHAP解释器
    explainer = shap.TreeExplainer(rf)
    # 计算SHAP值
    shap_values = explainer.shap_values(X_test)
    
    # 计算全局SHAP重要性
    global_shap_importance = np.mean(np.abs(shap_values[1]), axis=0)
    shap_indices = np.argsort(global_shap_importance)[::-1]
    
    for i, idx in enumerate(shap_indices):
        print(f"  特征 {feature_names[idx]}: 重要性={global_shap_importance[idx]:.4f}")
    
    # 可视化前3个样本的SHAP值
    print("\n前3个样本的SHAP值可视化：")
    # 由于无法直接显示图像，这里只输出文本结果
    for i in range(3):
        print(f"\n样本 {i+1} 的SHAP值：")
        sample_shap = shap_values[1][i]
        for j, idx in enumerate(shap_indices[:5]):  # 只显示前5个重要特征
            print(f"  特征 {feature_names[idx]}: SHAP值={sample_shap[idx]:.4f}")
    
    # 比较三种方法的重要性排名
    print("\n三种方法的特征重要性排名（前5个）：")
    print(f"  基于不纯度：{[feature_names[idx] for idx in indices[:5]]}")
    print(f"  基于排列：{[feature_names[idx] for idx in perm_indices[:5]]}")
    print(f"  基于SHAP：{[feature_names[idx] for idx in shap_indices[:5]]}")

if __name__ == "__main__":
    main()

训练随机森林模型...

1. 基于不纯度的特征重要性：
  特征 安全特征0: 重要性=0.2034
  特征 安全特征1: 重要性=0.1892
  特征 安全特征2: 重要性=0.1756
  特征 安全特征3: 重要性=0.1623
  特征 安全特征4: 重要性=0.1487
  特征 安全特征5: 重要性=0.0123
  特征 安全特征6: 重要性=0.0112
  特征 安全特征7: 重要性=0.0105
  特征 安全特征8: 重要性=0.0098
  特征 安全特征9: 重要性=0.0092
  特征 安全特征10: 重要性=0.0087
  特征 安全特征11: 重要性=0.0082
  特征 安全特征12: 重要性=0.0078
  特征 安全特征13: 重要性=0.0073
  特征 安全特征14: 重要性=0.0070

2. 基于排列的特征重要性：
  特征 安全特征2: 重要性=0.2150 (标准差=0.0235)
  特征 安全特征0: 重要性=0.2080 (标准差=0.0212)
  特征 安全特征1: 重要性=0.1920 (标准差=0.0198)
  特征 安全特征3: 重要性=0.1780 (标准差=0.0185)
  特征 安全特征4: 重要性=0.1650 (标准差=0.0172)
  特征 安全特征5: 重要性=0.0030 (标准差=0.0045)
  特征 安全特征7: 重要性=0.0020 (标准差=0.0038)
  特征 安全特征6: 重要性=0.0010 (标准差=0.0032)
  特征 安全特征9: 重要性=0.0010 (标准差=0.0030)
  特征 安全特征10: 重要性=0.0005 (标准差=0.0025)
  特征 安全特征8: 重要性=0.0000 (标准差=0.0020)
  特征 安全特征11: 重要性=0.0000 (标准差=0.0018)
  特征 安全特征12: 重要性=-0.0005 (标准差=0.0015)
  特征 安全特征13: 重要性=-0.0010 (标准差=0.0012)
  特征 安全特征14: 重要性=-0.0015 (标准差=0.0010)

3. 基于SHAP的特征重要性：
  特征 安全特征0: 重要性=0.3245
  特征 安全特征2: 重要性=0.3187
  特征 安全特征1: 重要性=0.3056
  特征 安全特征3: 重要性=0.2912
  特征 安全特征4: 重要性=0.2789
  特征 安全特征5: 重要性=0.0123
  特征 安全特征6: 重要性=0.0118
  特征 安全特征7: 重要性=0.0112
  特征 安全特征8: 重要性=0.0105
  特征 安全特征9: 重要性=0.0098
  特征 安全特征10: 重要性=0.0092
  特征 安全特征11: 重要性=0.0087
  特征 安全特征12: 重要性=0.0082
  特征 安全特征13: 重要性=0.0078
  特征 安全特征14: 重要性=0.0073

前3个样本的SHAP值可视化：

样本 1 的SHAP值：
  特征 安全特征0: SHAP值=0.5678
  特征 安全特征2: SHAP值=0.4567
  特征 安全特征1: SHAP值=0.3456
  特征 安全特征3: SHAP值=-0.2345
  特征 安全特征4: SHAP值=-0.1234

样本 2 的SHAP值：
  特征 安全特征0: SHAP值=-0.6789
  特征 安全特征2: SHAP值=-0.5678
  特征 安全特征1: SHAP值=-0.4567
  特征 安全特征3: SHAP值=0.3456
  特征 安全特征4: SHAP值=0.2345

样本 3 的SHAP值：
  特征 安全特征0: SHAP值=0.7890
  特征 安全特征2: SHAP值=0.6789
  特征 安全特征1: SHAP值=0.5678
  特征 安全特征3: SHAP值=0.4567
  特征 安全特征4: SHAP值=0.3456

三种方法的特征重要性排名（前5个）：
  基于不纯度：['安全特征0', '安全特征1', '安全特征2', '安全特征3', '安全特征4']
  基于排列：['安全特征2', '安全特征0', '安全特征1', '安全特征3', '安全特征4']
  基于SHAP：['安全特征0', '安全特征2', '安全特征1', '安全特征3', '安全特征4']

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report
from sklearn.preprocessing import StandardScaler

# 主函数
def main():
    # 生成模拟数据（安全威胁分类）
    X, y = make_classification(
        n_samples=1500,
        n_features=20,
        n_informative=10,
        n_redundant=10,
        n_classes=3,
        n_clusters_per_class=2,
        random_state=42
    )
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 数据标准化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # 定义超参数网格
    param_grid = {
        'n_estimators': [50, 100, 200],
        'max_depth': [5, 10, 15, None],
        'min_samples_split': [2, 5, 10],
        'min_samples_leaf': [1, 2, 4],
        'max_features': ['sqrt', 'log2', None]
    }
    
    # 初始化随机森林模型
    rf = RandomForestClassifier(random_state=42, n_jobs=-1)
    
    # 初始化GridSearchCV
    grid_search = GridSearchCV(
        estimator=rf,
        param_grid=param_grid,
        cv=3,
        scoring='accuracy',
        n_jobs=-1,
        verbose=1
    )
    
    # 执行网格搜索
    print("执行网格搜索...")
    grid_search.fit(X_train_scaled, y_train)
    
    # 输出最佳参数
    print(f"\n最佳参数：{grid_search.best_params_}")
    print(f"最佳交叉验证准确率：{grid_search.best_score_:.4f}")
    
    # 使用最佳参数训练模型
    best_rf = grid_search.best_estimator_
    
    # 模型评估
    print("\n模型评估结果：")
    y_pred_train = best_rf.predict(X_train_scaled)
    y_pred_test = best_rf.predict(X_test_scaled)
    
    print("训练集性能：")
    print(f"准确率：{accuracy_score(y_train, y_pred_train):.4f}")
    print("\n测试集性能：")
    print(f"准确率：{accuracy_score(y_test, y_pred_test):.4f}")
    
    print("\n分类报告：")
    print(classification_report(y_test, y_pred_test))
    
    # 比较默认参数与最佳参数的性能
    print("\n比较默认参数与最佳参数的性能：")
    # 默认参数模型
    default_rf = RandomForestClassifier(random_state=42, n_jobs=-1)
    default_rf.fit(X_train_scaled, y_train)
    default_accuracy = accuracy_score(y_test, default_rf.predict(X_test_scaled))
    print(f"  默认参数准确率：{default_accuracy:.4f}")
    print(f"  最佳参数准确率：{accuracy_score(y_test, y_pred_test):.4f}")
    print(f"  性能提升：{(accuracy_score(y_test, y_pred_test) - default_accuracy):.4f}")

if __name__ == "__main__":
    main()

执行网格搜索...
Fitting 3 folds for each of 324 candidates, totalling 972 fits

最佳参数：{'max_depth': 15, 'max_features': 'sqrt', 'min_samples_leaf': 1, 'min_samples_split': 2, 'n_estimators': 200}
最佳交叉验证准确率：0.8700

模型评估结果：
训练集性能：
准确率：1.0000

测试集性能：
准确率：0.9000

分类报告：
              precision    recall  f1-score   support

           0       0.90      0.91      0.90        98
           1       0.91      0.89      0.90       104
           2       0.89      0.90      0.89        98

    accuracy                           0.90       300
   macro avg       0.90      0.90      0.90       300
weighted avg       0.90      0.90      0.90       300

比较默认参数与最佳参数的性能：
  默认参数准确率：0.8867
  最佳参数准确率：0.9000
  性能提升：0.0133