アンサンブル学習の全体像 — Bagging・Boosting・Stacking

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 BaggingClassifierScratch:
    """Bagging分類器のスクラッチ実装"""

    def __init__(self, base_estimator=None, n_estimators=10, random_state=42):
        self.base_estimator = base_estimator or DecisionTreeClassifier(max_depth=None)
        self.n_estimators = n_estimators
        self.random_state = random_state
        self.estimators_ = []
        self.oob_score_ = None

    def fit(self, X, y):
        rng = np.random.RandomState(self.random_state)
        n_samples = X.shape[0]
        self.classes_ = np.unique(y)
        self.estimators_ = []

        # OOB予測を蓄積する配列
        oob_predictions = np.zeros((n_samples, len(self.classes_)))
        oob_counts = np.zeros(n_samples)

        for i in range(self.n_estimators):
            # ブートストラップサンプルを作成
            indices = rng.randint(0, n_samples, size=n_samples)
            oob_indices = np.setdiff1d(np.arange(n_samples), np.unique(indices))

            X_boot, y_boot = X[indices], y[indices]

            # 基本学習器を学習
            from sklearn.base import clone
            estimator = clone(self.base_estimator)
            estimator.fit(X_boot, y_boot)
            self.estimators_.append(estimator)

            # OOB予測を蓄積
            if len(oob_indices) > 0:
                oob_pred = estimator.predict(X[oob_indices])
                for idx, pred in zip(oob_indices, oob_pred):
                    class_idx = np.where(self.classes_ == pred)[0][0]
                    oob_predictions[idx, class_idx] += 1
                    oob_counts[idx] += 1

        # OOBスコアを計算
        valid = oob_counts > 0
        if valid.sum() > 0:
            oob_pred_labels = self.classes_[np.argmax(oob_predictions[valid], axis=1)]
            self.oob_score_ = np.mean(oob_pred_labels == y[valid])

        return self

    def predict(self, X):
        """多数決で予測"""
        predictions = np.array([est.predict(X) for est in self.estimators_])
        # 各サンプルについて最頻値を取得
        result = np.zeros(X.shape[0], dtype=self.classes_.dtype)
        for i in range(X.shape[0]):
            values, counts = np.unique(predictions[:, i], return_counts=True)
            result[i] = values[np.argmax(counts)]
        return result

    def predict_proba(self, X):
        """確率予測（各学習器の予測の平均）"""
        all_proba = np.array([
            est.predict_proba(X) for est in self.estimators_
        ])
        return np.mean(all_proba, axis=0)


# データ生成と実験
np.random.seed(42)
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
)

# 単一の決定木
single_tree = DecisionTreeClassifier(max_depth=None, random_state=42)
single_tree.fit(X_train, y_train)
print(f"単一決定木 - 訓練精度: {single_tree.score(X_train, y_train):.4f}")
print(f"単一決定木 - テスト精度: {single_tree.score(X_test, y_test):.4f}")

# Bagging（スクラッチ実装）
bagging = BaggingClassifierScratch(
    base_estimator=DecisionTreeClassifier(max_depth=None),
    n_estimators=50,
    random_state=42
)
bagging.fit(X_train, y_train)
bagging_train_acc = np.mean(bagging.predict(X_train) == y_train)
bagging_test_acc = np.mean(bagging.predict(X_test) == y_test)
print(f"\nBagging(50本) - 訓練精度: {bagging_train_acc:.4f}")
print(f"Bagging(50本) - テスト精度: {bagging_test_acc:.4f}")
print(f"Bagging(50本) - OOBスコア: {bagging.oob_score_:.4f}")

