集成学习：Bagging, Boosting, Stacking

import numpy as np
import matplotlib.pyplot as plt

def bias_variance_decomposition():
    """可视化偏差-方差分解"""
    np.random.seed(42)

    # 真实函数
    def true_function(x):
        return np.sin(1.5 * x)

    # 生成多组训练集，训练不同复杂度模型
    n_datasets = 50
    x_range = np.linspace(-3, 3, 200)
    y_true = true_function(x_range)

    fig, axes = plt.subplots(1, 3, figsize=(15, 4))

    for idx, (degree, title) in enumerate([
        (1, "高偏差（欠拟合）"),
        (3, "均衡（良好）"),
        (15, "高方差（过拟合）")
    ]):
        predictions = np.zeros((n_datasets, len(x_range)))

        for i in range(n_datasets):
            # 每次用不同随机样本训练
            X_train = np.random.uniform(-3, 3, 20)
            noise = np.random.normal(0, 0.3, 20)
            y_train = true_function(X_train) + noise
            coeffs = np.polyfit(X_train, y_train, degree)
            predictions[i] = np.polyval(coeffs, x_range)

        avg_prediction = np.mean(predictions, axis=0)
        variance = np.var(predictions, axis=0)
        bias_sq = (avg_prediction - y_true) ** 2

        ax = axes[idx]
        ax.plot(x_range, y_true, 'k-', linewidth=2, label='真实函数')
        for i in range(min(10, n_datasets)):
            ax.plot(x_range, predictions[i], 'b-', alpha=0.15)
        ax.plot(x_range, avg_prediction, 'r--', linewidth=2,
                label='平均预测')
        ax.set_title(f"{title}\nBias^2={np.mean(bias_sq):.3f}, "
                     f"Var={np.mean(variance):.3f}")
        ax.legend()
        ax.set_ylim(-3, 3)

    plt.tight_layout()
    plt.show()

bias_variance_decomposition()

import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
from sklearn.tree import DecisionTreeRegressor
from sklearn.datasets import make_friedman1

