Kaggle의 "House Prices: Advanced Regression Techniques" 대회 데이터를 기반으로 주택 가격을 예측하는 회귀 분석을 수행하였다. 이 과정에서 회귀 문제를 해결할 때 기본으로 가져가면 좋을 베이스 로직을 정리해 보았다. 이진 분류 문제 구조화하기도 따로 작성해 보았으니 필요하다면 참고하는 것도 좋을 것 같다.
회귀 분석에서도 마찬가지로 각 단계를 함수화하여 구조화하는 데에 많은 시간이 들었다.
전처리나 분석 과정보다는 회귀 문제를 해결하는 과정에서 도출할 수 있었던 기술적인 인사이트를 위주로 하고 있다. 각 함수 동작의 핵심을 최대한 설명할 수 있도록 주석을 달아 두었다.
글의 볼륨이 크지만 그래도 (나처럼) 회귀 분석이, 혹은 캐글 대회가 아직 낯설다면 아래의 내용 중 얻어 갈 부분이 조금은 있을 거라 생각한다.
- 회귀 분석 목차 및 실험 방향
- 모델별 Baseline 성능 비교 표 생성 자동화
- XGBoost/LightGBM 조기 종료 로직 분기 적용
- 트리 기반 모델 하이퍼파라미터 튜닝(GridSearchCV/RandomizedSearchCV)
- 스태킹(Ensemble Stacking) 적용
- Kaggle 제출용 스태킹 메타 모델 학습
분석에 정답은 없기에, 이런 방향으로도 바라볼 수 있구나 하는 가벼운 마음으로 살펴봐도 좋을 것 같다.
해당 문제를 해결하면서 정의한 함수를 포함한 자세한 모델링 과정과 전처리 로직은 GitHub repository에서 확인할 수 있다. 더 효율적인 방법이 있거나, 왜 이렇게 구현한 건지 의문이 드는 부분이 있다면 언제든 알려주기 바란다 :)
🖇 목차 및 실험 방향
해당 회귀 분석 수행 노트북은 다음과 같은 목차로 구성하였다.
회귀 문제를 해결하는 하나의 방향으로 참고가 될까 하여 정리해 두었다. 데이터 전처리 및 튜닝 과정도 상황에 맞게 적절히 변형하여 선택적으로 사용할 수 있다.
- 데이터 로드 및 확인
1.1 라이브러리 및 데이터 불러오기
1.2 데이터 확인
1.3 결측치 및 이상치 확인
- 데이터 전처리 로그 변환
2.1 로그 변환
2.2 결측치 처리
2.3 One-Hot Encoding
2.4 이상치 처리
2.5 Train/Test 데이터 분리
- 모델 학습 및 평가 함수 정의
3.1 회귀 모델 성능 평가 함수
3.2 회귀 모델 학습 및 평가 함수
3.3 개별 모델 실행 함수
3.4 회귀 계수 비교
- 모델 비교 및 선택
4.1 LinearRegression
4.2 Ridge
4.3 Lasso
4.4 RandomForest
4.5 XGBoost
4.6 LightGBM
4.7 Baseline 모델 성능 비교
- 하이퍼파라미터 튜닝
5.1 GridSearch: LightGBM
5.2 GridSearch: XGBoost
5.3 RandomSearch: LightGBM
5.4 RandomSearch: XGBoost
- Fine-tuning
6.1 LightGBM
6.2 XGBoost
- Stacking
- Submission
🖇 모델 학습 및 평가 준비
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
%matplotlib inline
from sklearn.model_selection import train_test_split
from sklearn.model_selection import KFold
from sklearn.base import clone
from sklearn.metrics import mean_squared_error, mean_absolute_error
from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from xgboost import XGBRegressor
from lightgbm import LGBMRegressor
import warnings
warnings.filterwarnings("ignore")회귀 모델 성능 평가 함수
- 회귀 성능 지표 함수
from sklearn.metrics import mean_squared_error
# RMSLE 계산 함수
def rmsle_func(y, pred):
"""
로그 변환을 통해 실제값과 예측값 간의 상대적인 오차를 측정한다.
"""
log_y = np.log1p(y)
log_pred = np.log1p(pred)
return np.sqrt(np.mean((log_y - log_pred) ** 2)) # 음수 방지
# RMSE 계산 함수
def rmse_func(y, pred):
"""
예측값과 실제값 사이의 평균 제곱 오차의 제곱근을 구해 절대 오차의 평균적인 크기를 측정한다.
"""
return np.sqrt(mean_squared_error(y, pred))
# 통합 평가 출력 함수
def evaluate_regr(y, pred):
"""
RMSLE, RMSE, MSE 세 가지 회귀 평가 지표를 계산하고 출력한다.
Parameters
----------
y : array-like of shape (n_samples,)
실제 타겟값 (로그 복원된 상태)
pred : array-like of shape (n_samples,)
예측값 (로그 복원된 상태)
"""
rmsle_val = rmsle_func(y, pred)
rmse_val = rmse_func(y, pred)
mse_val = mean_squared_error(y, pred)
print(f'RMSLE: {rmsle_val:.3f}, RMSE: {rmse_val:.3f}, MSE: {mse_val:.3f}')
return rmsle_va- 예측 오차 분석 함수
def get_top_error_data(y_test, pred, n_tops = 5):
"""
예측값과 실제값 간의 오차가 가장 큰 샘플 데이터를 상위 n개 출력한다.
Parameters
----------
y_test : array-like of shape (n_samples,)
테스트 데이터 실제 타겟값
pred : array-like of shape (n_samples,)
모델 예측값
n_tops : int, default=5
Prints
------
DataFrame
실제 가격, 예측 가격, 그리고 이들의 차이를 포함한 결과 상위 n개 샘플
"""
# 실제 타겟값(y_test)을 기반 DataFrame 생성
result_df = pd.DataFrame(y_test.values, columns=['real_sale_price'])
result_df['predicted_sale_price']= np.round(pred)
result_df['diff'] = np.abs(result_df['real_sale_price'] - result_df['predicted_sale_price'])
print(result_df.sort_values('diff', ascending=False)[:n_tops])- 평가 지표 출력 함수
# 모델 성능 비교 결과 저장 리스트
model_score = []
def evaluate_model_result(model, model_name, y_test, pred, feature_names, is_log_transform=True):
"""
회귀 모델의 예측 결과를 평가하고, 주요 성능 지표와 오차가 큰 샘플을 출력한다.
평가 결과를 전역 리스트(model_score)에 저장한다.
Parameters
----------
model : object
model_name : str
y_test : array-like of shape (n_samples,)
테스트 데이터 실제 타겟값 (로그 변환된 상태일 수 있음)
pred : array-like of shape (n_samples,)
모델 예측값 (로그 변환된 상태일 수 있음)
feature_names : list of str
입력 피처 이름 리스트 (중요도 시각화 시 사용)
is_log_transform : bool, default=True
타겟값이 로그 변환되었는지 여부 (True일 경우 expm1으로 복원)
Returns
-------
None
주요 성능 지표를 출력하고, 결과를 전역 리스트 model_score에 저장한다.
"""
# 로그 복원
if is_log_transform:
y_test = np.expm1(y_test)
pred = np.expm1(pred)
# 결과 출력
print(f'### {model_name} ###')
rmsle_val, rmse_val, mse_val = evaluate_regr(y_test, pred)
get_top_error_data(pd.Series(y_test), pred)
# # 선형 모델에 한해 중요도 시각화
# # feature importance 출력 (coef_이 있는 모델에만)
# if hasattr(model, 'coef_'):
# plot_feature_importance(model, feature_names)
# 평가 결과 저장
model_score.append({
'Model': model_name,
'RMSLE': round(rmsle_val, 3),
'RMSE': round(rmse_val, 3),
'MSE': round(mse_val, 3)
})회귀 모델 학습 및 평가 함수
- 스케일링 함수
from sklearn.preprocessing import StandardScaler
def scale_features(X_train, X_test):
"""
학습용 X_train 기준으로 StandardScaler를 학습하고,
X_train과 X_test를 스케일링하여 반환한다.
Returns
-------
X_train_scaled, X_test_scaled : ndarray
"""
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
return X_train_scaled, X_test_scaled- 조기 종료 적용 함수
from lightgbm import LGBMRegressor, early_stopping, log_evaluation
from xgboost import XGBRegressor
def apply_early_stopping(model, model_name, X_train, y_train, X_test, y_test):
"""
LightGBM/XGBoost 회귀 모델에 대해 조기 종료(Early Stopping)를 적용하여 학습을 수행한다.
Parameters
----------
model : object
model_name : str
모델 이름 (에러 메시지 출력용)
X_train : array-like of shape (n_samples, n_features)
학습용 입력 데이터
y_train : array-like of shape (n_samples,)
학습용 타겟 데이터
X_test : array-like of shape (n_samples, n_features)
검증용 입력 데이터
y_test : array-like of shape (n_samples,)
검증용 타겟 데이터
Returns
-------
None
모델 학습은 내부적으로 수행되며 반환값은 없다.
조기 종료가 적용된 상태로 모델이 학습된다.
Raises
------
ValueError
지원하지 않는 모델 타입일 경우 예외 발생
"""
eval_set = [(X_train, y_train), (X_test, y_test)]
if isinstance(model, LGBMRegressor):
model.fit(
X_train, y_train,
eval_set=eval_set,
eval_metric='rmse',
callbacks=[
early_stopping(stopping_rounds=100, verbose=False),
log_evaluation(period=0)
]
)
elif isinstance(model, XGBRegressor):
model.fit(
X_train, y_train,
eval_set=eval_set,
# eval_metric='rmse',
# early_stopping_rounds=50,
verbose=False
)
else:
raise ValueError(f'Early stopping not supported for mo- 모델 학습 및 평가 함수
def train_and_predict_model(model, model_name, X_train, X_test, y_train, y_test, feature_names, is_log_transform=True):
"""
회귀 모델을 학습하고 테스트 데이터를 예측한 뒤, 결과를 평가하는 통합 함수
XGBoost 또는 LightGBM 모델의 경우 조기 종료(Early Stopping)를 적용하며,
해당 모델들은 NumPy 배열 입력이 필요하므로 내부에서 변환이 이루어진다.
예측 결과는 로그 변환된 상태로 반환되며, 평가 시 로그 역변환 여부는 is_log_transform 인자에 따라 처리된다.
Parameters
----------
model : object
model_name : str
X_train : array-like of shape (n_samples, n_features)
학습용 입력 데이터
X_test : array-like of shape (n_samples, n_features)
테스트용 입력 데이터
y_train : array-like of shape (n_samples,)
학습용 타겟값
y_test : array-like of shape (n_samples,)
테스트용 타겟값
feature_names : list of str
피처 이름 리스트 (선형 모델 계수 시각화 등에 사용)
is_log_transform : bool, default=True
타겟값의 로그 변환 여부. True인 경우 평가 시 로그 역변환 수행
Returns
-------
model : object
학습이 완료된 회귀 모델 객체
"""
# XGB, LGBM: NumPy 변환 필요
# XGBoost / LightGBM은 NumPy 배열 사용
if isinstance(model, (XGBRegressor, LGBMRegressor)):
X_train, X_test = X_train.values, X_test.values
y_train, y_test = y_train.values, y_test.values
# 모델 학습
if isinstance(model, (XGBRegressor, LGBMRegressor)):
apply_early_stopping(model, model_name, X_train, y_train, X_test, y_test)
else:
model.fit(X_train, y_train)
# 예측
pred = model.predict(X_test)
# 평가 시에는 로그 역변환만 수행
evaluate_model_result(
model=model,
model_name=model_name,
y_test=y_test, # log 상태
pred=pred, # log 상태
feature_names=feature_names,
is_log_transform=is_log_transform
)
return model개별 모델 실행 함수
- 개별 회귀 모델을 실행하고 학습 및 예측 결과를 평가하는 함수
"""
각 함수는 회귀 모델을 정의한 후, train_and_predict_model()을 호출하여 다음 과정을 자동으로 수행한다.
1. 모델 학습 (필요 시 조기 종료 적용)
2. 테스트 데이터 예측
3. 성능 평가 지표 출력 (RMSLE, RMSE, MSE)
4. 오차가 큰 예측 샘플 출력
5. 평가 결과를 전역 리스트 model_score에 저장
Parameters
----------
X_train, X_test : array-like of shape (n_samples, n_features)
학습/테스트 입력 데이터
y_train, y_test : array-like of shape (n_samples,)
학습/테스트 타겟값 (log 변환 여부는 is_log_transform에 따름)
feature_names : list of str
피처 이름 목록 (선형 모델의 중요도 시각화 시 사용)
alpha : float, optional
릿지/라쏘 모델의 정규화 강도 (해당 모델 함수에서만 사용됨)
n_estimators : int, optiona
트리 기반 모델의 트리 개수 (RandomForest, XGBoost, LightGBM 전용)
is_log_transform : bool, optional (default: True)
타겟값의 로그 변환 여부. True인 경우 평가 시 로그 역변환 수행
Returns
-------
model : object
학습이 완료된 회귀 모델 객체
"""
def run_linear_regression(X_train, X_test, y_train, y_test, feature_names, is_log_transform=True):
model = LinearRegression()
return train_and_predict_model(model, 'LinearRegression',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)
def run_ridge(X_train, X_test, y_train, y_test, feature_names, alpha=1.0, is_log_transform=True):
model = Ridge(alpha=alpha)
return train_and_predict_model(model, 'Ridge',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)
def run_lasso(X_train, X_test, y_train, y_test, feature_names, alpha=1.0, is_log_transform=True):
model = Lasso(alpha=alpha)
return train_and_predict_model(model, 'Lasso',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)
def run_random_forest(X_train, X_test, y_train, y_test, feature_names, n_estimators=500, is_log_transform=True):
model = RandomForestRegressor(n_estimators=n_estimators, n_jobs=-1, random_state=42)
return train_and_predict_model(model, 'RandomForest',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)
def run_xgboost(X_train, X_test, y_train, y_test, feature_names, n_estimators=500, is_log_transform=True):
model = XGBRegressor(n_estimators=n_estimators,
eval_metric='rmse',
early_stopping_rounds=50,
n_jobs=-1,
random_state=42)
return train_and_predict_model(model, 'XGBoost',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)
def run_lightgbm(X_train, X_test, y_train, y_test, feature_names, n_estimators=500, is_log_transform=True):
model = LGBMRegressor(n_estimators=n_estimators,
verbosity=-1,
n_jobs=-1,
random_state=42)
return train_and_predict_model(model, 'LightGBM',
X_train, X_test, y_train, y_test,
feature_names,
is_log_transform)회귀 계수 비교
피처별 회귀 계수를 시각화하여 모델별로 어떠한 피처의 회귀 계수로 구성되는지 확인한다.
- 모델 계수에서 중요 피처 추출 함수
def get_top_bottom_coef(model, feature_names, top_n=10):
"""
선형 회귀 모델의 계수(coef_)를 바탕으로 가장 영향력 있는 피처들을 추출한다.
top_n 개수만큼 가장 양(+)의 영향을 주는 피처와 가장 음(-)의 영향을 주는 피처를 반환한다.
Returns
-------
coef_high : pd.Series
계수가 가장 큰 상위 top_n 피처
coef_low : pd.Series
계수가 가장 작은 하위 top_n 피처
"""
coef = pd.Series(model.coef_, index=feature_names)
coef_high = coef.sort_values(ascending=False).head(top_n)
coef_low = coef.sort_values(ascending=True).head(top_n)
return coef_high, coef_low- 모델별 회귀 계수 시각화 함수
def visualize_coefficient(models, feature_names):
"""
주어진 회귀 모델 리스트의 계수를 시각화한다.
각 모델별로 subplot에 상위/하위 영향력 피처를 barplot으로 표시한다.
Parameters
----------
models : list
학습이 완료된 선형 회귀 모델 객체들의 리스트
feature_names : list or Index
모델 학습에 사용된 피처 이름 리스트 (각 모델의 coef_과 매핑됨)
"""
fig, axs = plt.subplots(figsize=(24, 10), nrows=1, ncols=len(models))
fig.tight_layout()
for i_num, model in enumerate(models):
# 상위/하위 영향 피처 추출
coef_high, coef_low = get_top_bottom_coef(model, feature_names)
# 하나의 시리즈로 병합 후 정렬 (시각화 순서 통일)
coef_concat = pd.concat([coef_high, coef_low]).sort_values(ascending=False)
# 서브플롯 타이틀 및 레이블 조정
axs[i_num].set_title(model.__class__.__name__ + ' Coefficients', size=25)
axs[i_num].tick_params(axis="y", direction="in", pad=-120)
for label in (axs[i_num].get_xticklabels() + axs[i_num].get_yticklabels()):
label.set_fontsize(22)
sns.barplot(x=coef_concat.values, y=coef_concat.index, ax=axs[i_num])- alpha 값 및 교차검증 RMSE 출력 함수: GirdSearch
from sklearn.model_selection import GridSearchCV
def print_best_params(model, params):
"""
Ridge 또는 Lasso 모델에 대해 GridSearchCV를 수행하여
최적의 alpha 값을 찾고, 교차 검증 기반 RMSE를 출력한다.
Parameters
----------
model : object
Ridge 또는 Lasso 등 선형 회귀 모델 객체
params : dict
alpha 값 리스트를 포함하는 파라미터 딕셔너리
e.g., {'alpha': [0.1, 1, 10]}
Returns
-------
best_model : object
GridSearchCV를 통해 선택된 최적의 모델 객체 (best_estimator_)
"""
# GridSearchCV로 alpha 튜닝 수행 (neg_mean_squared_error 기준)
grid_model = GridSearchCV(model, param_grid=params,
scoring='neg_mean_squared_error', cv=5)
grid_model.fit(X_features, y_target_log)
# 평균 RMSE 계산
rmse = np.sqrt(-1* grid_model.best_score_)
# 결과 출력
print('{0} 5 CV 시 최적 평균 RMSE 값: {1}, 최적 alpha:{2}'.format(model.__class__.__name__,
np.round(rmse, 4), grid_model.best_params_))
return grid_model.best_estimator_- 선형 회귀 기반 모델 학습 및 계수 시각화
각 모델의 피처 계수를 시각화하여 변수 중요도 비교한다.
# default alpha
lin_reg = LinearRegression()
lin_reg.fit(X_train, y_train)
ridge = Ridge(alpha=1)
ridge.fit(X_train, y_train)
lasso = Lasso(alpha=1)
lasso.fit(X_train, y_train)
models = [lin_reg, ridge, lasso]
visualize_coefficient(models, feature_names=X_features.columns)
# Ridge/Lasso 모델의 alpha 값 최적화를 위한 후보 파라미터 리스트 정의
ridge_params = { 'alpha':[0.05, 0.1, 1, 5, 8, 10, 12, 15, 20] }
lasso_params = { 'alpha':[0.001, 0.005, 0.008, 0.05, 0.03, 0.1, 0.5, 1,5, 10] }
# 최적 모델 객체
best_rige = print_best_params(ridge, ridge_params)
best_lasso = print_best_params(lasso, lasso_params)
# 최적 alpha
lin_reg = LinearRegression()
lin_reg.fit(X_train, y_train)
ridge = Ridge(alpha=5)
ridge.fit(X_train, y_train)
lasso = Lasso(alpha=0.001)
lasso.fit(X_train, y_train)
models = [lin_reg, ridge, lasso]
visualize_coefficient(models, feature_names=X_features.columns)🖇 모델 비교 및 선택
# 모델 실행 함수에 공통적으로 전달할 기본 argument
default_args = {
'X_train': X_train,
'X_test': X_test,
'y_train': y_train,
'y_test': y_test,
'feature_names': X_features.columns,
'is_log_transform': True
}
# 스케일링 적용 argument
scaled_args = default_args.copy()
scaled_args['X_train'], scaled_args['X_test'] = scale_features(
default_args['X_train'], default_args['X_test']
)동일한 학습/테스트 데이터와 설정을 유지하면서 회귀 모델별 성능을 비교한다.
# LinearRegression
lin_reg = run_linear_regression(**scaled_args)
# Ridge
ridge_reg = run_ridge(**scaled_args, alpha=10)
# Lasso
lasso_reg = run_lasso(**scaled_args, alpha=0.01)
# 랜덤 포레스트
rf_reg = run_random_forest(**default_args, n_estimators=500)
# XGBoost
xgb_reg = run_xgboost(**default_args, n_estimators=500)
# LightGBM
lgbm_reg = run_lightgbm(**default_args, n_estimators=500)Baseline 모델 성능 비교
모델 학습 및 평가 후 각 모델의 점수를 바로 비교할 수 있도록
evaluate_model_result() 함수에서 표 생성 과정을 자동화해 두었다.
model_score_df = pd.DataFrame(model_score)
model_score_df
model_score_df.sort_values(by=['RMSLE', 'RMSE', 'MSE'])
# 모델 성능 결과 초기화 - 필요 시 실행
model_score.clear()
model_score_df = pd.DataFrame(model_score)🖇 하이퍼파라미터 튜닝
- 조기 종료 파라미터 반환 함수
def get_fit_params_with_early_stopping(model, X_train, X_test, y_train, y_test):
"""
LightGBM 또는 XGBoost 모델의 학습 시 조기 종료(Early Stopping)를 적용하기 위한
fit() 파라미터 딕셔너리를 생성하여 반환한다.
GridSearchCV 또는 RandomizedSearchCV의 fit() 호출 시 fit_params 인자로 사용된다.
Parameters
----------
model : 학습 대상 모델 객체
X_train, X_test, y_train, y_test : 학습/검증 데이터
Returns
-------
fit_params : dict
모델 학습 시 fit()에 전달할 파라미터
"""
eval_set = [(X_train, y_train), (X_test, y_test)]
if isinstance(model, LGBMRegressor):
return {
'eval_set': eval_set,
'eval_metric': 'rmse',
'callbacks': [
early_stopping(stopping_rounds=50, verbose=False),
log_evaluation(period=0)
]
}
elif isinstance(model, XGBRegressor):
return {
'eval_set': eval_set,
'verbose': False
}
else:
return {} # 일반 회귀 모델은 특별한 fit_params 필요 없음- 튜닝된 최적 모델 평가 함수
from sklearn.metrics import mean_squared_error, mean_squared_error
def rmsle(y, pred):
log_y = np.log1p(y)
log_pred = np.log1p(np.maximum(pred, 0)) # 예측값 음수 방지
return np.sqrt(np.mean((log_y - log_pred) ** 2))
def evaluate_best_model(grid_cv, X_test, y_test):
"""
GridSearchCV/RandomizedSearchCV로 튜닝된 최적 회귀 모델을
테스트 세트에 대해 평가하고, 주요 회귀 지표를 출력한다.
Parameters
----------
grid_cv : GridSearchCV 또는 RandomizedSearchCV 객체
사전에 fit()이 완료된 객체로, 최적 모델(best_estimator_)을 포함하고 있어야 함
X_test : array-like of shape (n_samples, n_features)
테스트 피처 (원핫 인코딩 상태)
y_test : array-like of shape (n_samples,)
테스트 타겟 (log 변환 상태)
Prints
------
- 최적 하이퍼파라미터 (best_params_)
- 교차 검증 기반 최고 성능 (neg-MSE 기준)
- 로그 복원된 테스트 세트에 대한 RMSLE, RMSE, MSE
"""
print('Best Hyperparameters:', grid_cv.best_params_)
print('Best Cross-Validation Score (neg-MSE 기준):', grid_cv.best_score_)
best_model = grid_cv.best_estimator_
pred = best_model.predict(X_test)
# 로그 복원
y_test_exp = np.expm1(y_test)
pred_exp = np.expm1(pred)
print('\nTest Evaluation')
print(f'RMSLE: {rmsle(y_test_exp, pred_exp):.4f}')
print(f'RMSE : {np.sqrt(mean_squared_error(y_test_exp, pred_exp)):.4f}')
print(f'MSE : {mean_squared_error(y_test_exp, pred_exp):.4f}')
return best_modelGirdSearch: LightGBM
from sklearn.model_selection import GridSearchCV
from lightgbm import LGBMRegressor
# 튜닝 대상 파라미터
lgbm_params = {
'n_estimators': [30, 500, 700, 1000],
'learning_rate': [0.01, 0.05, 0.1],
'max_depth': [4, 6, 8],
# 'num_leaves': [31, 64, 128],
}
# 모델 정의
lgbm_reg = LGBMRegressor(
boost_from_average=False,
random_state=42,
n_jobs=-1,
verbosity=-1
)
# 조기 종료용 fit_params
fit_params = get_fit_params_with_early_stopping(lgbm_reg, X_train, X_test, y_train, y_test)
# GridSearchCV 정의
grid_cv_lgbm = GridSearchCV(lgbm_reg, lgbm_params, cv=3, scoring='neg_mean_squared_error')
# 학습 시 추가 fit_params 전달
grid_cv_lgbm.fit(X_train, y_train, **fit_params)
# 최적 모델로 예측 및 평가
best_lgbm_grid = evaluate_best_model(grid_cv_lgbm, X_test, y_test)GirdSearch: XGBoost
from sklearn.model_selection import GridSearchCV
from xgboost import XGBRegressor
# 튜닝 대상 파라미터
xgb_params = {
'n_estimators': [100, 200],
'learning_rate': [0.05, 0.1],
'max_depth': [3, 5, 7],
'subsample': [0.8, 1.0],
'colsample_bytree': [0.8, 1.0],
}
# 모델 정의
xgb_reg = XGBRegressor(
eval_metric='rmse',
early_stopping_rounds=10,
use_label_encoder=False,
random_state=42,
n_jobs=-1
)
# 조기 종료용 fit_params
fit_params = get_fit_params_with_early_stopping(xgb_reg, X_train, X_test, y_train, y_test)
# GridSearchCV 정의
grid_cv_xgb = GridSearchCV(xgb_reg, xgb_params, cv=3, scoring='neg_mean_squared_error')
# 학습 시 추가 fit_params 전달
grid_cv_xgb.fit(X_train, y_train, **fit_params)
# 최적 모델로 예측 및 평가
best_xgb_grid = evaluate_best_model(grid_cv_xgb, X_test, y_test)RandomSearch: LightGBM
from sklearn.model_selection import RandomizedSearchCV
from lightgbm import LGBMRegressor
# 튜닝 대상 파라미터
lgbm_param_dist = {
'n_estimators': [30, 500, 700, 1000],
'learning_rate': [0.01, 0.05, 0.1],
'max_depth': [4, 6, 8],
# 'num_leaves': [31, 64, 128],
}
# 모델 정의
lgbm_reg = LGBMRegressor(
boost_from_average=False,
random_state=42,
n_jobs=-1,
verbosity=-1
)
# 조기 종료용 fit_params
fit_params = get_fit_params_with_early_stopping(lgbm_reg, X_train, X_test, y_train, y_test)
# RandomizedSearchCV 정의
rand_cv_lgbm = RandomizedSearchCV(
estimator=lgbm_reg,
param_distributions=lgbm_param_dist,
n_iter=20,
scoring='neg_mean_squared_error',
cv=3,
n_jobs=-1,
verbose=1,
random_state=42
)
# 학습
rand_cv_lgbm.fit(X_train, y_train, **fit_params)
# 평가
best_lgbm_rand = evaluate_best_model(rand_cv_lgbm, X_test, y_test)RandomSearch: XGBoost
from sklearn.model_selection import RandomizedSearchCV
from xgboost import XGBRegressor
# 튜닝 대상 파라미터
xgb_param_dist = {
'n_estimators': [100, 200],
'learning_rate': [0.05, 0.1],
'max_depth': [3, 5, 7],
'subsample': [0.8, 1.0],
'colsample_bytree': [0.8, 1.0],
}
# 모델 정의
xgb_reg = XGBRegressor(
eval_metric='rmse',
early_stopping_rounds=10,
random_state=42,
n_jobs=-1
)
# 조기 종료용 fit_params
fit_params = get_fit_params_with_early_stopping(
xgb_reg, X_train, X_test, y_train, y_test
)
# RandomizedSearchCV 정의
random_cv_xgb = RandomizedSearchCV(
estimator=xgb_reg,
param_distributions=xgb_param_dist,
n_iter=20,
scoring='neg_mean_squared_error',
cv=3,
n_jobs=-1,
verbose=1,
random_state=42
)
# 학습 시 추가 fit_params 전달
random_cv_xgb.fit(X_train, y_train, **fit_params)
# 최적 모델로 예측 및 평가
best_xgb_rand = evaluate_best_model(random_cv_xgb, X_test, y_test)🖇 Fine-tuning
하이어파라미터 튜닝을 통해 도출한 최적 조합을 기반으로
파라미터 조합을 수동으로 fine-tuning 하여 모델 성능의 추가 향상을 기대한다.
def evaluate_custom_model(model, X_test, y_test):
"""
수동 설정한 회귀 모델을 테스트 세트에 대해 평가한다.
Parameters
----------
model : 학습 완료된 회귀 모델 객체
X_test : 테스트 피처 (원핫 인코딩 포함, ndarray 또는 DataFrame)
y_test : 테스트 타겟 (log 변환된 상태)
Returns
-------
model : 입력된 모델 그대로 반환
"""
pred = model.predict(X_test)
y_test_exp = np.expm1(y_test)
pred_exp = np.expm1(pred)
print('\nCustom Model Evaluation')
print(f'RMSLE: {rmsle(y_test_exp, pred_exp):.4f}')
print(f'RMSE : {np.sqrt(mean_squared_error(y_test_exp, pred_exp)):.4f}')
print(f'MSE : {mean_squared_error(y_test_exp, pred_exp):.4f}')
return modelLightGBM
from lightgbm import LGBMRegressor
from sklearn.metrics import mean_squared_error, mean_absolute_error
# {'n_estimators': 1000, 'max_depth': 4, 'learning_rate': 0.01}
# {'n_estimators': 700, 'max_depth': 4, 'learning_rate': 0.1}
# 최적 파라미터를 기반으로 수동 조정한 LightGBM 모델
custom_lgbm = LGBMRegressor(
# 기본 구조
n_estimators=1000,
learning_rate=0.01, # 0.01
max_depth=4, # 4
num_leaves=64,
# # 규제 및 분할 관련
# min_child_samples=30,
# min_child_weight=1.0,
# reg_alpha=0,
# reg_lambda=0,
# # 샘플링 비율
subsample=0.8,
colsample_bytree=0.8,
# 기타 설정
boost_from_average=False, # 로그 변환 시 False가 안정적
random_state=42,
n_jobs=-1,
verbosity=-1
)
fit_params = get_fit_params_with_early_stopping(custom_lgbm, X_train, X_test, y_train, y_test)
custom_lgbm.fit(X_train, y_train, **fit_params)
best_lgbm_custom = evaluate_custom_model(custom_lgbm, X_test, y_test)XGB
from xgboost import XGBRegressor
# 최적 파라미터를 기반으로 수동 조정한 XGBoost 모델
custom_xgb = XGBRegressor(
# 핵심 튜닝 파라미터
n_estimators=1000,
learning_rate=0.03,
max_depth=6,
min_child_weight=3,
gamma=0.1,
# 과적합 방지용 샘플링
subsample=0.8,
colsample_bytree=0.8,
# 정규화
reg_alpha=0.1,
reg_lambda=1.0,
# 조기 종료 및 평가
early_stopping_rounds=50,
eval_metric='rmse',
# 기타 설정
random_state=42,
n_jobs=-1,
verbosity=0
)
fit_params = get_fit_params_with_early_stopping(custom_xgb, X_train, X_test, y_train, y_test)
custom_xgb.fit(X_train, y_train, **fit_params)
best_xgb_custom = evaluate_custom_model(custom_xgb, X_test, y_test)🖇 Stacking
미리 언급하고 넘어가자면
해당 문제에서는 성능 비교 결과 LinearRegression, Ridge, Lasso, XGBoost, LightGBM, RandomForest 모델로 스태킹 모델을 구성하였다.
- 스태킹 데이터셋 생성 함수
from sklearn.model_selection import KFold
from sklearn.base import clone
def get_stacking_base_datasets_by_train_func(
model, model_name, X_train, y_train, X_test,
feature_names, is_log_transform=True, n_folds=5
):
"""
개별 모델을 KFold 방식으로 학습 및 예측하여,
메타 모델 학습을 위한 스태킹 데이터셋(train/test)을 생성하는 함수입니다.
이 함수는 train_and_predict_model()을 내부에서 호출하며,
Fold별로 학습한 모델을 기반으로 검증 세트와 테스트 세트에 대한 예측값을 생성합니다.
반환된 예측값은 메타 모델의 입력 feature로 사용됩니다.
Parameters
----------
model : estimator object
학습에 사용할 회귀 모델 객체
model_name : str
모델의 이름 (로그 및 디버깅 용도)
X_train : pd.DataFrame
학습용 feature 데이터 (로그 변환 여부와 관계 없음)
y_train : pd.Series
학습용 타겟 데이터 (로그 변환된 상태일 수 있음)
X_test : pd.DataFrame
테스트용 feature 데이터 (평가용 test split)
feature_names : list or pd.Index
is_log_transform : bool, default=True
n_folds : int, default=5
Returns
-------
train_fold_pred : np.ndarray of shape (n_train, 1)
각 fold의 validation set에 대한 예측값을 모은 배열. 메타 모델 학습에 사용됨
test_pred_mean : np.ndarray of shape (n_test, 1)
각 fold에서 테스트 세트에 대해 예측한 값의 평균. 메타 모델 테스트에 사용됨
"""
kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
train_fold_pred = np.zeros((X_train.shape[0], 1))
test_fold_pred = np.zeros((X_test.shape[0], n_folds))
print(f'\n[{model_name}] 스태킹용 KFold 시작')
for fold_idx, (train_idx, valid_idx) in enumerate(kf.split(X_train)):
print(f'\t[Fold {fold_idx + 1}/{n_folds}]')
X_tr = X_train.iloc[train_idx]
y_tr = y_train.iloc[train_idx]
X_val = X_train.iloc[valid_idx]
y_val = y_train.iloc[valid_idx]
model_fold = clone(model)
trained_model = train_and_predict_model(
model=model_fold,
model_name=f'{model_name}_F{fold_idx+1}',
X_train=X_tr,
X_test=X_val,
y_train=y_tr,
y_test=y_val,
feature_names=feature_names,
is_log_transform=is_log_transform
)
val_pred = trained_model.predict(X_val)
train_fold_pred[valid_idx, :] = val_pred.reshape(-1, 1)
test_fold_pred[:, fold_idx] = trained_model.predict(X_test)
test_pred_mean = np.mean(test_fold_pred, axis=1).reshape(-1, 1)
return train_fold_pred, test_pred_mean- 스태킹을 위한 개별 회귀 모델 구성
각 모델은 하이퍼파라미터 수동 조절할 수 있도록 정의
# 모델 정의
lin_reg = LinearRegression()
ridge_reg = Ridge(alpha=20) # 고정
lasso_reg = Lasso(alpha=0.001) # 고정
# {'colsample_bytree': 0.8, 'learning_rate': 0.05, 'max_depth': 3, 'n_estimators': 200, 'subsample': 0.8}
xgb_reg = XGBRegressor(
n_estimators=1000, # 1000
learning_rate=0.01, # 0.01
max_depth=4, # 4
subsample=0.8, # 0.8
colsample_bytree=0.8, # 0.8
random_state=42
)
lgbm_reg = LGBMRegressor(
n_estimators=1000, # 1000
learning_rate=0.05, # 0.05
max_depth=3,
num_leaves=50, # 50
subsample=0.9, # 0.9
colsample_bytree=0.9,
random_state=42
)
rf_reg = RandomForestRegressor(
n_estimators=300,
max_depth=8,
random_state=42)- 스태킹 기반 학습/예측용 데이터 생성
"""
각 개별 모델을 기반으로 KFold 방식의 교차검증을 수행하여 스태킹 학습에 사용할 예측 결과를 생성한다.
- train_stack: 각 모델의 KFold 검증 세트 예측값 → 메타 모델 학습용
- test_stack : 테스트 세트 예측값 (Fold 평균) → 메타 모델 예측용
생성된 예측값은 모델명(key) 기준으로 딕셔너리에 저장된다.
"""
# 스태킹에 사용할 모델
models_to_stack = [
('Linear', lin_reg),
('Ridge', ridge_reg),
('Lasso', lasso_reg),
('XGBoost', xgb_reg),
('LightGBM', lgbm_reg),
('RandomForest', rf_reg),
]
# 결과 저장
stack_train_preds = {}
stack_test_preds = {}
# 각 모델에 대해 스태킹용 학습/예측 데이터 생성
for name, model in models_to_stack:
train_stack, test_stack = get_stacking_base_datasets_by_train_func(
model=model,
model_name=name,
X_train=X_train,
y_train=y_train,
X_test=X_test,
feature_names=X_features.columns,
is_log_transform=True,
n_folds=5
)
stack_train_preds[name] = train_stack
stack_test_preds[name] = test_stack
lin_train, lin_test = stack_train_preds['Linear'], stack_test_preds['Linear']
ridge_train, ridge_test = stack_train_preds['Ridge'], stack_test_preds['Ridge']
lasso_train, lasso_test = stack_train_preds['Lasso'], stack_test_preds['Lasso']
xgb_train, xgb_test = stack_train_preds['XGBoost'], stack_test_preds['XGBoost']
lgbm_train, lgbm_test = stack_train_preds['LightGBM'], stack_test_preds['LightGBM']
rf_train, rf_test = stack_train_preds['RandomForest'], stack_test_preds['RandomForest']- 메타 모델 학습을 위한 최종 스태킹 데이터 결합 및 학습
"""
앞서 생성한 개별 모델의 예측값(lin, ridge, lasso, xgb, lgbm, rf)을 수평 결합하여
메타 모델의 학습 및 테스트 입력으로 사용할 최종 피처 행렬을 구성한다.
- Stack_final_X_train: KFold 기반 개별 모델들의 train 예측값 → 메타 모델 학습용
- Stack_final_X_test : KFold 기반 개별 모델들의 test 예측값 → 메타 모델 예측용
메타 모델로는 Lasso 회귀를 사용하며, 규제 강도는 alpha=0.001로 설정되어 있다.
"""
# 스태킹 결합 시 함께 포함
Stack_final_X_train = np.concatenate(
(lin_train, ridge_train, lasso_train, xgb_train, lgbm_train, rf_train), axis=1
)
Stack_final_X_test = np.concatenate(
(lin_test, ridge_test, lasso_test, xgb_test, lgbm_test, rf_test), axis=1
)
# 메타 모델 Lasso 사용
meta_model = Lasso(alpha=0.001)
# meta_model = XGBRegressor(n_estimators=1000, learning_rate=0.01, max_depth=3, random_state=42)
meta_model.fit(Stack_final_X_train, y_train)
final_pred = meta_model.predict(Stack_final_X_test)- 스태킹 모델 성능 평가: 로그 변환 여부에 따른 전후 성능 모두 비교
- 로그 상태 성능은 모델 튜닝 및 내부 비교에 유용
- 역변환은 실제 캐글 제출 시의 평가 기준(RMSE)과 직접적으로 대응
# 평가
from sklearn.metrics import mean_squared_error, mean_absolute_error # 제출 전 MAE 비교
# 로그 복원
final_exp = np.expm1(final_pred)
y_test_exp = np.expm1(y_test)
# 평가 함수 정의
def rmsle(y_true, y_pred):
return np.sqrt(np.mean((np.log1p(y_true) - np.log1p(np.maximum(0, y_pred))) ** 2))
# 로그 상태
rmse_log = np.sqrt(mean_squared_error(y_test, final_pred))
mae_log = mean_absolute_error(y_test, final_pred)
# 로그 복원 상태
rmse = np.sqrt(mean_squared_error(y_test_exp, final_exp))
mae = mean_absolute_error(y_test_exp, final_exp)
rmsle_val = rmsle(y_test_exp, final_exp)
# 출력
print('스태킹 회귀 모델 평가 결과\n')
print(f'RMSE (log): {rmse_log:.4f}')
print(f'MAE (log): {mae_log:.4f}\n')
print(f'RMSLE : {rmsle_val:.4f}')
print(f'RMSE : {rmse:.4f}')
print(f'MAE : {mae:.4f}')성능 비교 결과, 해당 문제에서는 스태킹 모델을 다음과 같이 구성하였다.
- Base 모델: LinearRegression, Ridge, Lasso, XGBoost, LightGBM, RandomForest
- 메타 모델: Lasso(alpha=0.001)
🖇 Submission
Kaggle 제출 전, 기존의 X_train만을 대상으로 학습한 모델을 train 데이터 전체에 대해 학습하도록 스태킹 메타 모델 학습을 수행한다.
# 전체 학습 데이터 사용 (split 제거)
X_full = X_features
y_full = y_target_log # 로그 변환된 타겟
# 결과 저장
stack_train_preds = {}
stack_test_preds = {}
for name, model in models_to_stack:
train_stack, test_stack = get_stacking_base_datasets_by_train_func(
model=model,
model_name=name,
X_train=X_full, # 전체 데이터
y_train=y_full,
X_test=test_ohe2, # 실제 제출용 test 데이터 전처리 결과
feature_names=X_full.columns,
is_log_transform=True,
n_folds=5
)
stack_train_preds[name] = train_stack
stack_test_preds[name] = test_stack
# 합치기
Stack_final_X_train = np.concatenate(
[stack_train_preds[name] for name, _ in models_to_stack], axis=1
)
Stack_final_X_test_submission = np.concatenate(
[stack_test_preds[name] for name, _ in models_to_stack], axis=1
)# 1. base 모델 재학습 (훈련된 clone이 없어졌으므로 다시 fit 필요)
lin_reg.fit(X_features, y_target_log)
ridge_reg.fit(X_features, y_target_log)
lasso_reg.fit(X_features, y_target_log)
xgb_reg.fit(X_features, y_target_log)
lgbm_reg.fit(X_features, y_target_log)
rf_reg.fit(X_features, y_target_log)
# 2. 예측값 생성
lin_test_sub = lin_reg.predict(test_ohe2).reshape(-1, 1)
ridge_test_sub = ridge_reg.predict(test_ohe2).reshape(-1, 1)
lasso_test_sub = lasso_reg.predict(test_ohe2).reshape(-1, 1)
xgb_test_sub = xgb_reg.predict(test_ohe2).reshape(-1, 1)
lgbm_test_sub = lgbm_reg.predict(test_ohe2).reshape(-1, 1)
rf_test_sub = rf_reg.predict(test_ohe2).reshape(-1, 1)
# 3. 예측값 합쳐서 최종 입력 생성
Stack_final_X_test_submission = np.concatenate(
(lin_test_sub, ridge_test_sub, lasso_test_sub, xgb_test_sub, lgbm_test_sub, rf_test_sub),
axis=1
)
# 4. 메타 모델 예측 및 로그 복원
final_submission_log = meta_model.predict(Stack_final_X_test_submission)
final_submission_exp = np.expm1(final_submission_log)
submission['SalePrice'] = final_submission_exp이제 submission.csv 파일을 생성해서 제출하면 score를 확인할 수 있다.
회귀 문제에 적용할 수 있는 전체 흐름을 정리하려고 하다 보니 글이 길어졌다.
어떤 문제든 반드시 정답인 코드는 없다고 생각한다. 이 글에서 담은 로직은 아직 경험이 부족한 내가 정리한 것이니 당연히 더 효율적인 방법이 많을 것이다. (공유는 언제나 환영이다 ㅎㅎ) 이것이 토대가 되어 앞으로 더욱 고도화해 갈 수 있는 방향을 제시할 수 있을 거라 기대한다.
추후 버전 업데이트가 생기면 수정해 두겠다.
AI에 관심을 가지고, 데이터로 가치를 만들어 나가는 과정을 기록합니다.