class AdaBoostClassifierScratch:
    """AdaBoost分類器のスクラッチ実装"""

    def __init__(self, n_estimators=50, random_state=42):
        self.n_estimators = n_estimators
        self.random_state = random_state
        self.estimators_ = []
        self.alphas_ = []

    def fit(self, X, y):
        n_samples = X.shape[0]
        # ラベルを{-1, +1}に変換
        self.classes_ = np.unique(y)
        y_binary = np.where(y == self.classes_[0], -1, 1)

        # 重みの初期化
        weights = np.ones(n_samples) / n_samples

        self.estimators_ = []
        self.alphas_ = []

        for m in range(self.n_estimators):
            # 弱学習器（決定株）を重み付きデータで学習
            stump = DecisionTreeClassifier(max_depth=1, random_state=self.random_state + m)
            stump.fit(X, y_binary, sample_weight=weights)

            # 予測
            pred = stump.predict(X)

            # 重み付き誤り率
            incorrect = (pred != y_binary)
            epsilon = np.sum(weights * incorrect) / np.sum(weights)

            # εが0.5以上なら学習を停止
            if epsilon >= 0.5:
                break

            # 学習器の重み
            alpha = 0.5 * np.log((1 - epsilon) / (epsilon + 1e-10))

            # データの重み更新
            weights = weights * np.exp(-alpha * y_binary * pred)
            weights = weights / np.sum(weights)

            self.estimators_.append(stump)
            self.alphas_.append(alpha)

        return self

    def predict(self, X):
        """加重投票で予測"""
        weighted_sum = np.zeros(X.shape[0])
        for alpha, estimator in zip(self.alphas_, self.estimators_):
            weighted_sum += alpha * estimator.predict(X)
        predictions = np.sign(weighted_sum)
        return np.where(predictions == -1, self.classes_[0], self.classes_[1])

    def staged_predict(self, X):
        """各ステップでの予測を返す（学習過程の可視化用）"""
        weighted_sum = np.zeros(X.shape[0])
        for alpha, estimator in zip(self.alphas_, self.estimators_):
            weighted_sum += alpha * estimator.predict(X)
            predictions = np.sign(weighted_sum)
            yield np.where(predictions == -1, self.classes_[0], self.classes_[1])


# AdaBoostの実験
adaboost = AdaBoostClassifierScratch(n_estimators=100, random_state=42)
adaboost.fit(X_train, y_train)
ada_train_acc = np.mean(adaboost.predict(X_train) == y_train)
ada_test_acc = np.mean(adaboost.predict(X_test) == y_test)
print(f"\nAdaBoost(100ステップ) - 訓練精度: {ada_train_acc:.4f}")
print(f"AdaBoost(100ステップ) - テスト精度: {ada_test_acc:.4f}")

# 学習曲線（ステップ数 vs 精度）
train_scores = []
test_scores = []
for pred_train in adaboost.staged_predict(X_train):
    train_scores.append(np.mean(pred_train == y_train))
for pred_test in adaboost.staged_predict(X_test):
    test_scores.append(np.mean(pred_test == y_test))

plt.figure(figsize=(8, 5))
steps = range(1, len(train_scores) + 1)
plt.plot(steps, train_scores, label="訓練精度", linewidth=2)
plt.plot(steps, test_scores, label="テスト精度", linewidth=2)
plt.xlabel("ステップ数（弱学習器の数）", fontsize=12)
plt.ylabel("精度", fontsize=12)
plt.title("AdaBoostの学習曲線", fontsize=14)
plt.legend(fontsize=11)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("adaboost_learning_curve.png", dpi=150, bbox_inches="tight")
plt.show()

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import KFold

class StackingClassifierScratch:
    """Stacking分類器のスクラッチ実装（交差検証ベース）"""

    def __init__(self, base_estimators, meta_estimator, cv=5, random_state=42):
        self.base_estimators = base_estimators  # (名前, モデル) のリスト
        self.meta_estimator = meta_estimator
        self.cv = cv
        self.random_state = random_state
        self.fitted_base_ = []

    def fit(self, X, y):
        from sklearn.base import clone
        kf = KFold(n_splits=self.cv, shuffle=True, random_state=self.random_state)

        n_samples = X.shape[0]
        n_estimators = len(self.base_estimators)

        # メタ特徴量行列
        meta_features = np.zeros((n_samples, n_estimators))

        # 交差検証でメタ特徴量を生成
        for train_idx, val_idx in kf.split(X):
            X_tr, X_val = X[train_idx], X[val_idx]
            y_tr = y[train_idx]

            for k, (name, estimator) in enumerate(self.base_estimators):
                est = clone(estimator)
                est.fit(X_tr, y_tr)
                meta_features[val_idx, k] = est.predict(X_val)

        # メタ学習器を学習
        self.meta_estimator_ = clone(self.meta_estimator)
        self.meta_estimator_.fit(meta_features, y)

        # 全データでベース学習器を再学習（テスト時用）
        self.fitted_base_ = []
        for name, estimator in self.base_estimators:
            est = clone(estimator)
            est.fit(X, y)
            self.fitted_base_.append((name, est))

        return self

    def predict(self, X):
        """テスト時の予測"""
        n_samples = X.shape[0]
        meta_features = np.zeros((n_samples, len(self.fitted_base_)))

        for k, (name, estimator) in enumerate(self.fitted_base_):
            meta_features[:, k] = estimator.predict(X)

        return self.meta_estimator_.predict(meta_features)


