XGBoost 퀀트 매매 신호 예측

왜 XGBoost인가?

XGBoost(Extreme Gradient Boosting)는 퀀트 트레이딩에서 가장 널리 쓰이는 머신러닝 알고리즘 중 하나입니다. 결정 트리 앙상블 기반으로 비선형 패턴을 포착하면서도, 과적합 방지 기능이 내장되어 있어 금융 데이터의 노이즈가 많은 환경에서 강력한 성능을 발휘합니다.

딥러닝(LSTM, Transformer)에 비해 학습 속도가 빠르고, 피처 중요도를 직관적으로 해석할 수 있어 실전 자동매매 시스템에 적합합니다. 이 글에서는 XGBoost로 매매 신호를 예측하고 자동매매에 활용하는 전체 파이프라인을 구축합니다.

피처 엔지니어링: 매매 신호의 원재료

모델 성능의 80%는 피처 품질에 달려 있습니다. 가격, 거래량, 기술적 지표를 조합하여 예측력 있는 피처를 설계합니다.

import pandas as pd
import numpy as np
import ta

def create_features(df):
    """OHLCV 데이터에서 기술적 지표 피처 생성"""
    features = pd.DataFrame(index=df.index)

    # 수익률 기반 피처
    for period in [1, 3, 5, 10, 20]:
        features[f'return_{period}d'] = df['close'].pct_change(period)

    # 변동성 피처
    features['volatility_10'] = df['close'].pct_change().rolling(10).std()
    features['volatility_20'] = df['close'].pct_change().rolling(20).std()
    features['vol_ratio'] = features['volatility_10'] / features['volatility_20']

    # 이동평균 크로스 피처
    features['sma_5_20'] = (
        df['close'].rolling(5).mean() / df['close'].rolling(20).mean() - 1
    )
    features['sma_10_50'] = (
        df['close'].rolling(10).mean() / df['close'].rolling(50).mean() - 1
    )

    # RSI
    features['rsi_14'] = ta.momentum.RSIIndicator(df['close'], 14).rsi()
    features['rsi_7'] = ta.momentum.RSIIndicator(df['close'], 7).rsi()

    # MACD
    macd = ta.trend.MACD(df['close'])
    features['macd_diff'] = macd.macd_diff()
    features['macd_signal_dist'] = macd.macd() - macd.macd_signal()

    # 볼린저 밴드 위치
    bb = ta.volatility.BollingerBands(df['close'], 20, 2)
    features['bb_position'] = (
        (df['close'] - bb.bollinger_lband()) /
        (bb.bollinger_hband() - bb.bollinger_lband())
    )

    # 거래량 피처
    features['volume_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
    features['volume_trend'] = df['volume'].rolling(5).mean() / df['volume'].rolling(20).mean()

    # ATR (Average True Range)
    features['atr_ratio'] = (
        ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], 14).average_true_range()
        / df['close']
    )

    return features.dropna()

핵심은 절대값보다 비율(ratio)을 사용하는 것입니다. 가격 수준이 달라져도 모델이 일반화할 수 있도록 정규화된 피처를 설계합니다.

레이블 생성: 무엇을 예측할 것인가

단순히 “오를까 내릴까”보다는, 일정 수익률 이상 움직임을 예측하는 3-class 분류가 실전에 더 유용합니다.

def create_labels(df, forward_period=5, threshold=0.02):
    """
    미래 수익률 기반 레이블 생성
    - 1: 상승 (수익률 > threshold)
    - 0: 중립 (-threshold ~ threshold)
    - -1: 하락 (수익률 < -threshold)
    """
    future_return = df['close'].pct_change(forward_period).shift(-forward_period)

    labels = pd.Series(0, index=df.index)
    labels[future_return > threshold] = 1
    labels[future_return < -threshold] = -1

    return labels

# 사용 예시
# labels = create_labels(df, forward_period=5, threshold=0.02)
# 5일 후 2% 이상 상승 → 매수 신호
# 5일 후 2% 이상 하락 → 매도 신호

threshold는 거래 비용을 고려하여 설정합니다. 너무 낮으면 노이즈에 반응하고, 너무 높으면 신호가 부족해집니다.

시계열 교차 검증과 모델 학습

금융 데이터에서 일반적인 k-fold 교차 검증을 사용하면 미래 정보가 학습에 유입(data leakage)됩니다. 반드시 시계열 분할(TimeSeriesSplit)을 사용해야 합니다.

