保守性を考慮した機械学習モデル

前職

前職はソフトウェア開発(Java, C#)でした。
そこでは、効率良く・綺麗なコードを書く、という指針の元、コードレビューも細かく実施していました。

現職

データ分析〜機械学習を担当しています。
Pythonで記載しますが、製品版ではない場合はコードレビューもないですし、動けばよいという雰囲気になっています。。
確かに分析対象、モデルやパラメータが頻繁に変わるので、設計までしっかりして取り組むよりまずは結果かと考えます。

オブジェクトの導入

とはいえ、１つのJupyter notebookに、複数のモデル(例えば重回帰、SVR、Lasso、RandomForestRegressor)を定義して、それぞれ流すコードは嫌いですし、モデル毎にnotebookを分けるのも、ファイル増えて管理の手間なので嫌いです。
なので、動かすnotebookは１つにしていじらない(結果確認用)、modelはconfigurationから指定して実行、のような形で少しオブジェクト指向っぽくしています。
「機械学習＆notebookのモデルはそうじゃない」という方もいると思いますが。。

全体像

• Jupyter notebook上は、固有のモデル名を記載しない。以下で言えば、ClassCreatorで生成した結果、モデルが出来上がる。
• このmodelが何であろうと、jupyter notebook側は知らない。定義された抽象メソッドを呼ぶだけ。

    clazz_path = parser.get(path="regression", key="class_path")
    model = ClassCreator.create(clazz_path)

• configファイルにて、以下のように定義する

[regression]
class_path = utility.SVRModel.SVRModel

• ClassCreator内で、リフレクションでインスタンス化
  こちらを参考にさせていただきました。

import sys

class ClassCreator():
    @staticmethod
    def create(class_path):
        try:
            print("class:", class_path)
            component_path = str(class_path).split('.')
            package_path   = component_path[:-1]
            package_name   = ".".join(package_path)
            class_name     = component_path[-1]
            __import__(str(package_name))
            cls = getattr(sys.modules[package_name], class_name)
            return cls()
        except Exception as e:
            print('=== エラー内容 ===')
            print('type:' + str(type(e)))
            print('args:' + str(e.args))
            print('e自身:' + str(e))

• 抽象クラス
  中身は自由で良いですが、サンプルとしてはこんな感じです。

from abc import ABCMeta, abstractmethod


class AbstractModel(metaclass=ABCMeta):

    @abstractmethod
    def run_grid_search(self, x_train_scaled, y_train):
        pass

    @abstractmethod
    def create_and_pred(self, x_train_scaled, y_train, x_test_scaled):
        pass

    @abstractmethod
    def print_eval(self, x_train_scaled, x_test_scaled, y_train, y_test, df):
        pass

    @abstractmethod
    def get_score(self, x_test_scaled, y_test):
        pass

    @abstractmethod
    def get_rmse(self, y_test, pred):
        pass

    @abstractmethod
    def get_modeltype(self):
        pass

    @abstractmethod
    def get_best_params(self):
        pass

• 具象クラス(例: SVR)
  SVRではGridSearchCVを呼び出す必要があるため、以下のように定義しています。
  呼び出す抽象メソッドはPoC向けにえいやなので、「それいらないでしょう」というのもありますが。

from sklearn.svm import SVR
from sklearn.model_selection import GridSearchCV
import numpy as np
from sklearn.metrics import mean_squared_error
from utility.AbstractModel import AbstractModel

class SVRModel(AbstractModel):

    def __init__(self):
        self.clr = SVR()
        self.regr = None
        self.grid_search = None

    def run_grid_search(self, x_train_scaled, y_train):
        param_grid = {'C': [0.005, 0.0075, 0.1, 0.25, 0.4, 0.5, 0.6, 0.75, 1],
                      'epsilon': [0.000001, 0.00005, 0.00001, 0.0001, 0.0005, 0.001]}
        self.grid_search = GridSearchCV(self.clr, param_grid, cv=5)
        self.grid_search.fit(x_train_scaled, y_train)
        print("best param: {}".format(self.grid_search.best_params_))
        print("best score: {}".format(self.grid_search.best_score_))
        return self.grid_search

    def create_and_pred(self, x_train_scaled, y_train, x_test_scaled):
        self.regr = SVR(C=self.grid_search.best_params_["C"], epsilon=self.grid_search.best_params_["epsilon"])
        self.regr.fit(x_train_scaled, y_train)
        return self.regr.predict(x_test_scaled)

    def print_eval(self, x_train_scaled, x_test_scaled, y_train, y_test, df):
        if self.regr is None:
            raise Exception("needs to run 'create_and_pred' method before call this method.")

        print("テストデータにフィット")
        print("学習データの精度(決定係数r^2 score, 相関) =", self.regr.score(x_train_scaled, y_train))
        print("テストデータの精度(決定係数r^2 score, 相関) =", self.regr.score(x_test_scaled, y_test))
        pred = self.regr.predict(X=x_test_scaled)
        print("RMSE:", np.sqrt(mean_squared_error(y_test, pred)))

    def get_score(self, x_test_scaled, y_test):
        return self.regr.score(x_test_scaled, y_test)

    def get_rmse(self, y_test, pred):
        return np.sqrt(mean_squared_error(y_test, pred))

    def get_modeltype(self):
        return type(self.clr)

    def get_best_params(self):
        return self.grid_search.best_params_

もちろん更に良い形はあると思います。
何にしても、システム開発と違い、トライアンドエラーで繰り返すコードのため、
後で忘れないようにするのが大切かと思います。

この形を流用して、今後はサンプルデータでモデル構築をしてみます。