MultilabelPredictor 만들어보기 (python)

0. 왜 필요해?

동일한 테이블 데이터셋에서 예측하고 싶은 값이 여러 개일 때가 가끔 생긴다. 예를 들면, 기상 상태와 등장하는 선수의 스탯 등으로 야구 경기를 예측하는데 안타 수와 홈런 수를 동시에 예측해보고 싶을 수 있다. 그러면 데이터는 하나인데 각각의 모델을 별개로 만들어줘야한다. 물론 어려운 게 아니기 때문에 모델 각각 만들어서 예측하거나 혹은 scikit-learn의 `MultiOutputRegressor` 등으로도 만들 수 있지만 전자는 번거롭고 후자는 커스텀이 아쉽다. 그래서 번거로움을 줄일 수 있으면서 내가 직접 커스텀할 수 있는 `MultilabelPredictor` 클래스를 만들어보았다.

 

 

1. `MultilabelPredictor` 클래스

1-1. 라이브러리 호출

본인의 입맛에 맞게 수정하자. 나는 회귀 모델에 대한 Predictor를 만들기 위해 `RandomForestRegressor`, `XGBRegressor` 그리고 `LGBMRegressor`를 호출했다. (autogluon처럼 label의 종류에 따라 regressor, classifier를 자동으로 선정해주는 방법도 충분히 가능하다.)

import numpy as np
import pandas as pd
import joblib
import os
from scipy.stats import pearsonr
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score, mean_absolute_percentage_error
from xgboost import XGBRegressor
from lightgbm import LGBMRegressor

 

 

1-2. `MultilabelPredictor` 클래스 정의

class MultilabelPredictor:
    """
    labels: y인자 컬럼명
    eval_metrics: list or str
        XGBoost: {"reg:squarederror", "reg:absoluteerror"}, default:"reg:squarederror" (objective)
        RandomForest: {"squared_error", "absolute_error"}, default:"squared_error" (criterion)
        LightGBM: default:None
    consider_labels_correlation: default:True
        lables=[l1, l2, l3]일 때
        l1의 학습 features = l1, l2, l3를 제외한 모든 features
        l2의 학습 features = l2, l3를 제외한 모든 features (l1 포함)
        l3의 학습 features = l3를 제외한 모든 features (l1, l2 포함)
    model: {"XGBoost", "RandomForest", "LightGBM"}, default:"XGBoost"
    params: parameters of model
    """
    
    def __init__(self, labels, eval_metrics=None, consider_labels_correlation=True, model='XGBoost', **params):
        if len(labels) < 2:
            raise ValueError("MultilabelPredictor is only intended for predicting MULTIPLE labels (columns).")
        if (eval_metrics is not None) and (len(eval_metrics) != len(labels)):
            raise ValueError("If provided, `eval_metrics` must have same length as `labels`")
        self.labels = labels
        self.consider_labels_correlation = consider_labels_correlation
        self.predictors = {}  # key = label, value = model
        self.feature_names = {}
        self.model = model
        
        if eval_metrics is None and model=='XGBoost':
            self.eval_metrics = {labels[i] : 'reg:squarederror' for i in range(len(labels))}
        elif eval_metrics is None and model=='RandomForest':
            self.eval_metrics = {labels[i] : 'squared_error' for i in range(len(labels))}
        elif eval_metrics is None and model=='LightGBM':
            self.eval_metrics = {labels[i] : None for i in range(len(labels))}
        elif isinstance(eval_metrics, list):
            self.eval_metrics = {labels[i] : eval_metrics[i] for i in range(len(labels))}
        elif isinstance(eval_metrics, dict):
            self.eval_metrics = {label: eval_metrics[label] for label in labels}
            
        for label in labels:
            if self.model == 'XGBoost':
                self.predictors[label] = XGBRegressor(objective=self.eval_metrics[label], **params)
            elif self.model == 'RandomForest':
                self.predictors[label] = RandomForestRegressor(criterion=self.eval_metrics[label], **params)
            elif self.model == 'LightGBM':
                self.predictors[label] = LGBMRegressor(objective=self.eval_metrics[label], **params)

나는 최대한 단순하게 모든 예측할 label마다 모델을 통일시켰는데, 다 다르게 커스텀할 수도 있을 것 같다.

그리고 `eval_metrics`도 사용자 편의성을 위해 `rmse`나 `mae` 등을 넣으면 알아서 변환시켜주는 방법도 있을텐데, 코드가 너무 길어져서 최대한 단순화시켰다.

 

