개요
- 취업 준비생 들에게 필요한 캐글 연습 코드 클래스로 구현함
- 클래스에 대한 기본적인 이해가 있다는 전제하에 작성
- 전체 코드는 다음과 같다.
import numpy as np
import pandas as pd
import shap
import matplotlib.pyplot as plt
# 데이터 처리 및 모델링 라이브러리
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler, LabelEncoder, OrdinalEncoder
from sklearn.impute import SimpleImputer
from sklearn.metrics import (
mean_squared_error, mean_absolute_error,
r2_score, mean_absolute_percentage_error
)
from sklearn.ensemble import RandomForestRegressor
# 부스팅 모델
import xgboost as xgb
import lightgbm as lgb
from catboost import CatBoostRegressor
# Matplotlib 설정
plt.rcParams.update({'font.size': 9}) # 원하는 크기로 설정 (예: 10)
class DataPreprocessor:
"""
데이터 전처리 클래스
"""
def __init__(self, train_path, test_path):
self.train_data = pd.read_csv(train_path)
self.test_data = pd.read_csv(test_path)
def preprocess(self, target_column):
"""
데이터 전처리 메서드
- 결측치 처리 전략 개선
- 문자열 컬럼 Ordinal Encoding
"""
# 타겟 분리
X = self.train_data.drop(columns=[target_column])
y = self.train_data[target_column]
X_test = self.test_data.copy()
# 결측치 처리 전략 개선
def handle_missing_values(df):
# 숫자형 변수 - 중앙값 대체
numeric_columns = df.select_dtypes(include=['int64', 'float64']).columns
df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].median())
# 범주형 변수 - 최빈값 대체
categorical_columns = df.select_dtypes(include=['object']).columns
df[categorical_columns] = df[categorical_columns].fillna(df[categorical_columns].mode().iloc[0])
return df
# 결측치 처리
X = handle_missing_values(X)
X_test = handle_missing_values(X_test)
# 문자열 컬럼 식별
categorical_columns = X.select_dtypes(include=['object']).columns
# Ordinal Encoder 초기화
ordinal_encoder = OrdinalEncoder(
handle_unknown='use_encoded_value',
unknown_value=-1
)
# 훈련 및 테스트 데이터에 인코딩 적용
if len(categorical_columns) > 0:
# 전체 데이터 결합하여 인코딩
combined_categorical = pd.concat([X[categorical_columns], X_test[categorical_columns]])
# 인코더 훈련
ordinal_encoder.fit(combined_categorical)
# 훈련 데이터 인코딩
X[categorical_columns] = ordinal_encoder.transform(X[categorical_columns])
# 테스트 데이터 인코딩
X_test[categorical_columns] = ordinal_encoder.transform(X_test[categorical_columns])
return X, y, X_test, ordinal_encoder
class EvaluationMetrics:
"""
모델 평가 지표 클래스
"""
@staticmethod
def rmse(y_true, y_pred):
return np.sqrt(mean_squared_error(y_true, y_pred))
@staticmethod
def mae(y_true, y_pred):
return mean_absolute_error(y_true, y_pred)
@staticmethod
def r2(y_true, y_pred):
return r2_score(y_true, y_pred)
@staticmethod
def mape(y_true, y_pred):
return mean_absolute_percentage_error(y_true, y_pred)
class SHAPExplainer:
"""
SHAP 설명 및 시각화 클래스
"""
def __init__(self, model, X):
self.model = model
self.X = X
def tree_explainer(self):
"""트리 기반 모델용 SHAP 설명"""
explainer = shap.TreeExplainer(self.model)
shap_values = explainer.shap_values(self.X)
return explainer, shap_values
def plot_feature_importance(self, shap_values, feature_names):
"""특성 중요도 플롯"""
plt.figure(figsize=(10, 6))
shap.summary_plot(shap_values, self.X, feature_names=feature_names)
plt.title("SHAP Feature Importance")
plt.tight_layout()
plt.show()
class BaseMLDL:
"""
기본 ML/DL 모델 베이스 클래스
"""
def __init__(self, X, y, test_size=0.2, random_state=42):
# 데이터 분할
self.X_train, self.X_val, self.y_train, self.y_val = train_test_split(
X, y, test_size=test_size, random_state=random_state
)
# 스케일링
self.scaler = StandardScaler()
self.X_train_scaled = self.scaler.fit_transform(self.X_train)
self.X_val_scaled = self.scaler.transform(self.X_val)
self.model = None
self.predictions = None
def train(self):
raise NotImplementedError("하위 클래스에서 구현해야 합니다.")
def predict(self):
raise NotImplementedError("하위 클래스에서 구현해야 합니다.")
def evaluate(self):
"""모델 평가"""
self.predictions = self.predict()
return {
'RMSE': EvaluationMetrics.rmse(self.y_val, self.predictions),
'MAE': EvaluationMetrics.mae(self.y_val, self.predictions),
'R2': EvaluationMetrics.r2(self.y_val, self.predictions),
'MAPE': EvaluationMetrics.mape(self.y_val, self.predictions)
}
# 각 모델 클래스 구현 (RandomForest, XGBoost, LightGBM, CatBoost 등)
class RandomForestModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = RandomForestRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class XGBoostModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = xgb.XGBRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class LightGBMModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = lgb.LGBMRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class CatBoostModel(BaseMLDL):
def train(self, iterations=100, **kwargs):
self.model = CatBoostRegressor(iterations=iterations, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class KaggleSubmission:
"""
Kaggle 제출 자동화 클래스
"""
def __init__(self, preprocessor):
self.preprocessor = preprocessor
def submit(self, model, X_test, submission_path, id_column, target_column):
"""제출 파일 생성"""
# 테스트 데이터 스케일링
X_test_scaled = model.scaler.transform(X_test)
# 예측
predictions = model.model.predict(X_test_scaled)
# 제출 파일 생성
submission = pd.DataFrame({
'id' : X_test['id'],
'Premium Amount' : predictions
})
submission.to_csv(submission_path, index=False)
print(f"제출 파일 생성: {submission_path}")
def compare_shap_feature_importance(models, X, save_path='shap_comparison.png'):
"""
여러 모델의 SHAP 디테일 특성 중요도를 한 그래프에 시각화 및 저장
Args:
models (list): 비교할 모델 리스트
X (pd.DataFrame): 특성 데이터
save_path (str): 저장할 파일 경로
"""
# 모델별 SHAP 값 저장할 딕셔너리
model_shap_values = {}
# 각 모델의 SHAP 값 계산
for model_info in models:
model = model_info['model']
model_name = model_info['name']
try:
# SHAP 설명자 생성
shap_explainer = SHAPExplainer(model.model, X)
_, shap_values = shap_explainer.tree_explainer()
# SHAP 값 저장
model_shap_values[model_name] = shap_values
except Exception as e:
print(f"{model_name} SHAP 분석 중 오류: {e}")
# 모델 수에 따른 동적 서브플롯 설정
n_models = len(model_shap_values)
# 행과 열 계산 (정사각형에 가깝게)
import math
n_cols = math.ceil(math.sqrt(n_models))
n_rows = math.ceil(n_models / n_cols)
# 큰 피규어 생성 (디테일 플롯만)
plt.figure(figsize=(20*n_cols, 5*n_rows))
# 각 모델의 SHAP 특성 중요도 시각화
for idx, (model_name, shap_values) in enumerate(model_shap_values.items(), 1):
plt.subplot(n_rows, n_cols, idx)
# SHAP 디테일 플롯
shap.summary_plot(
shap_values,
X,
show=False,
)
plt.gca().set_xlabel('')
# 제목 추가
plt.title(f"SHAP Detail - {model_name}", fontsize=11)
plt.tight_layout()
# 피규어 저장 (디테일 플롯만)
plt.savefig(save_path, dpi=1000, bbox_inches='tight')
print(f"SHAP 디테일 비교 그래프가 {save_path}에 저장되었습니다.")
# 선택적으로 화면에 표시
plt.show()
def main():
# 데이터 전처리
preprocessor = DataPreprocessor('playground-series-s4e12/train.csv', 'playground-series-s4e12/test.csv')
X, y, X_test, ordinal_encoder = preprocessor.preprocess('Premium Amount')
# 모델 학습 및 평가
models = [
RandomForestModel(X, y),
XGBoostModel(X, y),
LightGBMModel(X, y),
CatBoostModel(X, y),
]
# Kaggle 제출 클래스 초기화
submission_handler = KaggleSubmission(preprocessor)
# 모델별 성능 저장할 딕셔너리
model_performances = {}
model_shap_info = []
# 모델별 성능 평가 및 SHAP 분석
for model in models:
try:
# 모델 학습
model.train()
# 성능 평가
metrics = model.evaluate()
print(f"{model.__class__.__name__} 성능:")
for metric, value in metrics.items():
print(f"{metric}: {value}")
# 모델 성능 저장 (RMSE를 기준으로)
model_performances[model.__class__.__name__] = {
'model': model,
'rmse': metrics['RMSE']
}
# SHAP 정보 저장
model_shap_info.append({
'name': model.__class__.__name__,
'model': model
})
except Exception as model_error:
print(f"{model.__class__.__name__} 처리 중 오류: {model_error}")
continue
# 모델 간 SHAP 특성 중요도 비교
compare_shap_feature_importance(model_shap_info, X)
# 최적 모델 선택 (RMSE 기준 최소값)
if model_performances:
best_model_name = min(model_performances, key=lambda k: model_performances[k]['rmse'])
best_model = model_performances[best_model_name]['model']
print(f"\n최적 모델: {best_model_name}")
print(f"최적 모델 RMSE: {model_performances[best_model_name]['rmse']}")
# 최적 모델로 Kaggle 제출
submission_handler.submit(
best_model,
X_test,
'best_model_submission.csv',
'id',
'Premium Amount'
)
# 모든 모델의 성능 비교 CSV로 저장
performance_df = pd.DataFrame.from_dict(
{name: {'RMSE': data['rmse']} for name, data in model_performances.items()},
orient='index'
)
performance_df.to_csv('model_performance_comparison.csv')
print("\n모델 성능 비교 결과가 'model_performance_comparison.csv'에 저장되었습니다.")
else:
print("모델 학습에 실패했습니다.")
if __name__ == "__main__":
main()
- 코드의 주요 기능은 다음과 같음
- 데이터 전처리
- 다양항 모델 학습 및 평가 (Random Forest, XGBoost, LightGBM, CatBoost)
- SHAP(Shapley Additive Explanations)을 활용한 특성 중요도 분석
- 모델 성능 비교 및 최적 모델 선택
- 최적 모델 기반 Kaggle 제출 파일 생성
주요 클래스 설명
DataPreprocessor
- 데이터 전처리를 담당하는 클래스입니다.
- 주요 기능:
- 결측치 처리: 숫자형 데이터는 중앙값, 범주형 데이터는 최빈값으로 대체.
- Ordinal Encoding: 문자열 데이터를 수치형으로 변환.
- 훈련 데이터(
train.csv)와 테스트 데이터(test.csv)를 받아 전처리 후 반환.
class DataPreprocessor:
"""
데이터 전처리 클래스
"""
def __init__(self, train_path, test_path):
self.train_data = pd.read_csv(train_path)
self.test_data = pd.read_csv(test_path)
def preprocess(self, target_column):
"""
데이터 전처리 메서드
- 결측치 처리 전략 개선
- 문자열 컬럼 Ordinal Encoding
"""
# 타겟 분리
X = self.train_data.drop(columns=[target_column])
y = self.train_data[target_column]
X_test = self.test_data.copy()
# 결측치 처리 전략 개선
def handle_missing_values(df):
# 숫자형 변수 - 중앙값 대체
numeric_columns = df.select_dtypes(include=['int64', 'float64']).columns
df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].median())
# 범주형 변수 - 최빈값 대체
categorical_columns = df.select_dtypes(include=['object']).columns
df[categorical_columns] = df[categorical_columns].fillna(df[categorical_columns].mode().iloc[0])
return df
# 결측치 처리
X = handle_missing_values(X)
X_test = handle_missing_values(X_test)
# 문자열 컬럼 식별
categorical_columns = X.select_dtypes(include=['object']).columns
# Ordinal Encoder 초기화
ordinal_encoder = OrdinalEncoder(
handle_unknown='use_encoded_value',
unknown_value=-1
)
# 훈련 및 테스트 데이터에 인코딩 적용
if len(categorical_columns) > 0:
# 전체 데이터 결합하여 인코딩
combined_categorical = pd.concat([X[categorical_columns], X_test[categorical_columns]])
# 인코더 훈련
ordinal_encoder.fit(combined_categorical)
# 훈련 데이터 인코딩
X[categorical_columns] = ordinal_encoder.transform(X[categorical_columns])
# 테스트 데이터 인코딩
X_test[categorical_columns] = ordinal_encoder.transform(X_test[categorical_columns])
return X, y, X_test, ordinal_encoder
EvaluationMetrics
- 모델 성능을 평가하기 위한 지표를 제공합니다.
- 값이 작을수록 모델의 예측이 실제 값과 더 가까움을 의미합니다.
- 제공되는 지표:
- RMSE (Root Mean Squared Error)
- MAE (Mean Absolute Error) : 예측값과 실제값 간 절대 오차의 평균
- R² Score (결정 계수) : 모델의 예측이 실제 데이터를 얼마나 잘 설명하는지
- MAPE (Mean Absolute Percentage Error) : 예측값과 실제값 간의 상대적인 오차를 백분율
class EvaluationMetrics:
"""
모델 평가 지표 클래스
"""
@staticmethod
def rmse(y_true, y_pred):
return np.sqrt(mean_squared_error(y_true, y_pred))
@staticmethod
def mae(y_true, y_pred):
return mean_absolute_error(y_true, y_pred)
@staticmethod
def r2(y_true, y_pred):
return r2_score(y_true, y_pred)
@staticmethod
def mape(y_true, y_pred):
return mean_absolute_percentage_error(y_true, y_pred)
@staticmethod 적용 이유EvaluationMetrics 클래스의 메서드는 입력값 y_true, y_pred만 필요하며, 클래스의 상태(속성)와 무관.- 따라서 이 메서드를 정적 메서드로 정의하여 클래스의 다른 부분과 독립적으로 동작할 수 있도록 설계
- 직접 호출 가능
클래스명.메서드명() 형태로 호출
SHAPExplainer
- SHAP 분석 및 시각화를 담당
- 주요 기능
tree_explainer(): 트리 기반 모델(XGBoost, LightGBM, Random Forest)에서 SHAP 값을 계산.
class SHAPExplainer:
"""
SHAP 설명 및 시각화 클래스
"""
def __init__(self, model, X):
self.model = model
self.X = X
def tree_explainer(self):
"""트리 기반 모델용 SHAP 설명"""
explainer = shap.TreeExplainer(self.model)
shap_values = explainer.shap_values(self.X)
return explainer, shap_values
BaseMLDL
- 머신러닝 및 딥러닝 모델의 공통 동작을 정의하는 베이스 클래스
- 주요 메서드:
train(): 모델 학습. (하위 클래스에서 구현)predict(): 모델 예측. (하위 클래스에서 구현)evaluate(): 평가 지표를 계산.
class BaseMLDL:
"""
기본 ML/DL 모델 베이스 클래스
"""
def __init__(self, X, y, test_size=0.2, random_state=42):
# 데이터 분할
self.X_train, self.X_val, self.y_train, self.y_val = train_test_split(
X, y, test_size=test_size, random_state=random_state
)
# 스케일링
self.scaler = StandardScaler()
self.X_train_scaled = self.scaler.fit_transform(self.X_train)
self.X_val_scaled = self.scaler.transform(self.X_val)
self.model = None
self.predictions = None
def train(self):
raise NotImplementedError("하위 클래스에서 구현해야 합니다.")
def predict(self):
raise NotImplementedError("하위 클래스에서 구현해야 합니다.")
def evaluate(self):
"""모델 평가"""
self.predictions = self.predict()
return {
'RMSE': EvaluationMetrics.rmse(self.y_val, self.predictions),
'MAE': EvaluationMetrics.mae(self.y_val, self.predictions),
'R2': EvaluationMetrics.r2(self.y_val, self.predictions),
'MAPE': EvaluationMetrics.mape(self.y_val, self.predictions)
}
주요 모델 클래스 정의
BaseMLDL을 상속받아 각 알고리즘에 맞게 동작을 구현한 클래스.RandomForestModel: Random Forest Regressor 모델.XGBoostModel: XGBoost Regressor 모델.LightGBMModel: LightGBM Regressor 모델.CatBoostModel: CatBoost Regressor 모델.
- 각 클래스의 train을 재정의할 수 있습니다. 하이퍼파라미터 튜닝을 시행할 수 있다.
# 각 모델 클래스 구현 (RandomForest, XGBoost, LightGBM, CatBoost 등)
class RandomForestModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = RandomForestRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class XGBoostModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = xgb.XGBRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class LightGBMModel(BaseMLDL):
def train(self, n_estimators=100, **kwargs):
self.model = lgb.LGBMRegressor(n_estimators=n_estimators, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
class CatBoostModel(BaseMLDL):
def train(self, iterations=100, **kwargs):
self.model = CatBoostRegressor(iterations=iterations, **kwargs)
self.model.fit(self.X_train_scaled, self.y_train)
def predict(self):
return self.model.predict(self.X_val_scaled)
KaggleSubmission
