Stackingで用いるNN model備忘録

Background

テーブルデータの回帰分析コンペに参加していて、トップ層に追いつくためにstackingに行き着いたのがきっかけ。
調べていると、nn modelを組み込むことで、LGBMやCat Boostとは異なる観点でデータを解釈するので精度が上がる場合があるとのことで色々試したいモデルを記録

Models

備忘録のため、基本的にはAIの出力をそのまま貼り付け。実装に合わせてアップデートするかも。
共通

• skorchを通してStackingRegressorで使いやすい形にする
• float64 -> float32への変換を行う

Simple Regressor

• 後述するTabularNNのようにdembedding層を設けていないので、カテゴリ変数のnumber順を意味のあるものとして捉える

# pip install skorch
import torch
import numpy as np
from torch import nn
from skorch import NeuralNetRegressor
from sklearn.ensemble import StackingRegressor
from lightgbm import LGBMRegressor
from sklearn.linear_model import RidgeCV

# 1. PyTorchでシンプルなMLPを定義
class RegressorModule(nn.Module):
    def __init__(self, input_dim=20):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1)
        )
    def forward(self, X):
        return self.net(X)

# 2. skorchでScikit-learn互換のモデルにラップ
# ※入力データは float32 にしておく必要があります
nn_base = NeuralNetRegressor(
    RegressorModule,
    max_epochs=20,
    lr=0.1,
    optimizer=torch.optim.Adam,
    device='cuda' if torch.cuda.is_available() else 'cpu'
)

# 3. スタッキングのベースモデルとして設定
base_models = [
    ('lgbm', LGBMRegressor(n_estimators=100)),
    ('nn_base', nn_base)  # ここにNNを投入
]

stacking_model = StackingRegressor(
    estimators=base_models,
    final_estimator=RidgeCV()
)

# 学習 (X_trainはfloat32のnumpy配列)
# stacking_model.fit(X_train.astype(np.float32), y_train.astype(np.float32))

TabularNN

• カテゴリ変数に対してEmbedding層を噛ませているので、カテゴリ間の距離を自身で学習する（合っているのか？この理解）

import torch
import torch.nn as nn
from skorch import NeuralNetRegressor

class TabularNN(nn.Module):
    def __init__(self, n_cont, cat_dims, emb_dims):
        super().__init__()
        self.n_cont = n_cont
        # カテゴリ変数の数だけEmbedding層を作成
        self.embeddings = nn.ModuleList([
            nn.Embedding(num_embeddings=d, embedding_dim=e) 
            for d, e in zip(cat_dims, emb_dims)
        ])
        
        total_input_dim = n_cont + sum(emb_dims)
        self.fc = nn.Sequential(
            nn.Linear(total_input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1)
        )

    def forward(self, X):
        # Xは (batch_size, features) の形
        # 1. 数値変数とカテゴリ変数を分離 (スライス)
        # 前半が数値変数、後半がカテゴリ変数と仮定（ColumnTransformerの順序）
        x_cont = X[:, :self.n_cont].float()
        x_cat = X[:, self.n_cont:].long()
        
        # 2. カテゴリ変数をEmbedding
        x_emb = [emb(x_cat[:, i]) for i, emb in enumerate(self.embeddings)]
        x_emb = torch.cat(x_emb, dim=1)
        
        # 3. 結合して全結合層へ
        x = torch.cat([x_cont, x_emb], dim=1)
        return self.fc(x).squeeze(-1) # 出力を (batch_size,) に
