대회명 : Loan Approval Prediction
https://www.kaggle.com/competitions/playground-series-s4e10/overview
Loan Approval Prediction
Playground Series - Season 4, Episode 10
www.kaggle.com
대회 설명 : 신청자가 대출에 승인될지 예측
평가 방식: 예측 확률과 실제 목표를 사용하여 ROC 곡선 아래의 면적을 통해 평가(AUC)
수상작 구현
https://www.kaggle.com/code/trupologhelper/boosting-synergy-six-model-blend-for-loan-predict
Boosting Synergy: Six-Model Blend for Loan Predict
Explore and run machine learning code with Kaggle Notebooks | Using data from multiple data sources
www.kaggle.com
- 데이터 전처리
import sys
import matplotlib.pyplot as plt
import pandas as pd
from joblib import dump, load
from sklearn.metrics import roc_auc_score
from tqdm.auto import tqdm
train = pd.read_csv(f'{data_path}/train.csv', index_col = 0)
train['original'] = False
test = pd.read_csv(f'{data_path}/test.csv', index_col = 0)
test['original'] = False
original = pd.read_csv(f'{data_path}/credit_risk_dataset.csv')
original['original'] = True
target = 'loan_status'
train.head()
- 단순 대치 방법이 아닌, 다른 열의 관계를 활용해 결측값을 보다 정교하게 예측
- 범주형 데이터와 수치형 데이터를 결합하여 통합적으로 처리
# 필요한 라이브러리 로드
import pandas as pd
import numpy as np
from sklearn.experimental import enable_iterative_imputer
from sklearn.impute import IterativeImputer
from sklearn.ensemble import RandomForestRegressor
from catboost import CatBoostRegressor
from lightgbm import LGBMRegressor
from xgboost import XGBRegressor
from sklearn.base import BaseEstimator, RegressorMixin
# 사용자 정의 앙상블 모델: EnsembleRegressor
class EnsembleRegressor(BaseEstimator, RegressorMixin): # sklearn 워크플로우와의 호환,통합을 위해서??? 라이브러리의 패키지처럼? 진짜면 이사람 ㄹㅇ 개변태..
"""
4개의 앙상블 모델(Random Forest, CatBoost, LightGBM, XGBoost)을 결합하여
하나의 회귀 모델로 구현.
- 각 모델의 예측값을 평균내어 최종 예측값을 계산.
- Iterative Imputer에서 결측값 예측 모델로 사용.
"""
def __init__(self):
# 각 모델 초기화
self.rf = RandomForestRegressor(n_estimators=300, random_state=0, n_jobs=-1)
self.catboost = CatBoostRegressor(iterations=300, random_state=0, verbose=False, thread_count=-1)
self.lightgbm = LGBMRegressor(n_estimators=300, random_state=0, n_jobs=-1, verbosity=-1)
self.xgboost = XGBRegressor(n_estimators=300, random_state=0, n_jobs=-1)
self.models = [self.rf, self.catboost, self.lightgbm, self.xgboost]
def fit(self, X, y):
"""
각 모델을 독립적으로 학습.
- X: 학습 데이터
- y: 목표 변수
"""
for model in self.models:
model.fit(X, y) # 각 모델에 대해 학습 수행
return self
def predict(self, X):
"""
각 모델의 예측값을 계산한 뒤 평균값을 반환.
- X: 입력 데이터
"""
predictions = []
for model in self.models:
predictions.append(model.predict(X)) # 각 모델의 예측값 저장
return np.mean(predictions, axis=0) # 예측값의 평균 반환
# 1. 결측값 전처리를 위한 준비 단계
columns_to_impute = [col for col in original.columns if col not in ['loan_status', 'original']]
"""
결측값 처리를 적용할 열을 선택.
- `loan_status`와 `original` 열은 결측값 처리 대상에서 제외.
이유:
- `loan_status`: 목표 변수이므로 결측값 처리가 불필요.
- `original`: 원본 데이터 식별자, 모델 학습 불필요.
"""
# 숫자형 열과 범주형 열 구분
numeric_columns = original[columns_to_impute].select_dtypes(include=['int64', 'float64']).columns
categorical_columns = original[columns_to_impute].select_dtypes(include=['object']).columns
"""
숫자형 열과 범주형 열을 구분하는 이유:
- Iterative Imputer는 숫자형 데이터에서만 직접 동작할 수 있음.
- 따라서 범주형 데이터를 숫자로 변환한 뒤 결측값을 처리해야 함.
"""
# 결측값 처리 대상 열 복사
df_imputed = original[columns_to_impute].copy()
"""
원본 데이터를 복사하여 결측값 처리 작업을 수행.
- 원본 데이터를 안전하게 유지.
"""
# 2. 범주형 데이터를 정수로 변환
category_mappings = {}
for col in categorical_columns:
# 각 범주형 열에 대해 고유 값 -> 정수 매핑 생성
category_mappings[col] = {v: k for k, v in enumerate(df_imputed[col].unique())}
df_imputed[col] = df_imputed[col].map(category_mappings[col])
"""
범주형 데이터를 숫자로 변환:
- Iterative Imputer가 숫자형 데이터에서만 작동하므로,
범주형 데이터를 정수로 매핑하여 처리 가능하도록 변환.
- 고유 값에 정수를 매핑하는 방식은 데이터의 범주 정보를 유지.
"""
# 3. Iterative Imputer로 결측값 처리
imputer = IterativeImputer(estimator=EnsembleRegressor(), max_iter=10, random_state=0)
"""
Iterative Imputer 설정:
- `EnsembleRegressor`: 결측값 예측 모델로, 앙상블 모델을 결합하여 사용.
- `max_iter=10`: 결측값 처리의 최대 반복 횟수.
- `random_state=0`: 재현성을 보장하기 위한 난수 고정.
"""
imputed_data = imputer.fit_transform(df_imputed)
"""
결측값 처리:
- Iterative Imputer가 각 열의 결측값을 예측하여 채움.
- 다른 열의 정보를 사용해 결측값을 반복적으로 예측.
"""
# 4. 처리된 데이터프레임 생성
df_result = pd.DataFrame(imputed_data, columns=df_imputed.columns)
# 범주형 데이터를 원래 값으로 복원
for col in categorical_columns:
reverse_mapping = {v: k for k, v in category_mappings[col].items()}
df_result[col] = df_result[col].round().map(reverse_mapping)
"""
범주형 데이터 복원:
- 정수로 변환된 범주형 데이터를 원래의 범주 값으로 복원.
- `round()`를 사용해 정수로 변환 후, 매핑된 값을 역변환.
"""
# 5. 추가 열 복사 및 데이터 업데이트
df_result['loan_status'] = original['loan_status'] # 목표 변수 복사
df_result['original'] = original['original'] # 원본 데이터 식별자 복사
# 전처리 완료된 데이터로 원본 업데이트
original = df_result.copy()
# 데이터 구조 출력
original.info()
"""
전처리가 완료된 데이터:
- 결측값이 처리되었으며, 범주형 데이터는 원래 값으로 복원됨.
- 데이터의 구조와 결측값 여부를 확인.
"""
- 파생 변수(Feature)을 생성하여 머신러닝 모델의 성능을 향상
- PolynomialFeatures 이용해 다항식과 상호작용 항목을 자동으로 추가
import numpy as np
import pandas as pd
from sklearn.preprocessing import PolynomialFeatures
def create_features(train, test):
for df in [train, test]:
# 기본 비율 특성 생성
df['income_to_age'] = df['person_income'] / df['person_age']
df['loan_to_income'] = df['loan_amnt'] / df['person_income']
df['rate_to_loan'] = df['loan_int_rate'] / df['loan_amnt']
# 비선형 변환
df['age_squared'] = df['person_age'] ** 2
df['log_income'] = np.log1p(df['person_income'])
# 상호작용 특성
df['age_credit_history_interaction'] = df['person_age'] * df['cb_person_cred_hist_length']
# 연속형 데이터를 범주화
df['age_category'] = pd.cut(df['person_age'], bins=[0, 25, 35, 45, 55, 100], labels=['Very Young', 'Young', 'Middle', 'Senior', 'Elder'])
df['income_category'] = pd.qcut(df['person_income'], q=5, labels=['Very Low', 'Low', 'Medium', 'High', 'Very High'])
# 이진 플래그 생성
df['high_loan_to_income'] = (df['loan_percent_income'] > 0.5).astype(int)
df['is_new_credit_user'] = (df['cb_person_cred_hist_length'] < 2).astype(int)
df['high_interest_rate'] = (df['loan_int_rate'] > df['loan_int_rate'].mean()).astype(int)
# 범주형 데이터 상호작용 ## onehotencoding을 위해서 이렇게 했던 거구나... 맞나??
df['intent_grade_interaction'] = df['loan_intent'].astype(str) + '_' + df['loan_grade'].astype(str)
df['default_grade_interaction'] = df['cb_person_default_on_file'].astype(str) + '_' + df['loan_grade'].astype(str)
# 복잡한 계산
df['loan_to_employment'] = df['loan_amnt'] / (df['person_emp_length'] + 1)
df['risk_score'] = df['loan_percent_income'] * df['loan_int_rate'] * (5 - df['loan_grade'].map({'A':5, 'B':4, 'C':3, 'D':2, 'E':1, 'F':0, 'G':0}))
df['stability_score'] = (df['person_emp_length'] * df['person_income']) / (df['loan_amnt'] * (df['cb_person_cred_hist_length'] + 1))
# 추가 상호작용
df['age_to_credit_history'] = df['person_age'] / (df['cb_person_cred_hist_length'] + 1)
df['age_to_employment'] = df['person_age'] / (df['person_emp_length'] + 1)
# 범주형 특성 변형
df['home_ownership_intent'] = df['person_home_ownership'].astype(str) + '_' + df['loan_intent'].astype(str)
df['home_ownership_loan_interaction'] = df['person_home_ownership'].astype(str) + '_' + pd.qcut(df['loan_amnt'], q=5, labels=['Very Small', 'Small', 'Medium', 'Large', 'Very Large']).astype(str)
# 비선형 변환
df['age_sin'] = np.sin(2 * np.pi * df['person_age'] / 100)
df['age_cos'] = np.cos(2 * np.pi * df['person_age'] / 100)
# 다항식 및 상호작용 항목 생성
num_features = ['person_age', 'person_income', 'person_emp_length', 'loan_amnt', 'loan_int_rate', 'loan_percent_income']
poly = PolynomialFeatures(degree=2, include_bias=False, interaction_only=True) # 2차 다항식(degree=2)과 상호작용 항목 포함한 행렬 생성
poly_features = poly.fit_transform(df[num_features])
try:
poly_features_names = poly.get_feature_names_out(num_features)
except AttributeError: # For older versions of scikit-learn
poly_features_names = poly.get_feature_names(num_features)
for i, name in enumerate(poly_features_names[len(num_features):]):
df[f'poly_{name}'] = poly_features[:, len(num_features) + i]
return train, test
# 함수 호출로 새로운 특성 생성
train, test = create_features(train, test)
import pandas as pd
import numpy as np
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
# 1. 목표 변수 분리
target_col = train['loan_status'].reset_index(drop=True)
# 2. 학습 데이터에서 목표 변수 제거
train = train.drop('loan_status', axis=1)
# 3. 학습 데이터와 테스트 데이터를 결합
combined_data = pd.concat([train, test], axis=0, ignore_index=True)
# 4. 범주형 및 수치형 열 구분
categorical_features = combined_data.select_dtypes(include=['object']).columns
numerical_features = combined_data.select_dtypes(include=['int64', 'float64']).columns
"""
- 범주형 열(categorical_features)과 수치형 열(numerical_features)을 자동으로 탐지.
- select_dtypes: 데이터 타입을 기준으로 열을 선택.
"""
# 5. 전처리 파이프라인 구성
preprocessor = ColumnTransformer(
transformers=[
('num', StandardScaler(), numerical_features), # 수치형 열: 표준화
('cat', OneHotEncoder(sparse_output=False, handle_unknown='ignore'), categorical_features) # 범주형 열: 원-핫 인코딩
])
"""
- ColumnTransformer: 서로 다른 열에 대해 다른 전처리 방법을 적용 가능.
- StandardScaler: 수치형 데이터를 평균 0, 표준편차 1로 변환.
- OneHotEncoder:
- 범주형 데이터를 원-핫 인코딩으로 변환.
- sparse_output=False: 희소 행렬이 아닌 밀집 행렬 반환.
--> 일부 머신러닝 알고리즘은 희소 행렬을 직접 처리할 수 있음.
그러나 많은 Scikit-learn 모델과 일부 신경망 알고리즘은 밀집 행렬만 지원.(만약 toarray()를 추가적으로 사용할꺼라면 희소행렬로도 사용이 가능할듯함.)
- handle_unknown='ignore': 테스트 데이터에서 학습에 없는 새로운 범주가 나올 경우 무시.
"""
# 6. 데이터 전처리 수행
preprocessed_data = preprocessor.fit_transform(combined_data)
"""
- fit_transform: 학습 데이터를 기준으로 변환 규칙을 학습(fit)하고 변환(transform) 수행.
- 수치형 데이터는 표준화되고, 범주형 데이터는 원-핫 인코딩됨.
- 결과는 NumPy 배열 형태로 반환.
"""
# 7. 원-핫 인코딩된 범주형 열 이름 가져오기
onehot_encoder = preprocessor.named_transformers_['cat']
cat_feature_names = onehot_encoder.get_feature_names_out(categorical_features)
"""
- named_transformers_['cat']: 범주형 열에 적용된 원-핫 인코더를 가져옴.
- get_feature_names_out: 원-핫 인코딩으로 생성된 새로운 열 이름 반환.
"""
# 8. 최종 열 이름 생성
feature_names = list(numerical_features) + list(cat_feature_names)
"""
- 최종 열 이름:
- 원래 수치형 열 이름(numerical_features).
- 원-핫 인코딩으로 생성된 범주형 열 이름(cat_feature_names).
"""
# 9. 전처리된 데이터를 데이터프레임으로 변환
preprocessed_df = pd.DataFrame(preprocessed_data, columns=feature_names)
"""
- 전처리된 NumPy 배열(preprocessed_data)을 데이터프레임으로 변환.
- 열 이름은 feature_names로 설정.
"""
# 10. 원래 데이터를 보존한 열 추가
for col in combined_data.columns:
preprocessed_df[f'original_{col}'] = combined_data[col].values
"""
- 원래 데이터를 보존하기 위해 전처리 이전의 값을 새로운 열로 추가.
- 열 이름은 original_ 접두사가 붙음.
"""
# 11. 인덱스 초기화
preprocessed_df = preprocessed_df.reset_index(drop=True)
"""
- 전처리된 데이터프레임의 인덱스를 초기화.
- 데이터 병합 및 슬라이싱 시 인덱스 정합성 유지.
"""
# 12. 전처리된 데이터를 학습 및 테스트 데이터로 분리
train_preprocessed = preprocessed_df.iloc[:len(train)]
test_preprocessed = preprocessed_df.iloc[len(train):]
"""
- train_preprocessed: 학습 데이터에 해당하는 행.
- test_preprocessed: 테스트 데이터에 해당하는 행.
- 슬라이싱은 결합된 데이터프레임의 크기(len(train))를 기준으로 수행.
"""
# 13. 목표 변수 추가
train_preprocessed = pd.concat([train_preprocessed, target_col], axis=1)
"""
- 원래 저장해 둔 목표 변수(target_col)를 전처리된 학습 데이터에 추가.
"""
# 14. 결과 출력
print("Shape of preprocessed train data:", train_preprocessed.shape)
print("Shape of preprocessed test data:", test_preprocessed.shape)
"""
- 학습 데이터와 테스트 데이터의 최종 크기를 출력하여 변환 결과 확인.
"""
# 15. 전처리 결과를 원본 변수로 업데이트
train = train_preprocessed.copy()
test = test_preprocessed.copy()
"""
- 전처리된 학습 데이터와 테스트 데이터를 train 및 test로 업데이트.
"""
### 코드의 주요 목적
"""
데이터 전처리
- 학습 데이터(train)와 테스트 데이터(test)를 하나로 결합한 뒤 동일한 전처리를 적용.
- 수치형 열은 표준화(Standardization), 범주형 열은 원-핫 인코딩.
원본 데이터 보존
- 전처리된 데이터에 원래 데이터를 보존한 열을 추가.
학습 및 테스트 데이터 분리
- 결합된 데이터에서 전처리가 완료된 학습 데이터와 테스트 데이터를 다시 분리.
Scikit-learn의 파이프라인 활용
- ColumnTransformer와 Pipeline을 사용하여 전처리 과정을 모듈화.
목표 변수와 데이터 통합
- 목표 변수(loan_status)를 분리하고 전처리 후 다시 통합.
---
### 결과
- 전처리된 학습 데이터(train_preprocessed):
- 수치형 데이터는 표준화되고, 범주형 데이터는 원-핫 인코딩된 상태.
- 원래의 데이터도 보존된 열로 포함.
- 전처리된 테스트 데이터(test_preprocessed):
- 학습 데이터와 동일한 전처리 적용.
- 데이터가 모델 학습 및 평가에 적합한 형태로 준비.
"""
- 범주형 데이터 변환 : object 또는 category 열을 숫자 형태로 변환하거나 새로운 열로 생성
- 수치형 데이터 변환 : float 및 int 데이터를 범주형(category) 데이터로 변환
- 결합된 데이터 처리 : 학습 및 테스트 데이터를 결합하여 동일한 전처리를 수행한 후 다시 분리
def convert_data_types(train, test, target_column):
train = train.copy() # 원본 데이터 보존
test = test.copy()
# 타겟 변수 분리
target = train[target_column].copy()
train = train.drop(columns=[target_column])
# 학습 및 테스트 데이터 결합 (일관된 데이터 전처리 위함)
combined = pd.concat([train, test], keys=['train', 'test'])
# 새로운 열 목록
new_columns = []
for column in combined.columns:
# 범주형 데이터 처리
if combined[column].dtype == 'object' or combined[column].dtype.name == 'category':
try:
new_column = f'{column}_numeric'
combined[new_column] = pd.to_numeric(combined[column], errors='coerce') # to_numeric**를 사용하여 숫자로 변환 시도, 실패시 errors='coerce'로 NaN 처리
new_columns.append(new_column) # 변환된 열을 new_columns에 추가
except ValueError: ## 예외처리라고 보기는 좀 그렇다
pass
# 수치형 데이터 처리 (수치형 데이터를 범주형 데이터로 변환)
elif np.issubdtype(combined[column].dtype, np.number):
new_column = f'{column}_category'
if combined[column].dtype == float: # float 처리
combined[new_column] = combined[column].round().fillna(-999999).astype(int).astype('category') # 결측값(NaN)을 -999999로 대체한 후, 정수형(int)으로 변환.
combined[new_column] = combined[new_column].cat.add_categories('NaN') # 범주형(category)으로 변환 후, 'NaN' 범주를 추가
combined.loc[combined[new_column] == -999999, new_column] = 'NaN' #원래 결측값 위치를 'NaN'으로 복원
else:
combined[new_column] = combined[column].astype('category') # 나머지 int 타입 처리
new_columns.append(new_column)
# 결합된 데이터에서 학습/테스트 데이터 분리
new_train = combined.loc['train'].copy()
new_test = combined.loc['test'].copy()
# 타겟 변수 복원
new_train[target_column] = target
# 변환된 데이터 반환
return new_train, new_test
# 함수 호출
train, test = convert_data_types(train, test, target)import pandas as pd
import numpy as np
def convert_data_types(train, test, target_column):
train = train.copy()
test = test.copy()
# 타겟 변수 분리
target = train[target_column].copy()
train = train.drop(columns=[target_column])
# 학습 및 테스트 데이터 결합
combined = pd.concat([train, test], keys=['train', 'test'])
# 열 처리 함수 정의
def process_column(df, col):
if df[col].dtype == 'object': # object 처리
df[col] = df[col].fillna('unk').astype('category') # 결측값(NaN)을 'unk'로 대체
elif df[col].dtype.name == 'category': # 범주형 category 처리
if df[col].isnull().any(): #null이 없다면 pass (결측값 여부)
df[col] = df[col].cat.add_categories('unk').fillna('unk') #결측값(NaN)을 'unk'로 대체
return df
# 모든 열에 대해 전처리 수행
for column in combined.columns:
combined = process_column(combined, column)
# 학습 및 테스트 데이터로 다시 분리
new_train = combined.loc['train'].copy()
new_test = combined.loc['test'].copy()
# 타겟 변수 복원
new_train[target_column] = target
# 변환된 데이터 반환
return new_train, new_test
# 함수 호출
train, test = convert_data_types(train, test, target)
categorical_features = train.select_dtypes(include=['object', 'category']).columns.tolist()
categorical_features
모델 학습
- LightGBM을 활용한 모델 학습, 최적화 및 평가를 수행하며, Optuna를 사용해 하이퍼파라미터 최적화를 구현
import numpy as np
import optuna
import pandas as pd
from lightgbm import LGBMClassifier, early_stopping
from optuna.samplers import TPESampler
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
from typing import Dict, List, Tuple, Union
class Model_gbdt:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, categorical_feats: List[str], base_params=None):
"""
초기화 메서드: 데이터 및 기본 설정을 초기화.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- categorical_feats: 범주형 열의 리스트.
- base_params: 기본 하이퍼파라미터 설정 (옵션).
"""
if base_params is None:
base_params = {}
self.train = train # 학습 데이터
self.test = test # 테스트 데이터
self.model_dict: Dict[str, LGBMClassifier] = {} # Fold별 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # Fold별 테스트 데이터 예측 결과 저장
self.categorical_feats = categorical_feats # 범주형 열 리스트
self.target = target # 타겟 변수 열 이름
self.base_params = base_params # 기본 하이퍼파라미터 설정
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna에서 호출되는 목적 함수. 주어진 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Returns:
- AUC 점수 (float).
"""
params = {
"max_depth": trial.suggest_int("max_depth", 2, 10), # 최대 깊이
"num_leaves": trial.suggest_int("num_leaves", 2, 256), # 리프 노드 수
"learning_rate": trial.suggest_float("learning_rate", 1e-3, 0.3, log=True), # 학습률
"n_estimators": trial.suggest_int("n_estimators", 50, 1500), # 트리 수
"min_child_samples": trial.suggest_int("min_child_samples", 1, 100), # 최소 샘플 수
"feature_fraction": trial.suggest_float("feature_fraction", 0.1, 1.0), # 피처 샘플링 비율
"subsample": trial.suggest_float("subsample", 0.5, 1.0), # 데이터 샘플링 비율
"lambda_l1": trial.suggest_float("lambda_l1", 0, 1), # L1 정규화 # feature 선택(모델 단순화, 특정 feature의 importance를 0으로 만들 수 있음)
"lambda_l2": trial.suggest_float("lambda_l2", 0, 1), # L2 정규화 # 복잡도 감소(균등 규제, 과적합 방지)
**self.base_params
}
scores, _, _ = self.fit(params) # 주어진 하이퍼파라미터로 모델 학습
return np.mean(scores) # AUC 점수 반환
def fit(self, params: Dict[str, Union[int, float, str, bool, List[str]]]) -> Tuple[List[float], List[np.ndarray], np.ndarray]:
"""
모델 학습 및 교차 검증 수행. OOF 예측 및 테스트 데이터 예측 생성.
Args:
- params: LightGBM 하이퍼파라미터 설정.
Returns:
- scores: 각 fold별 AUC 점수 리스트.
- test_predict_list: 테스트 데이터 예측 결과 리스트.
- oof_valid_preds: OOF 예측 결과.
"""
label_columns = [self.target] # 타겟 변수 열 이름
train_cols = [col for col in self.train.columns.to_list() if col not in label_columns] # 입력 변수 열 이름
scores = [] # AUC 점수 저장
mskf = StratifiedKFold(n_splits=10, shuffle=True, random_state=42) # Stratified K-Fold
oof_valid_preds = np.zeros((self.train[train_cols].shape[0], 1)) # OOF 예측 결과 저장
for fold, (train_idx, valid_idx) in enumerate(mskf.split(self.train[train_cols], self.train[label_columns])): # 학습데이터와 검증데이터로 분리
# 학습 및 검증 데이터 분리
X_train, y_train = self.train[train_cols].iloc[train_idx], self.train[label_columns].iloc[train_idx].values.ravel()
X_valid, y_valid = self.train[train_cols].iloc[valid_idx], self.train[label_columns].iloc[valid_idx].values.ravel()
# LightGBM 모델 학습
model = LGBMClassifier(**params)
model.fit(X_train, y_train, eval_set=[(X_valid, y_valid)], callbacks=[early_stopping(250)])
# 검증 데이터 예측 및 저장
valid_preds = model.predict_proba(X_valid)[:, 1]
oof_valid_preds[valid_idx] = valid_preds.reshape(-1, 1)
# 테스트 데이터 예측 및 저장
test_predict = model.predict_proba(self.test[train_cols])[:, 1]
self.test_predict_list.append(test_predict)
# 모델 저장
self.model_dict[f'fold_{fold}'] = model
# OOF AUC 점수 계산
oof_score = roc_auc_score(self.train[label_columns], oof_valid_preds)
scores.append(oof_score)
print(f'The average ROC AUC score is {np.mean(scores)}')
return scores, self.test_predict_list, oof_valid_preds
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str, bool]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- best_params: 최적의 하이퍼파라미터 설정.
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # AUC 점수 최대화를 목표로 설정
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 수행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적의 AUC 점수
print(" Params:") # 최적의 하이퍼파라미터
for key, value in trial.params.items():
print(f" {key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환
- goss 모델 학습
class Model_goss:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, categorical_feats: List[str],
base_params=None):
"""
클래스 초기화: 데이터를 받아서 모델 학습 환경을 설정.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- categorical_feats: 범주형 열 이름 리스트.
- base_params: 기본 하이퍼파라미터 설정 (선택).
"""
self.train = train
self.test = test
self.model_dict: Dict[str, LGBMClassifier] = {} # fold별로 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # 테스트 데이터 예측 결과 저장
self.categorical_feats = categorical_feats # 범주형 열 리스트
self.target = target # 타겟 변수 열 이름
if base_params is None: # 가변객체의 값이 비었을 경우를 처리
base_params = {}
self.base_params = base_params # LightGBM의 기본 하이퍼파라미터 설정.
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna의 목적 함수: 주어진 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Returns:
- AUC 점수 (float).
"""
params = {
"max_depth": trial.suggest_int("max_depth", 2, 10), # 최대 깊이
"num_leaves": trial.suggest_int("num_leaves", 2, 256), # 리프 노드 수
"learning_rate": trial.suggest_float("learning_rate", 1e-3, 0.3, log=True), # 학습률
"n_estimators": trial.suggest_int("n_estimators", 50, 1500), # 트리 수
"feature_fraction": trial.suggest_float("feature_fraction", 0.1, 1.0), # 피처 샘플링 비율
"subsample": trial.suggest_float("subsample", 0.5, 1.0), # 데이터 샘플링 비율
**self.base_params # 기본 하이퍼파라미터 병합
}
scores, _, _ = self.fit(params) # 하이퍼파라미터로 모델 학습
return np.mean(scores) # AUC 점수 반환
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str, bool]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- 최적의 하이퍼파라미터 설정 (Dict).
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # 최대화 목표
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 실행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적 AUC 점수
print(" Params:")
for key, value in trial.params.items():
print(f" {key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환
- catboost 모델 학습
import numpy as np
import optuna
import pandas as pd
from catboost import CatBoostClassifier
from optuna.samplers import TPESampler
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
from typing import Dict, List, Tuple, Union
class Model_loss:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, categorical_feats: List[str], base_params=None):
"""
CatBoost 모델 학습 및 최적화를 위한 클래스 초기화.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- categorical_feats: 범주형 열 이름 리스트.
- base_params: 기본 하이퍼파라미터 설정 (선택).
"""
self.train = train
self.test = test
self.model_dict: Dict[str, CatBoostClassifier] = {} # Fold별 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # 각 fold별 테스트 데이터 예측 저장
self.target = target
self.categorical_feats = categorical_feats
if base_params is None:
base_params = {}
self.base_params = base_params # 기본 하이퍼파라미터 설정
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna를 통해 주어진 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Returns:
- AUC 점수 (float).
"""
params = {
'iterations': trial.suggest_int('iterations', 50, 1500), # 학습 반복 수
'learning_rate': trial.suggest_float('learning_rate', 1e-3, 0.3, log=True), # 학습률
**self.base_params # 기본 하이퍼파라미터 병합
}
scores, _, _ = self.fit(params) # 모델 학습 및 검증 수행
return np.mean(scores) # AUC 점수 반환
def fit(self, params: Dict[str, Union[int, float, str]]) -> Tuple[List[float], List[np.ndarray], np.ndarray]:
"""
CatBoost 모델 학습 및 교차 검증 수행.
Args:
- params: CatBoost 하이퍼파라미터 설정.
Returns:
- scores: 각 fold별 AUC 점수 리스트.
- test_predict_list: 테스트 데이터 예측 결과 리스트.
- oof_valid_preds: OOF 예측 결과.
"""
target_columns = [self.target]
train_cols = [col for col in self.train.columns.to_list() if col not in target_columns]
scores = [] # AUC 점수 저장
kf = StratifiedKFold(n_splits=10, shuffle=True, random_state=42) # Stratified K-Fold 교차 검증
oof_valid_preds = np.zeros((self.train[train_cols].shape[0], 1)) # OOF 예측 저장 공간
for fold, (train_idx, valid_idx) in enumerate(kf.split(self.train[train_cols], self.train[target_columns])):
# 학습 및 검증 데이터 분리
X_train, y_train = self.train[train_cols].iloc[train_idx], self.train[target_columns].iloc[train_idx]
X_valid, y_valid = self.train[train_cols].iloc[valid_idx], self.train[target_columns].iloc[valid_idx]
# CatBoost 모델 학습
model = CatBoostClassifier(**params)
model.fit(X_train, y_train, eval_set=(X_valid, y_valid), early_stopping_rounds=250, verbose=100,
cat_features=self.categorical_feats)
# 검증 데이터 예측
valid_preds = model.predict_proba(X_valid)[:, 1]
oof_valid_preds[valid_idx] = valid_preds.reshape(-1, 1)
# 테스트 데이터 예측
test_predict = model.predict_proba(self.test[train_cols])[:, 1]
self.test_predict_list.append(test_predict)
# 모델 저장
self.model_dict[f'fold_{fold}'] = model
# OOF AUC 점수 계산
oof_score = roc_auc_score(self.train[target_columns], oof_valid_preds)
scores.append(oof_score)
print(f'The average ROC AUC score is {np.mean(scores)}')
return scores, self.test_predict_list, oof_valid_preds
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- 최적의 하이퍼파라미터 설정 (Dict).
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # AUC 점수 최대화 목표
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 수행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적 AUC 점수
print(" Params:")
for key, value in trial.params.items():
print(f" {key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환
- 분류모델 수행
import numpy as np
import optuna
import pandas as pd
from catboost import CatBoostClassifier
from optuna.samplers import TPESampler
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
from typing import Dict, List, Tuple, Union
#분류모델 수행
class Model_sym:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, categorical_feats: List[str], base_params=None):
"""
CatBoost 모델 학습 및 최적화를 위한 클래스 초기화.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- categorical_feats: 범주형 열 이름 리스트.
- base_params: 기본 하이퍼파라미터 설정 (선택).
"""
self.train = train
self.test = test
self.model_dict: Dict[str, CatBoostClassifier] = {} # Fold별 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # 각 fold별 테스트 데이터 예측 저장
self.target = target # 타겟 변수
self.categorical_feats = categorical_feats # 범주형 피처
if base_params is None:
base_params = {}
self.base_params = base_params # 기본 하이퍼파라미터 설정
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna 목적 함수: 주어진 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Returns:
- AUC 점수 (float).
"""
params = {
'iterations': trial.suggest_int('iterations', 50, 1500), # 학습 반복 수
'learning_rate': trial.suggest_float('learning_rate', 1e-3, 0.3, log=True), # 학습률
**self.base_params # 기본 하이퍼파라미터 병합
}
scores, _, _ = self.fit(params) # 모델 학습 및 검증 수행
return np.mean(scores) # AUC 점수 반환
def fit(self, params: Dict[str, Union[int, float, str]]) -> Tuple[List[float], List[np.ndarray], np.ndarray]:
"""
CatBoost 모델 학습 및 교차 검증 수행.
Args:
- params: CatBoost 하이퍼파라미터 설정.
Returns:
- scores: 각 fold별 AUC 점수 리스트.
- test_predict_list: 테스트 데이터 예측 결과 리스트.
- oof_valid_preds: OOF 예측 결과.
"""
target_columns = [self.target]
train_cols = [col for col in self.train.columns.to_list() if col not in target_columns]
scores = [] # AUC 점수 저장
kf = StratifiedKFold(n_splits=10, shuffle=True, random_state=42) # Stratified K-Fold 교차 검증
oof_valid_preds = np.zeros((self.train[train_cols].shape[0], 1)) # OOF 예측 저장 공간
for fold, (train_idx, valid_idx) in enumerate(kf.split(self.train[train_cols], self.train[target_columns])):
# 학습 및 검증 데이터 분리
X_train, y_train = self.train[train_cols].iloc[train_idx], self.train[target_columns].iloc[train_idx]
X_valid, y_valid = self.train[train_cols].iloc[valid_idx], self.train[target_columns].iloc[valid_idx]
# CatBoost 모델 학습
model = CatBoostClassifier(**params)
model.fit(X_train, y_train, eval_set=(X_valid, y_valid), early_stopping_rounds=250, verbose=100,
cat_features=self.categorical_feats)
# 검증 데이터 예측
valid_preds = model.predict_proba(X_valid)[:, 1]
oof_valid_preds[valid_idx] = valid_preds.reshape(-1, 1)
# 테스트 데이터 예측
test_predict = model.predict_proba(self.test[train_cols])[:, 1]
self.test_predict_list.append(test_predict)
# 모델 저장
self.model_dict[f'fold_{fold}'] = model
# OOF AUC 점수 계산
oof_score = roc_auc_score(self.train[target_columns], oof_valid_preds)
scores.append(oof_score)
print(f'The average ROC AUC score is {np.mean(scores)}')
return scores, self.test_predict_list, oof_valid_preds
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- 최적의 하이퍼파라미터 설정 (Dict).
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # AUC 점수 최대화 목표
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 수행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적 AUC 점수
print(" Params:")
for key, value in trial.params.items():
print(f" {key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환import numpy as np
import optuna
import pandas as pd
from catboost import CatBoostClassifier
from optuna.samplers import TPESampler
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
from typing import Dict, List, Tuple, Union
## depth-> tree model.
class Model_depth:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, categorical_feats: List[str], base_params=None):
"""
CatBoost 모델 학습 및 최적화를 위한 클래스 초기화.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- categorical_feats: 범주형 열 이름 리스트.
- base_params: 기본 하이퍼파라미터 설정 (선택).
"""
self.train = train
self.test = test
self.model_dict: Dict[str, CatBoostClassifier] = {} # Fold별 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # 각 fold별 테스트 데이터 예측 저장
self.target = target # 타겟 변수
self.categorical_feats = categorical_feats # 범주형 피처
if base_params is None:
base_params = {}
self.base_params = base_params # 기본 하이퍼파라미터 설정
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna : 주어진 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Returns:
- AUC 점수 (float).
"""
params = {
'iterations': trial.suggest_int('iterations', 50, 1500), # 학습 반복 수
'learning_rate': trial.suggest_float('learning_rate', 1e-3, 0.3, log=True), # 학습률
**self.base_params # 기본 하이퍼파라미터 병합
}
scores, _, _ = self.fit(params) # 모델 학습 및 검증 수행
return np.mean(scores) # AUC 점수 반환
def fit(self, params: Dict[str, Union[int, float, str]]) -> Tuple[List[float], List[np.ndarray], np.ndarray]:
"""
CatBoost 모델 학습 및 교차 검증 수행.
Args:
- params: CatBoost 하이퍼파라미터 설정.
Returns:
- scores: 각 fold별 AUC 점수 리스트.
- test_predict_list: 테스트 데이터 예측 결과 리스트.
- oof_valid_preds: OOF 예측 결과.
"""
target_columns = [self.target]
train_cols = [col for col in self.train.columns.to_list() if col not in target_columns]
scores = [] # AUC 점수 저장
kf = StratifiedKFold(n_splits=10, shuffle=True, random_state=42) # Stratified K-Fold 교차 검증
oof_valid_preds = np.zeros((self.train[train_cols].shape[0], 1)) # OOF 예측 저장 공간
for fold, (train_idx, valid_idx) in enumerate(kf.split(self.train[train_cols], self.train[target_columns])):
# 학습 및 검증 데이터 분리
X_train, y_train = self.train[train_cols].iloc[train_idx], self.train[target_columns].iloc[train_idx]
X_valid, y_valid = self.train[train_cols].iloc[valid_idx], self.train[target_columns].iloc[valid_idx]
# CatBoost 모델 학습
model = CatBoostClassifier(**params)
model.fit(X_train, y_train, eval_set=(X_valid, y_valid), early_stopping_rounds=250, verbose=100,
cat_features=self.categorical_feats)
# 검증 데이터 예측
valid_preds = model.predict_proba(X_valid)[:, 1]
oof_valid_preds[valid_idx] = valid_preds.reshape(-1, 1)
# 테스트 데이터 예측
test_predict = model.predict_proba(self.test[train_cols])[:, 1]
self.test_predict_list.append(test_predict)
# 모델 저장
self.model_dict[f'fold_{fold}'] = model
# OOF AUC 점수 계산
oof_score = roc_auc_score(self.train[target_columns], oof_valid_preds)
scores.append(oof_score)
print(f'The average ROC AUC score is {np.mean(scores)}')
return scores, self.test_predict_list, oof_valid_preds
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- 최적의 하이퍼파라미터 설정 (Dict).
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # AUC 점수 최대화 목표
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 수행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적 AUC 점수
print(" Params:")
for key, value in trial.params.items():
print(f"{key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환import numpy as np
import optuna
import pandas as pd
from optuna.samplers import TPESampler
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
from typing import Dict, List, Tuple, Union
from xgboost import XGBClassifier
class Model_gbtree:
def __init__(self, train: pd.DataFrame, test: pd.DataFrame, target: str, base_params=None):
"""
XGBoost 모델 학습 및 최적화를 위한 클래스 초기화.
Args:
- train: 학습 데이터 (Pandas DataFrame).
- test: 테스트 데이터 (Pandas DataFrame).
- target: 타겟 변수의 열 이름.
- base_params: 기본 하이퍼파라미터 설정 (선택).
"""
self.train = train
self.test = test
self.model_dict: Dict[str, XGBClassifier] = {} # Fold별 학습된 모델 저장
self.test_predict_list: List[np.ndarray] = [] # 각 fold별 테스트 데이터 예측 저장
self.target = target # 타겟 변수
if base_params is None:
base_params = {}
self.base_params = base_params # 기본 하이퍼파라미터 설정
def objective(self, trial: optuna.Trial) -> float:
"""
Optuna 하이퍼파라미터로 모델 학습 후 AUC 점수를 반환.
Args:
- trial: Optuna의 Trial 객체.
Return:
- AUC 점수 (float).
"""
params = {
"learning_rate": trial.suggest_float("learning_rate", 1e-3, 0.3, log=True), # 학습률
"n_estimators": trial.suggest_int("n_estimators", 50, 1500), # 트리 수
**self.base_params # 기본 하이퍼파라미터 병합
}
scores, _, _ = self.fit(params) # 하이퍼파라미터로 모델 학습
return np.mean(scores) # AUC 점수 반환
def fit(self, params: Dict[str, Union[int, float, str, bool]]) -> Tuple[List[float], List[np.ndarray], np.ndarray]:
"""
모델 학습 및 교차 검증 수행.
Args:
- params: XGBoost 하이퍼파라미터 설정.
Returns:
- scores: 각 fold별 AUC 점수 리스트.
- test_predict_list: 테스트 데이터 예측 결과 리스트.
- oof_valid_preds: OOF 예측 결과.
"""
label_columns = [self.target] # 타겟 변수 열 이름
train_cols = [col for col in self.train.columns.to_list() if col not in label_columns] # 입력 변수 열 이름
scores = [] # AUC 점수 저장
mskf = StratifiedKFold(n_splits=10, shuffle=True, random_state=42) # Stratified K-Fold 교차 검증
oof_valid_preds = np.zeros((self.train[train_cols].shape[0], 1)) # OOF 예측 저장 공간
for fold, (train_idx, valid_idx) in enumerate(mskf.split(self.train[train_cols], self.train[label_columns])):
# 학습 및 검증 데이터 분리
X_train, y_train = self.train[train_cols].iloc[train_idx], self.train[label_columns].iloc[train_idx]
X_valid, y_valid = self.train[train_cols].iloc[valid_idx], self.train[label_columns].iloc[valid_idx]
# XGBoost 모델 학습
model = XGBClassifier(**params)
model.fit(X_train, y_train, eval_set=[(X_valid, y_valid)], verbose=False)
# 검증 데이터 예측
valid_preds = model.predict_proba(X_valid)[:, 1]
oof_valid_preds[valid_idx] = valid_preds.reshape(-1, 1)
# 테스트 데이터 예측
test_predict = model.predict_proba(self.test[train_cols])[:, 1]
self.test_predict_list.append(test_predict)
# 모델 저장
self.model_dict[f'fold_{fold}'] = model
# OOF AUC 점수 계산
oof_score = roc_auc_score(self.train[label_columns], oof_valid_preds)
scores.append(oof_score)
print(f'The average ROC AUC score is {np.mean(scores)}')
return scores, self.test_predict_list, oof_valid_preds #각 fold의 AUC 점수, 데이터 예측값, oof 검증데이터 예측값
def optimize(self, n_trials: int = 100) -> Dict[str, Union[int, float, str, bool]]:
"""
Optuna를 사용하여 하이퍼파라미터 최적화 수행 -> AUC 점수 최대화.
Args:
- n_trials: 최적화 반복 횟수.
Returns:
- 최적의 하이퍼파라미터 설정 (Dict).
"""
study = optuna.create_study(direction="maximize", sampler=TPESampler(seed=42)) # AUC 점수 최대화 목표
study.optimize(self.objective, timeout=n_trials, show_progress_bar=True) # 최적화 수행
print("Best trial:")
trial = study.best_trial
print(" Value:", trial.value) # 최적 AUC 점수
print(" Params:")
for key, value in trial.params.items():
print(f" {key}: {value}")
return study.best_params # 최적 하이퍼파라미터 반환
- 학습 수행 및 config 관련 함수화
from typing import Any, List, Optional, Tuple
import os
from joblib import dump, load
import torch
# get_model_configs: 모델 구성과 기본 하이퍼파라미터를 반환
def get_model_configs(device: str, categorical_feats: List[str], num_gpus: int) -> List[Tuple[Any, dict, bool, str]]:
"""
모델 구성 및 기본 하이퍼파라미터 반환.
Args:
- device: 'cpu' 또는 'gpu' (사용할 디바이스).
- categorical_feats: 범주형 피처 목록.
- num_gpus: 사용 가능한 GPU의 수.
Returns:
- 모델 클래스, 하이퍼파라미터, 범주형 피처 사용 여부, 모델 이름을 담은 튜플 리스트.
"""
base_config = {
'cpu': {
'lgb': {"random_state": 42, "n_jobs": -1, "metric": "auc", 'cat_features': categorical_feats, 'verbosity': -1},
'cat': {'task_type': 'CPU', 'devices': '0', 'random_seed': 42, 'eval_metric': 'AUC', 'thread_count': -1},
'xgb': {"random_state": 42, "n_jobs": -1, "objective": "binary:logistic", "eval_metric": "auc", 'enable_categorical': True, 'early_stopping_rounds': 250}
},
'gpu': {
'lgb': {"random_state": 42, "n_jobs": -1, "metric": "auc", 'cat_features': categorical_feats, 'verbosity': -1, 'device': 'gpu'},
'cat': {'task_type': 'GPU', 'devices': ':'.join(map(str, range(num_gpus))), 'random_seed': 42, 'eval_metric': 'AUC', 'thread_count': -1},
'xgb': {"random_state": 42, "n_jobs": -1, "objective": "binary:logistic", "eval_metric": "auc", 'enable_categorical': True, 'early_stopping_rounds': 250}
}
}
return [
(Model_goss, {**base_config[device]['lgb'], "boosting_type": 'goss'}, True, 'lightgbm_goss'),
(Model_gbdt, {**base_config[device]['lgb'], "boosting_type": 'gbdt'}, True, 'lightgbm_gbdt'),
(Model_loss, {**base_config[device]['cat'], 'grow_policy': 'Lossguide'}, True, 'catboost_lossguide'),
(Model_sym, {**base_config[device]['cat'], 'grow_policy': 'SymmetricTree'}, True, 'catboost_symmetric'),
(Model_depth, {**base_config[device]['cat'], 'grow_policy': 'Depthwise'}, True, 'catboost_depthwise'),
(Model_gbtree, {**base_config[device]['xgb'], "booster": "gbtree"}, False, 'xgboost')
]
# model_optimization: 각 모델에 대해 최적화 및 학습을 수행하고, 결과를 저장
def model_optimization(train: Any, test: Any, target: str, categorical_feats: List[str], time_per_model: int, device: str) -> None:
"""
여러 모델에 대해 하이퍼파라미터 최적화 및 학습 수행.
Args:
- train: 학습 데이터셋.
- test: 테스트 데이터셋.
- target: 타겟 변수 열 이름.
- categorical_feats: 범주형 피처 목록.
- time_per_model: 각 모델에 대해 할당할 최적화 시간.
- device: 'cpu' 또는 'gpu' (사용 디바이스).
"""
# 사용 가능한 GPU 개수 확인
num_gpus = torch.cuda.device_count() if torch.cuda.is_available() else 0
print(f"Number of available GPUs: {num_gpus}")
# 모델 구성과 기본 하이퍼파라미터 불러오기
models = get_model_configs(device, categorical_feats, num_gpus)
# 중요 피처 로드
n_imp_features = {model_type: load(os.path.join(base_path, 'features', f'n_imp_features_{model_type.lower()}.joblib')) for _, _, _, model_type in models}
# 출력 디렉토리 생성
os.makedirs(os.path.join(base_path, 'preds'), exist_ok=True)
os.makedirs(os.path.join(base_path, 'params'), exist_ok=True)
# 각 모델에 대해 최적화 및 학습 수행
for i, (model_cls, base_params, use_cat_feats, model_type) in enumerate(models):
print(f"Model {i} ({model_type}) optimization started.")
# 중요 피처 선택
selected_features = n_imp_features[model_type]
train_selected = train[selected_features + [target]]
test_selected = test[selected_features]
cat_feats_selected = list(set(categorical_feats) & set(selected_features)) if use_cat_feats else None
# 모델 초기화 및 최적화 수행
if use_cat_feats:
opt_model = model_cls(train_selected, test_selected, target, cat_feats_selected, base_params)
else:
opt_model = model_cls(train_selected, test_selected, target, base_params)
best_params = opt_model.optimize(n_trials=time_per_model) # 하이퍼파라미터 최적화
dump(best_params, os.path.join(base_path, 'params', f'best_params_{i}.joblib')) # 최적화 결과 저장
# 최적화된 파라미터로 모델 학습
if use_cat_feats:
train_model = model_cls(train_selected, test_selected, target, cat_feats_selected)
else:
train_model = model_cls(train_selected, test_selected, target)
print(base_params)
print(best_params)
# 최적화된 파라미터 병합 후 모델 학습
best_params.update(base_params)
scores, preds, oof_valid_preds = train_model.fit(best_params)
score = np.mean(scores)
# 학습 결과 저장
for item, filename in zip([score, preds, oof_valid_preds, train_model],
[f'score{i}.joblib', f'preds_{i}.joblib', f'oof_valid_preds_{i}.joblib', f'model_{i}.joblib']):
dump(item, os.path.join(base_path, 'preds', filename))
print(f"Model {i} ({model_type}) optimization completed. Mean ROC AUC score: {score}")
print(f"Number of selected features: {len(selected_features)}")
print(f"Selected features: {selected_features}")
- blending 수행을 통한 최종 예측값 생성
import os
from functools import partial
import optuna
import numpy as np
from sklearn.metrics import roc_auc_score
from typing import Any, List, Optional, Tuple
# base_path = '/kaggle/input/' # 데이터 경로 설정
base_path = 'G:/--backup--/Desktop/zero-base/data/loan_approval_data' # 데이터 경로 설정
def blending(train: Any, target: str, n_trials: int) -> Tuple[np.ndarray, List[Any], np.ndarray]:
"""
모델 블렌딩을 수행하여 최종 예측값 생성.
Args:
- train: 학습 데이터셋.
- target: 타겟 변수 이름.
- n_trials: Optuna 최적화 반복 횟수.
Returns:
- final_predictions: 블렌딩된 최종 예측값.
- weights: 각 모델의 최적 가중치.
"""
# 총 6개의 모델 인덱스를 정의 (사전 학습된 모델이 저장되어 있다고 가정)
indices = range(6)
# # OOF(Out-of-Fold) 검증 예측값 및 테스트 예측값 로드
# oof_valid_preds = [load(os.path.join(base_path, 'preds-loan', f'oof_valid_preds_{i}.joblib')) for i in indices]
# preds = [load(os.path.join(base_path, 'preds-loan', f'preds_{i}.joblib')) for i in indices]
oof_valid_preds = [load(os.path.join(base_path, 'preds', f'oof_valid_preds_{i}.joblib')) for i in indices]
preds = [load(os.path.join(base_path, 'preds', f'preds_{i}.joblib')) for i in indices]
# OptunaWeights 객체 생성 및 가중치 최적화
ow = OptunaWeights(random_state=42, n_trials=n_trials)
ow.fit(train[target], y_preds=oof_valid_preds)
# 선택된 모델의 예측값만 블렌딩에 포함
selected_predictions = [pred for pred, selected in zip(preds, ow.selected_preds) if selected]
# 가중치와 예측값을 이용해 최종 블렌딩 예측값 계산
final_predictions = np.sum([weight * np.mean(pred, axis=0) for weight, pred in zip(ow.weights, selected_predictions)], axis=0)
return final_predictions, np.array(ow.weights)class OptunaWeights:
def __init__(self, random_state: int, n_trials: int = 5000):
"""
Optuna를 사용하여 모델 가중치를 최적화하기 위한 클래스.
Args:
- random_state: 랜덤 시드 값.
- n_trials: Optuna 최적화 반복 횟수.
"""
self.study: Optional[optuna.Study] = None # Optuna 학습 스터디 객체
self.weights: Optional[List[float]] = None # 최적화된 가중치
self.random_state = random_state
self.n_trials = n_trials
self.selected_preds: Optional[List[bool]] = None # 선택된 모델 여부
def _objective(self, trial: optuna.Trial, y_true: np.ndarray, y_preds: List[np.ndarray]) -> float:
"""
Optuna 목적 함수: 최적 가중치를 탐색.
Args:
- trial: Optuna의 Trial 객체.
- y_true: 실제 타겟 값.
- y_preds: 모델 예측값 리스트.
Returns:
- AUC 점수.
"""
# 각 모델에 대한 가중치를 0~1 사이에서 탐색
weights = [trial.suggest_float(f"weight{n}", 0, 1) for n in range(len(y_preds))]
# 선택 여부 (현재 모든 모델을 선택함)
selected_preds = [True for n in range(len(y_preds))]
# 선택된 가중치와 예측값 필터링
selected_weights = [w for w, s in zip(weights, selected_preds) if s]
weight_sum = sum(selected_weights)
if weight_sum == 0:
return 0.0
# 가중치 정규화
norm_weights = [w / weight_sum for w in selected_weights]
# 선택된 예측값에 대해 가중 평균
selected_y_preds = [pred for pred, s in zip(y_preds, selected_preds) if s]
weighted_pred = np.average(np.array(selected_y_preds), axis=0, weights=norm_weights)
# AUC 점수 반환
return roc_auc_score(y_true, weighted_pred)
def fit(self, y_true: np.ndarray, y_preds: List[np.ndarray]) -> None:
"""
Optuna를 사용하여 모델 가중치를 최적화.
Args:
- y_true: 실제 타겟 값.
- y_preds: 모델 예측값 리스트.
"""
optuna.logging.set_verbosity(optuna.logging.ERROR) # Optuna 로그 레벨 설정
self.study = optuna.create_study(
sampler=optuna.samplers.CmaEsSampler(seed=self.random_state), # CMA-ES 샘플러 사용
pruner=optuna.pruners.HyperbandPruner(), # 학습 조기 종료
study_name="OptunaWeights",
direction='maximize' # AUC 점수 최대화 목표
)
# Partial 함수로 목적 함수에 고정된 인자 전달
objective_partial = partial(self._objective, y_true=y_true, y_preds=y_preds)
self.study.optimize(objective_partial, n_trials=self.n_trials, show_progress_bar=True)
# 최적 가중치 및 선택된 모델 저장
weights = [self.study.best_params[f"weight{n}"] for n in range(len(y_preds))]
self.selected_preds = [True for n in range(len(y_preds))] # 모든 모델 선택
selected_weights = [w for w, s in zip(weights, self.selected_preds) if s]
weight_sum = sum(selected_weights)
if weight_sum == 0:
raise ValueError("All weights are zero. Unable to normalize.")
self.weights = [w / weight_sum for w in selected_weights] # 정규화된 가중치 저장final_pred, final_weights = blending(train, target, 2200)
Lesson Learned
나름 다양한 코드를 보았지만, 이번 수상작 구현은 상당히 흥미롭고(사실 어렵고) 배울점이 많았다.
class의 타입 지정과 구성을 보면 마치 python 관련 라이브러리를 만드는것 같은 느낌을 받았을 정도로 난이도 있는 코드였다고 생각한다.
해당 kaggle notebook에 약간의 (dir관련 정도?) 오류가 존재하지만 그걸 감안하더라도 상당히 난이도 높은 코드이었고, 그렇기에 배울점 또한 많았다.
'kaggle study' 카테고리의 다른 글
| Regression of Used Car Prices (4) | 2024.12.09 |
|---|---|
| Big Data Derby 2022 (3) | 2024.12.05 |
| Data Science and MLOps Landscape in Industry (0) | 2024.12.03 |