# ベース学習器の定義
base_estimators = [
    ("RandomForest", RandomForestClassifier(n_estimators=50, random_state=42)),
    ("SVM", SVC(kernel="rbf", random_state=42)),
    ("KNN", KNeighborsClassifier(n_neighbors=5)),
    ("DecisionTree", DecisionTreeClassifier(max_depth=5, random_state=42)),
]

# メタ学習器（ロジスティック回帰）
meta_estimator = LogisticRegression(random_state=42)

# Stacking
stacking = StackingClassifierScratch(
    base_estimators=base_estimators,
    meta_estimator=meta_estimator,
    cv=5,
    random_state=42
)
stacking.fit(X_train, y_train)
stack_train_acc = np.mean(stacking.predict(X_train) == y_train)
stack_test_acc = np.mean(stacking.predict(X_test) == y_test)
print(f"\nStacking - 訓練精度: {stack_train_acc:.4f}")
print(f"Stacking - テスト精度: {stack_test_acc:.4f}")

# 各ベース学習器の単体性能
print("\n--- ベース学習器の単体性能 ---")
for name, estimator in base_estimators:
    from sklearn.base import clone
    est = clone(estimator)
    est.fit(X_train, y_train)
    print(f"{name}: 訓練={est.score(X_train, y_train):.4f}, テスト={est.score(X_test, y_test):.4f}")

fig, axes = plt.subplots(1, 4, figsize=(20, 5))

# 決定境界描画用のグリッド
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))
grid = np.c_[xx.ravel(), yy.ravel()]

models = [
    ("単一決定木", single_tree),
    ("Bagging (50本)", bagging),
    ("AdaBoost (100ステップ)", adaboost),
    ("Stacking (4種)", stacking),
]

for ax, (name, model) in zip(axes, models):
    Z = model.predict(grid).reshape(xx.shape)
    ax.contourf(xx, yy, Z, alpha=0.3, cmap="RdBu")
    ax.scatter(X_test[:, 0], X_test[:, 1], c=y_test, cmap="RdBu",
               edgecolors="k", s=20)
    test_acc = np.mean(model.predict(X_test) == y_test)
    ax.set_title(f"{name}\nテスト精度: {test_acc:.3f}", fontsize=11)
    ax.set_xlim(x_min, x_max)
    ax.set_ylim(y_min, y_max)

plt.suptitle("アンサンブル手法の決定境界比較", fontsize=14, y=1.02)
plt.tight_layout()
plt.savefig("ensemble_decision_boundaries.png", dpi=150, bbox_inches="tight")
plt.show()

# アンサンブルサイズの影響
ensemble_sizes = [1, 2, 5, 10, 20, 50, 100, 200]

bagging_scores = []
adaboost_scores = []

for size in ensemble_sizes:
    # Bagging
    bag = BaggingClassifierScratch(
        base_estimator=DecisionTreeClassifier(max_depth=None),
        n_estimators=size, random_state=42
    )
    bag.fit(X_train, y_train)
    bagging_scores.append(np.mean(bag.predict(X_test) == y_test))

    # AdaBoost
    ada = AdaBoostClassifierScratch(n_estimators=size, random_state=42)
    ada.fit(X_train, y_train)
    adaboost_scores.append(np.mean(ada.predict(X_test) == y_test))