여기서 `consider_labels_correlation` 파라미터는 라벨끼리 상관 관계가 있어 예측된 라벨로 다른 라벨을 다시 예측할 것인지를 지정한다. 예를 들어 True로 지정한 경우, 위에서 이야기했던 야구 상황에서 "안타가 많으면 홈런도 많다"는 가정을 넣으면 기존 변수들(기상 상황, 선수 기록)을 가지고 안타를 예측하고, 다시 기존 변수들+예측된 안타 값으로 홈런을 예측한다. False면 둘은 관계가 없다고 가정하여 각각을 예측한다. (autogluon의 MultilabelPredictor에서 참고하였다.)

 

 

 

1-3. 모델 학습(`fit`)

    def fit(self, train_data, **kwargs):
        """
        train_data: pandas DataFrame
        **kwargs: keyword arguments of predictor.fit()
        """
        
        train_data_og = train_data.copy()
        for i in range(len(self.labels)):
            label = self.labels[i]
            predictor = self.predictors[label]
            if not self.consider_labels_correlation:
                labels_to_drop = [l for l in self.labels if l != label]
            else:
                labels_to_drop = [self.labels[j] for j in range(i+1, len(self.labels))]
            
            train_data = train_data_og.drop(columns=labels_to_drop)
            
            X = train_data.drop(columns=label)
            y = train_data[label]
            predictor.fit(X.values, y.values, **kwargs)
            
            self.predictors[label] = predictor
            self.feature_names[label] = X.columns.tolist()
                        
        return self

`X`, `y`를 나눠서 입력할 필요 없이 훈련 데이터셋을 통으로 넣어주면 앞서 클래스를 선언할 때 입력했던 라벨 정보와 모델로 학습을 진행한다.

그리고 `**kwargs`를 통해 모델 학습 파라미터를 입력해주도록 하였다. (일종의 튜닝 용도일까...)

 

1-4. 모델 예측(`predict`) 및 평가(`evaluate`)

    def predict(self, data, **kwargs):
        """
        data: pandas DataFrame
        **kwargs: keyword arguments of predictor.predict()
        """
        X_test = data.copy()
        
        result = pd.DataFrame()
        for label in self.labels:
            predictor = self.predictors[label]
            feature_names = self.feature_names[label]
            pred = predictor.predict(X_test[feature_names].values, **kwargs)
            result[label] = pred
            if self.consider_labels_correlation:
                X_test[label] = pred
            
        result.index = data.index
        
        return result


    def evaluate(self, data, **kwargs):
        """
        data: pandas DataFrame
        **kwargs: keyword arguments of predictor.predict()
        """
        eval_dict = {}
        
        preds = self.predict(data, **kwargs)
        for label in self.labels:
            pred = preds[label]
            real = data[label]
            
            mae = mean_absolute_error(real, pred)
            rmse = np.sqrt(mean_squared_error(real, pred))
            r2 = r2_score(real, pred)
            pearson_corr, _ = pearsonr(real, pred)
            mape = mean_absolute_percentage_error(real, pred)
        
            eval_dict[label] = {'mae':mae, 'rmse':rmse, 'mape': mape, 'r2':r2, 'pearsonr':pearson_corr}
            
        return pd.DataFrame(eval_dict)

`predict`에서는 테스트 데이터를 받아 예측한다. `consider_labels_correlation`이 `True`면 예측한 라벨 값을 테스트 데이터에 덮어쓰기(혹은 컬럼을 생성)하여 다음 라벨을 예측한다. `fit`에서와 마찬가지로 `**kwargs`를 통해 `predict` 파라미터를 입력할 수 있도록 하였다.

 

`evaluate`에서는 내가 확인하고자 하는 평가지표를 넣어주었다. 회귀에서 주로 보는 MAE, RMSE를 비롯하여 MAPE, R2 score, 선형상관계수를 넣어주었다.

 

 

