048 다중 출력 회귀

키워드: 다중 출력, multi-output, 다중 타겟

개요

다중 출력 회귀(Multi-Output Regression)는 여러 타겟 변수를 동시에 예측하는 문제입니다. 예를 들어 주택의 가격과 임대료를 동시에 예측하거나, 제품의 판매량과 수익을 함께 예측할 수 있습니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: flaml[automl], scikit-learn, numpy

pip install flaml[automl] scikit-learn numpy matplotlib

다중 출력 회귀 개념

문제 유형

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split

# 048 다중 출력 데이터 생성
np.random.seed(42)

# 3개의 타겟을 가진 회귀 문제
X, Y = make_regression(
    n_samples=1000,
    n_features=10,
    n_targets=3,  # 다중 출력
    noise=0.1,
    random_state=42
)

print("다중 출력 회귀 데이터:")
print(f"  특성 shape: {X.shape}")
print(f"  타겟 shape: {Y.shape}")
print(f"  타겟 수: {Y.shape[1]}")

실제 예시 데이터

# 048 주택 데이터에서 다중 타겟 시뮬레이션
from sklearn.datasets import fetch_california_housing

data = fetch_california_housing()
X = data.data

# 048 가격에서 파생된 추가 타겟 생성
y_price = data.target
y_rental = y_price * 0.004 + np.random.randn(len(y_price)) * 0.001  # 월세 (가격의 0.4%)
y_tax = y_price * 0.012 + np.random.randn(len(y_price)) * 0.002  # 재산세 (가격의 1.2%)

# 048 다중 타겟으로 결합
Y_multi = np.column_stack([y_price, y_rental, y_tax])
target_names = ['가격', '월세', '재산세']

print("\n주택 다중 타겟 데이터:")
print(f"  타겟: {target_names}")

# 048 타겟 간 상관관계
correlation = np.corrcoef(Y_multi.T)
print("\n타겟 간 상관관계:")
for i, name1 in enumerate(target_names):
    for j, name2 in enumerate(target_names):
        if i < j:
            print(f"  {name1} - {name2}: {correlation[i,j]:.4f}")

다중 출력 회귀 방법

방법 1: 개별 모델 학습

from flaml import AutoML
from sklearn.metrics import r2_score, mean_squared_error

# 048 데이터 분할
X_train, X_test, Y_train, Y_test = train_test_split(
    X, Y_multi, test_size=0.2, random_state=42
)

# 048 각 타겟별로 개별 모델 학습
individual_models = {}
individual_predictions = []

print("방법 1: 개별 모델 학습")
print("-" * 50)

for i, name in enumerate(target_names):
    automl = AutoML()
    automl.fit(
        X_train, Y_train[:, i],
        task="regression",
        time_budget=30,
        metric="r2",
        verbose=0
    )
    individual_models[name] = automl

    y_pred = automl.predict(X_test)
    individual_predictions.append(y_pred)

    r2 = r2_score(Y_test[:, i], y_pred)
    print(f"  {name}: R² = {r2:.4f}, 모델 = {automl.best_estimator}")

# 048 예측 결합
Y_pred_individual = np.column_stack(individual_predictions)

방법 2: MultiOutputRegressor

from sklearn.multioutput import MultiOutputRegressor
from sklearn.ensemble import RandomForestRegressor

print("\n방법 2: MultiOutputRegressor")
print("-" * 50)

# 048 래퍼 사용
base_model = RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
multi_model = MultiOutputRegressor(base_model)
multi_model.fit(X_train, Y_train)

Y_pred_multi = multi_model.predict(X_test)

for i, name in enumerate(target_names):
    r2 = r2_score(Y_test[:, i], Y_pred_multi[:, i])
    print(f"  {name}: R² = {r2:.4f}")

방법 3: 체인 회귀 (Chained Regression)

from sklearn.multioutput import RegressorChain

print("\n방법 3: RegressorChain (타겟 의존성 활용)")
print("-" * 50)

# 048 타겟 간 의존성 학습
chain_model = RegressorChain(
    RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1),
    order=[0, 1, 2]  # 가격 → 월세 → 재산세 순서
)
chain_model.fit(X_train, Y_train)

Y_pred_chain = chain_model.predict(X_test)

for i, name in enumerate(target_names):
    r2 = r2_score(Y_test[:, i], Y_pred_chain[:, i])
    print(f"  {name}: R² = {r2:.4f}")

FLAML로 다중 출력 구현

커스텀 다중 출력 래퍼

class FLAMLMultiOutput:
    """FLAML 기반 다중 출력 회귀"""

    def __init__(self, time_budget_per_target=30):
        self.time_budget = time_budget_per_target
        self.models = {}
        self.n_targets = None

    def fit(self, X, Y, target_names=None):
        self.n_targets = Y.shape[1]
        self.target_names = target_names or [f'target_{i}' for i in range(self.n_targets)]

        print("FLAML 다중 출력 학습:")
        for i, name in enumerate(self.target_names):
            automl = AutoML()
            automl.fit(
                X, Y[:, i],
                task="regression",
                time_budget=self.time_budget,
                metric="r2",
                verbose=0
            )
            self.models[name] = automl
            print(f"  {name}: {automl.best_estimator}")

        return self

    def predict(self, X):
        predictions = []
        for name in self.target_names:
            pred = self.models[name].predict(X)
            predictions.append(pred)
        return np.column_stack(predictions)

    def score(self, X, Y):
        Y_pred = self.predict(X)
        scores = {}
        for i, name in enumerate(self.target_names):
            scores[name] = r2_score(Y[:, i], Y_pred[:, i])
        return scores

