[ML] scikit-learn: Custom Transformer

scikit-learn에서 Custom Transformer 구현하기 

scikit-learn에서는 데이터 전처리를 위한 다양한 transformer를 제공하지만,

경우에 따라서는 특정 요구 사항에 맞는 custom transformer를 직접 구현해야 할 때가 있음.

fit를 통해 학습데이터에서
특정 parameters의 값을 구해야할 필요가 없다면
sklearn.preprocessing.FunctionTransformer 를 사용하는게 보다 나음.

 

이 경우,

• BaseEstimator와
• TransformerMixin을 상속받아 구현함.

반드시 구현해야하는 methods 는

• __init__(self, ...) 생성자 ** : explicit parameters (=hyper-parameters)
• fit(self, X, y=None) 메서드 **
• transform(self, X) 메서드 **
• get_feature_names_out(self, input_features=None) 메서드**
• 필요할 경우, inverse_transform(self, X) 메서드도 구현되어야 함. **

반드시 설정해야하는 instance attributes는

• n_features_in_ : 입력데이터의 각 샘플의 features의 갯수 (fit 에서).
• feature_names_in_ : 입력데이터의 features 의 이름들.

---

참고: FunctionTransformer

2024.10.03 - [Programming/ML] - [ML] scikit-learn: FunctionTransformer

[ML] scikit-learn: FunctionTransformer

---

---

이 글에서는 StandardScalerClone이라는 Custom Transformer를 구현하는 방법을 단계별로 소개한다.

---

1. 기본 구조: BaseEstimator와 TransformerMixin

scikit-learn에서 custom transformer를 만들기 위해서는 BaseEstimator와 TransformerMixin을 상속받아야 구현.

scikit-learn은 duck typing이므로 필요한 methods를 구현하면 되지만,
이들을 상속하는 것이 훨씬 편하게 구현가능하므로 상속을 권함.

• BaseEstimator:
  • scikit-learn의 모든 estimator의 기반이 되는 클래스: estimator 라면 상속하는 게 편함.
  • hyper-parameters 관리 및 automatic hyper-parameter tuning을 가능하게해 줌.
  • __init__ constructor의 explicit parameters로 제공되는 hyper-parameters에 대해 getter와 setter를 제공.
    • get_params()
    • set_params()
• TransformerMixin:
  • fit_transform 메서드를 자동으로 제공하는 클래스.
  • fit과 transform 메서드를 구현하면 fit_transform도 자동으로 제공됨.

---

2. __init__ 메서드: 초기화

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.utils.validation import check_array, check_is_fitted

class StandardScalerClone(BaseEstimator, TransformerMixin):
    def __init__(self, with_mean=True):
        self.with_mean = with_mean

• __init__:
  • 이 메서드는 transformer의 constructor.
  • 여기에서 hyper-parameters를 설정.
  • 예시에서는 with_mean이라는 파라미터를 사용하여, 평균을 기준으로 데이터를 정규화할지 여부를 결정합니다.
• __init__ 생성자에서 주의사항.
  • *args나 **kwargs 같은 형태로 parameters를 제공해선 안 됨.
  • BaseEstimator를 상속받고, __init__에서 각 parameters를 명시적으로 정의해야만
  • scikit-learn의 Pipeline에서 파라미터 검색이나 Cross validation을 원활하게 수행할 수 있음.

---

3. fit 메서드: 학습 기능

def fit(self, X, y=None):
    X_orig = X
    X = check_array(X)  # X가 배열 형태인지 확인
    self.mean_ = X.mean(axis=0)
    self.scale_ = X.std(axis=0)
    self.n_features_in_ = X.shape[1]
    if hasattr(X_orig, "columns"):
        self.feature_names_in_ = np.array(X_orig.columns, dtype=object)
    return self  # 항상 self를 반환해야 함

• fit 메서드는 데이터를 기반으로 필요한 parameters 에 해당하는 instance variable을 계산하고 저장하는 역할을 수행.
• y 라벨값이 사용되지 않더라도 반드시 parameter로 있어야 함: 사용치 않을 경우를 위해 default로 None을 가지도록 처리 필요.
• 위의 예에서는 입력 데이터 X의 평균(mean_)과 표준편차(scale_)를 계산하여, 나중에 transform()가 호출될 때 사용할 수 있도록 저장.
• check_array 함수는 scikit-learn.utils.validation에서 제공하는 유틸리티 함수임.
  • 입력 데이터가 올바른 형식(배열 형태)인지 확인함.
• 학습된 transformer 의 경우 fit 메서드에서는 항상 n_features_in_ 속성(입력 데이터의 특성 개수)을 저장해야 함.
  • 이는 scikit-learn에서 모든 추정기(estimator)와 호환되도록 보장하는 중요한 규칙임.
  • n_features_in_ 의 값은 transform()과 predict()에서의 features의 수와 같아야함.
• if hasattr(X_orig, "columns")로 입력 데이터가 DataFrame일 경우 가지고 있는 각 feature의 이름을 저장.
  • hasattr는 Python 내장 함수로, 객체가 특정 속성(어트리뷰트)을 가지고 있는지를 확인하는 데 사용됨.
  • feature_names_in_ 은 입력데이터의 columns의 이름을 저장하는 ndarray 객체이며,
  • estimator의 경우에 입력이 DataFrame 객체일 경우 반드시 설정되어야 함.
