수치 예측 프로젝트

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.datasets import fetch_california_housing

# 데이터 로드
housing = fetch_california_housing(as_frame=True)
df = housing.frame
print(f"데이터 크기: {df.shape}")
print(f"\n기술 통계:\n{df.describe()}")

# 타겟 분포 확인
fig, axes = plt.subplots(1, 2, figsize=(12, 4))
df["MedHouseVal"].hist(bins=50, ax=axes[0], edgecolor="black")
axes[0].set_title("주택 가격 분포")
axes[0].set_xlabel("중위 주택 가격 ($100,000)")

np.log1p(df["MedHouseVal"]).hist(bins=50, ax=axes[1], edgecolor="black")
axes[1].set_title("주택 가격 분포 (로그 변환)")
axes[1].set_xlabel("log(중위 주택 가격)")
plt.tight_layout()
plt.show()

# 상관관계 히트맵
plt.figure(figsize=(10, 8))
sns.heatmap(df.corr(), annot=True, fmt=".2f", cmap="RdBu_r", center=0)
plt.title("변수 간 상관관계")
plt.tight_layout()
plt.show()

from sklearn.model_selection import train_test_split

X = df.drop(columns=["MedHouseVal"])
y = df["MedHouseVal"]

# 학습/테스트 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 특성 공학 함수
def create_features(df):
    """도메인 기반 새로운 특성 생성"""
    df = df.copy()

    # 방 관련 비율 특성
    df["rooms_per_household"] = df["AveRooms"] / df["AveOccup"]
    df["bedrooms_ratio"] = df["AveBedrms"] / df["AveRooms"]

    # 인구 밀도 관련
    df["population_per_household"] = df["Population"] / df["HouseAge"]

    # 위치 기반 특성 (클러스터 중심까지의 거리)
    # LA 중심 (34.05, -118.25)
    df["dist_to_la"] = np.sqrt(
        (df["Latitude"] - 34.05)**2 + (df["Longitude"] + 118.25)**2
    )
    # SF 중심 (37.77, -122.42)
    df["dist_to_sf"] = np.sqrt(
        (df["Latitude"] - 37.77)**2 + (df["Longitude"] + 122.42)**2
    )

    # 소득 구간화 (소득이 가격에 가장 큰 영향)
    df["income_cat"] = pd.cut(
        df["MedInc"],
        bins=[0, 2, 4, 6, 8, np.inf],
        labels=[1, 2, 3, 4, 5]
    ).astype(float)

    return df

X_train_fe = create_features(X_train)
X_test_fe = create_features(X_test)

print(f"원본 특성 수: {X_train.shape[1]}")
print(f"공학 후 특성 수: {X_train_fe.shape[1]}")
print(f"\n새로 추가된 특성: {set(X_train_fe.columns) - set(X_train.columns)}")

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression, Ridge, Lasso, ElasticNet
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.model_selection import cross_val_score
from lightgbm import LGBMRegressor

# 비교할 모델
models = {
    "선형 회귀": LinearRegression(),
    "Ridge": Ridge(alpha=1.0),
    "Lasso": Lasso(alpha=0.01),
    "ElasticNet": ElasticNet(alpha=0.01, l1_ratio=0.5),
    "랜덤 포레스트": RandomForestRegressor(n_estimators=100, random_state=42),
    "그래디언트 부스팅": GradientBoostingRegressor(n_estimators=200, random_state=42),
    "LightGBM": LGBMRegressor(n_estimators=200, random_state=42, verbose=-1),
}

# 교차검증 비교
results = {}
for name, model in models.items():
    pipe = Pipeline([
        ("scaler", StandardScaler()),
        ("model", model),
    ])

    # R² 점수로 평가
    r2_scores = cross_val_score(
        pipe, X_train_fe, y_train, cv=5, scoring="r2"
    )
    # RMSE로도 평가
    rmse_scores = cross_val_score(
        pipe, X_train_fe, y_train, cv=5,
        scoring="neg_root_mean_squared_error"
    )

    results[name] = {
        "R²": f"{r2_scores.mean():.4f} (+/- {r2_scores.std():.4f})",
        "RMSE": f"{-rmse_scores.mean():.4f} (+/- {rmse_scores.std():.4f})",
    }

# 결과 테이블
results_df = pd.DataFrame(results).T
print(results_df)

from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import randint, uniform

# LightGBM 튜닝 (교차검증 점수가 높은 모델)
pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LGBMRegressor(random_state=42, verbose=-1)),
])

param_dist = {
    "model__n_estimators": randint(100, 500),
    "model__max_depth": randint(3, 15),
    "model__learning_rate": uniform(0.01, 0.2),
    "model__num_leaves": randint(15, 63),
    "model__min_child_samples": randint(5, 30),
    "model__subsample": uniform(0.6, 0.4),
    "model__colsample_bytree": uniform(0.6, 0.4),
}

random_search = RandomizedSearchCV(
    pipe, param_dist,
    n_iter=50, cv=5,
    scoring="neg_root_mean_squared_error",
    random_state=42, n_jobs=-1,
)
random_search.fit(X_train_fe, y_train)

print(f"최적 RMSE: {-random_search.best_score_:.4f}")
print(f"최적 파라미터: {random_search.best_params_}")

from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

# 최종 모델로 테스트 세트 평가
y_pred = random_search.predict(X_test_fe)

# 평가 지표
rmse = np.sqrt(mean_squared_error(y_test, y_pred))
mae = mean_absolute_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)

print(f"테스트 RMSE: {rmse:.4f}")
print(f"테스트 MAE: {mae:.4f}")
print(f"테스트 R²: {r2:.4f}")

# 잔차 분석
residuals = y_test - y_pred

fig, axes = plt.subplots(1, 3, figsize=(18, 5))

# 1. 실제 vs 예측
axes[0].scatter(y_test, y_pred, alpha=0.3, s=10)
axes[0].plot([0, 5], [0, 5], "r--", linewidth=2)
axes[0].set_xlabel("실제 가격")
axes[0].set_ylabel("예측 가격")
axes[0].set_title("실제 vs 예측")

# 2. 잔차 분포
axes[1].hist(residuals, bins=50, edgecolor="black")
axes[1].axvline(0, color="r", linestyle="--")
axes[1].set_xlabel("잔차")
axes[1].set_ylabel("빈도")
axes[1].set_title("잔차 분포")

# 3. 예측값 vs 잔차 (이분산성 확인)
axes[2].scatter(y_pred, residuals, alpha=0.3, s=10)
axes[2].axhline(0, color="r", linestyle="--")
axes[2].set_xlabel("예측 가격")
axes[2].set_ylabel("잔차")
axes[2].set_title("예측값 vs 잔차")

plt.tight_layout()
plt.show()

# 특성 중요도
best_model = random_search.best_estimator_.named_steps["model"]
feat_imp = pd.Series(
    best_model.feature_importances_,
    index=X_train_fe.columns
).sort_values(ascending=True)

feat_imp.tail(15).plot(kind="barh", figsize=(10, 6))
plt.title("특성 중요도 (상위 15개)")
plt.xlabel("중요도")
plt.tight_layout()
plt.show()