# パラメータの設定例
num_features = len(num_cols)
cat_features_dims = [len(m) + 1 for m in category_number_maps.values()] # 各カテゴリのユニーク数+1
emb_dims = [min(50, (d + 1) // 2) for d in cat_features_dims] # 適切な埋め込み次元

# 1. skorchでラップ
nn_estimator = NeuralNetRegressor(
    module=TabularNN,
    module__n_cont=num_features,
    module__cat_dims=cat_features_dims,
    module__emb_dims=emb_dims,
    criterion=nn.MSELoss,
    optimizer=torch.optim.Adam,
    lr=0.001,
    max_epochs=50,
    batch_size=64,
    train_split=None, # Stacking内でCVされるため、内部の検証分割は不要
    device='cuda' if torch.cuda.is_available() else 'cpu',
)

# 2. StackingRegressor の定義
from sklearn.ensemble import StackingRegressor
from lightgbm import LGBMRegressor
from sklearn.linear_model import RidgeCV

base_models = [
    ('lgbm', LGBMRegressor(n_estimators=100, random_state=42)),
    ('nn', nn_estimator) # カスタムNNを投入
]

stacking_model = StackingRegressor(
    estimators=base_models,
    final_estimator=RidgeCV(), # メタ学習器
    cv=5
)

# 3. 実行（preprocessorを通した後のデータを渡す）
# X_transformed = preprocessor.fit_transform(X)
# stacking_model.fit(X_transformed.astype(np.float32), y.astype(np.float32))

Entity Embedding MLP

import torch
import torch.nn as nn
from skorch import NeuralNetRegressor

class EntityEmbeddingMLP(nn.Module):
    def __init__(self, n_cont, cat_dims, emb_dims):
        super().__init__()
        self.n_cont = n_cont
        # 各カテゴリ変数ごとの埋め込み層
        self.embeddings = nn.ModuleList([
            nn.Embedding(num_embeddings=d, embedding_dim=e) 
            for d, e in zip(cat_dims, emb_dims)
        ])
        
        # 入力サイズ = 数値変数の数 + 埋め込みベクトルの合計サイズ
        total_input_dim = n_cont + sum(emb_dims)
        
        self.mlp = nn.Sequential(
            nn.Linear(total_input_dim, 128),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )

    def forward(self, X):
        # 1. データの分離 (ユーザー様のpreprocessor順: 数値 -> カテゴリ)
        x_cont = X[:, :self.n_cont].float()
        x_cat = X[:, self.n_cont:].long()
        
        # 2. Embeddingの適用
        x_emb = [emb(x_cat[:, i]) for i, emb in enumerate(self.embeddings)]
        x_emb = torch.cat(x_emb, dim=1)
        
        # 3. 結合してMLPへ
        out = torch.cat([x_cont, x_emb], dim=1)
        return self.mlp(out).squeeze(-1)

# skorchでのラップ
model_embedding = NeuralNetRegressor(
    module=EntityEmbeddingMLP,
    module__n_cont=len(num_cols),
    module__cat_dims=[len(m) + 1 for m in category_number_maps.values()],
    module__emb_dims=[min(50, (len(m) + 1) // 2) for m in category_number_maps.values()],
    max_epochs=50,
    lr=0.001,
    optimizer=torch.optim.Adam,
    device='cuda' if torch.cuda.is_available() else 'cpu'
)

Tabnet

• pytorch_tabnetにclassがあるので実装が楽

# !pip install pytorch-tabnet
from pytorch_tabnet.tab_model import TabNetRegressor

# カテゴリ変数のインデックスを特定 (数値変数の後ろに結合されている場合)
cat_idxs = [len(num_cols) + i for i in range(len(category_cols))]
cat_dims = [len(category_number_maps[col]) + 1 for col in category_cols]

model_tabnet = TabNetRegressor(
    cat_idxs=cat_idxs,
    cat_dims=cat_dims,
    cat_emb_dim=2, # 各カテゴリの埋め込み次元
    optimizer_fn=torch.optim.Adam,
    optimizer_params=dict(lr=2e-2),
    scheduler_params={"step_size":50, "gamma":0.9},
    scheduler_fn=torch.optim.lr_scheduler.StepLR,
    mask_type='entmax' # 疎な特徴量選択を可能にする設定
)

以下はパラメータチューニングの方針

スタッキングのための実戦アドバイス

virtual_batch_size の活用: TabNetには batch_size と virtual_batch_size (Ghost Batch Normalization用) の2つがあります。

batch_size: 1024や2048など大きめにする。

virtual_batch_size: 128や256など小さめにする。 これにより、学習の安定性と汎化性能が向上します。

あえて「尖った」設定にする: スタッキングでは、すべてのモデルが正解することよりも、**「他のモデルが苦手なパターンを拾えるか」**が重要です。

LGBMが複雑なデータで苦戦しているなら、TabNetの n_steps を多め（7〜10）にして、より深い関係性を探らせる。

特徴量が多い場合は、lambda_sparse を大きめにして、特定の重要な変数にフォーカスさせる。

事前学習（Pre-training）: もしラベルのないデータが大量にある場合、TabNetは自己教師あり学習（Pre-training）が可能です。これにより、スタッキングのベースモデルとしての性能がさらに一段階上がります。

まずは n_da（幅）と n_steps（深さ）を固定して、lambda_sparse を動かしてみるだけでも、予測値の「質」が変わるのを実感できるはずです。

import optuna
from pytorch_tabnet.tab_model import TabNetRegressor
from sklearn.metrics import mean_squared_error

def objective(trial):
    # パラメータの探索範囲を定義
    # n_d と n_a は同じ値にするのが一般的
    n_da = trial.suggest_int('n_da', 8, 64, step=8)
    
    params = {
        "n_d": n_da,
        "n_a": n_da,
        "n_steps": trial.suggest_int('n_steps', 3, 10),
        "gamma": trial.suggest_float('gamma', 1.0, 2.0),
        "lambda_sparse": trial.suggest_float('lambda_sparse', 1e-6, 1e-1, log=True),
        "n_independent": trial.suggest_int('n_independent', 1, 5),
        "n_shared": trial.suggest_int('n_shared', 1, 5),
        "optimizer_fn": torch.optim.Adam,
        "optimizer_params": dict(lr=2e-2),
        "mask_type": trial.suggest_categorical('mask_type', ['sparsemax', 'entmax']),
    }

    model = TabNetRegressor(**params)
    
    # 学習 (検証データでの early stopping を推奨)
    model.fit(
        X_train=X_train_pre, y_train=y_train,
        eval_set=[(X_val_pre, y_val)],
        patience=10, max_epochs=100,
        batch_size=1024, virtual_batch_size=128
    )
    
    preds = model.predict(X_val_pre)
    rmse = np.sqrt(mean_squared_error(y_val, preds))
    return rmse

# 探索開始
# study = optuna.create_study(direction='minimize')
# study.optimize(objective, n_trials=50)

推奨パラメータの初期値

n_d, n_a: 24 〜 48

カテゴリ数が多いため、少し広めの次元（32以上）から始めるのが良いです。

n_steps: 3 〜 5

あまり深くしすぎると過学習します。10万行なら5ステップ程度が上限です。

lambda_sparse: 1e-4 〜 1e-3

カテゴリ変数が多い場合、不要な変数を無視させるために少し強めに設定します。

cat_emb_dim: 2 〜 4

各カテゴリの次元を大きくしすぎると、入力層が巨大化して計算が終わりません。各カテゴリ1〜4次元程度で十分です。

NNのLoss Functionの工夫

通常の L2 Loss（MSE）は「外れ値」に強く引きずられますが、この評価関数は「比率（Percentage Error）」を気にしています。そのため、NNの学習には L1 Loss（MAE） または Huber Loss を使う方が、評価指標の特性に近くなります。

import torch

# TabNetRegressorの初期化時に指定
model_tabnet = TabNetRegressor(
    optimizer_fn=torch.optim.Adam,
    optimizer_params=dict(lr=2e-2),
    loss_fn=torch.nn.L1Loss(), # MSEより評価関数に近い挙動になりやすい
    # ... 他のパラメータ
)

MLPRegressor

• こちらもsklearnにクラスがあるので楽

import numpy as np
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
from sklearn.ensemble import StackingRegressor, RandomForestRegressor
from sklearn.linear_model import RidgeCV
from sklearn.svm import SVR
from sklearn.neural_network import MLPRegressor
from sklearn.metrics import mean_squared_error

# 1. データの生成
X, y = make_regression(n_samples=1000, n_features=20, noise=0.1, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 2. ベースモデル（Level 0）の定義
base_models = [
    ('ridge', RidgeCV()),
    ('rf', RandomForestRegressor(n_estimators=100, random_state=42)),
    ('svr', SVR(C=1.0, epsilon=0.2))
]

# 3. メタ学習器（Level 1）としてのNNを定義
# ここでは MLPRegressor を使用します
meta_learner = MLPRegressor(
    hidden_layer_sizes=(16, 8), 
    activation='relu', 
    solver='adam', 
    max_iter=500, 
    random_state=42
)

# 4. スタッキングモデルの作成
stacking_model = StackingRegressor(
    estimators=base_models,
    final_estimator=meta_learner,
    cv=5  # 5-fold cross-validation
)

# 5. 学習
stacking_model.fit(X_train, y_train)