# 048 사용
flaml_multi = FLAMLMultiOutput(time_budget_per_target=30)
flaml_multi.fit(X_train, Y_train, target_names)

scores = flaml_multi.score(X_test, Y_test)
print("\nFLAML 다중 출력 성능:")
for name, score in scores.items():
    print(f"  {name}: R² = {score:.4f}")

공유 특성 학습

class SharedFeatureFLAML:
    """특성 공유 다중 출력 모델"""

    def __init__(self, time_budget=60):
        self.time_budget = time_budget
        self.main_model = None
        self.target_models = {}

    def fit(self, X, Y, target_names=None):
        self.n_targets = Y.shape[1]
        self.target_names = target_names or [f'target_{i}' for i in range(self.n_targets)]

        # 첫 번째 타겟으로 메인 특성 학습
        print("1단계: 메인 모델 학습")
        self.main_model = AutoML()
        self.main_model.fit(
            X, Y[:, 0],
            task="regression",
            time_budget=self.time_budget,
            verbose=0
        )
        print(f"   메인 모델: {self.main_model.best_estimator}")

        # 나머지 타겟은 메인 예측을 특성으로 추가
        main_pred_train = self.main_model.predict(X).reshape(-1, 1)
        X_augmented = np.hstack([X, main_pred_train])

        print("2단계: 보조 타겟 학습 (특성 공유)")
        for i in range(1, self.n_targets):
            model = AutoML()
            model.fit(
                X_augmented, Y[:, i],
                task="regression",
                time_budget=self.time_budget // 2,
                verbose=0
            )
            self.target_models[self.target_names[i]] = model
            print(f"   {self.target_names[i]}: {model.best_estimator}")

        return self

    def predict(self, X):
        predictions = []

        # 메인 타겟 예측
        main_pred = self.main_model.predict(X)
        predictions.append(main_pred)

        # 보조 타겟 예측
        X_augmented = np.hstack([X, main_pred.reshape(-1, 1)])
        for name in self.target_names[1:]:
            pred = self.target_models[name].predict(X_augmented)
            predictions.append(pred)

        return np.column_stack(predictions)

# 048 사용 (시간이 오래 걸릴 수 있음)
# 048 shared_model = SharedFeatureFLAML(time_budget=30)
# 048 shared_model.fit(X_train, Y_train, target_names)

방법 비교

# 048 결과 비교
methods = {
    'Individual FLAML': Y_pred_individual if 'Y_pred_individual' in dir() else None,
    'MultiOutputRegressor': Y_pred_multi,
    'RegressorChain': Y_pred_chain,
}

print("\n" + "=" * 60)
print("방법별 성능 비교")
print("=" * 60)

results = []
for method, Y_pred in methods.items():
    if Y_pred is not None:
        row = {'Method': method}
        for i, name in enumerate(target_names):
            row[f'{name}_R2'] = r2_score(Y_test[:, i], Y_pred[:, i])
        row['Avg_R2'] = np.mean([row[f'{name}_R2'] for name in target_names])
        results.append(row)

results_df = pd.DataFrame(results)
print(results_df.to_string(index=False))

시각화

# 048 예측 비교 시각화
fig, axes = plt.subplots(1, 3, figsize=(15, 5))

for i, name in enumerate(target_names):
    axes[i].scatter(Y_test[:, i], Y_pred_multi[:, i], alpha=0.3, s=10)
    min_val = min(Y_test[:, i].min(), Y_pred_multi[:, i].min())
    max_val = max(Y_test[:, i].max(), Y_pred_multi[:, i].max())
    axes[i].plot([min_val, max_val], [min_val, max_val], 'r--')
    axes[i].set_xlabel(f'Actual {name}')
    axes[i].set_ylabel(f'Predicted {name}')
    r2 = r2_score(Y_test[:, i], Y_pred_multi[:, i])
    axes[i].set_title(f'{name} (R² = {r2:.4f})')

plt.tight_layout()
plt.show()

언제 다중 출력을 사용할까?

guide = {
    '상황': ['타겟 독립적', '타겟 상관 높음', '타겟 순서 있음', '계산 자원 제한'],
    '권장 방법': ['개별 모델', 'MultiOutput', 'RegressorChain', 'MultiOutput'],
    '이유': [
        '타겟 간 정보 공유 불필요',
        '공유 표현 학습 가능',
        '순차적 의존성 활용',
        '한 번 학습으로 여러 예측'
    ]
}

print("\n다중 출력 방법 선택 가이드:")
print(pd.DataFrame(guide).to_string(index=False))

정리

• 다중 출력 회귀: 여러 타겟을 동시에 예측
• 개별 모델: 가장 단순, 타겟 독립 시 적합
• MultiOutputRegressor: 기본 모델 래핑, 효율적
• RegressorChain: 타겟 간 의존성 학습
• FLAML은 각 타겟별로 최적 모델 탐색 가능
• 타겟 상관관계가 높으면 공유 학습이 유리

다음 글 예고

다음 글에서는 회귀 모델 해석에 대해 알아보겠습니다. SHAP, 부분 의존성 플롯 등 회귀 모델을 해석하는 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #048