Contents

• 2022년 업무 요약
• 블로그 시작
• 원격 근무 느낀점
• 이직.. 해치웠나?
• 2023년 목표

2022년 업무 💻

2022년에는 농산물 거래 플랫폼을 위한 데이터 프로덕트를 주로 개발했습니다. 프로젝트를 3가지로 짧고 굵게 요약해 봤습니다.

1) 농산물 가격 예측 모델링

B2B 온라인 농산물 거래 플랫폼을 활성화시키기 위한 일환으로 사내에 농산물 시황을 파악할 수 있는 서비스가 필요했습니다. 그래서 농산물 거래 가격 트렌드와 예측 가격 현황을 파악할 수 있는 프로덕트를 만들어야 했고, 이것을 위한 개발을 전반적으로 진행했습니다.

• 시계열 예측을 경험할 기회가 생겨서 좋았습니다. 통계적 모형부터 딥러닝을 활용한 모델링까지 여러 실험을 시도할 수 있는 시간이 주어진 덕분에 시계열 예측에서의 알고리즘별 특징을 이해하게 되었습니다. 특히 국내 농산물 가격 예측 연구 논문을 참고하여 회사 비즈니스에 필요한 형태로 개발하다보니 Applied Data Scientist로서 역할을 많이 하게 된 것 같습니다.
• 농산물별로 예측 변수는 매우 다양합니다. 출하량, 가격, 날씨, 재배면적 등의 변수를 모든 작물에 동일 적용하기보다는, 각 품목별로 가격에 영향을 주는 변수를 먼저 연구하면 더 좋았을 것 같습니다. 작물이 다양한만큼 농산물 가격 변동을 설명하기 위한 연구 분야는 여전히 어려운 과제 중 하나긴 합니다.
• 이런 부분에서 data-centric AI가 더 중요하다고 느꼈습니다. 비슷한 학습 데이터로부터 여러 알고리즘 내에서 최적화를 진행하기보다는, 데이터 자체를 보강해야 한다는 점을 충분히 배운 프로젝트였습니다.

2) ML 파이프라인 개발

시계열 예측뿐만 아니라 다양한 ML/DL 모델 훈련을 스케줄링을 통해 배치로 수행하는 파이프라인을 만들었습니다. 사내 인프라는 AWS를 사용하고 있어서 SageMaker와 Batch를 통해 나름대로 안정적인 ML 파이프라인을 설계하여 운영했습니다.

• Conda 환경을 제공하는 SageMaker 기반으로 '인스턴스 생성' → '모델 훈련' → '예측 결과 저장' → '인스턴스 중지 및 삭제'를 수행하는 DAG 템플릿을 개발했습니다.
• 특히 팀 내 ML Engineer 덕분에 많은 걸 배울 수 있는 기회였습니다. CT(Continuous Training)를 수행할 때 학습된 모델은 MLflow을 통해 Model Registry를 따로 하며, 등록된 모델을 BentoML을 통해 API로 호출하여 Inference를 수행하도록 설계한 덕분에 MLOps 디자인 패턴과 아키텍처를 더 깊이 이해할 수 있어 좋았습니다.
• CT를 통해 새롭게 학습된 최신 버전의 모델만을 사용하기보다는 Metric을 모니터링하면서 Model Selection에 개입을 할 수 있는 트리거를 만들고 싶었지만, 이 부분을 못한 게 아쉬웠습니다.

3) 데이터 엔지니어링

2022년 한 해의 절반 이상은 데이터 엔지니어링에 투자할 수밖에 없던 시간이었던 것 같네요.

• 농산물 거래 데이터는 공공기관에서만 제공하기에 외부 API에 의존해야 하는 개발 환경이었습니다. 그러다 보니 장애를 빠르게 확인할 수 있는 Alert 기능부터 Backfill 장치까지 모두 만드느라 상반기는 거의 DE로 지냈습니다.
• 전 직장에서는 GCP를 사용했기에 BigQuery와 Composer(Airflow) 조합으로 모든 걸 끝낼 수 있었던 것에 반해, 여기에서는 AWS를 사용하다 보니 서비스 레이어가 상당히 많았습니다. MWAA(Airflow), S3, Athena, Glue, Redshift, Batch, SageMaker 등등 신경 써야 할 게 많았지만 그래도 AWS 기반의 데이터 엔지니어링 역량을 한껏 올린 것 같아 좋았네요.

블로그 시작 📒

연말까지 꾸준히 쓰진 못했지만 그래도 블로그를 시작한 덕분에 이렇게 회고도 쓸 수 있게 된 것 같습니다. 23년 목표가 있다면 한 달에 최소 1개는 쓰는 것인데, 이번엔 꼭 지켜보려고 합니다.

제 블로그의 성격은 업무를 통해 익힌 노하우와 활용 사례를 공유하는 것이다 보니 논문 리뷰 등과 같은 지식 전달은 약해서 멈칫하는 때가 많았습니다. 하지만 제가 더 좋아하고 잘 할 수 있는 장르로 써야 계속 연재할 수 있을 것 같았습니다.

공개된 블로그인만큼 나만 알아볼 수 있는 표현과 문맥으로 가득 찬 메모 같은 느낌의 포스팅보다는, 적어도 데이터 분야에 관심 있는 사람이 방문했을 때 이해할 수 있는 범용적인 글을 쓰려고 노력해 보고 있습니다. 여하튼 22년에 쓴 것보다는 더 많이 포스팅을 해볼 계획입니다.

원격 근무 느낀점 🏠

2022년은 거의 재택으로 일을 했습니다. 주 1회 사무실로 출근하는 빈도였으니까요. 출퇴근 시간을 아낄 수 있으니 여유가 생긴 만큼 조금 더 일에 집중할 수 있던 것 같습니다. 한 해를 재택 하면서 느낀 점을 써봤습니다.

• 여유로운 식사 + 운동 빈도 증가
  • 출퇴근 시간을 줄일 수 있기 때문에 아침 식사와 저녁 식사를 여유롭게 챙겨 먹을 수 있어 좋았습니다. 덕분에 운동 빈도도 늘어났어요.
• 적절한 휴식
  • 업무를 하다가 피로가 심하다면 언제든 누워서 쉴 수 있어 좋았습니다. 허리가 좋지 않다 보니 누워서 쉬거나 주기적인 스트레칭 시간이 필요한데, 이런 점에서도 좋았습니다.
• 정리 노하우 증가
  • 동료와 만나는 시간이 줄어든 만큼 소통도 줄어들어 업무 진척 속도가 느리지 않을까 싶었지만, 그만큼 본인이 하는 일을 잘 정리해서 기록하는 빈도가 늘어났습니다. 물론 업무 기록은 기본적인 요소이지만 조금 더 꼼꼼하게 정리하는 습관이 생긴 것 같네요.
• 정서적 교류 감소
  • 만나는 시간이 줄어들어서 동료와 많이 친해지긴 어려웠습니다. 코로나가 시작하기 전에 다녔던 직장에서는 항상 동료와 티키타카를 하면서 지내다 보니 내집단 의식이 강했던 반면, 이번엔 각자도생 하는 느낌이 강해서 아쉽긴 했습니다.

이직.. 해치웠나? 🏢

저에게 2022년 빅 이벤트를 꼽자면 이직입니다. 11월 초부터 본격적으로 준비했었는데 다행히 연말 전에 새로운 보금자리를 찾았습니다. 새해의 시작을 새로운 직장에서 이어가게 되었네요.

이번 이직은 저의 정체성을 찾아가기 위한 관문이기도 했습니다. 예전 글에서도 쓴 내용이지만, 개인의 성장이 멈추지 않으려면 '내가 잘하는 것 + 잘하고 싶은 것'과 '회사에서 필요한 것'의 교집합을 계속 찾아야 합니다. 저는 이 교집합의 영역이 다소 넓지 않음을 느끼다 보니 이직을 결심했고, 운이 좋게도 이것을 만족시킬만한 곳을 찾게 되었습니다.

그리고 주변에서 말하듯이 요즘 이직이 쉽지 않은 걸 확실히 느꼈습니다. 과제도 만만하지 않았고, 실무 면접에서도 진땀을 뺐습니다. 쉽지 않은 과정이었지만 최종 문턱을 넘었기에 나름 간절했던 것 같습니다. 이제는 데이터사이언티스트 혹은 머신러닝 엔지니어로서 경쟁력을 갖추려면 추천, 자연어, 비전 등의 도메인에서 최소 한 가지는 압도적으로 잘해야 할 것 같습니다. 블로그 같은 포트폴리오도 분명 가점이 있겠지만, 점점 도메인별 모델러나 엔지니어 경험이 훨씬 중요해지는 것 같습니다.

마무리 🌷

회고를 작성할 때마다 늘 빼먹지 않고 남기는 구절이 있는데, 이번에도 우려먹어 봅니다. 헤헤.

걱정은 흔들의자와 같다. 계속 움직이지만 아무 데도 가지 않는다.

인생의 절반은 걱정하는 데 시간을 쓴다고 합니다. 그래서 많은 걱정을 하지 말라는 말이죠. 근데 그게 쉽나요. 다이어트라는 평생 숙제를 달고 사는 세상이 되어버린 이상 걱정이란 것을 안 할 수 없습니다. 그런데 2022년은 나름대로 만족스러운 나날의 연속이었습니다. 걱정만 하진 않았기 때문입니다.

이 걱정을 줄여 나가는 과정에는 '실행력'이 한몫 했습니다. '아 잘 모르겠지만 일단 먼저 해보자'라는 마인드셋이 강했습니다. 때로는 돌다리도 두드려보며 건너는 조심스러운 때도 있었지만, 2022년은 대부분 '일단 고!'였습니다.

몰랐던 것도 일단 하다 보니 알게 되고, 알던 것도 다시 배우게 되는 시간이 많았습니다. 그래서 확실히 느꼈습니다. 잘 해야될 때도 분명 있지만 그냥 해야할 때가 더 많다는 것을요.

그래서 걱정이 그렇게 나쁘지만은 않다고 생각합니다. 불안함과 초조함에서 비롯된 걱정이 잠시 제자리에 있을진 모르겠지만, 일단 해결하려고 시도하면 분명 인생을 꽃길로 물들여줄 씨앗이 되어주기도 하는 것 같거든요.

2023년에도 이 자세를 최대한 유지해보려고 합니다. 이렇게 된다면 아마도 올 해 또 생길 걱정 씨앗들이 향기로운 꽃이 되는 순간을 볼 거라 믿습니다. 더 나아가 달콤한 열매까지 수확할 수 있는 시간이 될 겁니다.

여러분도 걱정 씨앗을 뿌렸다면 제대로 수확한 2022년이 되었길 바라며, 2023은 더 풍족해지시길 바라겠습니다.

아직 많이 부족하지만 실전을 통해 체득한 지식과 노하우를 기록해보려고 합니다. 잘못된 내용이 있다면 지적도 부탁드립니다! 🤚

✍ 이번 포스팅의 목적

요즘 시계열 예측을 LSTM으로 시도해보고 있습니다. 늘 하면서 느끼지만 하나의 모듈로 만들어 놓지 않으면 직성이 안 풀리는 성격인지라(대체 왜 그런 걸까?🤔) 만들어 놓은 기능을 이번 기회에 정리해 보려고 합니다. LSTM(RNN)의 개념은 이미 많은 자료에서 소개되고 있으므로 생략하고, 단변량(univariate)뿐만 아니라 다변량(multivariate)에서도 활용할 수 있는 LSTM Neural Network 시계열 예측 모듈 구현 과정을 공유해 보겠습니다.

🕰 LSTM 시계열 예측

딥러닝으로 시계열 예측을 한다면 LSTM이 뼈대가 되는 게 일반적이고 여기에 CNN을 추가하여 CNN-LSTM 등으로 확장된 네트워크 구조로 여러 실험을 진행하는 초석이 된다고 보면 될 것 같습니다. 물론 딥러닝 기반의 시계열 예측이라고 해서 일반적인 모델링 과정과 크게 다르지 않습니다.

사진의 3번에 해당하는 Sequential Dataset으로 모델을 학습시킨다는 것이 핵심인데, 이 Sequence Data 모양에 따라 학습 결과가 상당히 달라질 수 있습니다. 이처럼 Sequence Data 형태에 따라 모델 성능을 확인해야 하므로 Sequence 길이가 하나의 하이퍼파라미터인 셈이며 이것 말고도 노란색 박스로 표시된 부분을 모두 진행해야 제대로된 모델링을 했다고 볼 수 있습니다.

이 과정을 수행할 수 있는 코드를 소개할 거라 아마도 글이 길어질 것 같습니다. 그래서 두 편으로 나누어 포스팅할 것인데, 여기서는 6번까지만 설명하고 나머지는 다음 게시글에서 이어서 써보려고 합니다.

⛓ Sequential Dataset

Sequential Dataset을 부연 설명하자면, 예측에 사용할 Feature의 길이(Sequence Length)와 예측할 길이(Step)에 따라 연속된 Sequence Data를 만들어내고 이것을 합친 Dataset을 생성하는 것입니다. 사진의 예시에서 Sequence Length는 5, Step도 5가 되는 것이죠. 보통 Sequence Length보다는 Window Size라는 표현을 많이 사용하지만 여기에서는 직관적인 단어로 사용하겠습니다. 😅

💻 모듈 소개

전체 코드를 한 블록에 붙여 넣으면 가독성이 떨어질 것 같아서 각 기능을 하나의 클래스에 담아내는 방식으로 작성했습니다. 그래서 위에서 설명한 모델링 순서대로 적어봤는데, 혹시 보시는 분 중 댓글로 의견 남겨주시면 적극 반영하여 수정하겠습니다! 🙆‍♂️

클래스 생성

먼저 필요한 라이브러리를 호출하고 클래스를 하나 만들어 줍니다. 생성자에는 굳이 다른 attribute를 미리 선언해 놓진 않았고 보기 좋게(?) random_seed만 선언 해봤습니다.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import tensorflow as tf
from tensorflow.keras.layers import Dense
from tensorflow.keras.layers import LSTM
from tensorflow.keras.layers import Dropout
from tensorflow.keras.layers import Activation
from tensorflow.keras.models import Sequential
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.losses import MSE
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.callbacks import ReduceLROnPlateau


class ForecastLSTM:
    def __init__(self, random_seed: int = 1234):
        self.random_seed = random_seed

Input Dataset

해당 모듈에서는 Raw Dataset이 Pandas DataFrame인 경우를 가정했는데, 많은 데이터 분석가와 데이터 과학자가 이것으로 데이터를 핸들링하는 것에 기반하여 만들었습니다. 저 역시도 그렇고요!

또, 예측할 값의 컬럼의 이름을 y로 고정했는데요, 시계열에서 예측할 값을 일반적으로 y로 사용하는 것에 기반한 것입니다. 필요하다면 DataFrame의 컬럼명을 검사하는 Validator 함수를 넣으려고 했으니 굳이 추가하진 않았습니다. 이 함수를 통해 Sequential Dataset 생성에 필요한 Numpy Array로 변환합니다.

def reshape_dataset(self, df: pd.DataFrame) -> np.array:
    # y 컬럼을 데이터프레임의 맨 마지막 위치로 이동
    if "y" in df.columns:
        df = df.drop(columns=["y"]).assign(y=df["y"])
    else:
        raise KeyError("Not found target column 'y' in dataset.")

    # shape 변경
    dataset = df.values.reshape(df.shape)
    return dataset

ForecastLSTM.reshape_dataset = reshape_dataset

Sequential Dataset

변환된 데이터셋으로 Sequence Length와 Step에 따라 Sequence Data를 만든 후 이것을 합친 Dataset을 생성하여 반환하는 함수입니다.

def split_sequences(
    self, dataset: np.array, seq_len: int, steps: int, single_output: bool
) -> tuple:

    # feature와 y 각각 sequential dataset을 반환할 리스트 생성
    X, y = list(), list()
    # sequence length와 step에 따라 sequential dataset 생성
    for i, _ in enumerate(dataset):
        idx_in = i + seq_len
        idx_out = idx_in + steps
        if idx_out > len(dataset):
            break
        seq_x = dataset[i:idx_in, :-1]
        if single_output:
            seq_y = dataset[idx_out - 1 : idx_out, -1]
        else:
            seq_y = dataset[idx_in:idx_out, -1]
        X.append(seq_x)
        y.append(seq_y)
    return np.array(X), np.array(y)


ForecastLSTM.split_sequences = split_sequences

여기서 single_output=True은 예측할 값이 한개인 경우를 의미합니다. 만약 step=5이면서 single_output=True라면 아래의 Sequential Dataset이 만들어지는 것입니다.

Split Dataset

시퀀스 데이터셋으로부터 모델 훈련에 사용할 Train dataset, Validation dataset을 분리하는 함수를 따로 추가해 줍니다.

def split_train_valid_dataset(
    self,
    df: pd.DataFrame,
    seq_len: int,
    steps: int,
    single_output: bool,
    validation_split: float = 0.3,
    verbose: bool = True,
) -> tuple:
    # dataframe을 numpy array로 reshape
    dataset = self.reshape_dataset(df=df)

    # feature와 y를 sequential dataset으로 분리
    X, y = self.split_sequences(
        dataset=dataset,
        seq_len=seq_len,
        steps=steps,
        single_output=single_output,
    )

    # X, y에서 validation dataset 분리
    dataset_size = len(X)
    train_size = int(dataset_size * (1 - validation_split))
    X_train, y_train = X[:train_size, :], y[:train_size, :]
    X_val, y_val = X[train_size:, :], y[train_size:, :]
    if verbose:
        print(f" >>> X_train: {X_train.shape}")
        print(f" >>> y_train: {y_train.shape}")
        print(f" >>> X_val: {X_val.shape}")
        print(f" >>> y_val: {y_val.shape}")
    return X_train, y_train, X_val, y_val


ForecastLSTM.split_train_valid_dataset = split_train_valid_dataset

Build LSTM

LSTM 모델을 생성하고 Dense layer의 Unit과 Dropout을 원하는만큼 이어붙일 수 있도록 설계했습니다. 또한, LSTM()에서 return_sequences=True 혹은 return_sequences=False 파라미터를 선택할 수 있도록 변경하면 더욱 좋습니다.

문제 정의에 따라 다르지만, 예측 기간이 길면서 Multi Output을 산출해야 하는 모델이라면 many-to-many 방식으로 가중치를 갱신하는 return_sequences=True를 지정하는 것도 좋은 방법입니다.

(return_sequences의 자세한 이해는 이 게시글을 추천합니다!)

def build_and_compile_lstm_model(
    self,
    seq_len: int,
    n_features: int,
    lstm_units: list,
    learning_rate: float,
    dropout: float,
    steps: int,
    metrics: str,
    single_output: bool,
    last_lstm_return_sequences: bool = False,
    dense_units: list = None,
    activation: str = None,
):
    """
    LSTM 네트워크를 생성한 결과를 반환한다.

    :param seq_len: Length of sequences. (Look back window size)
    :param n_features: Number of features. It requires for model input shape.
    :param lstm_units: Number of cells each LSTM layers.
    :param learning_rate: Learning rate.
    :param dropout: Dropout rate.
    :param steps: Length to predict.
    :param metrics: Model loss function metric.
    :param single_output: Whether 'yhat' is a multiple value or a single value.
    :param last_lstm_return_sequences: Last LSTM's `return_sequences`. Allow when `single_output=False` only.
    :param dense_units: Number of cells each Dense layers. It adds after LSTM layers.
    :param activation: Activation function of Layers.
    """
    tf.random.set_seed(self.random_seed)
    model = Sequential()

    if len(lstm_units) > 1:
        # LSTM -> ... -> LSTM -> Dense(steps)
        model.add(
            LSTM(
                units=lstm_units[0],
                activation=activation,
                return_sequences=True,
                input_shape=(seq_len, n_features),
            )
        )
        lstm_layers = lstm_units[1:]
        for i, n_units in enumerate(lstm_layers, start=1):
            if i == len(lstm_layers):
                if single_output:
                    return_sequences = False
                else:
                    return_sequences = last_lstm_return_sequences
                model.add(
                    LSTM(
                        units=n_units,
                        activation=activation,
                        return_sequences=return_sequences,
                    )
                )
            else:
                model.add(
                    LSTM(
                        units=n_units,
                        activation=activation,
                        return_sequences=True,
                    )
                )
    else:
        # LSTM -> Dense(steps)
        if single_output:
            return_sequences = False
        else:
            return_sequences = last_lstm_return_sequences
        model.add(
            LSTM(
                units=lstm_units[0],
                activation=activation,
                return_sequences=return_sequences,
                input_shape=(seq_len, n_features),
            )
        )

    if single_output:  # Single Step, Direct Multi Step
        if dense_units:
            for n_units in dense_units:
                model.add(Dense(units=n_units, activation=activation))
        if dropout > 0:
            model.add(Dropout(rate=dropout))
        model.add(Dense(1))
    else:  # Multiple Output Step
        if last_lstm_return_sequences:
            model.add(Flatten())
        if dense_units:
            for n_units in dense_units:
                model.add(Dense(units=n_units, activation=activation))
        if dropout > 0:
            model.add(Dropout(rate=dropout))
        model.add(Dense(units=steps))

    # Compile the model
    optimizer = Adam(learning_rate=learning_rate)
    model.compile(optimizer=optimizer, loss=MSE, metrics=metrics)
    return model


ForecastLSTM.build_and_compile_lstm_model = build_and_compile_lstm_model

Model Training

이제 위에서 추가한 함수들을 호출하여 모델을 훈련하는 기능입니다. 이 부분이 사실상 메인에 해당하다 보니 함수에 파라미터가 많습니다 😅. 조금 복잡해 보이지만 구분해보면 딱 필요한 것(?)만 들어있습니다.

• LSTM 모델 하이퍼라라미터

• 훈련 데이터셋

• Sequence Data 길이 (Sequence Length)

• 예측 기간 (Step)

  def fit_lstm(
    self,
    df: pd.DataFrame,
    steps: int,
    lstm_units: list,
    activation: str,
    dropout: float = 0,
    seq_len: int = 16,
    single_output: bool = False,
    epochs: int = 200,
    batch_size: int = None,
    steps_per_epoch: int = None,
    learning_rate: float = 0.001,
    patience: int = 10,
    validation_split: float = 0.3,
    last_lstm_return_sequences: bool = False,
    dense_units: list = None,
    metrics: str = "mse",
    check_point_path: str = None,
    verbose: bool = False,
    plot: bool = True,
  ):
    """
    LSTM 기반 모델 훈련을 진행한다.
  
    :param df: DataFrame for model train.
    :param steps: Length to predict.
    :param lstm_units: LSTM, Dense Layers
    :param activation: Activation function for LSTM, Dense Layers.
    :param dropout: Dropout ratio between Layers.
    :param seq_len: Length of sequences. (Look back window size)
    :param single_output: Select whether 'y' is a continuous value or a single value.
    """
  
    np.random.seed(self.random_seed)
    tf.random.set_seed(self.random_seed)
  
    # 훈련, 검증 데이터셋 생성
    (
        self.X_train,
        self.y_train,
        self.X_val,
        self.y_val,
    ) = self.split_train_valid_dataset(
        df=df,
        seq_len=seq_len,
        steps=steps,
        validation_split=validation_split,
        single_output=single_output,
        verbose=verbose,
    )
  
    # LSTM 모델 생성
    n_features = df.shape[1] - 1
    self.model = self.build_and_compile_lstm_model(
        seq_len=seq_len,
        n_features=n_features,
        lstm_units=lstm_units,
        activation=activation,
        learning_rate=learning_rate,
        dropout=dropout,
        steps=steps,
        last_lstm_return_sequences=last_lstm_return_sequences,
        dense_units=dense_units,
        metrics=metrics,
        single_output=single_output,
    )
  
    # 모델 적합 과정에서 best model 저장
    if check_point_path is not None:
        # create checkpoint
        checkpoint_path = f"checkpoint/lstm_{check_point_path}.h5"
        checkpoint = ModelCheckpoint(
            filepath=checkpoint_path,
            save_weights_only=False,
            save_best_only=True,
            monitor="val_loss",
            verbose=verbose,
        )
        rlr = ReduceLROnPlateau(
            monitor="val_loss", factor=0.5, patience=patience, verbose=verbose
        )
        callbacks = [checkpoint, EarlyStopping(patience=patience), rlr]
    else:
        rlr = ReduceLROnPlateau(
            monitor="val_loss", factor=0.5, patience=patience, verbose=verbose
        )
        callbacks = [EarlyStopping(patience=patience), rlr]
  
    # 모델 훈련
    self.history = self.model.fit(
        self.X_train,
        self.y_train,
        batch_size=batch_size,
        steps_per_epoch=steps_per_epoch,
        validation_data=(self.X_val, self.y_val),
        epochs=epochs,
        use_multiprocessing=True,
        workers=8,
        verbose=verbose,
        callbacks=callbacks,
        shuffle=False,
    )
  
    # 훈련 종료 후 best model 로드
    if check_point_path is not None:
        self.model.load_weights(f"checkpoint/lstm_{check_point_path}.h5")
  
    # 모델링 과정 시각화
    if plot:
        plt.figure(figsize=(12, 6))
        plt.plot(self.history.history[f"{metrics}"])
        plt.plot(self.history.history[f"val_{metrics}"])
        plt.title("Performance Metric")
        plt.xlabel("Epoch")
        plt.ylabel(f"{metrics}")
        if metrics == "mape":
            plt.axhline(y=10, xmin=0, xmax=1, color="grey", ls="--", alpha=0.5)
        plt.legend(["Train", "Validation"], loc="upper right")
        plt.show()
  
  

ForecastLSTM.fit_lstm = fit_lstm

