【Python実践】スタッキング（Stacking）で回帰問題と分類問題を解く（コピペで使えるサンプルコード付き）

pip install matplotlib
pip install pandas
pip install scikit-learn
pip install catboost

from sklearn.datasets import make_classification
import pandas as pd

# ダミーデータ作成
X, y = make_classification(
    n_samples=1000,
    n_features=7,
    n_informative=5,
    n_redundant=2,
    n_classes=4,
    n_clusters_per_class=1,
    flip_y=0.05,
    random_state=42
)

# データフレーム化
df = pd.DataFrame(X, columns=[f"特徴量_{i+1}" for i in range(7)])
df['正解データ'] = y

# ファイルに保存（任意）
df.to_csv("classification_data.csv", index=False)

import pandas as pd
import catboost as cb
from sklearn.ensemble import StackingClassifier, RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

def train_stacking_classifier(df, target, columns, base_models, meta_model, test_size=0.2, random_state=42):
    """
    スタッキングを用いた分類モデルを構築し、レポートを出力する関数。

    引数:
    df : pandas.DataFrame
        データを含むデータフレーム。
    target : str
        予測対象となるターゲット列名。
    columns : list of str
        使用する特徴量の列名のリスト。
    base_models : list of tuple
        ベースモデルのリスト。各モデルは ('model_name', model_object) の形式。
    meta_model : estimator
        スタッキングの最終モデル。
    test_size : float, optional (default=0.2)
        テストデータの割合。
    random_state : int, optional (default=42)
        乱数シード。

    戻り値:
    model : StackingClassifier
        訓練されたスタッキングモデルのオブジェクト。
    cm : numpy.ndarray
        混同行列を表す2次元配列。
    """
    columns = [col for col in columns if col != target]
    X = df[columns]
    y = df[target]

    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state)

    # スタッキング分類器の作成
    model = StackingClassifier(estimators=base_models, final_estimator=meta_model)

    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)

    accuracy = accuracy_score(y_test, y_pred)
    report = classification_report(y_test, y_pred, target_names=[str(i) for i in range(len(df[target].unique()))])
    cm = confusion_matrix(y_test, y_pred)

    print(f'正解率: {accuracy:.4f}')
    print('混同行列: ')
    print(cm)
    print('分類レポート: ')
    print(report)

    return model, cm


if __name__ == "__main__":
    df = pd.read_csv('classification_data.csv')

    # ベースモデルを関数外で作成
    catboost_params = {
        'loss_function': 'MultiClass',
        'iterations': 500,
        'learning_rate': 0.1,
        'depth': 6,
        'random_seed': 42
    }
    base_models = [
        ('catboost', cb.CatBoostClassifier(**catboost_params, verbose=False)),
        ('rf', RandomForestClassifier(n_estimators=100, random_state=42))
    ]
    
    # メタモデルを関数外で作成
    meta_model = LogisticRegression()

    # スタッキング分類器を訓練
    model, cm = train_stacking_classifier(df, target='正解データ', columns=df.columns, 
                                          base_models=base_models, meta_model=meta_model)

#---------------------------------------------------------------
# この部分に、前述した train_stacking_classifier関数 を張り付けて下さい
#---------------------------------------------------------------

import matplotlib.pyplot as plt
import seaborn as sns
from matplotlib import rcParams
import pandas as pd
import catboost as cb
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

rcParams['font.family'] = 'Meiryo'

def plot_confusion_matrix(cm, class_names=[]):
    """
    混同行列をプロットする関数。

    引数:
    cm : numpy.ndarray
        混同行列を表す2次元配列。
    class_names : list of str, optional
        クラス名のリスト。省略すると連番が表示される
    """
    plt.figure(figsize=(10, 7))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=class_names, yticklabels=class_names)
    plt.xlabel('予測ラベル')
    plt.ylabel('実際のラベル')
    plt.title('混同行列')
    plt.show()

if __name__ == "__main__":

    df = pd.read_csv('classification_data.csv')

    # ベースモデルを関数外で作成
    catboost_params = {
        'loss_function': 'MultiClass',
        'iterations': 500,
        'learning_rate': 0.1,
        'depth': 6,
        'random_seed': 42
    }
    base_models = [
        ('catboost', cb.CatBoostClassifier(**catboost_params)),
        ('rf', RandomForestClassifier(n_estimators=100, random_state=42))
    ]
    
    # メタモデルを関数外で作成
    meta_model = LogisticRegression()

    # スタッキング分類器を訓練
    model, cm = train_stacking_classifier(df, target='正解データ', columns=df.columns, 
                                          base_models=base_models, meta_model=meta_model)

    # クラス名を設定して混同行列をプロット
    class_names = df['正解データ'].unique().astype(str)
    plot_confusion_matrix(cm, class_names=class_names)

import numpy as np
import pandas as pd

# 特徴量の数
n_features = 7

# 各特徴量の標準偏差
std_devs = [1, 1, 1, 1, 1, 1, 1]

# 相関行列の設定（例: 1番目と2番目の特徴量が強い相関を持つ）
correlation_matrix = np.array([
    [1, 0.8, 0.2, 0.1, 0.1, 0.1, 0.1],
    [0.8, 1, 0.3, 0.2, 0.2, 0.2, 0.2],
    [0.2, 0.3, 1, 0.4, 0.3, 0.3, 0.3],
    [0.1, 0.2, 0.4, 1, 0.5, 0.5, 0.5],
    [0.1, 0.2, 0.3, 0.5, 1, 0.4, 0.4],
    [0.1, 0.2, 0.3, 0.5, 0.4, 1, 0.3],
    [0.1, 0.2, 0.3, 0.5, 0.4, 0.3, 1]
])

# 共分散行列の計算
covariance_matrix = np.outer(std_devs, std_devs) * correlation_matrix

# ダミーデータの生成
np.random.seed(42)
n_samples = 1000
data = np.random.multivariate_normal(np.zeros(n_features), covariance_matrix, size=n_samples)

# DataFrameに変換
df = pd.DataFrame(data, columns=[f'特徴量_{i+1}' for i in range(n_features)])

# 結果の確認
print(df.head())

# CSVファイルに保存
df.to_csv('regression_data.csv', index=False)

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import StackingRegressor
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score, mean_absolute_percentage_error

def train_stacking_regressor(df, target, columns, base_models, meta_model, 
                             test_size=0.2, random_state=42):
    """
    スタッキング回帰モデルを学習し、評価指標と特徴量重要度を出力する関数。
    """
    columns = [col for col in columns if col != target]
    X = df[columns]
    y = df[target]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, random_state=random_state
    )

    model = StackingRegressor(
        estimators=base_models,
        final_estimator=meta_model,
        passthrough=False,
        n_jobs=-1
    )
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)

    # 評価指標
    mse = mean_squared_error(y_test, y_pred)
    rmse = np.sqrt(mse)
    mae = mean_absolute_error(y_test, y_pred)
    r2 = r2_score(y_test, y_pred)
    mape = mean_absolute_percentage_error(y_test, y_pred)

    print(f'{target}の統計情報:')
    print(f'最小値(Min): {y.min()}')
    print(f'最大値(Max): {y.max()}')
    print(f'平均値(Mean): {y.mean()}')
    print(f'標準偏差(Std): {y.std()}')
    print(f'分散(Var): {y.var()}')

    print('\n精度評価:')
    print(f'平均絶対誤差 (MAE): {mae}')
    print(f'平均二乗誤差 (MSE): {mse}')
    print(f'ルート平均二乗誤差 (RMSE): {rmse}')
    print(f'平均絶対パーセンテージ誤差 (MAPE): {mape}')
    print(f'決定係数 (R^2): {r2}')

    print("\nベースモデルの特徴量重要度:")

    for name, estimator in model.named_estimators_.items():
        print(f"\n{name} の特徴量重要度:")
        try:
            if hasattr(estimator, 'get_feature_importance'):
                importances = estimator.get_feature_importance()
            elif hasattr(estimator, 'feature_importances_'):
                importances = estimator.feature_importances_
            else:
                print(" → このモデルは特徴量重要度をサポートしていません。")
                continue
            importance_df = pd.DataFrame({
                '特徴量': columns,
                '重要度': importances
            }).sort_values(by='重要度', ascending=False)
            print(importance_df)
        except Exception as e:
            print(f" → 取得できませんでした: {e}")

    return model, y_test, y_pred

if __name__ == "__main__":
    
    from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
    from sklearn.linear_model import LinearRegression
    from catboost import CatBoostRegressor
    import pandas as pd

    # ベースモデル定義
    base_models = [
        ('catboost', CatBoostRegressor(iterations=100, learning_rate=0.1, depth=6, verbose=0)),
        ('rf', RandomForestRegressor(n_estimators=100, random_state=42)),
        ('gb', GradientBoostingRegressor(n_estimators=100, random_state=42))
    ]

    # メタモデル定義
    meta_model = LinearRegression()

    # データ読み込みと学習
    df = pd.read_csv('regression_data.csv')
    model, y_test, y_pred = train_stacking_regressor(
        df,
        target='特徴量_1',
        columns=df.columns,
        base_models=base_models,
        meta_model=meta_model
    )

#------------------------------------------------------------------------
# この部分に、前述した train_stacking_regressor関数 を張り付けて下さい
#------------------------------------------------------------------------

import matplotlib.pyplot as plt
import seaborn as sns
from matplotlib import rcParams
import pandas as pd
import catboost as cb
from sklearn.model_selection import train_test_split
from sklearn.metrics import r2_score
from scipy.stats import norm
import numpy as np

rcParams['font.family'] = 'Meiryo'

def plot_figures(y_test, y_pred):

    residuals = y_test - y_pred

    fig, axs = plt.subplots(1, 4, figsize=(18, 5))

    axs[0].plot(y_test.values, label='実際の値 (Actual)', linestyle='-', marker=None)
    axs[0].plot(y_pred, label='予測値 (Predicted)', color='red', linestyle='-', marker=None)
    axs[0].set_xlabel('サンプルインデックス (Sample Index)')
    axs[0].set_ylabel('値 (Value)')
    axs[0].set_title('実際の値 VS 予測値\n(Comparison of Actual and Predicted Values)')
    axs[0].legend()

    axs[1].scatter(y_test, y_pred, alpha=0.3, label='予測値 (Predicted)')
    axs[1].plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--', label='対角線 (y=x)')
    slope, intercept = np.polyfit(y_test, y_pred, 1)
    r2 = r2_score(y_test, y_pred)
    axs[1].plot(y_test, slope * y_test + intercept, color='blue', linestyle='-', linewidth=2,
                label=f'回帰線 (y={slope:.2f}x + {intercept:.2f}, $R^2$={r2:.2f})')
    axs[1].set_xlabel('実際の値 (Actual)')
    axs[1].set_ylabel('予測値 (Predicted)')
    axs[1].set_title('実際の値 vs 予測値\n(Actual vs Predicted)')
    axs[1].legend()

    axs[2].scatter(y_pred, residuals, alpha=0.3, label='残差 (Residuals)')
    axs[2].hlines(y=0, xmin=y_pred.min(), xmax=y_pred.max(), linestyles='dashed', label='y=0')
    residuals_mean = np.mean(residuals)
    residuals_std = np.std(residuals)
    sorted_indices = np.argsort(y_pred)
    sorted_y_pred = y_pred[sorted_indices]
    sorted_upper = residuals_mean + residuals_std
    sorted_lower = residuals_mean - residuals_std
    axs[2].axhline(y=residuals_mean, color='blue', linestyle='--', label=f'平均 (Mean): {residuals_mean:.2f}')
    axs[2].fill_between(sorted_y_pred, sorted_lower, sorted_upper, color='lightblue', alpha=0.4,
                        label=f'標準偏差 (Std): {residuals_std:.2f}')
    axs[2].set_xlabel('予測値 (Predicted)')
    axs[2].set_ylabel('残差 (Residuals)')
    axs[2].set_title('残差プロット\n(Residual Plot)')
    axs[2].legend()

    axs[3].hist(residuals, bins=30, edgecolor='k', density=True, alpha=0.6, color='g')
    xmin, xmax = axs[3].get_xlim()
    x = np.linspace(xmin, xmax, 100)
    p = norm.pdf(x, residuals_mean, residuals_std)
    axs[3].plot(x, p, 'k', linewidth=2)
    axs[3].set_xlabel('残差 (Residuals)')
    axs[3].set_ylabel('頻度 (Frequency)')
    axs[3].set_title('残差のヒストグラム\n(Histogram of Residuals)')

    plt.tight_layout()
    plt.show()