import xgboost as xgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import classification_report, f1_score

class QuantXGBoostModel:
    def __init__(self, params=None):
        self.params = params or {
            'objective': 'multi:softprob',
            'num_class': 3,
            'max_depth': 4,
            'learning_rate': 0.05,
            'subsample': 0.8,
            'colsample_bytree': 0.8,
            'reg_alpha': 1.0,
            'reg_lambda': 1.0,
            'min_child_weight': 10,
            'eval_metric': 'mlogloss',
            'tree_method': 'hist',
            'seed': 42
        }
        self.model = None

    def train_with_cv(self, X, y, n_splits=5):
        """시계열 교차 검증으로 학습"""
        tscv = TimeSeriesSplit(n_splits=n_splits)
        cv_scores = []

        for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
            X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
            y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]

            # 레이블을 0, 1, 2로 매핑 (-1→0, 0→1, 1→2)
            y_train_mapped = y_train + 1
            y_val_mapped = y_val + 1

            dtrain = xgb.DMatrix(X_train, label=y_train_mapped)
            dval = xgb.DMatrix(X_val, label=y_val_mapped)

            model = xgb.train(
                self.params,
                dtrain,
                num_boost_round=500,
                evals=[(dval, 'val')],
                early_stopping_rounds=30,
                verbose_eval=False
            )

            preds = model.predict(dval).argmax(axis=1) - 1
            score = f1_score(y_val, preds, average='macro')
            cv_scores.append(score)
            print(f"Fold {fold+1} F1: {score:.4f}")

        print(f"n평균 F1: {np.mean(cv_scores):.4f} ± {np.std(cv_scores):.4f}")

        # 전체 데이터로 최종 모델 학습
        y_mapped = y + 1
        dtrain_full = xgb.DMatrix(X, label=y_mapped)
        self.model = xgb.train(self.params, dtrain_full, num_boost_round=300)

        return cv_scores

    def predict(self, X):
        """예측 확률 반환"""
        dtest = xgb.DMatrix(X)
        probs = self.model.predict(dtest)
        return probs  # [하락 확률, 중립 확률, 상승 확률]

과적합 방지를 위해 max_depth=4, min_child_weight=10, L1/L2 정규화를 적극 활용합니다. 이는 백테스트 과적합 방지의 핵심 원칙과 일맥상통합니다.

피처 중요도 분석

XGBoost의 가장 큰 장점 중 하나는 피처 중요도를 시각적으로 분석할 수 있다는 것입니다.

import matplotlib.pyplot as plt

def analyze_feature_importance(model, feature_names, top_n=15):
    """피처 중요도 분석 및 시각화"""
    importance = model.get_score(importance_type='gain')

    # 피처명 매핑
    mapped = {
        feature_names[int(k.replace('f', ''))]: v
        for k, v in importance.items()
    }

    sorted_imp = sorted(mapped.items(), key=lambda x: x[1], reverse=True)

    print("=== Top Feature Importance (Gain) ===")
    for name, score in sorted_imp[:top_n]:
        print(f"  {name:30s} {score:.2f}")

    # 불필요한 피처 제거 후 재학습하면 과적합 감소
    low_importance = [name for name, score in sorted_imp if score < 1.0]
    print(f"n제거 후보 ({len(low_importance)}개): {low_importance}")

    return sorted_imp

중요도가 낮은 피처를 제거하면 모델이 단순해지면서 일반화 성능이 향상됩니다. 보통 상위 10~15개 피처만으로도 전체 성능의 90% 이상을 재현할 수 있습니다.

확률 기반 포지션 사이징

XGBoost는 각 클래스의 확률을 출력합니다. 이 확률을 활용하면 확신도에 비례한 포지션 크기를 설정할 수 있습니다.