# 验证集成如何降低方差
X, y = make_friedman1(n_samples=2000, noise=0.5, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 单棵深树（高方差）
single_tree = DecisionTreeRegressor(max_depth=15, random_state=42)
single_tree.fit(X_train, y_train)
single_mse = mean_squared_error(y_test, single_tree.predict(X_test))

# 集成：多棵树的平均预测
n_estimators = 100
ensemble_preds = np.zeros(len(y_test))
for i in range(n_estimators):
    bootstrap_idx = np.random.choice(
        len(X_train), len(X_train), replace=True)
    X_boot, y_boot = X_train[bootstrap_idx], y_train[bootstrap_idx]
    tree = DecisionTreeRegressor(max_depth=15, random_state=i)
    tree.fit(X_boot, y_boot)
    ensemble_preds += tree.predict(X_test) / n_estimators

ensemble_mse = mean_squared_error(y_test, ensemble_preds)

print(f"单棵深树 MSE:  {single_mse:.4f}")
print(f"集成 {n_estimators} 棵树 MSE: {ensemble_mse:.4f}")
print(f"方差降低比例: {(1 - ensemble_mse/single_mse)*100:.1f}%")

import numpy as np
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import make_moons
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt

class SimpleBaggingClassifier:
    """从零实现 Bagging 分类器——理解核心逻辑"""

    def __init__(self, n_estimators=50, max_depth=None, random_state=42):
        self.n_estimators = n_estimators
        self.max_depth = max_depth
        self.random_state = random_state
        self.trees = []
        self.oob_scores = []

    def fit(self, X, y):
        np.random.seed(self.random_state)
        self.trees = []
        n = len(X)

        for i in range(self.n_estimators):
            # Bootstrap 采样（有放回）
            indices = np.random.choice(n, n, replace=True)
            X_boot, y_boot = X[indices], y[indices]

            # 训练决策树
            tree = DecisionTreeClassifier(
                max_depth=self.max_depth, random_state=i)
            tree.fit(X_boot, y_boot)
            self.trees.append(tree)

            # 计算 OOB 分数
            oob_mask = np.ones(n, dtype=bool)
            oob_mask[np.unique(indices)] = False
            if oob_mask.sum() > 0:
                oob_score = tree.score(X[oob_mask], y[oob_mask])
                self.oob_scores.append(oob_score)

        print(f"OOB 平均准确率: "
              f"{np.mean(self.oob_scores):.4f}")

    def predict(self, X):
        # 多数表决
        predictions = np.array(
            [tree.predict(X) for tree in self.trees])
        return np.round(predictions.mean(axis=0)).astype(int)

from sklearn.datasets import make_moons
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt
import numpy as np

# 生成非线性数据
X, y = make_moons(n_samples=500, noise=0.3, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 训练 Bagging 分类器
bagging = SimpleBaggingClassifier(n_estimators=50, max_depth=5)
bagging.fit(X_train, y_train)

# 绘制决策边界
def plot_decision_boundary(model, X, y, title):
    x_min, x_max = X[:, 0].min() - 0.5, X[:, 0].max() + 0.5
    y_min, y_max = X[:, 1].min() - 0.5, X[:, 1].max() + 0.5
    xx, yy = np.meshgrid(
        np.linspace(x_min, x_max, 200),
        np.linspace(y_min, y_max, 200))
    Z = model.predict(np.c_[xx.ravel(), yy.ravel()]).reshape(xx.shape)

    plt.contourf(xx, yy, Z, alpha=0.3, cmap='viridis')
    plt.scatter(X[:, 0], X[:, 1], c=y, cmap='viridis',
                edgecolors='k', s=30)
    plt.title(title)
    plt.show()

# 对比单棵树 vs Bagging
single_tree = DecisionTreeClassifier(max_depth=5, random_state=42)
single_tree.fit(X_train, y_train)
print(f"单棵树准确率: {single_tree.score(X_test, y_test):.4f}")
print(f"Bagging 准确率: {bagging.score(X_test, y_test):.4f}")

plot_decision_boundary(single_tree, X_test, y_test, "单棵决策树")
plot_decision_boundary(bagging, X_test, y_test, "Bagging (50 棵树)")

import numpy as np
from sklearn.tree import DecisionTreeClassifier

class SimpleAdaBoost:
    """从零实现 AdaBoost——理解权重更新机制"""

    def __init__(self, n_estimators=50):
        self.n_estimators = n_estimators
        self.stumps = []
        self.alpha = []  # 每个模型的投票权重

    def fit(self, X, y):
        n = len(X)
        # 初始权重均匀分布
        w = np.ones(n) / n

        for t in range(self.n_estimators):
            # 训练一个浅层决策树（桩）
            stump = DecisionTreeClassifier(max_depth=1)
            stump.fit(X, y, sample_weight=w)
            predictions = stump.predict(X)

            # 计算加权错误率
            error = np.sum(w[predictions != y]) / np.sum(w)
            error = np.clip(error, 1e-10, 1 - 1e-10)

            # 计算模型权重
            alpha_t = 0.5 * np.log((1 - error) / error)

            # 更新样本权重：分错的样本权重增加
            w *= np.exp(-alpha_t * y * (2 * predictions - 1))
            w /= np.sum(w)

            self.stumps.append(stump)
            self.alpha.append(alpha_t)

    def predict(self, X):
        # 加权投票
        predictions = np.zeros(len(X))
        for stump, alpha in zip(self.stumps, self.alpha):
            predictions += alpha * stump.predict(X)
        return np.sign(predictions)

from sklearn.ensemble import GradientBoostingClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt
import numpy as np

# 梯度提升的逐步学习过程
X, y = make_classification(n_samples=1000, n_features=10,
                           n_informative=5, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 不同树数量的效果
n_estimators_range = [1, 3, 5, 10, 20, 50, 100, 200]
train_scores = []
test_scores = []

for n_est in n_estimators_range:
    gb = GradientBoostingClassifier(
        n_estimators=n_est, max_depth=3,
        learning_rate=0.1, random_state=42)
    gb.fit(X_train, y_train)
    train_scores.append(gb.score(X_train, y_train))
    test_scores.append(gb.score(X_test, y_test))

plt.figure(figsize=(8, 5))
plt.plot(n_estimators_range, train_scores, 'b-o',
         label='训练集准确率')
plt.plot(n_estimators_range, test_scores, 'r-o',
         label='测试集准确率')
plt.xlabel('树的数量')
plt.ylabel('准确率')
plt.title('梯度提升：树数量 vs 性能')
plt.legend()
plt.grid(alpha=0.3)
plt.show()

# 找到最佳树数量
best_idx = np.argmax(test_scores)
print(f"最佳树数量: {n_estimators_range[best_idx]}")
print(f"最高测试准确率: {test_scores[best_idx]:.4f}")

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_predict, StratifiedKFold
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# 生成数据
X, y = make_classification(n_samples=2000, n_features=15,
                           n_informative=8, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 第一层：多个不同类型的基模型
level1_models = {
    'rf': RandomForestClassifier(n_estimators=50, random_state=42),
    'svm': SVC(probability=True, random_state=42),
    'knn': KNeighborsClassifier(n_neighbors=7),
}

# 用交叉验证生成元特征
kf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
meta_features_train = np.zeros((len(X_train), len(level1_models)))
meta_features_test = np.zeros((len(X_test), len(level1_models)))

for i, (name, model) in enumerate(level1_models.items()):
    # 训练集上的 OOF 预测
    meta_features_train[:, i] = cross_val_predict(
        model, X_train, y_train, cv=kf, method='predict_proba')[:, 1]
    # 在完整训练集上训练后预测测试集
    model.fit(X_train, y_train)
    meta_features_test[:, i] = model.predict_proba(X_test)[:, 1]
    print(f"{name} 单独准确率: "
          f"{accuracy_score(y_test, model.predict(X_test)):.4f}")

# 第二层：元学习器
meta_learner = LogisticRegression(random_state=42)
meta_learner.fit(meta_features_train, y_train)
stacking_pred = meta_learner.predict(meta_features_test)
print(f"\nStacking 集成准确率: "
      f"{accuracy_score(y_test, stacking_pred):.4f}")

import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import make_classification

# 可视化元学习器如何组合基模型
np.random.seed(42)

# 模拟 3 个基模型的预测概率
n_samples = 500
model_a_pred = np.clip(
    np.random.beta(3, 2, n_samples), 0.01, 0.99)
model_b_pred = np.clip(
    np.random.beta(2, 3, n_samples), 0.01, 0.99)
model_c_pred = np.clip(
    np.random.beta(2.5, 2.5, n_samples), 0.01, 0.99)

meta_X = np.column_stack(
    [model_a_pred, model_b_pred, model_c_pred])
meta_y = (model_a_pred * 0.5 + model_b_pred * 0.3
          + model_c_pred * 0.2 + np.random.normal(0, 0.05, n_samples) > 0.5
          ).astype(int)

meta_learner = LogisticRegression()
meta_learner.fit(meta_X, meta_y)

print("元学习器学到的权重:")
for name, weight in zip(
        ['模型 A', '模型 B', '模型 C'], meta_learner.coef_[0]):
    print(f"  {name}: {weight:.4f}")

# 不同组合策略对比
strategies = {
    '等权平均': (model_a_pred + model_b_pred + model_c_pred) / 3 > 0.5,
    'Stacking': meta_learner.predict(meta_X),
    '只看模型 A': model_a_pred > 0.5,
}
for name, pred in strategies.items():
    acc = np.mean(pred == meta_y)
    print(f"{name:>12s}: {acc:.4f}")

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

def analyze_ensemble_diversity():
    """分析集成中多样性和准确率的关系"""
    np.random.seed(42)
    n_models = 20
    n_samples = 1000

    # 模拟不同多样性水平下的集成效果
    diversities = np.linspace(0.1, 0.9, 20)
    ensemble_errors = []

    for div in diversities:
        # 模拟 n_models 个基模型的预测
        individual_error = 0.3  # 每个模型 30% 错误率

        # 多样性越高，错误重叠越少
        overlap = 1 - div
        errors_per_model = np.random.binomial(
            1, individual_error, (n_models, n_samples))

        # 通过重叠度控制模型间错误相关性
        for i in range(1, n_models):
            correlated = np.random.binomial(1, overlap, n_samples)
            independent = np.random.binomial(
                1, individual_error, n_samples)
            errors_per_model[i] = correlated * errors_per_model[0] +                 (1 - correlated) * independent

        # 多数表决
        ensemble_pred = np.round(
            errors_per_model.mean(axis=0)).astype(int)
        ensemble_error = np.mean(ensemble_pred)
        ensemble_errors.append(ensemble_error)

    plt.figure(figsize=(8, 5))
    plt.plot(diversities, ensemble_errors, 'b-o', linewidth=2)
    plt.xlabel('多样性水平')
    plt.ylabel('集成错误率')
    plt.title('多样性 vs 集成性能')
    plt.grid(alpha=0.3)
    plt.axhline(y=0.3, color='r', linestyle='--',
                label='单模型错误率 (30%)')
    plt.legend()
    plt.show()

analyze_ensemble_diversity()

from sklearn.ensemble import (
    RandomForestClassifier, VotingClassifier, BaggingClassifier)
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# 对比不同多样性策略的效果
X, y = make_classification(n_samples=2000, n_features=20,
                           n_informative=10, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 策略 1：同质模型（低多样性）
same_trees = [
    DecisionTreeClassifier(max_depth=5, random_state=i)
    for i in range(10)
]
voting_same = VotingClassifier(
    estimators=[(f'tree_{i}', m) for i, m in enumerate(same_trees)],
    voting='soft')
voting_same.fit(X_train, y_train)

# 策略 2：异构模型（高多样性）
diverse_models = [
    ('rf', RandomForestClassifier(n_estimators=20, random_state=42)),
    ('dt', DecisionTreeClassifier(max_depth=8, random_state=42)),
    ('knn', BaggingClassifier(
        DecisionTreeClassifier(max_depth=8),
        n_estimators=20, random_state=42)),
]
voting_diverse = VotingClassifier(
    estimators=diverse_models, voting='soft')
voting_diverse.fit(X_train, y_train)

print(f"同质模型集成准确率: "
      f"{accuracy_score(y_test, voting_same.predict(X_test)):.4f}")
print(f"异构模型集成准确率: "
      f"{accuracy_score(y_test, voting_diverse.predict(X_test)):.4f}")

# 随机森林：特征多样性
rf = RandomForestClassifier(
    n_estimators=100, max_features='sqrt', random_state=42)
rf.fit(X_train, y_train)
print(f"随机森林 (特征多样性): "
      f"{accuracy_score(y_test, rf.predict(X_test)):.4f}")

from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import (
    RandomForestClassifier, GradientBoostingClassifier,
    VotingClassifier)
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import accuracy_score
import time

# 性能 vs 成本对比实验
X, y = make_classification(n_samples=10000, n_features=30,
                           n_informative=15, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

models = {
    "逻辑回归": LogisticRegression(max_iter=1000),
    "决策树": DecisionTreeClassifier(random_state=42),
    "随机森林 (50)": RandomForestClassifier(
        n_estimators=50, random_state=42),
    "随机森林 (200)": RandomForestClassifier(
        n_estimators=200, random_state=42),
    "梯度提升 (100)": GradientBoostingClassifier(
        n_estimators=100, random_state=42),
    "集成投票": VotingClassifier([
        ('rf', RandomForestClassifier(n_estimators=50, random_state=42)),
        ('gb', GradientBoostingClassifier(n_estimators=50, random_state=42)),
        ('lr', LogisticRegression(max_iter=1000))
    ], voting='soft'),
}

print(f"{'模型':<20s} | {'准确率':>8s} | {'训练时间':>8s} | {'推理时间':>10s}")
print("-" * 55)

for name, model in models.items():
    t0 = time.time()
    model.fit(X_train, y_train)
    train_time = time.time() - t0

    t0 = time.time()
    model.predict(X_test)
    infer_time = time.time() - t0

    acc = accuracy_score(y_test, model.predict(X_test))
    print(f"{name:<20s} | {acc:>7.4f} | {train_time:>7.3f}s | "
          f"{infer_time*1000:>9.1f}ms")

import numpy as np
import matplotlib.pyplot as plt

def analyze_model_scaling():
    """分析模型数量增加时的性能收益递减"""
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.datasets import make_classification
    from sklearn.model_selection import train_test_split

    X, y = make_classification(n_samples=5000, n_features=20,
                               n_informative=10, random_state=42)
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=42)

    n_estimators_list = [1, 2, 5, 10, 20, 50, 100, 200, 500]
    accuracies = []

    for n in n_estimators_list:
        rf = RandomForestClassifier(
            n_estimators=n, random_state=42)
        rf.fit(X_train, y_train)
        acc = rf.score(X_test, y_test)
        accuracies.append(acc)
        print(f"n_estimators={n:>4d}: 准确率={acc:.4f}")

    plt.figure(figsize=(8, 5))
    plt.plot(n_estimators_list, accuracies, 'b-o', linewidth=2)
    plt.xscale('log')
    plt.xlabel('树的数量 (对数尺度)')
    plt.ylabel('测试集准确率')
    plt.title('模型数量 vs 性能收益递减')
    plt.grid(alpha=0.3)
    plt.show()

analyze_model_scaling()

from sklearn.ensemble import (
    VotingClassifier, StackingClassifier,
    RandomForestClassifier, GradientBoostingClassifier)
from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression
from sklearn.neighbors import KNeighborsClassifier
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report
import numpy as np

# 加载数据
data = load_breast_cancer()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.25, random_state=42)

# 基模型集合
estimators = [
    ('rf', RandomForestClassifier(n_estimators=100, random_state=42)),
    ('svm', SVC(probability=True, random_state=42)),
    ('knn', KNeighborsClassifier(n_neighbors=7)),
    ('gb', GradientBoostingClassifier(n_estimators=50, random_state=42)),
]

# 硬投票
hard_voting = VotingClassifier(estimators=estimators, voting='hard')
# 软投票
soft_voting = VotingClassifier(estimators=estimators, voting='soft')
# Stacking
stacking = StackingClassifier(
    estimators=estimators,
    final_estimator=LogisticRegression(max_iter=1000),
    cv=5, stack_method='predict_proba', n_jobs=-1)

# 评估对比
for name, model in [('硬投票', hard_voting),
                     ('软投票', soft_voting),
                     ('Stacking', stacking)]:
    cv_score = cross_val_score(model, X_train, y_train, cv=5).mean()
    model.fit(X_train, y_train)
    test_score = model.score(X_test, y_test)
    print(f"{name:>10s}: CV={cv_score:.4f}, Test={test_score:.4f}")

print(f"\nStacking 详细报告:")
print(classification_report(
    y_test, stacking.predict(X_test), target_names=data.target_names))

from sklearn.ensemble import (
    StackingClassifier, RandomForestClassifier,
    GradientBoostingClassifier)
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import numpy as np

# 高级 Stacking：多层次的元特征分析
X, y = make_classification(n_samples=3000, n_features=20,
                           n_informative=10, n_classes=2, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42)

# 第一层基模型：三种不同类型
level1 = [
    ('rf', RandomForestClassifier(n_estimators=50, random_state=42)),
    ('gb', GradientBoostingClassifier(n_estimators=50, random_state=42)),
]

# 第二层 Stacking：用 RF 做最终估计器
stacking_rf = StackingClassifier(
    estimators=level1,
    final_estimator=RandomForestClassifier(
        n_estimators=100, random_state=42),
    cv=5, n_jobs=-1)

# 对比：用逻辑回归做最终估计器（默认）
stacking_lr = StackingClassifier(
    estimators=level1,
    final_estimator=LogisticRegression(max_iter=1000),
    cv=5, n_jobs=-1)

for name, model in [('Stacking+LR', stacking_lr),
                     ('Stacking+RF', stacking_rf)]:
    model.fit(X_train, y_train)
    print(f"{name}: Train={model.score(X_train, y_train):.4f}, "
          f"Test={model.score(X_test, y_test):.4f}")

# 查看元特征重要性（仅适用于树型 final_estimator）
stacking_rf.fit(X_train, y_train)
if hasattr(stacking_rf.final_estimator_, 'feature_importances_'):
    importances = stacking_rf.final_estimator_.feature_importances_
    for i, imp in enumerate(importances):
        print(f"  基模型 {i+1} 重要性: {imp:.4f}")