회귀 예측

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')
 
# 재현 가능한 결과를 위한 시드 설정
np.random.seed(42)
 
# 고객 피처 데이터 생성
n_customers = 800
 
customer_features = pd.DataFrame({
    'user_id': range(1, n_customers + 1),
    'total_orders': np.random.poisson(5, n_customers) + 1,
    'total_items': np.random.poisson(15, n_customers) + 1,
    'avg_order_value': np.random.exponential(80, n_customers) + 20,
    'order_std': np.random.exponential(30, n_customers),
    'tenure_days': np.random.randint(30, 730, n_customers),
    'avg_order_gap': np.random.exponential(30, n_customers) + 5,
    'unique_categories': np.random.randint(1, 10, n_customers),
    'unique_brands': np.random.randint(1, 15, n_customers)
})
 
# CLV (타겟) 생성 - 피처와 관계가 있도록
customer_features['total_spent'] = (
    customer_features['total_orders'] * customer_features['avg_order_value'] +
    np.random.normal(0, 100, n_customers)
).clip(50, None)
 
# 결측치 처리
customer_features = customer_features.fillna(0)
 
print(f"고객 수: {len(customer_features)}")
print(f"평균 CLV: ${customer_features['total_spent'].mean():,.2f}")
print(f"CLV 중앙값: ${customer_features['total_spent'].median():,.2f}")
print(f"CLV 범위: ${customer_features['total_spent'].min():,.2f} ~ ${customer_features['total_spent'].max():,.2f}")

# 피처와 타겟 분리
feature_cols = ['total_orders', 'total_items', 'avg_order_value', 'order_std',
                'tenure_days', 'avg_order_gap', 'unique_categories', 'unique_brands']
X = customer_features[feature_cols]
y = customer_features['total_spent']
 
# 학습/테스트 분리
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
 
# 스케일링
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
 
print(f"학습 세트: {len(X_train)}건")
print(f"테스트 세트: {len(X_test)}건")
print(f"학습 CLV 평균: ${y_train.mean():,.2f}")
print(f"테스트 CLV 평균: ${y_test.mean():,.2f}")

y = β₀ + β₁x₁ + β₂x₂ + ... + βₙxₙ + ε

from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
 
# 모델 학습
lr_model = LinearRegression()
lr_model.fit(X_train_scaled, y_train)
 
# 예측
y_pred_lr = lr_model.predict(X_test_scaled)
 
# 평가
print("=== 선형 회귀 결과 ===")
print(f"MAE: ${mean_absolute_error(y_test, y_pred_lr):,.2f}")
print(f"RMSE: ${np.sqrt(mean_squared_error(y_test, y_pred_lr)):,.2f}")
print(f"R²: {r2_score(y_test, y_pred_lr):.3f}")

import matplotlib.pyplot as plt
 
# 피처별 계수
coef_df = pd.DataFrame({
    'feature': feature_cols,
    'coefficient': lr_model.coef_
}).sort_values('coefficient', key=abs, ascending=False)
 
print("\n피처별 계수 (영향력):")
print(coef_df.to_string(index=False))
 
# 시각화
plt.figure(figsize=(10, 6))
colors = ['green' if c > 0 else 'red' for c in coef_df['coefficient']]
plt.barh(coef_df['feature'], coef_df['coefficient'], color=colors)
plt.xlabel('계수')
plt.title('선형 회귀 피처 계수', fontsize=14, fontweight='bold')
plt.axvline(x=0, color='black', linestyle='-', linewidth=0.5)
plt.tight_layout()
plt.show()
 
# 해석 예시
top_feature = coef_df.iloc[0]['feature']
top_coef = coef_df.iloc[0]['coefficient']
print(f"\n해석: {top_feature}가 1 표준편차 증가하면 CLV가 ${top_coef:,.2f} 변화")

from sklearn.linear_model import Ridge
 
# 여러 alpha 값 테스트
alphas = [0.01, 0.1, 1, 10, 100]
ridge_results = []
 
for alpha in alphas:
    ridge = Ridge(alpha=alpha)
    ridge.fit(X_train_scaled, y_train)
    y_pred = ridge.predict(X_test_scaled)
    r2 = r2_score(y_test, y_pred)
    ridge_results.append({'alpha': alpha, 'r2': r2})
 
ridge_df = pd.DataFrame(ridge_results)
print("Ridge 회귀 alpha별 R²:")
print(ridge_df.to_string(index=False))
 
# 최적 alpha로 모델 학습
best_alpha = ridge_df.loc[ridge_df['r2'].idxmax(), 'alpha']
ridge_model = Ridge(alpha=best_alpha)
ridge_model.fit(X_train_scaled, y_train)
y_pred_ridge = ridge_model.predict(X_test_scaled)
 
print(f"\n최적 alpha: {best_alpha}")
print(f"Ridge R²: {r2_score(y_test, y_pred_ridge):.3f}")

from sklearn.linear_model import Lasso
 
# Lasso 회귀
lasso_model = Lasso(alpha=0.1, max_iter=10000)
lasso_model.fit(X_train_scaled, y_train)
y_pred_lasso = lasso_model.predict(X_test_scaled)
 
# 선택된 피처 (0이 아닌 계수)
selected_features = pd.DataFrame({
    'feature': feature_cols,
    'coefficient': lasso_model.coef_
})
selected_features = selected_features[selected_features['coefficient'] != 0]
print(f"Lasso 선택 피처 ({len(selected_features)}개):")
print(selected_features.to_string(index=False))
 
