はじめに
機械学習を始めた人がまず最初に手に触れるPythonのライブラリの一つがscikit-learn. なんて話はよくあると思います. scikit-learnには独自の設計思想があり,
- モデルのインスタンスを生成する. このときにハイパーパラメータを一緒に設定する.
-
.fit()でモデルを訓練する. -
.predict()で予測する.
という, 非常にわかりやすい一貫した作りになってます. そのため, 使い方も初心者向に易しい方だと思います.
一方, scikit-learnでは扱われていない機械学習アルゴリズムの１つである「ニューラルネットワーク」は, kerasやtensorflow, pytorchで実装するのが一般的です. これらのライブラリはそれぞれに設計が異なるため, 当たり前ですが使い方もまちまちです. これ, 面倒でよね. 全て.fit()でモデルを訓練できれば楽に作業できるし, 作業効率が上がるのに... こんな悩みを著者はよく抱きます. 今回は, kerasのニューラルネットワークをscikit-learnライクに扱えるように実装してみます.
サンプルデータ
サンプルデータにはscikit-learnのbostonデータを使います.
from sklearn.datasets import load_boston
boston = load_boston()
X = boston['data']
y = boston['target']
kerasによるニューラルネットワークの実装
まずkerasによるオーソドックスなneural networkの実装をおさらいしましょう. まず, データセットをトレーニングセットとバリデーションセットとテストセットに分割します.
from sklearn.model_selection import KFold
kf = KFold(n_splits=5, shuffle=True, random_state=1647)
_, te_idx = list(kf.split(X, y))[0]
tr_idx, va_idx = list(kf.split(X[_], y[_]))[0]
X_te, y_te = X[te_idx], y[te_idx]
X_tr, y_tr = X[_][tr_idx], y[_][tr_idx]
X_va, y_va = X[_][va_idx], y[_][va_idx]
説明変数を見ていきます.
pd.DataFrame(X).isnull().sum(axis=0).sum()
>> 0
説明変数に欠損値はありませんでした.
次に, 説明変数の分布を見ていきます.
pd.DataFrame(X).iloc[:,0].hist(bins=50)
説明変数が正規分布してません. ニューラルネットワークはこのような入力に弱いので, 説明変数を強制的に正規分布っぽくしていきます.
from sklearn.preprocessing import QuantileTransformer
transformer = QuantileTransformer(n_quantiles=100, random_state=0, output_distribution='normal')
X_train = transformer.fit_transform(X_tr)
X_valid = transformer.transform(X_va)
X_test = transformer.transform(X_te)
pd.DataFrame(X_train).iloc[:,0].hist(bins=50)
準備ができましたので, モデルの実装に移ります.
from keras.callbacks import EarlyStopping
from keras.layers.advanced_activations import ReLU
from keras.layers.core import Dense, Dropout
from tensorflow.keras.layers import BatchNormalization
from keras.models import Sequential
from tensorflow.keras.optimizers import Adam
model = Sequential()
# input layer
model.add(Dropout(0.001, input_shape=(X_train.shape[1],)))
# hidden layer
for i in range(2):
model.add(Dense(10))
model.add(BatchNormalization())
model.add(ReLU())
# out layer
model.add(Dense(1))
# optimizer
optimizer = Adam(lr=0.001, beta_1=0.9, beta_2=0.999, decay=0.)
# compile
model.compile(
loss='mean_squared_error',
optimizer=optimizer,
metrics=['mae']
)
# eaerly stopping
early_stopping = EarlyStopping(patience=100, restore_best_weights=True)
# train
history = model.fit(
X_train,
y_tr,
batch_size=32,
epochs=10000,
verbose=1,
validation_data=(X_valid, y_va),
callbacks=[early_stopping]
)
scikit-learnライクなニューラルネットワークの実装
以下のようにNNRegressorというクラスを新しく設計して, コンストラクタの引数にモデルのアーキテクチャに関する設定を記述すれば, scikit-learn風に走らせることができて便利です.
class NNRegressor:
def __init__(self, input_dropout=0.01, hidden_layers=2, hidden_units=256, hidden_dropout=0.01,
batch_norm="before_act", learning_rate=1e-5, batch_size=64, epochs=10000,
standardization=True
):
self.input_dropout = input_dropout # layer param
self.hidden_layers = int(hidden_layers) # layer param
self.hidden_units = int(hidden_units) # layer param
self.hidden_dropout = hidden_dropout # layer param
self.batch_norm = batch_norm # layer param
self.learning_rate = learning_rate # optimizer param
self.batch_size = int(batch_size) # fit param
self.epochs = int(epochs) # fit param
self.standardization = standardization
def fit(self, X_train, y_train, eval_set, early_stopping_rounds, eval_metric, verbose=1):
# Data standardization
if self.standardization:
self.transformer = QuantileTransformer(n_quantiles=100, random_state=0, output_distribution='normal')
X_train = self.transformer.fit_transform(X_train)
X_valid = self.transformer.transform(eval_set[0])
elif self.standardization is False:
X_valid = eval_set[0]
# Keras Wrapper for sklearn AIP
def create_model():
model = Sequential()
model.add(Dropout(self.input_dropout, input_shape=(X_train.shape[1],)))
for i in range(self.hidden_layers):
model.add(Dense(self.hidden_units))
if self.batch_norm == 'before_act':
model.add(BatchNormalization())
model.add(ReLU())
model.add(Dropout(self.hidden_dropout))
model.add(Dense(1))
# Optimazer
optimizer = Adam(lr=self.learning_rate, beta_1=0.9, beta_2=0.999, decay=0.)
# Compile
model.compile(
loss=eval_metric,
optimizer=optimizer,
metrics=['mae']
)
return model
early_stopping = EarlyStopping(patience=early_stopping_rounds, restore_best_weights=True)
self.model = create_model()
self.model.fit(
X_train,
y_train,
batch_size=self.batch_size,
epochs=1000,
verbose=1,
validation_data=(X_valid, eval_set[1]),
callbacks=[early_stopping]
)
def predict(self, x):
if self.standardization:
x = self.transformer.transform(x)
y_pred = self.model.predict(x).astype("float64")
y_pred = y_pred.flatten()
return y_pred
def get_model(self):
return self.model
def get_transformer(self):
if self.standardization:
return self.transformer
トレーニングは以下のように行います.
model = NNRegressor(
input_dropout=0.01, hidden_layers=2, hidden_units=256, hidden_dropout=0.01,
batch_norm="before_act", learning_rate=1e-3, batch_size=32, epochs=1000,
standardization=True
)
history = model.fit(
X_train=X_tr,
y_train=y_tr,
early_stopping_rounds=10,
eval_set=(X_va, y_va),
eval_metric='mean_absolute_error'
)