1-5. 모델 저장(`save`) 및 불러오기(`load`)

    def save(self, path):
        """
        path: 모델 저장 경로 지정
        """
        if not os.path.isdir(path):
            os.makedirs(path, exist_ok=True)

        # 모델 구성 저장
        config = {}
        
        config['model'] = self.model
        config['eval_metrics'] = self.eval_metrics
        config['consider_labels_correlation'] = self.consider_labels_correlation
        config['labels'] = self.labels
        config['feature_names'] = self.feature_names
        config_path = os.path.join(path, 'config.json')
        with open(config_path, 'w', encoding='utf-8-sig') as f:
            json.dump(config, f, ensure_ascii=False, indent=4)

        # 모든 모델 저장
        for label, predictor in self.predictors.items():
            model_path = os.path.join(path, f'{label}.joblib')
            joblib.dump(predictor, model_path)

    @classmethod
    def load(cls, path):
        """
        path: 저장된 모델 경로 지정
        """
        config_path = os.path.join(path, 'config.json')
        with open(config_path, 'r', encoding='utf-8-sig') as f:
            config = json.load(f)
        
        # 새 인스턴스 생성
        model_type = config.pop('model', 'XGBoost') 
        eval_metrics = config.pop('eval_metrics', None)
        instance = cls(labels=config['labels'], eval_metrics=eval_metrics,
                       consider_labels_correlation=config['consider_labels_correlation'], model=model_type)

        # 명시적으로 속성을 업데이트
        for key, value in config.items():
            setattr(instance, key, value)

        # 모든 모델을 로드하여 인스턴스에 설정
        instance.predictors = {}
        for label in instance.labels:
            model_path = os.path.join(path, f'{label}.joblib')
            instance.predictors[label] = joblib.load(model_path)

        return instance

만들어진 모델을 저장하고 불러오는 코드이다. `config`를 통해서 클래스가 갖고 있던 값들도 저장해주고 모델 인스턴스도 저장해준다. 불러올 때도 `config`를 통해서 값들을 불러온다. 나도 만들면서 배웠는데, `load`할 때 저렇게 `@classmethod` 데코레이터를 써준 뒤 아래처럼 코드를 작성하면 `predictor = MultilabelPredictor.load(path)`처럼 인스턴스를 선언하지 않고도 불러올 수 있다. (저렇게 하지 않으면 명시적으로 `labels`를 선언해줘서 `predictor = MultilabelPredictor(labels=labels).load(path)`를 해줘야 한다.

 

 

전체 코드: github

 

 

2. 예제

그럼 간단한 사용 예제를 만들어보자.

 

먼저 scikit-learn의 `make_regression`를 통해 임의의 데이터셋을 만들어주었다.

from sklearn.datasets import make_regression

X, y = make_regression(n_samples=1000, n_features=25, n_targets=3, )

y = (y - y.min()) / (y.max() - y.min())

이렇게 하면 데이터 수 1000개, 컬럼 수 25개, 라벨 수 3개인 데이터가 만들어진다.

 

그런데 위에서 만든 코드는 컬럼이 있는 dataframe을 받아야 하기 때문에 df로 바꿔주고 학습을 위해 train test로 나눠준다.

features = [f'f{i:02d}' for i in range(25)]
labels = [f'y{i}' for i in range(3)]

X = pd.DataFrame(X, columns=features)
y = pd.DataFrame(y, columns=labels)


from sklearn.model_selection import train_test_split

train_data, test_data = train_test_split(pd.concat([X, y], axis=1), test_size=0.2, random_state=42)

 

 

학습 준비가 완료되었으니 바로 학습과 평가를 진행한다.

predictor = MultilabelPredictor(labels=labels, model='XGBoost', random_state=24).fit(train_data)
predictor.evaluate(test_data)

`consider_labels_correlation=True`가 기본값이기 때문에 `features`로 `y0`을 예측하고, `features`+`y0`로 `y1`을 예측하고, `features`+`y0`+`y1`으로 `y2`를 예측했을 때의 평가지표다.

 

그럼 `consider_labels_correlation=False`로 하여 예측한 결과를 살펴보자. (위와 동일한 결과를 위해 `random_state`를 주었다.)

predictor = MultilabelPredictor(labels=labels, model='XGBoost', consider_labels_correlation=False, random_state=24).fit(train_data)
predictor.evaluate(test_data)

위와 거의 동일하지만 약간의 차이가 존재한다. `y0`, `y1`, `y2`가 서로 관계가 없다보니 아래 결과의 오차가 더 적은 것을 확인할 수 있다. (혹은 예측한 값으로 예측을 하다보니, 처음 예측한 결과가 부정확하면 이후 예측의 오차는 계속 커지는 결과를 낳아서 그럴 수도 있다. 마치 LSTM처럼)

 

 

마지막으로 모델 저장과 모델 불러오기가 잘 작동하는지도 확인해보자.

predictor.save('multilabel') # 모델 저장


predictor = MultilabelPredictor.load('multilabel') # 모델 호출

predictor.evaluate(test_data)

위와 정확히 동일한 결과가 나왔다. 모델이 잘 불러와진 것까지 확인할 수 있었다.

 

 

 

전체 코드: github

 

 

3. 참고

Predicting Multiple Columns in a Table (Multi-Label Prediction) - AutoGluon 1.0.0 documentation