print(f"\nLasso R²: {r2_score(y_test, y_pred_lasso):.3f}")

from sklearn.ensemble import RandomForestRegressor
 
# 모델 학습
rf_model = RandomForestRegressor(
    n_estimators=100,
    max_depth=10,
    min_samples_split=10,
    random_state=42,
    n_jobs=-1
)
rf_model.fit(X_train, y_train)
 
# 예측 (스케일링 불필요)
y_pred_rf = rf_model.predict(X_test)
 
# 평가
print("=== 랜덤 포레스트 회귀 결과 ===")
print(f"MAE: ${mean_absolute_error(y_test, y_pred_rf):,.2f}")
print(f"RMSE: ${np.sqrt(mean_squared_error(y_test, y_pred_rf)):,.2f}")
print(f"R²: {r2_score(y_test, y_pred_rf):.3f}")

# 피처 중요도
importance_df = pd.DataFrame({
    'feature': feature_cols,
    'importance': rf_model.feature_importances_
}).sort_values('importance', ascending=False)
 
print("피처 중요도:")
print(importance_df.to_string(index=False))
 
# 시각화
plt.figure(figsize=(10, 6))
plt.barh(importance_df['feature'], importance_df['importance'], color='forestgreen')
plt.xlabel('중요도')
plt.title('랜덤 포레스트 피처 중요도', fontsize=14, fontweight='bold')
plt.tight_layout()
plt.show()

from xgboost import XGBRegressor
 
# 모델 학습
xgb_model = XGBRegressor(
    n_estimators=100,
    max_depth=6,
    learning_rate=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    random_state=42
)
xgb_model.fit(X_train, y_train)
 
# 예측
y_pred_xgb = xgb_model.predict(X_test)
 
# 평가
print("=== XGBoost 회귀 결과 ===")
print(f"MAE: ${mean_absolute_error(y_test, y_pred_xgb):,.2f}")
print(f"RMSE: ${np.sqrt(mean_squared_error(y_test, y_pred_xgb)):,.2f}")
print(f"R²: {r2_score(y_test, y_pred_xgb):.3f}")

# 모델별 성능 비교
models = {
    'Linear Regression': y_pred_lr,
    'Ridge': y_pred_ridge,
    'Lasso': y_pred_lasso,
    'Random Forest': y_pred_rf,
    'XGBoost': y_pred_xgb
}
 
results = []
for name, y_pred in models.items():
    results.append({
        '모델': name,
        'MAE': mean_absolute_error(y_test, y_pred),
        'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)),
        'R²': r2_score(y_test, y_pred)
    })
 
results_df = pd.DataFrame(results).round(2)
print("=== 모델 성능 비교 ===")
print(results_df.to_string(index=False))

fig, axes = plt.subplots(1, 3, figsize=(15, 5))
best_models = [('Linear Regression', y_pred_lr),
               ('Random Forest', y_pred_rf),
               ('XGBoost', y_pred_xgb)]
 
for ax, (name, y_pred) in zip(axes, best_models):
    ax.scatter(y_test, y_pred, alpha=0.5, s=20)
    ax.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()],
            'r--', linewidth=2, label='완벽한 예측')
    ax.set_xlabel('실제 CLV ($)')
    ax.set_ylabel('예측 CLV ($)')
    ax.set_title(f'{name}\nR² = {r2_score(y_test, y_pred):.3f}')
    ax.legend()
 
plt.tight_layout()
plt.show()

# 잔차 분석 (최고 모델 기준)
residuals = y_test - y_pred_xgb
 
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
 
# 잔차 분포
axes[0].hist(residuals, bins=30, edgecolor='black', alpha=0.7)
axes[0].axvline(x=0, color='red', linestyle='--')
axes[0].set_xlabel('잔차 ($)')
axes[0].set_ylabel('빈도')
axes[0].set_title('잔차 분포')
 
# 잔차 vs 예측값
axes[1].scatter(y_pred_xgb, residuals, alpha=0.5, s=20)
axes[1].axhline(y=0, color='red', linestyle='--')
axes[1].set_xlabel('예측 CLV ($)')
axes[1].set_ylabel('잔차 ($)')
axes[1].set_title('잔차 vs 예측값')
 
plt.tight_layout()
plt.show()
 
# 잔차 통계
print(f"잔차 평균: ${residuals.mean():,.2f}")
print(f"잔차 표준편차: ${residuals.std():,.2f}")

from sklearn.model_selection import cross_val_score
 
# 5-Fold 교차 검증
cv_scores = cross_val_score(
    xgb_model, X, y,
    cv=5,
    scoring='r2'
)
 
print("=== 5-Fold 교차 검증 ===")
print(f"R² 점수: {cv_scores.round(3)}")
print(f"평균 R²: {cv_scores.mean():.3f} (+/- {cv_scores.std():.3f})")

from sklearn.model_selection import GridSearchCV
 
# 그리드 서치
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [4, 6, 8],
    'learning_rate': [0.05, 0.1, 0.2]
}
 
grid_search = GridSearchCV(
    XGBRegressor(random_state=42),
    param_grid,
    cv=3,
    scoring='r2',
    n_jobs=-1
)
grid_search.fit(X_train, y_train)
 
print("최적 파라미터:", grid_search.best_params_)
print(f"최고 R²: {grid_search.best_score_:.3f}")