sklearn을 활용한 Custom Outlier Transformer 만들어보기

광고 한 번씩 눌러주세요. 블로그 운영에 큰 힘이 됩니다 :)

특정 모델을 열심히 돌려보는데, Train data은 AUC가 100%를 찍지만, Test Data는 77%를 찍고 더 이상의 성능 향상이 되지 않았다. 이러한 점을 해결하기 위해서Weight으 정규화를 하던지, 전처리 방식을 바꿔봤지만, 동일한 성능이 계속 나왔다.
그래서 남은 방법은 머 많이 있겠지만, 그중에서 Outlier 부분을 고려해보기로 하였다.
하지만 실제로 적용하였지만, 오히려 더 overfitting을 시키는 것 같다 ㅎㅎㅎ 

그래도 이왕 Outlier Transformer를 만들었으니, 공개함.

Outlier 체크하는 부분은 일반적으로 통계에서 사용하는 IQR 방식을 적용하기로 하였다.

IQR이란 Quantile(0.75) - Quantile(0.25)를 의미하고, 그 다음에 최소, 최대를 구해서 제거하는 방식이다.
제거하는 방식을 적용하는 것은 데이터의 손실을 발생시키니, 일단은 결측값으로 만들고 나서, 평균 대체 후 표준화를 하기로 하였다. 그리고 sklearn을 활용해 Pipeline으로 만들어봤다.

1. 연속형 변수만 선택
2. IQR 방식으로 Outlier NAN 처리
3. 컬럼마다 평균 대체
4. 표준화 or 정규화

그러면 이제 코드 공개
기존 Transformer에는 변수 선택이나, Outlier Transformer가 없어 보여서 만듦.

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.base import BaseEstimator
from sklearn.base import TransformerMixin
from collections import Counter
from copy import deepcopy
class FeatureSelector(BaseEstimator, TransformerMixin):
    #Class Constructor 
    def __init__(self, feature_names):
        self.feature_names = feature_names
    #Return self nothing else to do here    
    def fit(self, X, y = None):
        return self
    #Method that describes what we need this transformer to do
    def transform(self, X, y = None):
        return X[self.feature_names]
    
class OutlierTransformer(BaseEstimator, TransformerMixin ):
    #Class constructor method that takes a boolean as its argument
    def __init__(self, new_features = True , multiple_outlier_n = 5):
        self.multiple_outlier_n = multiple_outlier_n
        self.new_features = new_features
    #Return self nothing else to do here    
    def fit( self, X:pd.DataFrame, y = None )-> dict:
        self.Detect_Outlier = {}
        df = deepcopy(X)
        outlier_indices = [] 
        # convert columns to numerical
        self.totalcol = df.columns.tolist()
        print(self.totalcol)
        for name in self.totalcol :
            selectvar = df[name]
            Q1, Q3 = selectvar.quantile([0.25,0.75]).values
            IQR = Q3 - Q1
            minimum , maximum = (Q1 - 1.5 * IQR) , (Q3 + 1.5 * IQR)
            select_idx = (selectvar  < minimum ) | (selectvar > maximum)
            outlier_list_col = selectvar[select_idx].index.tolist()
            outlier_indices.extend(outlier_list_col)
            self.Detect_Outlier[name] = [minimum , maximum]
        outlier_indices = Counter(outlier_indices) 
        multiple_outliers = list(k for k, v in outlier_indices.items() if v > self.multiple_outlier_n)
        print(f"{self.multiple_outlier_n} Column Dudicated Outlier Index : {len(multiple_outliers)}")
        return self
    
    def transform(self, X:pd.DataFrame , y = None ) -> pd.DataFrame :
        df = deepcopy(X)
        for name in self.totalcol :
            selectvar = df[name]
            Q1, Q3 = selectvar.quantile([0.25,0.75]).values
            IQR = Q3 - Q1
            minimum , maximum = self.Detect_Outlier[name]
            select_idx = (selectvar  < minimum ) | (selectvar > maximum)
            selectvar[select_idx] = np.nan
            df[name] = selectvar
        print(df.shape)
        return df    
    def get_feature(self) :
        return self.totalcol

기본 원칙에 따라서 Train에 적용 후 Test에 적용

Version 1 (Feature Selector 사용)

• pandas로 결과 출력

from sklearn.pipeline import FeatureUnion
numerical_steps = [
    ("num_select", FeatureSelector(num_var)) , 
    ('Outlier', OutlierTransformer(multiple_outlier_n=5)),
    ('imputer', SimpleImputer(strategy='mean')),
    ('scaler2', StandardScaler()),
]

pipeline_list = [("num",Pipeline(numerical_steps))]
preprocessing_pipeline  = FeatureUnion(transformer_list=pipeline_list)
preprocessing_pipeline.fit(Train_X)
Train_X[num_var] = preprocessing_pipeline.transform(Train_X)
Test_X[num_var] = preprocessing_pipeline.transform(Test_X)

Version 2 (ColumnTransformer 적용)

• numpy로 출력되기 때문에 적용해줘야함.

 

numeric_pipe = Pipeline([("Outlier", OutlierTransformer(multiple_outlier_n=5)),
                         ("imputer", SimpleImputer(strategy='mean')),
                         ("scaler2", MinMaxScaler(feature_range=(-2,2)))])
col_transformer = ColumnTransformer(transformers=[("nums",numeric_pipe , num_var)],n_jobs=None)
Train_X[num_var] = col_transformer.transform(Train_X)
Test_X[num_var] = col_transformer.transform(Test_X)

여기서 범주형 변수도 같이 처리할 수 있지만, 해당 글에서는 연속형 변수만 처리하는 것을 만들어봤다.

끝