## Forecast validation dataset
훈련 종료 이후 검증 데이터셋에 대한 예측 결과를 반환하는 함수도 필요합니다. Epochs에 따라 Validation Dataset의 Loss를 확인할 수 있지만, 검증 데이터셋에 대한 실제 값과 예측값으로부터 오차를 확인할 필요가 있기 때문입니다. 이 오차(Performance Metric)에 의해 최적의 하이퍼파라미터를 찾을 수 있게 되고, 최적의 모델로부터 Test Dataset 예측을 통해 최종 성능을 기록하는 과정을 수행해야 합니다.
```python
def forecast_validation_dataset(self) -> pd.DataFrame:
    # 검증 데이터셋의 실제 값(y)과, 예측 값(yhat)을 저장할 리스트 생성
    y_pred_list, y_val_list = list(), list()

    # 훈련된 모델로 validation dataset에 대한 예측값 생성
    for x_val, y_val in zip(self.X_val, self.y_val):
        x_val = np.expand_dims(
            x_val, axis=0
        )  # (seq_len, n_features) -> (1, seq_len, n_features)
        y_pred = self.model.predict(x_val)[0]
        y_pred_list.extend(y_pred.tolist())
        y_val_list.extend(y_val.tolist())
    return pd.DataFrame({"y": y_val_list, "yhat": y_pred_list})


ForecastLSTM.forecast_validation_dataset = forecast_validation_dataset

Performance Metric

예측 결과로부터 오차를 확인할 수 있는 함수를 하나 만들어 두면 좋습니다. 이것까지 클래스에 넣기 보다는 evaluation 용도로 따로 분리하는 게 적절한 것 같습니다. 아주 간단하게 만든 버전이고, mase, coverage, winkler score 등을 계산하는 것도 추가하면 좋습니다.

def calculate_metrics(df_fcst: pd.DataFrame) -> dict:
    true = df_fcst["y"]
    pred = df_fcst["yhat"]

    mae = (true - pred).abs().mean()
    mape = (true - pred).abs().div(true).mean() * 100
    mse = ((true - pred) ** 2).mean()
    return {
        "mae": mae,
        "mape": mape,
        "mse": mse,
    }

🤔 활용 예시

주(week) 단위의 시계열 데이터셋이 있다고 가정하고, 여기까지 소개한 코드를 사용한다면 아래와 같은 느낌입니다.

## 1) Train, Test 데이터 분리
cutoff = "2022-01-01"
df_train = df[df.index < cutoff]
df_test = df[df.index >= cutoff]

## 2) Sequence Length, 예측 기간(Step), Single Output 여부 등 정의
seq_len = 5  # 과거 5주의 데이터를 feature로 사용
steps = 5  # 향후 5주의 y를 예측
single_output = False  # 향후 5주차의 시점만이 아닌, 1~5주 모두 예측
metrics = "mse"  # 모델 성능 지표

## 3) LSTM 하이퍼파라미터 정의
lstm_params = {
    "seq_len": seq_len,
    "epochs": 300,  # epochs 반복 횟수
    "patience": 30,  # early stopping 조건
    "steps_per_epoch": 5,  # 1 epochs 시 dataset을 5개로 분할하여 학습
    "learning_rate": 0.01,
    "lstm_units": [64, 32],  # Dense Layer: 2, Unit: (64, 32)
    "activation": "relu",
    "dropout": 0,
    "validation_split": 0.3,  # 검증 데이터셋 30%
}

## 4) 모델 훈련
fl = ForecastLSTM()
fl.fit_lstm(
    df=df_train,
    steps=steps,
    single_output=single_output,
    metrics=metrics,
    **lstm_params,
)

## 5) Validation dataset 예측 성능
df_fcst_val = fl.forecast_validation_dataset()
val_loss = calculate_metrics(df_fcst=df_fcst_val)[metrics]
print(f"{metrics} of validation dataset: {val_loss.round(3)}")

이 코드를 실행하면 아래와 같이 모델 성능을 바로 확인할 수 있으며, 이러한 구조를 토대로 하이퍼파라미터 튜닝 코드까지 개발하여 진행한다면 LSTM 시계열 모델링은 어렵지 않게 할 수 있습니다. (참 쉽죠?)

여기까지가 모델 훈련을 위한 코드입니다. 이후엔 Test Dataset로 예측값과 실제값을 비교하여 모델의 최종 성능을 일반화할 수 있습니다. 다음 편에서 최종 성능 측정 과정을 소개하여 매듭을 지어보겠습니다.

⌛ 1편 마무리

LSTM을 사용한 시계열 예측 활용 사례를 찾아보면 Quick Start 수준에 머물러있는 느낌을 많이 받았습니다. 그래서 이번 글을 통해 LSTM 모델링을 추상화된 함수로 구현하여 용이하게 사용할 수 있는 사례를 공유해 보고자 작성해 봤습니다. (물론 이것도 거의 vanilla version에 해당하는 수준이긴 하지만..😇)

1편은 코드 기반의 설명이 주를 이었는데, 2편에서는 모델링 시 유의해야 할 점과 알면 좋을 것들에 대한 내용도 추가로 작성할 예정입니다!

그래서 이번 글은 5년 차를 맞이하는 기념(?)으로 저의 직무에 대한 회고를 간단히 적어 보면서 나약해진 정신을 재무장해보고자 합니다.

데이터 분석, 호기심이 반이다.

저는 데이터 분석가로 일을 시작했습니다. 타이틀은 분석이지만 데이터 집계, 추출 요청 대응에도 많은 시간을 할애해야 하는 직군이죠. 하지만 다행히 저에겐 이 시간이 낭비되었다고 생각하진 않았습니다. 이 과정을 거치면서 다양한 분석 주제를 접할 수 있었기 때문입니다.

이전 회사에서 일할 때 서비스, 마케팅 부서에 각각 소속되어 일을 해봤습니다. 그리고 데이터실이라는 중앙 조직이 따로 있었습니다. '아 나도 데이터 실에서 일하면 더 좋았을 것 같은데'라는 생각에 잠긴 적이 있었지만, 되돌아보니 오히려 현업 부서에 있던 것이 더 재밌는 분석을 할 기회가 많았습니다.

현업 부서에 속하다 보니 추출과 대응 업무가 끊임없는 환경이었음에도 이 일을 지속 가능하게 한 원동력은 저에게 있어서 호기심이었습니다. 이러한 호기심은 때때로 좋은 분석 주제로 물망에 오르곤 하는데, 현업에 계신 마케팅 전문가, 서비스 운영 전문가와 얘기하면서 가설이 나오기 마련이고, 이것을 직접 확인해 보는 과정과 결과를 통해 재미를 느꼈던 것 같습니다.

그래서 데이터 분석에서 중요한 것은 기초 통계와 데이터 정제 능력도 있겠지만, 불타오르는 호기심 천국 마인드를 유지하는 것도 중요한 것 같습니다. 이는 곧 프로덕트 관점에서의 가설 검정을 위한 실험 설계로 이어지기 마련이고, 조금씩 난이도가 있는 분석 주제를 헤쳐 나감으로써 실력을 한층 탄탄하게 할 수 있는 불씨가 되지 않나 생각합니다. 물론 이런 동기부여는 누군가 해주기보다는 스스로 끌어낼 수 있어야 하는데, 어쩌면 성격에 의해 좌우되는 것인가 싶기도 합니다. 🤔

데이터 엔지니어링, 흉내라도 내보자.

지금의 회사로 이직할 때 좋았던 점이 직접 데이터 인프라를 구성해 볼 수 있는 기회가 있다는 것이었습니다. 이전 회사는 데이터 레이크가 상당히 잘 구축되어 있었고, 다양한 Fact Table이 일 배치로 척척 잘 쌓이고 있었기 때문에 Mart Table을 구성하는 일이 그다지 어렵지 않았습니다. 그런데 이번에 몸담고 있는 회사를 왔을 땐 DW를 새롭게 개발해야 했는데, ODS → DataLake → Fact Table → Mart Table까지 구성하는 파이프라인을 AWS로 개발하는 좋은 경험을 할 수 있었습니다.

DataOps 경험 덕분에 머신러닝 모델링 과정 중 Feature Store에 등록할 데이터 마트 설계와 적재를 어렵지 않게 할 수 있는 능력을 갖추게 되었습니다. 이를 통해 찐 데이터 엔지니어가 얼마나 대단한 사람인지, 데이터 엔지니어가 있음으로써 데이터 과학자와 분석가가 얼마나 편하게 일할 수 있는 것인지도 새삼 깨닫게 되었습니다.

백종원이 착한 척도 오래 하다 보니까 삶이 됐다고 합니다. 저도 처음 배울 때는 엔지니어링 흉내만이라도 낼 수 있는 능력을 갖추자라고 생각했지만, 지금은 이 덕분에 End to end로 일하는 데이터 스페셜리스트가 되어 업무 생산성이 높아진 것 같습니다.

데이터 프로덕트를 만들자.

인생 첫 데이터 프로덕트 개발을 추천(Recommendation) 서비스를 통해 익혔습니다. 추천 알고리즘에 사용할 Feature 적재 → 모델 훈련 및 검증 → 랭킹 알고리즘 설계 → 추천 데이터 적재 → CVR 측정까지를 약 반년 정도 매진하며 처음으로 프로젝트의 시작과 끝을 소화했었습니다. 여기서 느낀 것이, '아 데이터 프로덕트 개발을 하면 Data Science 영역의 대부분을 경험하는구나'였습니다. 이 때문에 더더욱 데이터 프로덕트 개발에 눈을 돌리게 됐고, 이 프로젝트 덕분에 데이터 과학자로 이직에 성공하게 되었습니다.

하지만 데이터 프로덕트라는 것이 누군가에겐 허황으로 들릴 수 있습니다. '그게 있으면 어떤 기대 효과가 있는 건가요?'라는 질문에 답을 할 수 있어야 하죠. 그래서 기획 문서를 만들어 공유해 보기도 하고, 회의를 통해 아이디어를 얘기하면서 프로젝트 주제를 잡아 보려고 노력했지만 생각대로 잘 흘러가지 않은 경우가 다반사였습니다. 특히 이직 직후 과거에 했던 프로젝트에 매몰되어 비슷한 것만 추진하려는 것이 저의 문제였는데, 지금 있는 곳에서 뭘 할 수 있는지를 보여주는 것이 더 중요하다는 걸 이제서야 깨닫게 되었습니다.

이쯤 되어 제 머릿속에 내려진 결론은 '제가 예전에 이런 거 만들어 봤어요! 저희 비슷하게 해보면 어때요?'라는 제안 보다 직접 데모로 만들어서 눈으로 보여주는 것입니다. 말 그대로 백문이 불여일견을 실천하는 것이죠. 그런데, 이게 되려면 당장 눈앞에 해결해야 할 본업을 끝내고 본인의 시간을 더 써야 한다는 것입니다. 하지만, 이렇게 해서라도 회사에 도움이 되는 것이 하나 걸린다면 그것만으로 저는 만족하고 있습니다. 이런 식으로 기회를 스스로 만들어 내고, 내 경력에도 도움 될 수 있는 프로젝트를 실현하기 위해 추가시간을 투자하는 게 데이터 사이언티스트와 데이터 분석가의 현주소인 것 같기도 합니다. 이처럼 회사에 도움이 되는 것과 내가 잘할 수 있는 것의 교집합을 찾아나가는 시간이 필요합니다.

공부는 계속되어야 한다.

'일이 바쁜데 공부는 언제 하지?'라는 생각이 들 때를 가장 경계해야 하는데, 제가 지금 그렇습니다. 정말 물리적으로 업무량이 넘쳐서 일이 바쁠 때도 있긴 하지만, 이런 순간은 일 년 중 그렇게 많은 날이 아니었습니다. 좀 더 냉정하게 생각해 보니 내가 맡은 업무의 결과를 만들어 내는 데에 필요한 지식과 기술이 부족해서 발생하는 업무 적체인 경우가 많았습니다.

이런 생각이 차오를수록 흔히 슬럼프에 빠진다는 표현을 하는데, 슬럼프는 새로운 학습을 통해 극복하는 것이 가장 효과적인 것 같습니다. 지금 하는 일을 해결할 더 좋은 방법을 찾지 않고, 과거에 했던 방식을 그대로 차용하여 반복하다 보니 결과는 그저 그렇고 시간도 오래 걸리는 상황이기 때문인 것 같습니다.

특히 데이터 분석가로 시작하여 데이터 프로덕트 개발까지 하고 있는 저에겐 더더욱 공부가 필요한 상황이기도 합니다. 어쩌면 저와 같은 커리어를 걸으며 성장하는 일꾼들이 공통적으로 고민하는 지점이 아닐까 싶습니다. 다양한 프레임워크를 잘 활용하여 사용자에게 도움이 될만할 기능을 개발해야 하는데, 엔지니어링과 ML/DL 알고리즘까지 섭렵해야 하는 막중한 책임감을 느끼는 상황에 때로는 부담을 느끼는 그런 상황..😇 이유야 어찌되었든 Output이 좋으려면 Input도 좋아야한다는 것을 더 많이 느끼는 요즘입니다.

마무리⌛️

지난 4년간 많은 지식을 채우고, 많은 지혜를 배우며, 많은 프로젝트를 경험했습니다. 그럼에도 올해는 예전보다 걱정이 많았습니다. 직전 회사에서 짧은 기간을 다니고 했던 이직인 만큼 이번 선택을 후회하면 어쩌나 하는 걱정, 그리고 팀 빌딩의 첫 구성원으로서 느껴지는 부담감 등이 있었습니다. 하지만 결과적으로 만족스러운 나날의 연속이었습니다. 걱정만 하진 않았기 때문입니다.

제가 예전에 읽은 책에서 기억에 남는 구절이 있는데요, '걱정은 흔들의자 같다'입니다. 제자리에서 흔들리기만 할 뿐 앞으로는 나아가지 못하기 때문이라고 합니다. 걱정을 하지 말라는 말이죠. 근데 어디 그게 쉽나요.

돌이켜보면 걱정이 그렇게 나쁘지만은 않다고 생각합니다. 지금 하고 있는 걱정이 잠시 제자리에 있을진 모르겠지만, 분명 인생을 꽃길로 물들여줄 씨앗이 되어주기도 하는 것 같거든요. 여전히 주니어 연차에서 벗어나진 못했지만, 지금까지 품고 있던 걱정 씨앗이 앞으로는 프로의 향기를 내뱉는 꽃이 되도록 더 정진해야겠습니다.

5년 차에 접어들면서 지난 4년간의 농사가 과연 어땠는지 되돌아보기 위해 의식의 흐름대로 적어본 글이지만, 저와 비슷한 커리어를 걷는 분들에게 힘이 되는 글이 되길 희망해 봅니다!

아직 많이 부족하지만 실전을 통해 체득한 지식과 노하우를 기록해보려고 합니다. 잘못된 내용이 있다면 지적도 부탁드립니다! 🤚

이번 포스팅의 목적 ✍️

바야흐로 MLOps 시대입니다. MLOps 안에서도 AutoML 프레임워크가 상당히 추상화되었을 뿐만 아니라 예측 성능도 많이 좋아진 것 같습니다. 그래서 파라미터 튜닝이나 모델 성능 비교를 위한 코드 개발에 많은 시간을 들이지 않아도 되는 시점에 도래한 것 아닌가 생각해봅니다. 이제는 알고리즘의 이해는 기본이고 프레임워크도 잘 다룰 줄 알아야 하는 게 기본 역량이 되어가고 있습니다.

AI 생태계를 이끌어가는 데에는 고도화된 오픈 소스가 한몫하고 있지만 '작고 귀여운 수준일지언정 직접 비슷하게라도 만들어 보자'라는 마음이 아직 한편에 남아 있습니다. ML 알고리즘을 저수준(low level) 코드로 하나씩 직접 개발하는 것은 상당히 도전적인 일이지만, 라이브러리를 적절히 활용한다면 AutoML 비스름하게 만들어 볼 순 있습니다. (그저 객기에 불과할지도..🥲)

AutoML의 목표는 모델 최적화를 위한 반복적인 과정을 자동화하는 것인데, 이 파이프라인을 당장 자동화하긴 어렵다면 적어도 모델별 성능과 Feature Selection을 위한 정보를 간편하게 확인할 수 있는 장치가 필요했습니다. 그래서 이번 포스팅에서는 언젠간 써먹겠지라는 생각으로 만들어 두고 혼자 업무에만 사용하고 있던 기능을 간단히 정리해 봤습니다. 바로 분류 알고리즘 성능 비교 및 하이퍼파라미터 튜닝 모듈입니다.

이런 걸 만들었습니다

AutoML에 필요한 과정을 아래의 단계로 구분했고, 각각에 필요한 함수를 구현했습니다. 이 흐름을 자동화해준다면 AutoML 일 것이지만, 여기서는 직접 실행하면서 Best 모델을 찾아가는 방법을 시도해 봤습니다. (아니 밑장빼기도 아니고 Auto ML이라 했다가 Manual ML로 급선회..)

밑에서 다시 자세히 소개하겠지만 모듈에서 각 역할을 수행하는 기능을 나열해 보면 대략 아래와 같습니다.

• 교차 검증 (Cross Validation)
• 교차 검증 결과 시각화
• 하이퍼파라미터 튜닝 (Hyperparameter optimization)
• 하이퍼파라미터 튜닝 결과 시각화
• Feature Importance 시각화
• Permutation Importance 시각화
• Best Model 저장
• (Optional) Deicision Tree 시각화 등

예를 들면 알고리즘별 예측 성능과 수행 시간을 한눈에 비교해 볼 수 있습니다. 여기서 Y 축을 보면 비교하고 있는 알고리즘이 꽤 많은데, '굳이 자주 사용하지 않는 것까지 비교를 해야 하나?' 싶은 생각이 충분히 들 수 있습니다. 당연히 저렇게 많은 알고리즘을 매번 비교하는 건 무리지만, 이 글에서는 단순히 예시를 위해 많이 비교해 봤습니다. (좀 풍부하니까 있어보여서..😂)

마치 거창한 것을 만든 것처럼 얘기했지만 sklearn, xgboost, lightgbm를 활용하여 열심히 조립(?)해서 함수를 만든 것뿐입니다. 헤헤. 에.. 막상 쓰고 보니 제가 봐도 별로인 것 같기도 하고.. (글 망한듯..?)

모듈 소개

코드는 여기에서 확인하실 수 있습니다. 약간의 스압에 주의해주세요! 😋

교차 검증 실행 부분만 보자면 아주 간단합니다. 훈련에 사용할 데이터와 예측하려는 타깃(Y), 그리고 원하는 알고리즘을 지정하여 예측 결과를 비교하도록 설계했습니다. 다른 기능들도 마찬가지로 이런 형태입니다. 즉, 모델 튜닝에 필요한 옵션을 자유롭게 선택만 하도록 의도한 것이라 보면 되겠습니다.

from ml_classifiers import Classifiers

# 객체 생성 (Feature Scaler 지정 가능)
clf = Classifiers(feature_scaler="RobustScaler")

# 알고리즘별 교차검증 실행
clf.run_cross_validation(
    estimators=[
        "AdaBoostClassifier",
        "DecisionTreeClassifier",
        "ExtraTreesClassifier",
        "GaussianNB",
        "GaussianProcessClassifier",
        "GaussianProcessClassifier",
        "GradientBoostingClassifier",
        "KNeighborsClassifier",
        "LGBMClassifier",
        "LogisticRegression",
        "MLPClassifier",
        "QuadraticDiscriminantAnalysis",
        "RandomForestClassifier",
        "SVC",
        "XGBClassifier",
    ]
    X=X_train,
    y=y_train,
    target="target",  # 예측할 타겟 변수 컬럼
    kfold="KFold",  # train, test 분할 방법
    n_splits=5,  # K-Fold의 K 값
    scoring="accuracy",  # 모델 성능 지표
)

# 알고리즘별 성능
clf.show_cross_validation_result()

사용 예시 💻

여기서 부터는 주피터 노트북에서도 확인할 수 있지만, 좀 더 가벼운 버전으로 적어봤습니다.

Dataset

Scikit-learn에서 제공하는 기본 데이터셋으로, 이진 분류에 자주 사용하는 샘플입니다.

from sklearn.datasets import load_breast_cancer

breast_cancer = load_breast_cancer()
df_breast_cancer = pd.DataFrame(
    breast_cancer["data"], columns=breast_cancer["feature_names"]
)
df_breast_cancer["label"] = breast_cancer["target"]
df_breast_cancer

X_train = df_breast_cancer.iloc[:, :-1]
y_train = df_breast_cancer.iloc[:, -1]
feature_names = X_train.columns

Preparation

따로 개발한 모듈을 호출하여 객체를 먼저 생성합니다. 이때 Feature Scaler(StandardScaler, MinMaxScaler, RobustScaler, MaxAbsScaler, Normalizer)를 지정할 수 있습니다. 만약 이 외에 다른 표준화 방법을 적용하려면 데이터에 이미 적용이 되어 있음을 가정합니다.

from ml_classifier import Classifiers
clf = Classifiers(feature_scaler="RobustScaler")

Algorithms

모듈에서 사용할 수 있는 알고리즘입니다. 여기에 나온 알고리즘을 한꺼번에 비교하는 건 시간 소요도 크므로 그다지 좋은 방법은 아닙니다. 특히 GradientBoostingClassifier은 시간이 굉장히 오래 걸리는 편이라 실제 사례에서도 많이 사용하는 편은 아니기도 합니다.

원래 의도는 3~4개 알고리즘의 예측 성능과 수행 시간을 수시로 비교하여 최적의 모델을 선택하는 것입니다. 이 중에서 주로 사용하는 알고리즘은 RF, XGB, LGBM 정도입니다. 캐글(Kaggle)에서도 이 세 가지는 자주 사용하는 것을 확인할 수 있고, Decision Tree는 일종의 base model로 여겨지는 경향도 있어서 같이 사용하곤 합니다. 만약 이 외에 추가할 알고리즘이 있다면 모듈을 수정하여 사용할 수 있습니다.

list(clf._get_classifier_models())

['AdaBoostClassifier',
 'DecisionTreeClassifier',
 'ExtraTreesClassifier',
 'GaussianNB',
 'GaussianProcessClassifier',
 'GradientBoostingClassifier',
 'KNeighborsClassifier',
 'LGBMClassifier',
 'LogisticRegression',
 'MLPClassifier',
 'QuadraticDiscriminantAnalysis',
 'RandomForestClassifier',
 'SVC',
 'XGBClassifier']

Cross Validation

이제 위에서 확인한 알고리즘을 통해 교차 검증을 수행할 수 있습니다. 교차 검증은 반드시 필요한 부분이 아닐 순 있지만, 분류 모델을 만들기에 앞서 몇 개의 알고리즘을 선택하여 대략적인 성능을 비교해 보고 싶을 수 있습니다.

만약 하이퍼파라미터 탐색을 여러 개의 알고리즘으로 시도한다면 많은 시간이 필요한데, 저는 먼저 CV를 통해 성능을 비교하여 선택의 폭을 줄여나갈 수 있는 결과가 있으면 좋을 것 같았습니다. 그래서 알고리즘별 파라미터를 직접 지정하거나 혹은 디폴트 값을 그대로 사용하여 대략적인 성능을 보고, 1~2개 알고리즘을 선택하여 하이퍼파라미터 탐색을 시도하기 위해 만든 기능입니다.

# 알고리즘별 하이퍼파라미터 지정 (만약 사용하지 않는다면 디폴트 파라미터로 예측을 수행합니다.)
estimator_fit_params = {
    "DecisionTreeClassifier": {
        "criterion": "entropy",
        "splitter": "random",
        "max_depth": 100,
        "min_samples_split": 3,
        "min_samples_leaf": 3,
        "max_features": "sqrt",
    },
    "RandomForestClassifier": {
        "n_estimators": 100,
        "criterion": "entropy",
        "max_depth": 10,
        "min_samples_split": 3,
        "min_samples_leaf": 3,
        "max_features": "sqrt",
        "bootstrap": True,
    },
}

# 교차 검증 실행
estimators = [
    "DecisionTreeClassifier",
    "RandomForestClassifier",
    "XGBClassifier",
    "LGBMClassifier",
]
scoring = ["accuracy", "recall", "precision", "f1", "roc_auc"]

# 교차 검증 실행
clf.run_cross_validation(
    X=X_train,
    y=y_train,
    estimators=estimators,  # 예측 알고리즘
    estimator_params=estimator_params,  # 알고리즘별 파라미터
    scoring=scoring,  # 모델 성능 지표
    kfold="RepeatedStratifiedKFold",  # 훈련, 검증 데이터 분할 방법
    n_splits=5,  # K-Fold의 K
    n_repeats=5,  # 교차 검증 반복 횟수
)

# 교차 검증 결과 시각화
clf.show_cross_validation_result()

(F1-score, Precision, Recall, ROC-AUC 결과는 생략)

필요하다면 교차 검증 결과도 따로 확인할 수 있습니다.

clf.df_cv_result

Hyperparameters Optimization

교차 검증을 통해 알고리즘 후보군을 좁혔다면 해당 알고리즘별 하이퍼파라미터 튜닝을 통해 성능을 확인하는 단계로 넘어갑니다. 하이퍼파리미터 튜닝은 여러 방법이 있지만, 여기서는 Scikit-learn에서 제공하는 하이퍼파라미터 탐색 방법(GridSearchCV, RandomizedSearchCV, HalvingGridSearchCV, HalvingRandomSearchCV)을 선택할 수 있도록 개발했습니다.

# 알고리즘별 하이퍼파라미터 범위 지정
import numpy as np

hyperparams_space = {
    "DecisionTreeClassifier": {
        "criterion": ["gini", "entropy"],
        "splitter": ["best", "random"],
        "max_depth": np.arange(5, 105, 5).astype(int),
        "min_samples_split": np.arange(2, 11).astype(int),
        "min_samples_leaf": np.arange(2, 11).astype(int),
        "max_features": ["sqrt", "log2"],
    },
    "RandomForestClassifier": {
        "n_estimators": np.linspace(100, 1000, 10).astype(int),
        "criterion": ["gini", "entropy"],
        "max_depth": np.arange(5, 105, 5).astype(int),
        "min_samples_split": np.arange(2, 11).astype(int),
        "min_samples_leaf": np.arange(2, 11).astype(int),
        "max_features": ["sqrt", "log2"],
        "bootstrap": [True, False],
    },
    "XGBClassifier": {
        "n_estimators": np.linspace(100, 1000, 10).astype(int),
        "learning_rate": np.arange(0.001, 0.1, 0.01),
        "max_depth": np.arange(5, 105, 5).astype(int),
        "colsample_bytree": sp_uniform(loc=0.4, scale=0.6),
        "colsample_bytree": [0.5],
        "gamma": [i / 10.0 for i in range(3)],
        "fit_params": {"verbose": False},
        "eval_metric": ["logloss"],
        "early_stopping_rounds": [100],
    },
    "LGBMClassifier": {
        "n_estimators": np.linspace(100, 1000, 10).astype(int),
        "learning_rate": np.arange(0.001, 0.1, 0.01),
        "max_depth": np.arange(5, 105, 5).astype(int),
        "colsample_bytree": [0.5],
        "verbose": [-1],
        "fit_params": {
            "eval_metric": ["binary_logloss"],
            "callbacks": [early_stopping(100)],
        },
    },
}

# 하이퍼파라미터 탐색 실행
clf.search_hyperparameter(
    X=X_train,
    y=y_train,
    search_method="random",  # 하이퍼파라미터 탐색 방법 (grid, random, grid_halving, random_halving)
    hyperparams_space=hyperparams_space,  # 알고리즘별 하이퍼파라미터 범위
    scoring="roc_auc",  # 모델 성능 지표
    kfold="RepeatedStratifiedKFold",  # 훈련, 검증 데이터 분할 방법
    n_splits=5,  # K-Fold의 K
    n_repeats=5,  # 교차 검증 반복 횟수
    #     n_iter=10,  # 파라미터 조합 수
    #     factor=3,  # 파라미터 선택 수 (`search_method`가 `*_havling`일 때만 적용)
)

# 탐색 결과 시각화
clf.show_hyperparameter_search_result()

하이퍼파라미터 탐색을 끝내면 다음과 같은 로직을 통해 최적의 모델을 반환하도록 구현했는데, 아래의 코드로 모델의 성능과 파라미터를 조회해 볼 수 있습니다.

• 검증 데이터의 분류 성능(mean_test_score)이 가장 높은 모델
• 만약 동점의 모델이 있을 경우 검증 데이터의 분류 성능 표준편차(std_test_score)가 낮은 모델
• 두 번째에서도 동점이라면 모델 적합 시간(mean_fit_time)이 가장 짧은 모델

clf.get_best_model_info()

{'mean_test_score': 0.9906190645773979,
 'std_test_score': 0.01117774544765023,
 'mean_fit_time': 0.2448660612106323,
 'estimator_name': 'LGBMClassifier',
 'params': {'colsample_bytree': 0.5166582231544349,
  'learning_rate': 0.020999999999999998,
  'max_depth': 55,
  'n_estimators': 800,
  'objective': 'binary',
  'verbose': -1}}

위의 로직으로 반환된 최종 모델을 출력해 보면 다음과 같이 나타납니다. 만약 여기에서 끝낸다면 이 모델을 저장해놓고 인퍼런스로 사용할 수 있게 됩니다.

clf.get_best_classifier()

Feature Importance

자, 튜닝까지 끝났으면 Feature Importances를 확인해 봐야죠. 불필요 변수가 있을지, 있다면 제거를 할지 판단해야 합니다. 보통 이쯤되면 여러 개의 알고리즘을 계속 비교하진 않습니다. 거의 한 개의 모델로 추려지는데, 여기에서는 예시로 두 개의 알고리즘별로 Feature Importance를 각각 출력해봤습니다.

# 만약 `estimators`를 지정하지 않는다면 하이퍼파라미터 탐색을 실행한 모든 알고리즘의 Feature Importance가 출력됩니다.
clf.show_feature_importances(
    estimators=["XGBClassifier", "LGBMClassifier"],  # 알고리즘 선택
    n_features=20,  # 중요도가 높은 순으로 Feature 개수
)

Permutation Importance

마지막으로 Permutation Importance입니다. 훈련된 모델에서 특정 Feature를 사용하지 않을 때 이것이 성능 하락에 미치는 영향을 파악하는 방법으로, 모델 입장에서의 개별 Feature의 의존도를 확인하는 것입니다. Feature Importance까지만 보고 이것은 안 보는 경우가 종종 있는데, 오히려 이게 도움이 되는 정보라고 생각합니다. 추출 과정은 아래와 같습니다.

1. 기존 검증 데이터셋에서 하나의 변수(Feature)를 선택하여 값의 순서를 무작위로 섞은 후 새로운 검증 데이터셋을 생성
2. 새로운 검증 데이터셋으로 성능(score)을 측정
3. 기존 검증 데이터셋에 의한 성능 대비 새로운 검증 데이터셋에 의한 성능 비교
   • 성능 감소 ➡ 해당 변수가 중요하다는 의미
   • 성능 큰 변화 없음 ➡ 해당 변수는 그닥 중요하진 않다는 의미

따라서 성능 감소가 없는 변수 발견 시, 해당 변수를 제와한다면 두 가지 효과를 기대할 수 있습니다.

• 모델 성능 개선
• 변수 제거에 따른 추가 리소스 확보 (특히 변수가 상당히 많을 때 유용)

# 모델과 score를 따로 지정할 수 있습니다.
clf.show_permutation_importances(
    estimators=["LGBMClassifier"],
    X=X_train,
    y=y_train,
    target="label",
    scoring="roc_auc", # or ["f1", "roc_auc"]
)

(Appendix) Decision Tree 시각화

중요한 기능은 아니지만 Decision Tree를 튜닝한 경우, 트리 구조를 볼 수 있는 것도 만들어 두었습니다. 의사결정나무는 base model 성능으로 보는 경향도 있기 때문에 일반적으로 많이 사용하는데요, 분류 결과를 시각화할 수 있어서 설명하기 용이한 형태이기에 현업에서도 종종 사용하는 편입니다.

clf.show_decision_tree(
    feature_names=df_breast_cancer.columns[:-1],  # DT 모델에 적용한 Feature 
    class_names=breast_cancer.target_names  # Target 변수의 클래스(Label) Unique 값
)

어떻게 활용하면 좋을까 🤔

모델 튜닝, 변수 선택의 반복을 조금이라도 편하게 해보자 이겁니다!

분류 모델을 처음 시도한다면 3대장(LightGBM, XGBoost, RandomForest)과 그리고 Deicision Tree의 성능을 먼저 확인해 보면 좋은 것 같습니다. 그러면 어떤 알고리즘을 최종으로 사용할만할지, 그리고 어떤 파라미터를 튜닝하면 좋을지 대략 '감'이 잡히는 것 같습니다. 그래서 예시에서도 네 가지 알고리즘을 사용하여 비교했던 것이기도 합니다.

그 이후 하이퍼파라미터 튜닝과 변수 선택을 몇 번 반복하다 보면 그 상황에 맞는 최적의 모형을 도출할 수 있을 것이라 기대합니다. 개인적인 의견으로는 이 글에서 제시한 단계를 모두 거쳤다면 어느 정도 괜찮은 모델이 나오는 편입니다. 저 역시 이렇게 활용하고 있고요! 단지, 이것을 자동화하지 못해 슬펐을 뿐(...) 이 과정 덕택에 분류 문제 해결을 위한 노하우도 약간은 생긴 것 같습니다.

물론 여기에서 제공하는 알고리즘 보다 더 나은 방법은 항상 존재합니다. 그래서 저는 분류 모델을 만들 때 이 글에서 제시한 방법으로 먼저 시도해 보고, 이후에 딥러닝을 사용하여 업그레이드 하는 편입니다. 근데 솔직히 이제는 다 딥러닝으로 시작해도 괜찮은 것 같습니다. 😅

(아 그래서 Kaggle 몇 점이냐고요?.. 혼자 있고 싶네요..)

추가로 개발할 기능

아직 개발한 것은 아니지만 추가로 있으면 좋을 법한 기능을 나열해 봤습니다.

• 하이퍼파라미터 탐색 방법에 hyperopt, optuna를 사용할 수 있도록 추가
• Feature Importance, Permutation Importance 결과로부터 Feature Selection 자동화 함수 추가

마무리 ⌛️

예측 모형을 만들 때 때마다 느끼는 것이지만 하이퍼파라미터 탐색의 범위는 데이터 분석가나 데이터 과학자가 직접 정의하는 편이 많습니다. 저는 이것을 염두에 두고 모델링 할 때 약간의 편의성을 도모하고자 만들어 봤습니다. 쓰고 보니 AutoML까진 전혀 아니었지만(..) 모델 훈련, 변수 선택의 반복적인 과정을 쉽게 할 수 있는 기능이 있으면 좋겠다는 가벼운 호기심에서 출발한 게 이렇게 되었네요. (이런 호기심은 이제 그만 가지겠습니다..)

요즘 MLOps에 계속 관심을 갖고 있다 보니 이런 것(?)을 만드는 게 즐겁습니다. 그래봐야 머신러닝 라이브러리를 오픈 소스로 만들어 주신 훌륭한 분들 덕택에 저도 레고 조립하듯 가져다 쓴 것이지만요.😛 그래도 토이 프로젝트 처럼 나름 심도 있게 기능을 개발하고자 노력했기 때문에 정리를 하고 보니 약간의 뿌듯함이 있네요. 물론 여전히 리팩토링할 게 많고, 추가해야 할 것도 많으니.. 저는 부족한 부분을 더 채워서 AutoML의 A까지는 따라한다고 볼법한 기능으로 무장하여 다시 돌아오겠읍니다!

업종과 분야를 막론하고 우리는 항상 더 좋은 판단을 하기 위해 실험을 진행합니다. 특히 비즈니스 분석, 데이터 분석을 한다면 피해 갈 수 없는 업무가 있죠. 바로 A/B 테스트입니다.

• 이 푸시를 보내면 구매 전환율이 높아질까?
• 앱 화면을 이렇게 바꾸면 체류 시간이 늘어날까?
• 연관된 상품을 보여주면 더 많이 구매할까?

사소한 궁금증과 호기심에서 시작된 것이 실험 설계로 이어지고, 이것을 검증하기 위해 생각 보다 많은 시간을 할애합니다. 비교군과 실험군을 나누고, 일정 시간 동안 실험을 진행하여 각 집단에서 보인 관측 결과에 차이가 있는지를 검정하려면 최소 하루는 필요한 것 같습니다. Braze 같은 마케팅 솔루션을 이용한다고 해도 어쨌든 캠페인 기간을 최소 몇 시간은 설정하는 편이니 시간이 꽤 필요합니다.

그런데 실험을 통해 얻은 관측치만 보고 성급한 결론을 내리는 경향이 종종 있습니다. '어? 앱 화면을 이렇게 바꾸니까 하니까 체류 시간이 더 기네? 앞으로 이렇게 계속해야겠다.'라고 결론짓는 것이죠. 하지만 실험 결과에 의한 차이가 우연에 의한 것임을 놓칠 때가 있습니다. 즉, 똑같은 실험을 여러 번 했을 때도 과연 지금과 같은 결과일 것인지 확인해 볼 필요가 있기 때문입니다.

따라서 우리의 실험 결과를 보장해 줄 수 있는, 그니까 정말 우연이 아니라 찐이라는 것을 확인하기 위해 통계적 가설검정을 거쳐야 합니다. 가설검정 방법 중 가장 많이 사용하는 것이 t-검정입니다. 다만, 이 방법을 사용하기 위해선 몇 가지 전제가 있습니다. 그중 가장 중요한 것이 각 집단의 분포가 정규분포를 띄어야 한다는 것인데요, 현업에서 문제를 정의하고 실험을 설계하다 보면 이 방법을 이용하기 어려운 경우가 많습니다. 대부분 한쪽으로 치우친 분포의 데이터가 상당히 많기 때문입니다.

이럴 때 사용할 수 있는 방법 중 하나가 바로 부트스트랩 순열 검정입니다. 이 방법은 세 가지 이유 때문에 제가 좋아하는 검정 방법인데요,

• 실험 집단간 표본이 적고 정규성을 띄지 않는 데이터에도 검정을 시도할 수 있으며
• 데이터가 정규분포일 경우 t-검정의 검정력과 상당히 유사한 수준을 보이면서도
• 검정 통계량의 분포를 시각화함으로써 현업 부서를 비롯한 조직 내 동료에게 실험 결과의 검정력을 설명할 때 직관적으로 이해를 도울 수 있는 방법

특히 마지막 이유가 이 방법을 자주 사용하게 된 계기가 되었습니다. 그래서 이번 글에서는 t-검정보다는 휴리스틱 방법인 순열 검정을 소개하고 활용 방안을 공유해 보려고 합니다.

부트스트랩 순열 검정

검정 절차는 간단합니다.

1. 가설검정에 사용한 각 집단의 샘플 데이터들을 하나로 합쳐서 잘 섞는다.
2. 통제 집단(control group)과 동일한 크기의 표본을 복원 추출하고 통계량(A)을 계산한다.
3. 실험 집단(test group)과 동일한 크기의 표본을 복원 추출하고 통계량(B)을 계산한다.
4. A와 B의 차이를 기록한다.
5. 1~4번을 N번 반복하여 검정 통계량 분포를 생성하고, 두 집단의 기존 관측값의 차이가 해당 분포 내 어느 위치에 있는지 확인한다.

순열 검정의 원리로부터 두 가지 특징을 잘 이해하면 좋습니다.

• 가설검정에 사용한 각 집단을 하나로 합친다는 점에 주목해야 하는데요, 실험군과 대조군에 차이를 두지 않겠다는 의도를 내포합니다. '각 집단 실험 결과에 차이가 없는 동일한 그룹일 것이다'라는 일종의 귀무가설을 구체화한 것이죠. 즉, 원래 하나의 동일한 집단이 나타낼만한 분포로부터 실험을 통한 관측 값이 얼마나 극단에 있을지 확인해 보려는 것입니다. 이 관측 값이 극단에 있을수록 우연이 아님을 주장할 좋은 근거가 되기 때문입니다.
• 순열이 영어로 permute입니다. '순서를 바꾼다'라는 의미로, 집합 내 순서를 섞음으로써 기존에 관측된 순서로 인해 발생할 수 있는 편향 자체를 없애는 것입니다.

부트스트랩 (Bootstrap)

조금 더 파고들겠습니다. '복원 추출을 N 번 반복한다'라는 것을 이해할 필요가 있는데요, 이것은 부트스트랩(Bootstrap) 기법입니다. 현재 갖고 있는 데이터에서 발생 가능한 변동성을 알아보기 위해 반복적으로 표본을 추출하는 방법에서 출발하는 것이죠.

통계학에서는 일반적으로 우리가 보유하고 있는 데이터로 관측한 통계량은 '표본 통계량'으로 간주합니다. 실제 세계에서의 모집단은 알 수 없는 미지의 집단으로 보는 전제가 깔려있는 것입니다. 그래서 이 미지의 모집단 통계량(=모수)을 추정하기 위해 현재 보유한 데이터로 표본 통계량을 무수히 많이 생성해보는 방법을 시도하게 되는데, 이 표본 통계량 분포를 통해 모수를 추정할 수 있게 됩니다. 추론통계의 기반인 모집단 분포에 상관없이 표본평균의 분포가 정규분포로 수렴한다는 중심극한정리의 개념을 설명하는 것과 동일한 맥락입니다.

이때 표본 통계량을 생성하기 위해 무작위 복원 추출을 시도한다는 점이 중요한데, '각각의 데이터가 관측될 확률은 모두 독립적이고 동일하다'라는 가정을 만족시키기 위한 방법인 셈입니다. 따라서 현재 보유한 표본으로 많은 반복 추출을 통해 또 다른 표본 통계량을 생성한다는 것은, 현실 세계에서 한 번쯤은 발견할 만한 또 다른 표본을 만들겠다는 다소 휴리스틱 한 의도가 담겨있다고 볼 수 있습니다. (이런 점에서 표본은 다다익선입니다. 많을수록 모집단을 잘 추정할 수 있으니까요.)

이 과정을 통해 생성된 표본의 통계량 중 가장 많은 빈도로 관측된 값을 모수로 추정하는 것이고, 이 값의 변동성(or 불확실성)을 우리가 많이 들어본 신뢰구간(confidence level)으로 제공하는 것입니다. 이 신뢰구간은 표본 통계량 분포의 표준편차와 관련이 있습니다. 우리가 얻은 표본 통계량의 분포에서 이 표준편차가 작으면 작을수록 모수가 확실한 값일 테고, 클수록 불확실한 값이 되는 것이죠. 신뢰구간은 보통 분포 내의 95%로 제한하는 게 일반적인데, 표준편차의 약 2배 정도되는 구간으로 봐도 무방합니다.

마찬가지로 부트스트랩 순열 검정에서도 많은 검정 통계량을 생성하게 되는데, 실험을 통해 얻은 차이가 이 검정 통계량 분포 내의 표준편차 보다 훨씬 멀리 있다면 '우연히 관측될만한 값은 아니었구나!'라고 판단할 수 있는 것입니다. 우리가 가장 얻고 싶은 결론이기도 하지요. 😎

다시 말해 순열 검정 통계량 분포의 약 95%를 벗어난 곳에 있는 값이라면, 실험의 차이가 우연이었을 확률이 5%보다도 작다고 말할 수 있는 것입니다. 그만큼 신뢰할 만한 차이라는 것이죠. 그리고 이것이 바로 우리가 귀 따갑게 듣는 'p-value가 0.05 이하다'라는 것과 동일한 말로 해석할 수 있겠습니다.

예시 만들기

설명이 조금 길었지만 예시를 통해 원리를 이해하면 더욱 좋을 것 같습니다.

• 앱 내 상품 카테고리 UI를 기존 A에서 B로 바꾸어 노출하려고 합니다. 이때 B로 바꾸었을 때 체류시간이 증가하는지 확인해 보고 싶습니다.
• 고객 아이디를 무작위로 10만 개를 추출하여 동일한 표본 크기의 A, B 그룹을 생성하였고, 48시간 동안 각 고객의 세션 평균 체류시간을 측정해 보았습니다.
• 48시간이 지난 직후 그룹별 평균 체류시간을 계산해 보니 B가 A 보다 약 2분 높게 나왔습니다.
• 이번 실험이 결과가 우연인지 아닌지 확인해 보고, 우연이 아니라면 새롭게 디자인한 UI로 최종 배포를 진행하려고 합니다.

각 그룹의 체류시간 분포를 확인해 보니 아래와 같습니다. 위에서 설명한 순열 검정 방법을 그대로 적용해 보겠습니다.

# 샘플 데이터 생성 예시

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns


np.random.seed(2022)
n = 50000

a = abs(np.random.normal(loc=5, scale=4, size=n))
b = abs(np.random.normal(loc=7, scale=5, size=n))

df_sample_a = pd.DataFrame({"Group": ["A"] * n, "Duration time": a})
df_sample_b = pd.DataFrame({"Group": ["B"] * n, "Duration time": b})
df_samples = pd.concat([df_sample_a, df_sample_b], ignore_index=True)

duration_time_mean_A = (
    df_samples.loc[df_samples["Group"] == "A", "Duration time"].mean().round(2)
)
duration_time_mean_B = (
    df_samples.loc[df_samples["Group"] == "B", "Duration time"].mean().round(2)
)

sns.histplot(data=df_samples, x="Duration time", binwidth=0.5, hue="Group", kde=True)
plt.title(label="Time duration distribution")
plt.xlabel(xlabel="Time (minute)")
plt.ylabel(ylabel="Frequency")

plt.axvline(x=duration_time_mean_A, ymin=0, ymax=1, color="b", ls="--")
plt.axvline(x=duration_time_mean_B, ymin=0, ymax=1, color="r", ls="--")

plt.text(
    x=duration_time_mean_A,
    y=2500,
    s=str(duration_time_mean_A),
    bbox=dict(facecolor="b", alpha=0.2),
)
plt.text(
    x=duration_time_mean_B,
    y=2500,
    s=str(duration_time_mean_B),
    bbox=dict(facecolor="r", alpha=0.2),
)

plt.gca().yaxis.set_major_formatter(mpl.ticker.StrMethodFormatter("{x:,.0f}"))

순열 검정 통계량 분포 생성

부트스트랩 순열 검정의 원리를 코드로 간단히 구현하여 분포를 확인한 결과입니다. 표본 추출 횟수는 10,000번을 시행하였으며, 표준편차에 의한 변동성도 1 sigma, 2 sigma로 확인할 수 있습니다. 분포 자체를 생성하는 건 어렵지 않습니다. 이 검정 통계량 분포가 의미하는 바를 이해하는 게 중요합니다!

실험 집단과 통제 집단의 차이가 한 번쯤은 나타날만한 값(검정 통계량) 10,000개를 만들어 확인한 분포

## 순열 검정 통계량 분포 생성 예시
random_state = 2022

# 데이터 순서 섞기(shuffling)
df_samples = df_samples.sample(frac=1, ignore_index=True, random_state=random_state)

# A, B 그룹의 표본 개수 생성
n_sample_A, n_sample_B = df_samples["Group"].value_counts()

# 표본 통계량을 저장할 리스트
sample_mean_list = list()

# 표본 추출 시행 횟수
n = 10000

# 순열 검정 통계량 생성 
for _ in range(n):
    sample_A = df_samples.sample(n=n_sample_A, replace=True)["Duration time"]
    sample_B = df_samples.sample(n=n_sample_B, replace=True)["Duration time"]

    sample_mean_A = np.mean(sample_A)
    sample_mean_B = np.mean(sample_B)

    diff = sample_mean_B - sample_mean_A
    sample_mean_list.append(diff)

# 분포 확인
plt.figure(figsize=(12, 6))
ax = plt.hist(sample_mean_list, bins=20, alpha=0.8)

plt.title(label=f"distribution of sample mean differences (Number of trials: {n:,})")
plt.xlabel(xlabel="sample mean difference")
plt.ylabel(ylabel="frequency")

sigma_1_left = np.mean(sample_mean_list) - np.std(sample_mean_list)
sigma_1_right = np.mean(sample_mean_list) + np.std(sample_mean_list)

sigma_2_left = np.mean(sample_mean_list) - (np.std(sample_mean_list) * 2)
sigma_2_right = np.mean(sample_mean_list) + (np.std(sample_mean_list) * 2)

plt.axvline(x=sigma_1_left, ymin=0, ymax=1, color="g", ls="--")
plt.axvline(x=sigma_1_right, ymin=0, ymax=1, color="g", ls="--")

plt.axvline(x=sigma_2_left, ymin=0, ymax=1, color="r", ls="--")
plt.axvline(x=sigma_2_right, ymin=0, ymax=1, color="r", ls="--");

시행 횟수에 따른 검정 통계량 분포

시행 횟수가 많아질수록 표본 통계량의 분포가 점점 정규분포에 가까워질 뿐만 아니라, 표준편차에 의한 변동성도 더 안정적으로 나타나는 것을 확인할 수 있습니다. 어떤 검정 방법이든 마찬가지겠지만 표본의 수가 많고 반복 횟수가 많을수록 불확실성을 줄일 수 있다는 것은 자명합니다. 요즘 시대에 회사에서 다루는 표본의 개수 아무리 많다고 한들, 컴퓨팅 리소스가 부족할 정도로(..) 많지는 않을 것이기 때문에, 이 방법을 통해 검증하는 것에는 큰 문제가 없을 것이라 생각합니다. 😅

검정 결과

실험 결과로 나타난 차이가 우연히 발생하였을 확률이 2.5% 보다도 낮다!

예시로 만든 실험군과 대조군의 차이가 위에서 생성한 검정 통계량의 분포로부터 얼마나 떨어져 있는지 확인해 보니 상당히 멀리 있습니다. 이것이 바로 우연히 발생하지 않을만한, 그니까 의미 있는 차이라고 볼 수 있는 근거가 된다는 것입니다. 두 집단에 차이가 있다고 판단할 수 있는 지표로 보는 p-value가 0.05보다도 낮다고 보는 것과 동일한 맥락인 것이죠.

과연 얼마나 아슬아슬하게 우연이 아님을 피할 수 있었을까 싶었지만, 생각 보다 시시하게 너무나 우연이 아님을 확인할 수가 있었습니다. 이것은 예시로 만든 데이터이니 만큼 참고만 해주셨으면 좋겠습니다. 😋

활용 방안

• 순열 검정 통계량 분포 생성과 시각화까지 해줄 수 있는 모듈을 만들어 보기
• 가설 검정을 처음 접하는 동료에게 설명해야 하는 상황이라면 분포를 통해 설명해보기

순열 검정의 절차를 보시면 아시겠지만 특별한 가정을 두지 않습니다. 데이터 분포가 어떻든, 표본 개수가 어떻든 일단 표본 통계량을 많이 만들어 내는 것이죠. '아니 그러면 다른 검정 방법 다 필요 없고 얘만 쓰면 되겠네??'라고 생각할 수 있으나, 당연히 꼭 그렇진 않습니다. 😅

이와 관련한 질문을 저도 찾아봤는데요, 여기서 나온 답변을 토대로 얘기해 보자면 결국 표본이 많을수록 순열 검정의 결과가 다른 방법과 비교해도 검정력에서 문제가 없다는 의견인 것 같습니다. 이런 점에서 현시대에 데이터 표본이 부족할 일은 없기 때문에 충분히 믿고 사용할만한 방법이라고 생각합니다. 다만, 분포에 따라서 비교할 통계량을 무엇으로 볼지는 다를 수 있습니다. 예를 들어 왜도(skewness)가 심한 분포는 평균보다는 중위수로 그룹 간 그 차이를 비교하는 게 적절할 것입니다.

물론 t-검정과 같은 수식에 기반한 방법 사용하는 게 더 편할 수도 있습니다. 코드 몇 줄이면 되거든요. 혹은 정규 분포를 따르지 않는 비모수 집단의 비교에는 Wilcoxon 검정, Mann-Whitney 검정 등의 더 잘 알려진 방법도 있습니다. 이처럼 표본 개수, 정규성, 그룹 수 등 다양한 조건을 따지긴 해야하지만 통계적 검정은 이런 원칙적인 절차를 반드시 거쳐야 하는 것은 맞습니다. 어우 사진만 봐도 머리가 아프네요. 😇

다만, 데이터와 통계에 대해 잘 아는 사람들만 모여 있는 이상적인 조직이라면 그렇게만 해도 된다고 생각합니다. 근데 이런 오르비스 같은 조직이 얼마나 될까요..😂 우리는 늘 데이터와 통계를 통해 나온 수치를 다른 동료에게 이해시켜야 하는 상황을 흔하게 마주합니다. 과거에도 그랬고 현재도 그렇고 앞으로도 그럴 겁니다. 그럴 때마다 '이거 검정 결과가 차이가 없는 걸로 나왔어요'라고 끝내는 것과 '이게 왜 유의미한 차이가 아니냐면요 이 분포를 보시면..🔥'와 같은 설명을 통해 데이터 리터러시를 높이려는 시도를 하는 것과는 분명 다를 것입니다.

이런 점에서 저는 이 방법이 모두에게 설명하기 쉬우면서도, 해석하기 쉬운, 그리고 수식에 기반한 통계학이 빠지기 쉬운 형식주의로부터 벗어날 수 있는 접근법이라 좋습니다. 코드도 간단하기 때문에 A/B 테스트 데이터셋을 바꿔가면서 검정 결과를 확인할 수 있는 모듈을 만들어 놓는다면 더 좋을 것입니다. 👏👏👏

마무리 ⌛️

(흐아 드디어 다 썼다!! 💪)

요새는 공룡 기업뿐만 아니라 많은 회사에서 A/B 테스트는 기본으로 하고 강화학습 기반의 MAB(multi armed bandit) 시스템이 대세인 추세입니다. 그래서 MAB를 설명하는 글을 적어보고 싶었으나.. 쪼렙인 저는 기초를 다시 다지자는 마음으로 해당 내용을 정리해봤습니다. 대부분 기초 통계와 관련된 내용이 주를 이루었음에도 직접 써보니 쉽지 않네요. A/B 테스트에 포커스를 두고 설명한 것이긴 하지만, 다양한 가설 검정에도 사용할 수 있으리라 기대합니다.

진짜 마무리로 명언 하나 스윽 던지고 끝내봅니다..ㅋㅋㅋ

인생은 하나의 실험이다. 실험이 많아질수록 당신은 더 좋은 사람이 된다. (by 랄프 왈도 에머슨)

아직 많이 부족하지만 실전을 통해 체득한 지식과 노하우를 기록해보려고 합니다. 잘못된 내용이 있다면 지적도 부탁드립니다! 🤚

지금 다니는 회사에서는 AWS를 사용하고 있습니다. 전 회사에서는 BigQuery(빛☀️쿼리..), Composer(Airflow) 조합으로 GCP 기반 데이터 파이프라인을 구축하고 운영했었는데, 이번 기회에 AWS를 통해 ML Pipeline을 만들면서 겪은 시행착오(?)를 간단하게 기록해보려고 합니다.

MWAA

Amazon Managed Workflow of Apache Airflow (MWAA)는 관리형 오케스트레이션 서비스로, 사실상 Apache Airflow를 AWS에서 제공하는 것이라 볼 수 있습니다. 요즘엔 Airflow를 모르는 분이 없을 정도로 많이 사용하는 스케줄러인데요, Data Lakehouse을 구축하기 위한 DataOps과 머신러닝 파이프라인을 운영하기 위한 MLOps 영역에서 많이 사용하는 것 같습니다. 원래 작년 초부터 사용하고 싶었으나, Asia/Seoul 리전에는 지원이 안되고 있었는데 2021년 9월 경에 출시되어 드디어 사용하게 되었습니다.

SageMaker

SageMaker는 머신러닝에 특화된 완전 관리형 인프라로, 모델을 직접 구현하지 않아도 각종 컴포넌트를 통해 훈련, 검증, 배포를 손쉽게 할 수 있는 기능을 제공합니다. 특히 ML Engineer가 해야 할 업무를 SageMaker에서 제공하는 컴포넌트를 활용하여 구현할 수 있도록하여 거의 SaaS(Software-as-a-Service)로 여겨지는 추세입니다.

이처럼 ML에 필요한 각 단계를 컴포넌트화하여 추상화된 기능으로 제공합니다. 다만 이번 글에서는 해당 기능을 사용하기 보다는, 직접 개발한 ML 모듈을 SageMaker의 노트북 인스턴스를 통해 일배치로 실행하기 위한 파이프라인 설계 과정 위주로 작성했습니다.

MWAA + SageMaker 파이프라인

저는 아래의 단계를 통해 배치 파이프라인을 구현했으며, 각 단계를 모두 Airflow의 Task로 실행하도록 개발했습니다.

• (Task 1) 노트북 인스턴스 lifecycle configurations 생성
• (Task 2) 노트북 인스턴스 생성 및 ML 모듈 실행 (데이터 로드, 모델 학습, 예측 실행, 예측 값 DB 업데이트 등)
• (Task 3) 노트북 인스턴스 중지 및 삭제
• (Task 4) 노트북 인스턴스 lifecycle configurations 삭제

노트북 인스턴스 생성 & 삭제

그러면 각 Task를 통해 SageMaker 노트북 인스턴스의 생성과 삭제를 어떻게 실행하는지 살펴보겠습니다.

GCP와 마찬가지로 AWS는 각 서비스를 CLI로 명령할 수 있는 SDK를 제공하는데요, Python에서는 Boto3(공식 문서)라는 라이브러리로 이 기능을 동일하게 사용할 수 있습니다. 따라서 이것을 통해 SageMaker의 노트북 생성과 삭제를 직접 컨트롤할 수 있습니다. 즉, Boto3로부터 SageMaker 클라이언트 객체를 생성할 수 있고, 해당 객체에서 필요한 메서드를 호출하여 코드 레벨에서 명령할 수 있습니다.

import boto3
client = boto3.client('sagemaker')

이제 Task 순서 별로 필요한 함수를 만들어보겠습니다.

Task 1) lifecycle configurations 생성

노트북 인스턴스를 새로 생성하거나, 혹은 중지했던 것을 다시 시작할 때 해당 인스턴스 환경 설정 셋팅에 필요한 명령어를 사전에 정의해 놓을 수 있는 기능

SageMaker 노트북 인스턴스를 생성하여 들어가보면 사진과 같이 다양한 Conda 환경을 제공하는데요, 이 가상환경 별로 머신러닝에 필요한 라이브러리들이 사전에 설치되어 있는 형태입니다. 여기에는 범용적으로 사용하는 Scikit-learn, Tensorflow, Torch 등이 설치 되어있긴 하지만, 로컬에서 개발한 코드가 해당 라이브러리 버전과 일치하지 않을 수 있고 혹은 새롭게 설치가 필요한 라이브러리가 있을 수 있습니다.

그래서 로컬 개발 환경을 노트북 인스턴스에 동일하게 맞춰주기 위한 작업이 필요할 때가 있는데, 이것을 lifecycle configurations을 통해 자동화 할 수 있습니다. 예를 들어 새로운 Conda 가상환경을 생성하고, 해당 환경에 로컬에서 사용했던 requirements.txt를 그대로 설치할 수 있도록 정의할 수 있습니다.

#!/bin/bash
sudo -u ec2-user -i <<'EOF'
conda create -n new_conda_env_p38 python=3.8 -y --force # 새로운 가상환경 생성
source activate new_conda_env_p38

cd SageMaker # SageMaker 노트북 인스턴스에 default로 생성되어 있는 폴더 
mkdir new_folder # 새로운 폴더 생성
cd new_folder

# Github repo를 clone하여 필요한 라이브러리를 설치
username="my_github_account"
password="my_github_account_password"
GIT_URL="https://${username}:${password}@github.com/WORKSPACE/my-repository.git"
git clone ${GIT_URL}

pip install -r requirements.txt &

conda deactivate
EOF

그러면 이 작업을 DAG의 PythonOperator Task로 실행할 수 있도록 함수로 생성해야 합니다. 즉, 해당 shell을 읽어들여서 SageMaker의 lifecycle configurations를 생성할 수 있도록 하는 것입니다. 단, shell 스크립트 안에는 github 계정 등의 민감한 정보가 포함되어 있기 때문에 따로 파일로 관리하기 보다는 Airflow Variables을 통해 불러오도록 설계했습니다. 그래서 해당 함수에 content 파라미터를 만들었는데, Variables을 통해 읽어들인 shell을 string으로 전달받아서 그대로 lifecycle configurations를 생성하는 것입니다. (이것은 아래에서 다시 한번 설명합니다!)

import base64

def create_notebook_instance_lifecycle_config(NotebookInstanceLifecycleConfigName: str, content: str):
    content_bytes = content.encode("utf-8") 
    content_base64 = base64.b64encode(content_bytes)
    content_base64_str = content_base64.decode("utf-8")

    try:
        response = client.create_notebook_instance_lifecycle_config(
            NotebookInstanceLifecycleConfigName=NotebookInstanceLifecycleConfigName,
            OnStart=[
                {
                    "Content": content_base64_str
                },
            ]
        )
        print(" >>> NotebookInstanceLifecycleConfig is created. ")
        return response
    except Exception as err:
        return err

Task 2) 인스턴스 생성 및 ML 모듈 실행

앞서 설명한 lifecycle configurations를 토대로 노트북 인스턴스가 생성되게끔 해야합니다. 노트북 인스턴스로 사용할 이름과 lifecycle configurations를 입력 받도록 정의하면 됩니다. 특히 ML 모듈 실행을 위한 컴퓨팅 리소스를 잘 잡아줘야 하는데, 이것은 비용과 직결되는 문제이기 떄문에 상황에 맞게 선택하면 됩니다. 또, SubnetId, SecurityGroupIds, RoleArn의 파라미터도 필수로 입력해야 합니다. 이 부분은 VPC(네트워크 보안)와 관련된 문제이므로 본인이 사용하는 조직 내에서 사용하는 Security Group과 Role을 지정해주어야 합니다.

def create_notebook_instance(
        NotebookInstanceName: str, # 노트북 인스턴스로 사용할 이름
        LifecycleConfigName: str, # 노트북 인스턴스를 생성할 실행할 lifecycle configurations 
        RoleArn: str,
        SubnetId: str,
        SecurityGroupIds: str,
        InstanceType: str = "ml.t3.medium", # 노트북 인스턴스 타입
        DirectInternetAccess: str = "Enabled",
        VolumeSizeInGB: int = 50, # 노트북 인스턴스 용량
        DefaultCodeRepository: str = "my-github-repo", # 노트북 인스턴스가 생성되고 자동으로 Clone할 Repo
    ):
    try:
        response = client.create_notebook_instance(
            NotebookInstanceName=NotebookInstanceName,
            InstanceType=InstanceType,
            SubnetId=SubnetId,
            SecurityGroupIds=SecurityGroupIds,
            RoleArn=RoleArn,
            LifecycleConfigName=LifecycleConfigName,
            DirectInternetAccess=DirectInternetAccess,
            VolumeSizeInGB=VolumeSizeInGB,
            DefaultCodeRepository=DefaultCodeRepository,
            RootAccess=RootAccess
        )
        print(" >>> NotebookInstance is created. ")
        return response
    except Exception as err:
        return err

자 그런데.. 노트북 인스턴스까지 생성하는 건 알겠는데, 내가 실행하고자 하는 모듈은 어디에서 명령하는 걸까요? 🤔

가장 아름다운 그림은 노트북 인스턴스가 생성되면, Boto3로 생성한 client를 통해 해당 인스턴스 내의 터미널 CLI로 명령어를 전달하는 모습일 것입니다. 그런데 공식 문서에 나와있는 모든 함수를 다 찾아봤지만.. 이것을 제공하는 기능은 없는 것 같았습니다. SageMaker의 사상이 관리형이라 그런지는 몰라도 생성된 노트북 인스턴스 내에 특정 명령어를 전달할 수 있는 기능은 찾지 못했습니다. 🥲

결국.. lifecycle configurations를 활용하는 방법 뿐이 떠오르지 않았습니다. 노트북 인스턴스를 생성하면서 lifecycle configurations에 선언한 명령어를 실행할 때 파이썬 모듈까지 실행하도록 하는 것이죠..! 그래서 python 실행 코드가 담긴 shell을 실행하는 코드를 추가했습니다.

#!/bin/bash
sudo -u ec2-user -i <<'EOF'
conda create -n new_conda_env_p38 python=3.8 -y --force # 새로운 가상환경 생성
source activate new_conda_env_p38

cd SageMaker # SageMaker 노트북 인스턴스에 default로 생성되어 있는 폴더 
mkdir new_folder # 새로운 폴더 생성
cd new_folder

# Github repo를 clone하여 필요한 라이브러리를 설치
username="my_github_account"
password="my_github_account_password"
GIT_URL="https://${username}:${password}@github.com/WORKSPACE/my-repository.git"
git clone ${GIT_URL}

pip install -r requirements.txt

######## 추가된 부분 ^.^.. #########
chmod +x my_python_script.sh
nohup bash my_python_script.sh &
#################################

conda deactivate
EOF

전혀 아름다운 파이프라인이 아니긴 하지만, 직접 개발한 모듈을 실행하기 위해선 이 방법 밖엔 없었습니다. 흐흑. 뭐 그래도 여기까지는 그래도 괜찮았습니다. 문제는 lifecycle configurations은 최대 5분까지만 실행이 된다는 것입니다. 그래서 인스턴스가 생성되고 나서도 백그라운드에서 실행되게끔 nohup을 붙여줘야 했습니다. 😂

사실 여기서 더 중요한 것은 파이썬으로 실행하는 모듈의 구성과 순서입니다. 이 부분이 파이프라인을 만드는 가장 큰 이유이기 때문에 아래의 구성을 잘 고려하여 개발해야 합니다. 이 부분에 대한 내용도 추후에 따로 포스팅 해보려고 합니다.

• 데이터 로드
• 데이터 전처리
• 모델 훈련 및 검증
• 훈련된 모델로 예측 결과 생성
• 예측 결과 저장 (to AWS S3, Redshift, Athena 등)

Task 3) 인스턴스 삭제

모듈 실행이 끝났으면 인스턴스를 죽여..가 아니고 삭제해줘야 합니다. 회사에 돈이 많다면(...) 그냥 생성해놓고 쭉 놔둬도 상관은 없습니다. (만약 그렇다면 부럽습니다..)

하지만 인스턴스 타입에 GPU라도 포함되어 있다면 비용이 상당히 많이 발생합니다. 이처럼 비용까지 고려한다면 필요할 때만 노트북 인스턴스를 사용하고, 그 외에는 삭제하여 비용을 절감하는 것이 좋습니다.

그런데 SageMaker에서 노트북 인스턴스를 삭제하려면 중지(Stopped) 상태가 되어야 가능 합니다. '아 그러면 노트북 인스턴스를 중지하는 명령어를 실행하고, 그 이후에 삭제하면 되겠네!' 라고 생각하는 것이 자연스러운 사고의 흐름입니다. 에.. 근데 이게 안되겠더라고요. 왜냐면 위의 lifecycle configurations에서 실행한 파이썬 모듈이 끝날 때까지 기다려야 했기 때문이죠.

그래서 my_python_script.sh 스크립트에 파이썬 실행이 끝나면 노트북 인스턴스를 중지하는 CLI를 미리 추가 해줘야 했습니다.

#! /bin/bash
python ./run_my_script1.py
python ./run_my_script2.py
python ./run_my_script3.py
# 파이썬 모듈 실행이 끝나면 노트북 인스턴스를 중지하기 위한 명령어 추가.. ^.^..
aws sagemaker stop-notebook-instance --notebook-instance-name MY_NOTEBOOK_INSTANCE_NAME

그러면 여기서 필요한 것이 노트북 인스턴스의 상태를 계속 찔러보는 함수가 필요했습니다. (만두 속 찔러보는 것도 아니고 참내 이렇게까지 만들어야 하나 싶었지만 계속 만들었읍니다..)

def describe_notebook_instance(NotebookInstanceName: str = None):
    try:
        response = client.describe_notebook_instance(
            NotebookInstanceName=NotebookInstanceName
        )
        return response
    except Exception as err:
        return err

그리고 이 함수를 사용하여 주기적으로 상태를 조회하면서 'Stopped'가 되면 삭제를 실행하는 함수를 다시 만들었습니다.

import time

def delete_notebook_instance(NotebookInstanceName: str = None):
    try:
        response = client.delete_notebook_instance(
            NotebookInstanceName=NotebookInstanceName
        )
        print(" >>> NotebookInstance is deleting. ")
        return response
    except Exception as err:
        return err


def DeleteNotebookInstance(NotebookInstanceName: str, loop: int, time_sleep: int):
    for _ in range(loop):
        response = describe_notebook_instance(
            NotebookInstanceName=NotebookInstanceName
        )
        NotebookInstanceStatus = response["NotebookInstanceStatus"]
        if NotebookInstanceStatus == "Stopped":
            return delete_notebook_instance(
                NotebookInstanceName=NotebookInstanceName
            )
        else:
            print(f" >>> NotebookInstanceStatus is {NotebookInstanceStatus}. ")
            time.sleep(time_sleep)

DeleteNotebookInstance에서 loop와 time_sleep를 통해 노트북 인스턴스의 상태를 조회할 주기를 설정할 수 있도록 했는데, 예를 들면 아래와 같이 사용하도록 설계했습니다.

• loop=100, time_sleep=60 이라면 60초 간격으로 100회 동안 인스턴스의 상태를 조회하겠다는 것입니다. 즉, 총 6,000초니까 최대 1.6시간 동안 계속 조회하는 Task가 유지되는 것입니다. 이 부분은 인스턴스의 타입과 파이썬 모듈 스크립트 실행 시간을 적절히 고려햐여 배분하도록 의도했습니다.
• 위 함수를 통해 실행된 로그를 확인해보면 주기적으로 상태를 보면서 틈(?)을 노리다가 중지가 되면 삭제하는 것을 확인할 수 있습니다.

Task 4) lifecycle configurations 삭제

맨 처음에 생성한 lifecycle configurations까지 삭제를 해주면 파이프라인의 긴(?) 여정이 끝납니다. 해당 Task는 반드시 필요한 부분은 아니라고 생각하는데요, 어차피 계속 사용할 lifecycle configurations이라면 굳이 삭제할 필요는 없습니다.

다만, github 계정이나 repository 정보를 노출하지 않도록 하기 위해 해당 함수도 추가했습니다.

def delete_notebook_instance_lifecycle_config(NotebookInstanceLifecycleConfigName: str):
    try:
        response = client.delete_notebook_instance_lifecycle_config(
            NotebookInstanceLifecycleConfigName=NotebookInstanceLifecycleConfigName
        )
        print(" >>> NotebookInstanceLifecycleConfigName is deleted. ")
        return response
    except Exception as err:
        return err

Airflow DAG 생성

SageMaker 노트북 인스턴스 생성, 삭제에 필요한 함수는 모두 생성했으니 이것을 DAG에서 호출하여 Task로 만들면 원했던 파이프라인이 완성됩니다.

Variables

저는 DAG의 Task를 생성할 때 Operator에 전달할 파라미터의 값을 하드코딩하기 보다는 웬만하면 Variables로 빼놓고, 이것을 호출하여 파라미터로 전달하는 편입니다. 두 가지 이유 때문입니다.

• 첫째로 보안 때문입니다. Github에 DAG 코드를 올릴 때 계정 정보나 네트워크 정보가 포함되어 올라가지 않도록 해야 하기 때문입니다.
• 두 번째는 DAG 스크립트를 반복 사용할 수 있도록 하기 위함입니다. 해당 DAG.py를 복사해서 dag_id를 바꿔주고, Variables만 새로 추가 해준다면 새로운 파이프라인을 손쉽게 생성할 수 있기 때문입니다.

예시로 Variables을 아래와 같이 구성해봤습니다. 각 함수에 전달해야 할 파라미터를 key, value로 미리 정의 해놓는 것 뿐이라 전혀 어려운 작업은 아닙니다.

{
  "my_variable": {
    "NotebookInstanceLifecycleConfigName": "my-lifecycle-configurations",
    "NotebookInstanceName": "my-notebook-instance",
    "InstanceType": "ml.c5d.4xlarge",
    "RoleArn": "arn:aws:iam::*************:role/service-role/*************",
    "SubnetId": "****************",
    "SecurityGroupIds": [
      "sg-****************",
      "sg-****************"
    ],
    "content": "#!/bin/bash\nsudo ... 아까 위애서 작성한 shell을 여기에!! ... EOF"
  }
}

DAG

Variables를 DAG에서 호출하고, PythonOperator의 op_kwargs에 전달해주어 Task별 Flow만 선언해주면 원했던 파이프라인이 완성됩니다. 👏

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime, timedelta

from utils.sagemaker import SageMaker # 위에서 설계한 함수를 클래스로 만들어서 utils에 따로 저장

sagemaker = SageMaker()

## Variables
variables = Variable.get("my_variable", deserialize_json=True)

NotebookInstanceLifecycleConfigName = variables["NotebookInstanceLifecycleConfigName"]
NotebookInstanceName = variables["NotebookInstanceName"]
InstanceType = variables["InstanceType"]
SecurityGroupIds = variables["SecurityGroupIds"]
SubnetId = variables["SubnetId"]
RoleArn = variables["RoleArn"]
content = variables["content"]

## DAG
default_args={
    "depends_on_past": False,
    "email": ["airflow@example.com"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5)
}

dag = DAG(
    dag_id="my_dag_id",
    description="DAG for SageMaker Operating",
    schedule_interval="0 0 * * *",
    catchup=False,
    default_args=default_args
)

## Tasks
START = DummyOperator(dag=dag, task_id="START")
END = DummyOperator(dag=dag, task_id="END")

CreateNotebookInstanceLifecycleConfig = PythonOperator(
    task_id="CreateNotebookInstanceLifecycleConfig",
    python_callable=sagemaker.create_notebook_instance_lifecycle_config,
    provide_context=True,
    op_kwargs={
        "NotebookInstanceLifecycleConfigName": NotebookInstanceLifecycleConfigName,
        "content": content,
    },
    dag=dag
)

CreateNotebookInstance = PythonOperator(
    task_id="CreateNotebookInstance",
    python_callable=sagemaker.create_notebook_instance,
    provide_context=True,
    op_kwargs={
        "NotebookInstanceName": NotebookInstanceName,
        "LifecycleConfigName": NotebookInstanceLifecycleConfigName,
        "InstanceType": InstanceType,
        "RoleArn": RoleArn,
        "SubnetId": SubnetId,
        "SecurityGroupIds": SecurityGroupIds
    },
    dag=dag
)

DeleteNotebookInstance = PythonOperator(
    task_id="DeleteNotebookInstance",
    python_callable=sagemaker.DeleteNotebookInstance,
    provide_context=True,
    op_kwargs={
        "NotebookInstanceName": NotebookInstanceName,
        "time_sleep": 300, # 300초(5분) 마다 인스턴스 확인
        "loop": 12 * 6  # 최대 6 시간 동안 인스턴스 확인
    },
    dag=dag
)

DeleteNotebookInstanceLifecycleConfig = PythonOperator(
    task_id="DeleteNotebookInstanceLifecycleConfig",
    python_callable=sagemaker.delete_notebook_instance_lifecycle_config,
    provide_context=True,
    op_kwargs={
        "NotebookInstanceLifecycleConfigName": NotebookInstanceLifecycleConfigName
    },
    dag=dag
)

START >> \
CreateNotebookInstanceLifecycleConfig >> \
CreateNotebookInstance >> \
DeleteNotebookInstance >> \
DeleteNotebookInstanceLifecycleConfig >> \ 
END

마무리 ⌛️

MWAA와 SageMaker 조합으로 ML 파이프라인을 만들기 위해 겪은 시행착오를 간단히 공유해봤습니다. 글이 좀 장황해서(..) 간단해 보이진 않았지만, Airflow의 스케줄링 기능을 바탕으로 boto3로 SageMaker 노트북 인스턴스를 생성하고 삭제하는 과정이 전부이기 때문에 그리 복잡한 파이프라인은 아닐 것입니다. 😅

물론 SageMaker 장점을 잘 살린 파이프라인은 아니었습니다. 데이터 로드, 모델 훈련, 검증, 모델 배포까지의 흐름을 SageMaker에서 제공하는 컴포넌트를 활용하여 만드는 사례가 훨씬 많기 때문입니다. 다만, SageMaker의 컴포넌트를 사용하지 않고 직접 개발한 모듈을 배치로 실행하기 위해 노트북 인스턴스를 컴퓨팅 리소스로 사용하기 위한 사례 정도로 보면 될 것 같습니다.

기회가 된다면 SageMaker Studio를 활용하여 파이프라인을 새로이 구성해보려고 합니다. 이것도 무사히 만들게 된다면(과아연..) 시행착오의 여정을 공유해보겠습니다. 뭐어.. 잘 실행만 된다면 되는 거니깐요 하하.. 끄-읕 👋

아직 많이 부족하지만 실전을 통해 체득한 노하우를 기록해보려고 합니다. 잘못된 내용이 있다면 지적도 부탁드립니다! 🤚

이번 포스팅의 목적

시계열 예측도 모델 훈련, 검증을 통해 성능이 가장 좋은 모형을 만드는 머신러닝 방법에 속합니다. 그런데 시계열 예측에 활용할 수 있는 알고리즘은 매우 다양합니다.

가장 기본적인 smoothing 기법부터 ARIMA, SARIMA, ARIMAX, VAR 등의 통계적 예측 방법이 있으며, Meta(과거 Facebook)에서 개발한 Prophet, LinkedIn에서 개발한 Greykite, 그리고 딥러닝 기반의 LSTM, ELM, Attenion 등 시도해볼 방법이 너무나 많습니다.

요새는 대부분 딥러닝을 활용하는 추세이지만, 반드시 딥러닝만이 좋은 모델이라고 장담하긴 어렵습니다. 모형 적합에 사용할 변수와 예측 기간 등 문제 정의와 실험 설계에 따라 때로는 통계 기반의 시계열 모델이 더 좋은 성능을 보일 때도 있기 때문입니다. 그래서 지금까지 프로젝트를 해보면서 느낀 것은 '하나의 방법으로만 실험하는 건 별 의미가 없다'입니다.

가령, 프로젝트 구성원과 열띤 논의를 통해 D+1 예측으로 문제를 정의했다가도, 향후 D+7 예측이 필요한 상황으로 바뀔 수 있습니다. 심지어 단기 예측이 아닌 중・장기 예측이 필요한 상황이 추가될 수도 있죠. 그뿐만 아니라 비즈니스 도메인을 이해하면서 새롭게 추가할 변수를 찾게 된다면 또다시 실험을 해봐야 합니다. 이렇듯 요구사항과 데이터는 언제든 바뀔 수 있기에 어떤 알고리즘이 적절한지는 매번 테스트를 해봐야 하는 게 현실인 것 같습니다.

이처럼 문제 정의가 바뀌면 실험 설계도 달라질 가능성이 크기 때문에, 변화에 발맞춰 새로운 모델 성능을 빠르게 확인할 수 있는 환경이 필요했습니다.

그래서 이번 글에서는 알고리즘별 시계열 예측 성능 측정 모듈 개발 과정의 일부를 간단히 공유해보려고 합니다.

본 포스팅에서 사용한 데이터

시계열 샘플 데이터는 Kaggle, UCI Machine Learning Repository 등에서 쉽게 얻을 수 있습니다. 근데 이번 글에서는 코로나 확진자 데이터를 사용했습니다. (이제 실외 마스크 해제가 가능해졌으니 하루빨리 종식되길 바라며..)

원래는 주가 데이터를 사용해 보려고 했으나.. 제 주식을 떠올리니 눈물 젖은 글을 쓸 것 같아 꾹 참았습니다. (크흑..)

이런 걸 만들었습니다

결론부터 보여드리자면, 예측 시작 시점, 예측 기간, 시계열 예측 모델을 원하는 형태로 지정하여 교차검증(Cross Validation)을 수행하는 함수를 개발했습니다. Cutoff에 따라 훈련 & 검증 데이터셋을 나누어 교차 검증을 수행하되, 원하는 모델을 언제든 추가하여 실행할 수 있도록 확장성을 고려하여 모듈을 개발했습니다.

>>> from time_series_forecast import forecast_cross_validation

>>> df_fcst = forecast_cross_validation(
        df=df,                  # 예측에 사용할 pandas dataframe
        freq="D",               # 예측 주기 (D: Daily, W: Weekly, M: Monthly)
        cutoffs=["2022-04-01"], # 예측 시작 시점
        step=3,                 # 예측 기간 (2022-04-01 ~ 2022-04-03 예측)
        model="arima"           # 예측에 사용할 알고리즘 선택 (arima, prophet, lstm, ...)
    )

>>> print(df_fcst)
        model   cutoff      date        y     yhat     yhat_lower  yhat_upper  yhat_naive
    0   ARIMA   2022-04-01  2022-04-01  2224  2105.13  1944.76     2265.51     2224
    1   ARIMA   2022-04-01  2022-04-02  1760  1765.64  1564.36     1966.92     2224
    2   ARIMA   2022-04-01  2022-04-03  1715  1880.29  1654.24     2106.33     2224

>>> from time_series_forecast_evaluation import get_performace_metric

>>> df_perf = get_performace_metric(df=df_fcst) # 예측 성능 지표 계산

>>> print(df_perf)
    {'mae': 906.85, 'mape': 20.81, 'mase': 6.22, 'coverage': 19.04, 'winkler': 6688.22}

이처럼 cutoff, step, model을 다양하게 조합하여 성능을 비교할 수 있으며, 실행 결과를 아래의 사진처럼 만들 수 있습니다.

Cutoff는 무엇인가

시계열 데이터에서 특정 시점 전까지를 Train dataset으로 모델링하기 위해 사용하는 표현

시계열 예측에서 교차 검증은 흔히 알려진 Cross Validtion과 약간 차이가 있습니다. 너무 당연한 말이긴 하지만, 시계열은 시간의 순서를 유지해야 하므로 Row를 뒤섞어(suffling) 데이터셋을 나눌 수 없습니다.

그래서 예측 시작점을 따로 선택하고, 해당 시점 전까지를 훈련 데이터로 사용하여 그 이후를 step 만큼 예측하여 성능을 확인합니다.

Cutoff를 아래와 같이 10개를 설정한다면 10개의 훈련 데이터 셋이 생기고, 각각 예측한 결과로 시간의 흐름에 따른 성능을 비교할 수 있습니다.

모듈 구조 (Functions in Module)

위에서 예시로 보여드린 모듈(time_series_forecast)의 구조는 대략 아래와 같습니다.

• forecast_cross_validation(): 교차검증을 수행하는 main 함수 (forecast() 호출)
• forecast(): cutoff에 따라 train, test dataset 분리 후 예측 실행 (fit_*() 호출)
• fit_arima() : ARIMA 예측 모델
• fit_prophet() : Prophet 예측 모델
• fit_lstm() : LSTM 예측 모델
• fit_...() : 그 외 시계열 모델
• ...

forecast_cross_validation()에서 모든 작업을 통제합니다. cutoff와 사용할 모델을 지정하면 그에 맞게 예측을 수행하는 구조입니다. 이렇게 메인 함수는 어느 정도 추상화를 해놓고, 시계열 예측에 사용할 모델을 튜닝하거나 추가하여 확장성이 용이하도록 설계했습니다.

단변량 시계열 예측 모듈 만들어보기 (hands-on 💻)

데이터에 따라 다르겠지만 대부분의 경우 단변량 예측은 다변량 예측에 비해 성능이 떨어지기 마련입니다. 다만, 이번 글에서는 모듈의 구조와 성능 비교까지의 흐름을 공유하는 게 목적이기 때문에 이해하기 쉬운 단변량 예측으로 설명한 점을 참고해 주세요.

이제 모듈을 간단한 버전으로 만들어보겠습니다. 여기서는 auto_arima와 prophet 두 가지만 사용했습니다.

1) 모델 반환 함수

먼저 알고리즘별로 모델을 반환하는 함수를 만들어 줍니다.

1.1) ARIMA

ARIMA는 과거 관측값과 오차항만으로 적합한 모형을 도출하는데, 주로 단기 예측에서 많이 사용합니다. 최적의 모형을 추정하려면 정상성을 확인하는 단위근 검정, 계절성 차수를 결정하기 위한 AICc 최소화, 최대 가능도 추정(MLE) 등의 복잡한 과정을 거쳐야 합니다. 그래서 보통 단변량 & 단기 예측을 할 때 이러한 단계를 통해 모형을 도출하는 auto_arima()를 많이 사용합니다.

from pmdarima.arima import auto_arima, ARIMA


def fit_arima(
        df: pd.DataFrame,
        params: dict = None
    ) -> ARIMA:
    """
    ARIMA에 데이터를 fitting한 객체를 반환한다.
    """
    if params is None:
        ar = auto_arima(y=df["y"])
    else:
        ar = auto_arima(y=df["y"], **params)
    print(ar.summary())
    ar.fit(df["y"])
    return ar

1.2) Prophet

Prophet은 Meta(과거 Facebook)에서 개발한 시계열예측 라이브러리로, 학습 속도가 빠르고 계절성이 있는 데이터에 좋은 예측 성능을 보이며 seasonality, growth 등의 파라미터 튜닝이 가능한 모형입니다. ARIMA와 다르게 차분을 통해 정상성을 만족하는 형태로 변환할 필요가 없고, 결측치를 채워주지 않아도 예측이 가능하도록 설계된 라이브러리입니다.

특히, 휴일(Holiday)의 영향을 고려하는 것이 특징입니다. 특정 날짜를 이벤트로 인식하도록 feature를 추가로 넣어주면 해당 날짜 전후에 따른 영향을 학습하여 미래에 있을 같은 이벤트에 전후로도 그 영향을 반영해 주기 때문에 이것 역시 많이 사용합니다.

from prophet import Prophet


def fit_prophet(
        df: pd.DataFrame,
        params: dict = None
    ) -> Prophet:
    """
    Prophet에 데이터를 fitting한 객체를 반환한다.
    """
    if params is None:
        m = Prophet()
        add_seasonality = None
    else:
        if "add_seasonality" in params:
            add_seasonality = params["add_seasonality"]
            params_ = params.copy()
            params_.pop("add_seasonality")
            m = Prophet(**params_)
        else:
            m = Prophet(**params)
            add_seasonality = None

    if add_seasonality is not None:
        if type(add_seasonality) == list:
            for add in add_seasonality:
                m.add_seasonality(**add)
        elif type(add_seasonality) == dict:
            m.add_seasonality(**add_seasonality)
        else:
            raise TypeError("TypeError: 'add_seasonality' is allowed dictionary or list type.")

    m.add_country_holidays(country_name="KR")
    m.fit(df)
    return m

2) 예측 실행 함수

위에서 정의한 함수를 사용하여 예측 실행 함수를 만듭니다. 이 부분이 모델별 예측을 수행하는 함수이므로 새로운 시계열 예측 알고리즘을 추가할 때마다 여기에도 추가해 주는 방식으로 사용할 수 있습니다.

아래와 같은 아키텍쳐 보다는 예측 실행을 최종 담당하는 추상화된 클래스가 있고, 알고리즘별 예측 클래스를 상속받아 실행하는 형태로 개발한다면 더욱 좋습니다.

def forecast(
        df: pd.DataFrame,
        model: str,
        cutoff: str,
        steps: int,
        freq: str = "D",
        params: dict = None
    ) -> pd.DataFrame:
    """
    시계열 예측 모델을 호출하여 cutoff로부터 steps 만큼 예측한 결과를 반환한다.
    """
    series = pd.date_range(start=cutoff, periods=steps, freq=freq)
    df_future = pd.DataFrame(series)
    df_future.columns = ["date"]

    if model == "prophet":
        df = df.rename(columns={"date": "ds"})
        df_future = df_future.rename(columns={"date": "ds"})
        m = fit_prophet(df=df, params=params)
        df_fcst = m.predict(df_future)
        df_fcst = df_fcst.rename(columns={"ds": "date"})

    elif model == "arima":
        df = df.set_index("date").resample(rule=freq).ffill()
        m = fit_arima(df=df, params=params)
        pred, pi = m.predict(n_periods=steps, return_conf_int=True, alpha=.2)
        df_fcst = pd.DataFrame(pi, index=df_future["date"], columns=["yhat_lower", "yhat_upper"])
        df_fcst["yhat"] = pd.Series(pred, index=df_future["date"])
        df_fcst = df_fcst.reset_index()

    # 필요시 모델별 예측 조건 추가 가능
    # elif model == "lstm": ...
    # elif model == "lstm-stl": ...
    # elif model == "prophet-stl": ...

    else:
        raise ValueError(
            "Not found model. 'model' should be one of 'arima', 'prophet'."
        )
    return df_fcst

3) 교차검증 실행 함수

이제 여러 cutoff를 통해 예측한 결과와 실제값을 비교해 볼 수 있는 메인 함수를 만들면 큰 그림은 어느 정도 완성입니다. 이것을 사용하여 모델별 예측값과 실제값이 담긴 결과를 생성하고, 해당 결과로부터 성능 지표(performance metrics)를 만들기만 하면 본 포스팅의 목적을 달성하게 됩니다.

def add_date(date: str, delta: int) -> str:
    return (datetime.strptime(date, "%Y-%m-%d") + timedelta(days=delta)).strftime("%Y-%m-%d")

def forecast_cross_validation(
        df: pd.DataFrame,
        model: str,
        cutoffs: list,
        steps: int,
        freq: str = "D",
        params: dict = None
    ) -> pd.DataFrame:
    """
    forecast()를 사용하여 cutoff로부터 steps 만큼 예측하고, 실제값(y)과 예측값(yhat)을 비교할 수 있는 dataframe을 반환한다.
    """
    df_fcst_all = pd.DataFrame()
    for cutoff in cutoffs:
        print(f"\n ---------------------- cutoff: {cutoff} ---------------------- ")
        date_end = add_date(date=cutoff, delta=steps - 1)
        df_train = df[df["date"] < cutoff].reset_index(drop=True)
        df_test = df[df["date"].between(cutoff, date_end)].reset_index(drop=True)

        if len(df_test) == 0:
            print("There is no test dataset for this cutoff.")
            continue
        else:
            print(f"train dataset rows: {df_train.shape[0]}")
            print(f"test dataset rows: {df_test.shape[0]}\n")

        df_fcst = forecast(
            df=df_train,
            model=model,
            cutoff=cutoff,
            steps=steps,
            freq=freq,
            params=params
        )
        df_fcst = df_fcst.merge(df_test, on="date", how="left")
        df_fcst["cutoff"] = cutoff
        df_fcst["yhat_naive"] = df_train["y"].iloc[-1]
        df_fcst_all = pd.concat([df_fcst_all, df_fcst], ignore_index=True)
    df_fcst_all.insert(0, "model", model)
    df_fcst_all = df_fcst_all[["model", "cutoff", "date", "y", "yhat", "yhat_lower", "yhat_upper", "yhat_naive"]]
    return df_fcst_all

3.1) 사용 예시

위에서 보여드린 샘플 데이터를 활용하여 2021.12.01부터 3일간의 multi step 예측 결과를 확인해 볼 수 있습니다.

# auto_arima에 하이퍼파라미터를 직접 전달하여 예측
df_fcst = forecast_cross_validation(
    df=df,
    cutoffs=["2021-12-01"],
    steps=3,
    freq="D",
    model="arima",
    params={
        "seasonal": False,
        "stepwise": True,
        "trace": True,
        "start_p": 5,
        "max_p": 15,
        "max_q": 5,
        "max_order": 20,
        "m": 0
    }
)
+----+---------+------------+------------+------+---------+--------------+--------------+--------------+
|    | model   | cutoff     | date       |    y |    yhat |   yhat_lower |   yhat_upper |   yhat_naive |
|----+---------+------------+------------+------+---------+--------------+--------------+--------------|
|  0 | arima   | 2021-12-01 | 2021-12-01 | 5123 | 3928.96 |      3759.58 |      4098.33 |         3032 |
|  1 | arima   | 2021-12-01 | 2021-12-02 | 5266 | 4004.15 |      3784.21 |      4224.08 |         3032 |
|  2 | arima   | 2021-12-01 | 2021-12-03 | 4944 | 3774.39 |      3528.63 |      4020.15 |         3032 |
+----+---------+------------+------------+------+---------+--------------+--------------+--------------+

이때 cutoff를 여러 개로 지정하여 예측을 실행하는 것이 이 함수의 주요 역할입니다.

# prophet에 하이퍼파라미터를 직접 전달하여 예측
forecast_cross_validation(
    df=df,
    cutoffs=["2021-12-01", "2021-12-04"], # cutoff를 여러 개로 설정하는 것이 포인트!
    steps=3,
    freq="D",
    model="prophet",
    params={"yearly_seasonality": True, "weekly_seasonality": True, "daily_seasonality": False}
)
+----+---------+------------+------------+------+---------+--------------+--------------+--------------+
|    | model   | cutoff     | date       |    y |    yhat |   yhat_lower |   yhat_upper |   yhat_naive |
|----+---------+------------+------------+------+---------+--------------+--------------+--------------|
|  0 | prophet | 2021-12-01 | 2021-12-01 | 5123 | 3318.81 |      3046.54 |      3571.58 |         3032 |
|  1 | prophet | 2021-12-01 | 2021-12-02 | 5266 | 3324.61 |      3086.13 |      3578.71 |         3032 |
|  2 | prophet | 2021-12-01 | 2021-12-03 | 4944 | 3330.05 |      3059.28 |      3570.91 |         3032 |
|  3 | prophet | 2021-12-04 | 2021-12-04 | 5352 | 3673.2  |      3392.92 |      3975.52 |         4944 |
|  4 | prophet | 2021-12-04 | 2021-12-05 | 5128 | 3639.85 |      3346.52 |      3920.09 |         4944 |
|  5 | prophet | 2021-12-04 | 2021-12-06 | 4325 | 3568.89 |      3299.8  |      3860.35 |         4944 |
+----+---------+------------+------------+------+---------+--------------+--------------+--------------+

4) 성능 지표 계산 함수

시계열 예측 성능을 비교하는 몇 가지 metric이 있는데, 이것을 계산하는 함수까지 만들면 끝입니다!

아래는 시계열 예측에서 성능 지표로 많이 사용하는 것을 나열한 것인데, 여기에선 MAE, MAPE 두 가지만 사용하여 만들어보겠습니다.

• R-Squared
• Mean Absolute Error (MAE) ✅
• Mean Absolute Percentage Error (MAPE) ✅
• Mean Squared Error (MSE)
• Root Mean Squared Error(RMSE)
• Normalized Root Mean Squared Error (NRMSE)
• Winkler scores

def get_performance_metrics(df_fcst: pd.DataFrame) -> dict:
    """
    시계열 예측 결과가 담긴 DataFrame으로부터 performance metrics을 계산하여 반환한다.
    """
    true = df_fcst["y"]
    pred = df_fcst["yhat"]
    mae = (true - pred).abs().mean()
    mape = (true - pred).abs().div(true).mean() * 100
    n_cover = (((true - df_fcst["yhat_lower"]) > 0) & ((true - df_fcst["yhat_upper"]) < 0)).sum()
    coverage = (n_cover / (true > 0).sum()) * 100
    return {"mae": mae, "mape": mape, "coverage": coverage}

5) 모델별 성능 지표 비교하기

이 정도면 대충 준비는 끝났습니다. 2021년 11월의 매주 일요일을 Cutoff로 놓고 D+3을 예측하여 성능을 비교해 보겠습니다.

# cutoff 리스트 생성
cutoffs = ['2021-11-07', '2021-11-14', '2021-11-21', '2021-11-28']

# Prophet 예측
df_fcst_prophet = forecast_cross_validation(
    df=df,
    cutoffs=cutoffs,
    steps=3,
    freq="D",
    model="prophet",
    params={"yearly_seasonality": True, "weekly_seasonality": True, "daily_seasonality": False}
)

# ARIMA 예측
df_fcst_arima = forecast_cross_validation(
    df=df,
    cutoffs=cutoffs,
    steps=3,
    freq="D",
    model="arima",
    params={
        "seasonal": False,
        "stepwise": True,
        "trace": True,
        "start_p": 5,
        "max_p": 15,
        "max_q": 5,
        "max_order": 20,
        "m": 0
    }
)

# 예측 결과 합치기
df_fcst = pd.concat([df_fcst_prophet, df_fcst_arima], ignore_index=True)

# 각 cutoff의 모델별 예측 성능 지표 생성
df_perf = pd.DataFrame()
for model in df_fcst["model"].unique():
    for cutoff in cutoffs:
        df_fcst_tmp = df_fcst[(df_fcst["cutoff"]==cutoff) & (df_fcst["model"]==model)].reset_index(drop=True)
        metric_dict = get_performance_metrics(df_fcst_tmp)
        metric_dict["model"] = model
        metric_dict["cutoff"] = cutoff
        df_perf_tmp = pd.Series(metric_dict).to_frame().T
        df_perf = pd.concat([df_perf, df_perf_tmp], ignore_index=True)

# 각 cutoff의 naive 예측 성능 생성
for cutoff in cutoffs:
    df_fcst_tmp = df_fcst[df_fcst["cutoff"]==cutoff].reset_index(drop=True)
    metric_dict = get_performance_metrics(df_fcst_tmp.drop("yhat", axis=1).rename(columns={"yhat_naive": "yhat"}))
    df_perf_tmp = pd.Series(metric_dict).to_frame().T
    df_perf_tmp["cutoff"] = cutoff
    df_perf_tmp["model"] = "naive"
    df_perf = pd.concat([df_perf, df_perf_tmp], ignore_index=True)

df_perf = df_perf[["model", "cutoff", "mae", "mape", "coverage"]]

# Performance metrics 시각화
plt.figure(figsize=(16, 8))
sns.barplot(data=df_perf, x="model", y="mape", hue="cutoff")
_ = plt.yticks(np.linspace(0, 25, 11), fontsize=14)
_ = plt.yticks(fontsize=14);

대체로 ARIMA가 상대적으로 좋은 성능을 나타내는 것으로 보입니다. 이런 식으로 시간의 흐름에 따른 예측 성능을 지속 확인할 수 있다면, 현시점의 오차를 파악하는데 조금이라도 도움이 될 수 있을 것 같습니다. 또, 최소한 Naive 예측 결과 보다 좋은지를 확인하는 게 모델 선택의 기준점이 되기도 합니다. (해당 샘플 코드는 하나의 방법일 뿐, 절대적인 방법은 아닙니다!)

어떻게 활용하면 좋을까요 🤔

열심히 만들었는데 목적에 맞게 잘 활용하는 것도 중요하겠죠. 이 모듈의 큰 목적은 '시간의 흐름에 따라 모델의 성능이 어떻게 변화하는가'를 보는 것입니다.

만약 전반적으로 모든 모델의 성능이 큰 변동없이 비슷한 수준을 보이면서, 만족할만한 예측력이라면 그 중에서 best 모델을 선택하는 기준을 충분히 잡을 수 있을 것입니다.

혹은 예측하려는 Target의 변동성이 애초에 커서 전반적으로 모델 성능이 출렁인다면 모델 튜닝이 필요하거나, 현재의 데이터셋만으로 설명이 불가능한 또 다른 변수를 찾아야 한다는 신호를 감지할 수 있는 용도로 사용할 수도 있습니다. (제가 실제로 이렇게 사용하고 있습니다!)

추가해보면 좋을 것들

이 글에 적진 않았지만 따로 추가하면 더 좋을 것 같은 기능도 나열해 봤습니다.

• 다른 예측 모델을 추가해보면서 비교하면 더욱 좋습니다. (LSTM, LGBM, …)
• 모델별 하이퍼파라미터별 성능 비교 모듈이 있으면 좋습니다.
• Cutoff별 예측 실행을 병렬로 처리할 수 있으면 좋습니다.
• Cutoff를 일, 주, 월별로 반환하는 함수를 만들어서 사용하면 좋습니다.
• 데이터셋으로 사용하는 DataFrame에 반드시 있어야 할 컬럼명을 검증하는 Validator 함수가 있으면 좋습니다.
• 데이터셋 결측치를 Interpolation 해주는 함수를 만들어 놓으면 좋습니다.
• STL을 통해 시계열을 분해하여 Trend, Seasonality, Residual를 반환하는 함수가 있으면 좋습니다. 예측 시 특정 요소를 제거하여 시도해볼 수 있습니다.

마무리 ⌛️

(와 드디어 다 썼다!)

시계열 모델 성능 비교를 위해 이해하기 쉬운 단변량 예측에 초점을 두고 모듈 개발의 컨셉을 공유해 봤습니다.

사실 코사인이나 사인 그래프 마냥 아름다운 계절성과 트렌드를 보이는 시계열 데이터가 아닌 이상, 단변량으로 정확도 높은 예측을 해내는 것은 상당히 어려운 일입니다. 결국 다양한 변수를 추가해 보면서 성능을 끌어올리는 작업은 필연적이기 때문에 이 글에서 제안한 코드 보다는 훨씬 복잡한 수준이 될 것입니다. 또, 모델을 추가할수록 더 모듈 내 기능을 잘 분리해야 하죠. 그래서 더더욱 모델별 비교를 위한 환경을 잘 준비해야 하지 않을까 싶습니다.

시계열 예측을 처음 시작한 지 얼마 안 되었을 때, 모델별로 모듈을 각각 만들고 있었습니다. 그러다가 문득 '이거 이러다가 더 많은 모델을 실험하면 점점 비교가 어려워지겠다'라는 생각이 자연스럽게 들었고, 이것을 해결하기 위해 본 포스팅에서 작성한 흐름대로 모듈을 개발했습니다.

시계열 데이터는 시간이 지남에 따라 얼마든지 '좋은' 모델이 달라질 수 있습니다. 과거의 패턴을 특정 시점에만 학습을 시키고, 해당 모델로만 계속 예측을 하다 보면 분명히 불확실성이 높은 예측 구간이 발생하기 마련입니다. 그러므로 현재 만들고 있는 시계열 모델들을 언제든 비교할 수 있는 환경을 미리 만들어 놓는 것도 좋을 것 같습니다.

이 글을 보신 분들 중 시계열 예측을 처음 시작하시거나, 혹은 저처럼 어려움을 겪었던 분들에게 도움이 되었으면 좋겠네요!

'늦었다고 생각할 때가 너무 늦었다'라는 무한도전 명수옹의 말이 떠오릅니다. 회사를 다니면서 공부한 것을 기록하거나, 프로젝트를 기술한 내용을 써보겠다고 마음 먹은지가 언제인지 잘 기억도 안 나니 말입니다. 여튼 시작이 반이라고, 이렇게라도 시작했으니 머릿속에만 남겨져 있던 기억을 조금씩 기록으로 남겨보려고 합니다.

무엇을 쓸 것인가

저는 데이터 사이언티스트로 일하고 있습니다. 그래서 데이터와 관련된 글을 많이 쓸 것 같습니다. 일에 필요한 책을 읽으며 정리할 수도 있고, 프로젝트를 기술한 글을 쓸 수도 있고, 조직 문화와 관련된 글을 쓸 것 같습니다.

사실 지금 단계에선 글감의 주제가 중요하진 않다고 생각합니다. 쓰는 습관을 들이는 게 중요할 것 같군요. 여러번 작성하다 보면 블로그의 성격과 방향이 잡히지 않을까 하는 생각입니다. '어떤 것을 연재해볼까?'라는 생각에 잠길수록 블로그 시작의 발목을 잡는 것 같았습니다. 제가 그랬거든요.

어떤 글이 올라올지 저 조차 예상이 안 됩니다(ㅋㅋ). 무엇이 되었든 '기록'을 남기는 데에 당분간 초점을 두고 써보렵니다.

어느 주기로 써야할까

특정 요일에 맞춰서 글을 쓰기 보다는, 최소 2주에 한번은 글을 남기려고 합니다. 회사에서 보통 2주 단위로 스프린트를 하는데, 이 사이에 했던 업무나 경험을 주로 써볼 예정입니다. 사실 블로그를 통해 회사에서 했던 프로젝트를 여기에 남김으로써 따로 포트폴리오 파일을 만들지 않는 것이 목적이기도 하거든요. 여하튼 2주에 한 번은 쓴다!

Velog를 선택한 이유

원래는 개인 노션에 기록을 하고 있었습니다. 아무래도 개인 저장용으로만 사용하다 보니 다른 사람에게 노출이 되지 않아서 글의 의견을 받지 못한 것이 아쉬웠어요. 바꿔 말하자면 다른 사람에게 보이는 글을 써보고 싶었어요. 사진의 노션이 저의 초라한 개인 블로그입니다.. ㅠㅠ (아니, 였습니다..)

Velog 말고도 Tistory, Medium, Github 등 여러 개의 선택지가 있었지만 Velog가 여러모로 가입도 편하고 글을 쉽게 쓸 수 있는 플랫폼인 것 같아서 이것으로 선택했습니다. 물론 노션도 이제는 oopy를 통해 public으로 노출할 수 있는 것으로 알고 있지만, 새롭게 시작하려는 마음에 여기로 옮겨 보았습니다.

오늘은 여기까지

처음부터 너무 힘을 들이면 오래가지 못할 것 같은 느낌이 듭니다. 운동도 워밍업을 충분히 해야 메인 세트에서 힘을 잘 발휘하듯, 차근 차근 글을 늘려나가보겠습니다. 사실 제 블로그를 꾸준히 보는 사람이 아마도 없을테지만 그래도 열심히 남겨보렵니다. 끄읏.