class ProbabilityBasedTrader:
    def __init__(self, model, max_position=1.0, min_confidence=0.5):
        self.model = model
        self.max_position = max_position
        self.min_confidence = min_confidence

    def generate_signal(self, features):
        """확률 기반 매매 신호 및 포지션 크기 결정"""
        probs = self.model.predict(features)
        # probs: [P(하락), P(중립), P(상승)]

        p_down, p_neutral, p_up = probs[0]

        signal = {
            'direction': 0,
            'size': 0,
            'confidence': 0,
            'probabilities': {
                'down': round(p_down, 4),
                'neutral': round(p_neutral, 4),
                'up': round(p_up, 4)
            }
        }

        if p_up > self.min_confidence:
            signal['direction'] = 1  # 매수
            signal['confidence'] = p_up
            # 확신도에 비례한 포지션 크기
            signal['size'] = self.max_position * (
                (p_up - self.min_confidence) / (1 - self.min_confidence)
            )

        elif p_down > self.min_confidence:
            signal['direction'] = -1  # 매도
            signal['confidence'] = p_down
            signal['size'] = self.max_position * (
                (p_down - self.min_confidence) / (1 - self.min_confidence)
            )

        return signal

예를 들어 상승 확률 70%이면 포지션의 50%만 진입하고, 90%이면 전체를 투입합니다. 이 방식은 켈리 공식 자금 관리와 결합하면 더욱 정교한 자금 관리가 가능합니다.

Walk-Forward 최적화

실전에서는 모델을 한 번 학습하고 끝내는 것이 아니라, 주기적으로 재학습(Walk-Forward)하여 시장 변화에 적응해야 합니다.

class WalkForwardOptimizer:
    def __init__(self, train_window=252, test_window=21, retrain_every=21):
        self.train_window = train_window   # 학습 기간 (거래일)
        self.test_window = test_window     # 테스트 기간
        self.retrain_every = retrain_every # 재학습 주기

    def run(self, X, y):
        """Walk-Forward 방식으로 모델 평가"""
        all_preds = []
        all_actuals = []

        start = self.train_window
        while start + self.test_window <= len(X):
            # 학습 구간
            X_train = X.iloc[start - self.train_window:start]
            y_train = y.iloc[start - self.train_window:start]

            # 테스트 구간
            X_test = X.iloc[start:start + self.test_window]
            y_test = y.iloc[start:start + self.test_window]

            # 모델 학습
            model = QuantXGBoostModel()
            y_mapped = y_train + 1
            dtrain = xgb.DMatrix(X_train, label=y_mapped)
            model.model = xgb.train(model.params, dtrain, num_boost_round=200)

            # 예측
            preds = model.predict(X_test).argmax(axis=1) - 1
            all_preds.extend(preds)
            all_actuals.extend(y_test.values)

            start += self.retrain_every

        print(classification_report(all_actuals, all_preds,
              target_names=['하락', '중립', '상승']))

        return all_preds, all_actuals

Walk-Forward는 미래 데이터 유출 없이 실전과 동일한 조건에서 모델을 평가할 수 있어, 백테스트 결과의 신뢰도가 크게 높아집니다.

실전 파이프라인 구성

단계	내용	주기
데이터 수집	OHLCV + 온체인 데이터 수집	실시간
피처 생성	기술적 지표 계산	봉 마감 시
모델 예측	XGBoost 확률 출력	봉 마감 시
신호 생성	확률 기반 매수/매도/홀드	봉 마감 시
주문 실행	거래소 API 연동 주문	신호 발생 시
모델 재학습	Walk-Forward 방식 업데이트	월 1회
성과 모니터링	F1, 샤프비율, MDD 추적	일 1회

주의사항과 실전 팁

• 과적합 위험: 피처가 많을수록 과적합 위험 증가 — 상위 10~15개만 사용
• 레짐 변화: 시장 구조가 바뀌면 모델 성능 급락 — 성능 모니터링 필수
• 거래 비용: 백테스트에 슬리피지·수수료 반드시 반영
• 앙상블: XGBoost + LightGBM + CatBoost 앙상블로 안정성 향상 가능
• 대안 레이블: 고정 threshold 대신 Triple Barrier Method 활용 추천
• SHAP 분석: 피처 중요도보다 정교한 해석이 필요하면 SHAP 값 활용

마무리

XGBoost는 해석 가능성, 학습 속도, 과적합 방지를 모두 갖춘 실전 퀀트 트레이딩의 핵심 도구입니다. 피처 엔지니어링으로 양질의 입력을 만들고, 시계열 교차 검증과 Walk-Forward로 신뢰도 높은 평가를 수행하며, 확률 기반 포지션 사이징으로 리스크를 관리하는 것이 성공의 열쇠입니다.

다음 단계로는 리스크 관리 시스템과 통합하여 실시간 자동매매 파이프라인을 완성해 보시기 바랍니다.