if __name__ == "__main__":
    
    from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
    from sklearn.linear_model import LinearRegression
    from catboost import CatBoostRegressor
    import pandas as pd

    # ベースモデル定義
    base_models = [
        ('catboost', CatBoostRegressor(iterations=100, learning_rate=0.1, depth=6, verbose=0)),
        ('rf', RandomForestRegressor(n_estimators=100, random_state=42)),
        ('gb', GradientBoostingRegressor(n_estimators=100, random_state=42))
    ]

    # メタモデル定義
    meta_model = LinearRegression()

    # データ読み込みと学習
    df = pd.read_csv('regression_data.csv')
    model, y_test, y_pred = train_stacking_regressor(
        df,
        target='特徴量_1',
        columns=df.columns,
        base_models=base_models,
        meta_model=meta_model
    )

    plot_figures(y_test, y_pred)

import pandas as pd
import numpy as np
from sklearn.ensemble import StackingClassifier, RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from catboost import CatBoostClassifier
from xgboost import XGBClassifier
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.metrics import accuracy_score

# ❶ データの読み込みと前処理
df = pd.read_csv("classification_data.csv")

# 目的変数（target）を分離し、特徴量（X）を抽出
X = df.drop(columns=["特徴量_1"])  # "特徴量_1" がターゲット変数
y = df["特徴量_1"]

# 訓練データとテストデータに分割
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# ❷ ベースモデルのグリッドサーチ
# 各モデルの最適なハイパーパラメータを探索し、ベストモデルを取得
param_grid_catboost = {'learning_rate': [0.01, 0.1], 'depth': [4, 6, 8]}
grid_catboost = GridSearchCV(CatBoostClassifier(verbose=False), param_grid_catboost, cv=5, n_jobs=-1)
grid_catboost.fit(X_train, y_train)
best_catboost = grid_catboost.best_estimator_

param_grid_rf = {'n_estimators': [100, 200], 'max_depth': [None, 10, 20]}
grid_rf = GridSearchCV(RandomForestClassifier(random_state=42), param_grid_rf, cv=5, n_jobs=-1)
grid_rf.fit(X_train, y_train)
best_rf = grid_rf.best_estimator_

param_grid_xgb = {'learning_rate': [0.01, 0.1], 'max_depth': [3, 6, 9]}
grid_xgb = GridSearchCV(XGBClassifier(use_label_encoder=False, eval_metric='logloss'), param_grid_xgb, cv=5, n_jobs=-1)
grid_xgb.fit(X_train, y_train)
best_xgb = grid_xgb.best_estimator_

# ❸ メタモデル（最終分類器）のグリッドサーチ
param_grid_meta = {'C': [0.1, 1, 10]}
grid_meta = GridSearchCV(LogisticRegression(), param_grid_meta, cv=5, n_jobs=-1)
grid_meta.fit(X_train, y_train)
best_meta_model = grid_meta.best_estimator_

# ❹ スタッキングの構築
stacking_model = StackingClassifier(estimators=[
    ('catboost', best_catboost),
    ('rf', best_rf),
    ('xgb', best_xgb)
], final_estimator=best_meta_model, n_jobs=-1)

# ❺ モデルの学習と評価
stacking_model.fit(X_train, y_train)
y_pred = stacking_model.predict(X_test)

# ❻ 結果の表示
accuracy = accuracy_score(y_test, y_pred)
print(f"スタッキングモデルの正解率: {accuracy:.4f}")