080 시계열 실전 - 주가 예측

키워드: 주가 예측, stock prediction

개요

주가 예측은 시계열 분석의 대표적인 응용 분야입니다. 이 글에서는 주가 데이터의 특성을 이해하고, 예측 모델을 구축하는 방법을 다룹니다. 단, 실제 투자에는 신중한 접근이 필요합니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: pycaret[full]>=3.0

주가 예측의 어려움

주가 예측이 어려운 이유:

1. 효율적 시장 가설 (EMH)
   - 모든 정보가 이미 가격에 반영
   - 예측 불가능성

2. 비정상성
   - 평균, 분산이 시간에 따라 변함
   - 추세 변화, 변동성 클러스터링

3. 외부 요인
   - 뉴스, 정책, 글로벌 이벤트
   - 예측 불가능한 충격

4. 노이즈
   - 신호 대 잡음비 낮음
   - 과적합 위험

주의: 이 예제는 교육 목적입니다.
실제 투자 결정에 사용하지 마세요.

데이터 준비 (시뮬레이션)

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

np.random.seed(42)

# 3년간 일별 데이터 (주말 제외)
dates = pd.bdate_range('2021-01-01', periods=756)  # 약 3년 영업일

# 080 GBM (Geometric Brownian Motion) 시뮬레이션
initial_price = 100
mu = 0.0002  # 일별 수익률 기대값
sigma = 0.015  # 일별 변동성

returns = np.random.normal(mu, sigma, len(dates))
price = initial_price * np.cumprod(1 + returns)

# 080 OHLCV 데이터 생성
data = pd.DataFrame({
    'date': dates,
    'close': price,
    'open': price * (1 + np.random.uniform(-0.005, 0.005, len(dates))),
    'high': price * (1 + np.abs(np.random.normal(0, 0.01, len(dates)))),
    'low': price * (1 - np.abs(np.random.normal(0, 0.01, len(dates)))),
    'volume': np.random.lognormal(15, 0.5, len(dates)).astype(int)
})

# 080 high/low 조정
data['high'] = data[['open', 'close', 'high']].max(axis=1)
data['low'] = data[['open', 'close', 'low']].min(axis=1)

data.set_index('date', inplace=True)

print(f"데이터 기간: {data.index.min()} ~ {data.index.max()}")
print(f"시작가: ${data['close'].iloc[0]:.2f}")
print(f"종가: ${data['close'].iloc[-1]:.2f}")
print(f"수익률: {(data['close'].iloc[-1]/data['close'].iloc[0] - 1)*100:.1f}%")

탐색적 데이터 분석

import matplotlib.pyplot as plt

fig, axes = plt.subplots(2, 2, figsize=(15, 10))

# 1. 주가 차트
axes[0, 0].plot(data.index, data['close'], 'b-', linewidth=1)
axes[0, 0].set_title('Stock Price')
axes[0, 0].set_xlabel('Date')
axes[0, 0].set_ylabel('Price ($)')

# 2. 수익률 분포
returns = data['close'].pct_change().dropna()
axes[0, 1].hist(returns, bins=50, edgecolor='black', alpha=0.7)
axes[0, 1].axvline(x=0, color='red', linestyle='--')
axes[0, 1].set_title('Daily Returns Distribution')
axes[0, 1].set_xlabel('Return')
axes[0, 1].set_ylabel('Frequency')

# 3. 거래량
axes[1, 0].bar(data.index, data['volume'], width=1, alpha=0.7)
axes[1, 0].set_title('Trading Volume')
axes[1, 0].set_xlabel('Date')
axes[1, 0].set_ylabel('Volume')

# 4. 변동성 (20일 이동 표준편차)
volatility = returns.rolling(20).std() * np.sqrt(252)  # 연간화
axes[1, 1].plot(volatility.index, volatility.values)
axes[1, 1].set_title('20-Day Rolling Volatility (Annualized)')
axes[1, 1].set_xlabel('Date')
axes[1, 1].set_ylabel('Volatility')

plt.tight_layout()
plt.savefig('stock_eda.png', dpi=150)

기술적 지표 생성

import pandas as pd
import numpy as np