plt.figure(figsize=(8, 5))
plt.plot(ensemble_sizes, bagging_scores, "o-", label="Bagging", linewidth=2)
plt.plot(ensemble_sizes, adaboost_scores, "s-", label="AdaBoost", linewidth=2)
plt.axhline(y=single_tree.score(X_test, y_test), color="gray",
            linestyle="--", label="単一決定木")
plt.xlabel("アンサンブルサイズ（モデル数）", fontsize=12)
plt.ylabel("テスト精度", fontsize=12)
plt.title("アンサンブルサイズと性能の関係", fontsize=14)
plt.legend(fontsize=11)
plt.xscale("log")
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("ensemble_size_effect.png", dpi=150, bbox_inches="tight")
plt.show()

from sklearn.ensemble import VotingClassifier

# Votingの実験
voting_hard = VotingClassifier(
    estimators=[
        ("rf", RandomForestClassifier(n_estimators=50, random_state=42)),
        ("knn", KNeighborsClassifier(n_neighbors=5)),
        ("dt", DecisionTreeClassifier(max_depth=5, random_state=42)),
    ],
    voting="hard"
)
voting_hard.fit(X_train, y_train)

voting_soft = VotingClassifier(
    estimators=[
        ("rf", RandomForestClassifier(n_estimators=50, random_state=42)),
        ("knn", KNeighborsClassifier(n_neighbors=5)),
        ("dt", DecisionTreeClassifier(max_depth=5, random_state=42)),
    ],
    voting="soft"
)
voting_soft.fit(X_train, y_train)

print(f"Hard Voting - テスト精度: {voting_hard.score(X_test, y_test):.4f}")
print(f"Soft Voting - テスト精度: {voting_soft.score(X_test, y_test):.4f}")

# 特性比較表の作成
comparison_data = {
    "特性": [
        "学習方式", "主な効果", "基本学習器",
        "過学習リスク", "並列化", "計算コスト",
        "ハイパーパラメータ", "代表的手法"
    ],
    "Bagging": [
        "並列（独立）", "分散削減", "高バリアンス（深い木）",
        "低い", "容易", "中程度",
        "モデル数、サンプル率", "ランダムフォレスト"
    ],
    "Boosting": [
        "逐次（依存）", "バイアス削減", "高バイアス（浅い木）",
        "中〜高", "困難", "高い",
        "モデル数、学習率、木の深さ", "XGBoost, LightGBM"
    ],
    "Stacking": [
        "2段階", "バイアス+分散", "異種モデル",
        "中程度", "Level 0は可能", "最も高い",
        "ベース学習器の選択、CV数", "StackingClassifier"
    ]
}

# テーブル表示
print(f"{'特性':<20} {'Bagging':<25} {'Boosting':<25} {'Stacking':<25}")
print("=" * 95)
for i, prop in enumerate(comparison_data["特性"]):
    print(f"{prop:<20} {comparison_data['Bagging'][i]:<25} "
          f"{comparison_data['Boosting'][i]:<25} "
          f"{comparison_data['Stacking'][i]:<25}")

print("""
=== アンサンブル手法の選択フローチャート ===

1. まず単一モデル（決定木、ロジスティック回帰等）でベースライン
   ↓
2. バイアスが高い（訓練誤差が大きい）？
   → YES: Boosting（XGBoost, LightGBM）を試す
   → NO: ステップ3へ
   ↓
3. バリアンスが高い（訓練-テスト誤差の差が大きい）？
   → YES: Bagging（ランダムフォレスト）を試す
   → NO: ステップ4へ
   ↓
4. さらに精度を上げたい？
   → YES: 異種モデルでStacking
   → NO: 現在のモデルで十分

=== 実務でのTips ===
- ランダムフォレスト: 最初に試すべきアンサンブル。ほぼチューニング不要
- LightGBM/XGBoost: 表形式データの最高性能。ただし要チューニング
- Stacking: コンペ向け。実務では計算コストとメンテナンスコストを考慮
- Voting: 最もシンプル。既存の複数モデルを手軽に統合
""")