• method chain을 위해 반드시 self를 반환해야 함. **

---

4. transform 메서드: 변환 기능 수행 (필수구현)

def transform(self, X):
    check_is_fitted(self)  # fit() 가 호출되어 학습이 되었는지 확인.
    X = check_array(X)     # X가 float 로 구성된 array인지 확인.
    if self.n_features_in_ != X.shape[1]:
        raise ValueError("Unexpected number of features")
    if self.with_mean:
        X = X - self.mean_
    return X / self.scale_

• transform 메서드는
  • fit()메서드를 통해 학습된 속성(mean_, scale_)을 사용하여 데이터를 변환.
  • 예시에서는 입력 데이터 X에서 평균을 빼고, 표준편차로 나누어 표준화.
• check_is_fitted 함수는 fit()메서드를 통해 학습되었는지 확인.
• 변환할 데이터가 원래 학습한 데이터와 동일한 특성 수를 가졌는지도 검증.
  • 그렇지 않으면 ValueError를 발생.

---

5. inverse_transform 메서드:  역변환 기능 (필수는 아님)

def inverse_transform(self, X):
    check_is_fitted(self)
    X = check_array(X)
    if self.n_features_in_ != X.shape[1]:
        raise ValueError("Unexpected number of features")
    X = X * self.scale_
    return X + self.mean_ if self.with_mean else X

• inverse_transform 메서드는 transform에서 변환된 데이터를 원래 상태로 복원.
• transform의 역변환임.
• 이 메서드는 데이터 전처리 후 원래 값을 복원을 수행함: 역변환이 불가한 경우엔 구현하지 않아도 됨.

---

6. get_feature_names_out 메서드: feature names 반환 (필수)**

def get_feature_names_out(self, input_features=None):
    if input_features is None:
        return getattr(self, "feature_names_in_", [f"x{i}" for i in range(self.n_features_in_)])
    else:
        if len(input_features) != self.n_features_in_:
            raise ValueError("Invalid number of features")
        if hasattr(self, "feature_names_in_") and not np.all(self.feature_names_in_ == input_features):
            raise ValueError("input_features != feature_names_in_")
        return input_features

• get_feature_names_out는 transform 후 feature names을 반환하는 메서드.
  • 이 메서드는 Pipeline 에서 feature name을 추적하거나,
  • 변환 후 데이터에 대해 분석 등을 수행할 때 사용됨.
• 위의 예제에서는 input_features가 없으면 기본적으로 column 번호를 이름으로 사용되도록 구현됨.
• 일반적으로 input_features가 주어지면 이를 columns의 이름으로 사용하는게 일반적.
  • 기본적으로 input_features 가 가지고 있는 요소의 숫자가 같아야함.
• 이는 사실상 반드시 구현해야하는 메서드임.

---

사용 예시

import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.utils.estimator_checks import check_estimator

# 예시 데이터 생성
X = np.array([[1, 2], [3, 4], [5, 6], [7, 8]])

# 트랜스포머 초기화
scaler = StandardScalerClone(with_mean=True)

# Test
print(check_estimator(scaler))

# 파이프라인에 적용
pipeline = Pipeline([
    ('scaler', scaler)
])

# 데이터 학습 및 변환
pipeline.fit(X)
X_transformed = pipeline.transform(X)
print("Transformed data:\n", X_transformed)

# 역변환
X_original = pipeline.inverse_transform(X_transformed)
print("Inverse transformed data (original):\n", X_original)

# 특성 이름 출력
print("Feature names:", scaler.get_feature_names_out())

# 특성 이름 동작을 좀더 확인.
assert np.all(scaler.get_feature_names_out() == ["x0", "x1"])
assert np.all(scaler.get_feature_names_out(["a", "b"]) == ["a", "b"])

# DataFrame과 잘 동작하나?
df = pd.DataFrame({"a": np.random.rand(100), "b": np.random.rand(100)})
scaler = StandardScalerClone()
X_scaled = scaler.fit_transform(df)

assert np.all(scaler.feature_names_in_ == ["a", "b"])
assert np.all(scaler.get_feature_names_out() == ["a", "b"])

• sklearn.utils.estimator_checks의 check_estimator()란?
  • scikit-learn에서 제공하는 유틸리티 함수로,
    Custom Estimator가 scikit-learn의 규칙을 따르고 있는지 검증.
  • custom transformer 또는 custom model 을 작성할 때,
    해당 estimator 가 scikit-learn의 호환성 기준을 만족하는지 확인.
    • 반환값은 None
    • estimator 가 scikit-learn의 규칙을 따르지 않을 경우, 예외를 발생시킴:TypeError 또는 ValueError

---

실행 결과 예시

None
Transformed data:
 [[-1.34164079 -1.34164079]
 [-0.4472136  -0.4472136 ]
 [ 0.4472136   0.4472136 ]
 [ 1.34164079  1.34164079]]
Inverse transformed data (original):
 [[1. 2.]
 [3. 4.]
 [5. 6.]
 [7. 8.]]
Feature names: ['x0', 'x1']

---

같이보면 좋은 자료들

https://gist.github.com/dsaint31x/eaa0b0103aaf47b212c20e9cb289e63d

ml_custom_transformer.ipynb

---