def create_technical_indicators(df):
    """기술적 지표 생성"""
    df = df.copy()

    # 수익률
    df['return_1d'] = df['close'].pct_change()
    df['return_5d'] = df['close'].pct_change(5)
    df['return_20d'] = df['close'].pct_change(20)

    # 이동 평균
    df['sma_5'] = df['close'].rolling(5).mean()
    df['sma_20'] = df['close'].rolling(20).mean()
    df['sma_60'] = df['close'].rolling(60).mean()

    # 이동 평균 비율
    df['price_sma5_ratio'] = df['close'] / df['sma_5']
    df['price_sma20_ratio'] = df['close'] / df['sma_20']
    df['sma5_sma20_ratio'] = df['sma_5'] / df['sma_20']

    # 지수 이동 평균
    df['ema_12'] = df['close'].ewm(span=12).mean()
    df['ema_26'] = df['close'].ewm(span=26).mean()

    # MACD
    df['macd'] = df['ema_12'] - df['ema_26']
    df['macd_signal'] = df['macd'].ewm(span=9).mean()
    df['macd_hist'] = df['macd'] - df['macd_signal']

    # RSI
    delta = df['close'].diff()
    gain = delta.where(delta > 0, 0).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / (loss + 1e-10)
    df['rsi'] = 100 - (100 / (1 + rs))

    # 볼린저 밴드
    df['bb_middle'] = df['close'].rolling(20).mean()
    df['bb_std'] = df['close'].rolling(20).std()
    df['bb_upper'] = df['bb_middle'] + 2 * df['bb_std']
    df['bb_lower'] = df['bb_middle'] - 2 * df['bb_std']
    df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_middle']
    df['bb_position'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'] + 1e-10)

    # ATR (Average True Range)
    high_low = df['high'] - df['low']
    high_close = (df['high'] - df['close'].shift()).abs()
    low_close = (df['low'] - df['close'].shift()).abs()
    true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    df['atr'] = true_range.rolling(14).mean()
    df['atr_ratio'] = df['atr'] / df['close']

    # 변동성
    df['volatility_20d'] = df['return_1d'].rolling(20).std()
    df['volatility_60d'] = df['return_1d'].rolling(60).std()

    # 거래량 지표
    df['volume_sma_20'] = df['volume'].rolling(20).mean()
    df['volume_ratio'] = df['volume'] / df['volume_sma_20']

    # Lag 특성
    for lag in [1, 5, 10]:
        df[f'close_lag_{lag}'] = df['close'].shift(lag)
        df[f'return_lag_{lag}'] = df['return_1d'].shift(lag)

    return df

# 080 적용
data_features = create_technical_indicators(data)
print(f"생성된 특성 수: {data_features.shape[1]}")
print(data_features.columns.tolist())

PyCaret 모델링

from pycaret.time_series import *

# 080 타겟: 종가
ts_data = data[['close']].copy()

# 080 환경 설정
ts = setup(
    data=ts_data,
    target='close',
    fh=5,  # 5일 예측
    fold=5,
    session_id=42,
    verbose=False
)

# 080 모델 비교
print("=== 모델 비교 ===")
best = compare_models(n_select=3)

수익률 예측 접근법

from pycaret.regression import *
import pandas as pd
import numpy as np

# 080 수익률 예측 (분류/회귀)
data_ml = data_features.copy()
data_ml['target'] = data_ml['close'].shift(-5) / data_ml['close'] - 1  # 5일 후 수익률
data_ml = data_ml.dropna()

# 080 특성 선택
feature_cols = ['return_1d', 'return_5d', 'return_20d',
                'price_sma5_ratio', 'price_sma20_ratio', 'sma5_sma20_ratio',
                'macd', 'macd_hist', 'rsi', 'bb_position', 'bb_width',
                'atr_ratio', 'volatility_20d', 'volume_ratio',
                'return_lag_1', 'return_lag_5']

# 080 학습/테스트 분할 (시간 순서)
train_size = int(len(data_ml) * 0.8)
train = data_ml[:train_size]
test = data_ml[train_size:]

# 080 PyCaret 회귀
reg = setup(
    data=train[feature_cols + ['target']],
    target='target',
    session_id=42,
    verbose=False
)

# 080 모델 비교
best_reg = compare_models(n_select=3)

방향 예측 (분류)

from pycaret.classification import *
import pandas as pd
import numpy as np

# 080 방향 예측 (상승/하락)
data_class = data_features.copy()
data_class['direction'] = (data_class['close'].shift(-5) > data_class['close']).astype(int)
data_class = data_class.dropna()

# 080 학습/테스트 분할
train_size = int(len(data_class) * 0.8)
train_class = data_class[:train_size]
test_class = data_class[train_size:]

# 080 PyCaret 분류
clf = setup(
    data=train_class[feature_cols + ['direction']],
    target='direction',
    session_id=42,
    verbose=False
)

# 080 모델 비교
best_clf = compare_models(n_select=3)

# 080 테스트
predictions = predict_model(best_clf[0], data=test_class[feature_cols])
accuracy = (predictions['prediction_label'] == test_class['direction']).mean()
print(f"\n방향 예측 정확도: {accuracy:.2%}")

백테스팅

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

def simple_backtest(prices, signals, initial_capital=10000):
    """간단한 백테스팅"""

    position = 0  # 0: 현금, 1: 주식 보유
    capital = initial_capital
    shares = 0
    portfolio_values = []

    for i in range(len(prices)):
        price = prices[i]
        signal = signals[i]

        if signal == 1 and position == 0:  # 매수 신호
            shares = capital / price
            capital = 0
            position = 1
        elif signal == 0 and position == 1:  # 매도 신호
            capital = shares * price
            shares = 0
            position = 0

        # 포트폴리오 가치
        if position == 1:
            portfolio_values.append(shares * price)
        else:
            portfolio_values.append(capital)

    return portfolio_values

# 080 예측 기반 신호
test_prices = test_class['close'].values
if 'prediction_label' in predictions.columns:
    signals = predictions['prediction_label'].values
else:
    signals = np.random.binomial(1, 0.5, len(test_prices))  # 더미

# 080 백테스트
portfolio = simple_backtest(test_prices, signals)

# 080 벤치마크 (Buy & Hold)
buy_hold = 10000 * test_prices / test_prices[0]

# 080 시각화
plt.figure(figsize=(14, 6))
plt.plot(test_class.index, portfolio, label='Strategy', linewidth=2)
plt.plot(test_class.index, buy_hold, label='Buy & Hold', linewidth=2)
plt.xlabel('Date')
plt.ylabel('Portfolio Value ($)')
plt.title('Backtest: Strategy vs Buy & Hold')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('backtest.png', dpi=150)

# 080 성과 비교
strategy_return = (portfolio[-1] / portfolio[0] - 1) * 100
benchmark_return = (buy_hold[-1] / buy_hold[0] - 1) * 100

print(f"\n=== 백테스트 결과 ===")
print(f"전략 수익률: {strategy_return:.2f}%")
print(f"벤치마크 수익률: {benchmark_return:.2f}%")
print(f"초과 수익: {strategy_return - benchmark_return:.2f}%")

리스크 지표

import numpy as np
import pandas as pd

def calculate_risk_metrics(portfolio_values):
    """리스크 지표 계산"""

    returns = pd.Series(portfolio_values).pct_change().dropna()

    # 총 수익률
    total_return = (portfolio_values[-1] / portfolio_values[0] - 1) * 100

    # 연환산 수익률
    days = len(portfolio_values)
    annual_return = ((portfolio_values[-1] / portfolio_values[0]) ** (252 / days) - 1) * 100

    # 변동성 (연환산)
    volatility = returns.std() * np.sqrt(252) * 100

    # 샤프 비율 (무위험 이자율 2% 가정)
    risk_free_rate = 0.02
    sharpe = (annual_return / 100 - risk_free_rate) / (volatility / 100)

    # 최대 낙폭 (MDD)
    peak = pd.Series(portfolio_values).cummax()
    drawdown = (pd.Series(portfolio_values) - peak) / peak * 100
    max_drawdown = drawdown.min()

    return {
        'Total Return (%)': total_return,
        'Annual Return (%)': annual_return,
        'Volatility (%)': volatility,
        'Sharpe Ratio': sharpe,
        'Max Drawdown (%)': max_drawdown
    }

# 080 전략 리스크 지표
strategy_metrics = calculate_risk_metrics(portfolio)
benchmark_metrics = calculate_risk_metrics(buy_hold.tolist())

print("\n=== 리스크 지표 비교 ===")
metrics_df = pd.DataFrame({
    'Strategy': strategy_metrics,
    'Benchmark': benchmark_metrics
})
print(metrics_df.round(2))

주의사항

주가 예측 시 주의점:

1. 과적합 (Overfitting)
   - 과거에 맞춤, 미래에 실패
   - 아웃오브샘플 검증 필수

2. 룩어헤드 바이어스
   - 미래 정보 사용 금지
   - shift() 주의

3. 생존자 바이어스
   - 상장폐지 종목 제외된 데이터
   - 실제보다 좋아 보임

4. 거래 비용
   - 수수료, 슬리피지
   - 실제 수익 감소

5. 시장 영향
   - 대량 거래 시 가격 영향
   - 백테스트와 실제 괴리

실제 투자 결정에는
전문가 상담을 권장합니다.

정리

• 주가 예측: 높은 난이도, 많은 위험
• 기술적 지표: 이동평균, RSI, MACD 등
• 백테스팅: 전략 검증 필수
• 리스크 관리: 샤프 비율, MDD
• 교육 목적으로만 사용

다음 글 예고

다음 글에서는 **하이퍼파라미터 튜닝 (tune_model)**을 다룹니다.

---

PyCaret 머신러닝 마스터 시리즈 #080