백테스팅 기초 (Backtesting Basics)

과거 데이터로 투자 전략을 검증하는 방법

1. 백테스팅이란?

정의

백테스팅(Backtesting)은 투자 전략을 과거 데이터에 적용하여 해당 전략이 실제로 작동했을지 시뮬레이션하는 과정입니다.

목적

• 전략의 유효성 검증
• 수익률 및 리스크 측정
• 파라미터 최적화
• 전략 비교 및 선택

2. 백테스팅 기본 구조

"""
백테스팅 기본 템플릿
"""
import pandas as pd
import numpy as np
from datetime import datetime

class SimpleBacktest:
    def __init__(self, data: pd.DataFrame, initial_capital: float = 10000000):
        """
        Parameters:
        -----------
        data: OHLCV 데이터 (columns: date, open, high, low, close, volume)
        initial_capital: 초기 자본금 (기본 1000만원)
        """
        self.data = data.copy()
        self.initial_capital = initial_capital
        self.cash = initial_capital
        self.holdings = 0
        self.portfolio_value = []
        self.trades = []

    def generate_signals(self) -> pd.Series:
        """
        매매 신호 생성 (오버라이드 필요)
        Returns:
        --------
        pd.Series: 1 (매수), -1 (매도), 0 (홀드)
        """
        raise NotImplementedError("Subclass must implement generate_signals()")

    def execute_trade(self, date, signal, price):
        """거래 실행"""
        if signal == 1 and self.cash > 0:  # 매수
            shares = self.cash // price
            cost = shares * price
            self.cash -= cost
            self.holdings += shares
            self.trades.append({
                'date': date,
                'type': 'BUY',
                'price': price,
                'shares': shares,
                'value': cost
            })

        elif signal == -1 and self.holdings > 0:  # 매도
            revenue = self.holdings * price
            self.cash += revenue
            self.trades.append({
                'date': date,
                'type': 'SELL',
                'price': price,
                'shares': self.holdings,
                'value': revenue
            })
            self.holdings = 0

    def run(self) -> pd.DataFrame:
        """백테스트 실행"""
        signals = self.generate_signals()

        for i, row in self.data.iterrows():
            date = row['date']
            price = row['close']
            signal = signals.iloc[i] if i < len(signals) else 0

            self.execute_trade(date, signal, price)

            # 포트폴리오 가치 기록
            total_value = self.cash + (self.holdings * price)
            self.portfolio_value.append({
                'date': date,
                'cash': self.cash,
                'holdings_value': self.holdings * price,
                'total_value': total_value
            })

        return pd.DataFrame(self.portfolio_value)

    def get_metrics(self) -> dict:
        """성과 지표 계산"""
        pv = pd.DataFrame(self.portfolio_value)
        returns = pv['total_value'].pct_change().dropna()

        # 총 수익률
        total_return = (pv['total_value'].iloc[-1] / self.initial_capital) - 1

        # CAGR
        days = (pv['date'].iloc[-1] - pv['date'].iloc[0]).days
        years = days / 365
        cagr = (1 + total_return) ** (1/years) - 1 if years > 0 else 0

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

        # 샤프 비율
        sharpe = (cagr - 0.03) / volatility if volatility > 0 else 0

        # MDD
        cummax = pv['total_value'].cummax()
        drawdown = (pv['total_value'] - cummax) / cummax
        mdd = drawdown.min()

        return {
            'Total Return': f'{total_return:.2%}',
            'CAGR': f'{cagr:.2%}',
            'Volatility': f'{volatility:.2%}',
            'Sharpe Ratio': f'{sharpe:.2f}',
            'MDD': f'{mdd:.2%}',
            'Total Trades': len(self.trades)
        }

3. 간단한 전략 예시

3.1 이동평균 크로스오버 전략

class MovingAverageCrossover(SimpleBacktest):
    """
    골든크로스/데드크로스 전략
    - 단기 이평선이 장기 이평선을 상향 돌파 → 매수
    - 단기 이평선이 장기 이평선을 하향 돌파 → 매도
    """
    def __init__(self, data, short_window=20, long_window=60, **kwargs):
        super().__init__(data, **kwargs)
        self.short_window = short_window
        self.long_window = long_window

    def generate_signals(self) -> pd.Series:
        # 이동평균 계산
        short_ma = self.data['close'].rolling(self.short_window).mean()
        long_ma = self.data['close'].rolling(self.long_window).mean()

        # 신호 생성
        signals = pd.Series(0, index=self.data.index)

        # 골든크로스 (단기 > 장기로 전환)
        signals[(short_ma > long_ma) & (short_ma.shift(1) <= long_ma.shift(1))] = 1

        # 데드크로스 (단기 < 장기로 전환)
        signals[(short_ma < long_ma) & (short_ma.shift(1) >= long_ma.shift(1))] = -1

        return signals


# 사용 예시
# bt = MovingAverageCrossover(data, short_window=20, long_window=60)
# results = bt.run()
# metrics = bt.get_metrics()

3.2 RSI 전략

class RSIStrategy(SimpleBacktest):
    """
    RSI 과매도/과매수 전략
    - RSI < 30 → 매수 (과매도)
    - RSI > 70 → 매도 (과매수)
    """
    def __init__(self, data, period=14, oversold=30, overbought=70, **kwargs):
        super().__init__(data, **kwargs)
        self.period = period
        self.oversold = oversold
        self.overbought = overbought

    def calculate_rsi(self) -> pd.Series:
        delta = self.data['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=self.period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.period).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        return rsi

    def generate_signals(self) -> pd.Series:
        rsi = self.calculate_rsi()
        signals = pd.Series(0, index=self.data.index)

        signals[rsi < self.oversold] = 1   # 과매도 → 매수
        signals[rsi > self.overbought] = -1  # 과매수 → 매도

        return signals

4. 데이터 수집

4.1 한국 주식 (FinanceDataReader)

import FinanceDataReader as fdr

# 개별 종목
samsung = fdr.DataReader('005930', '2020-01-01', '2024-12-31')

# KOSPI 지수
kospi = fdr.DataReader('KS11', '2020-01-01', '2024-12-31')

# 종목 리스트
krx_list = fdr.StockListing('KRX')  # 전체
kospi_list = fdr.StockListing('KOSPI')
kosdaq_list = fdr.StockListing('KOSDAQ')

4.2 미국 주식 (yfinance)

import yfinance as yf

# 개별 종목
apple = yf.download('AAPL', start='2020-01-01', end='2024-12-31')

# 여러 종목
tickers = ['AAPL', 'GOOGL', 'MSFT', 'AMZN']
data = yf.download(tickers, start='2020-01-01', end='2024-12-31')

# 종목 정보
ticker = yf.Ticker('AAPL')
info = ticker.info
financials = ticker.financials

4.3 데이터 전처리

def prepare_data(df: pd.DataFrame) -> pd.DataFrame:
    """데이터 전처리"""
    df = df.copy()

    # 컬럼명 통일
    df.columns = [col.lower() for col in df.columns]

    # 날짜 인덱스 → 컬럼
    if isinstance(df.index, pd.DatetimeIndex):
        df = df.reset_index()
        df = df.rename(columns={'index': 'date'})

    # 결측치 처리
    df = df.dropna()

    # 정렬
    df = df.sort_values('date').reset_index(drop=True)

    return df

5. 성과 지표 상세

5.1 수익률 지표

def calculate_returns(portfolio_values: pd.Series) -> dict:
    """수익률 관련 지표"""

    # 일간 수익률
    daily_returns = portfolio_values.pct_change().dropna()

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

    # CAGR (연평균 수익률)
    n_years = len(portfolio_values) / 252
    cagr = (1 + total_return) ** (1 / n_years) - 1

    # 월간/연간 수익률
    monthly_returns = portfolio_values.resample('M').last().pct_change()
    yearly_returns = portfolio_values.resample('Y').last().pct_change()

    return {
        'daily_returns': daily_returns,
        'total_return': total_return,
        'cagr': cagr,
        'monthly_returns': monthly_returns,
        'yearly_returns': yearly_returns
    }

5.2 리스크 지표

def calculate_risk_metrics(returns: pd.Series, risk_free_rate: float = 0.03) -> dict:
    """리스크 관련 지표"""

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

    # 샤프 비율
    excess_return = returns.mean() * 252 - risk_free_rate
    sharpe_ratio = excess_return / volatility if volatility > 0 else 0

    # 소르티노 비율 (하방 변동성만 고려)
    downside_returns = returns[returns < 0]
    downside_std = downside_returns.std() * np.sqrt(252)
    sortino_ratio = excess_return / downside_std if downside_std > 0 else 0

    # 최대 낙폭 (MDD)
    cumulative = (1 + returns).cumprod()
    rolling_max = cumulative.cummax()
    drawdown = (cumulative - rolling_max) / rolling_max
    mdd = drawdown.min()

    # 칼마 비율 (CAGR / MDD)
    cagr = (cumulative.iloc[-1]) ** (252 / len(returns)) - 1
    calmar_ratio = cagr / abs(mdd) if mdd != 0 else 0

    return {
        'volatility': volatility,
        'sharpe_ratio': sharpe_ratio,
        'sortino_ratio': sortino_ratio,
        'mdd': mdd,
        'calmar_ratio': calmar_ratio
    }

5.3 거래 통계

def calculate_trade_stats(trades: list) -> dict:
    """거래 관련 통계"""
    if not trades:
        return {}

    df = pd.DataFrame(trades)

    # 매수/매도 분리
    buys = df[df['type'] == 'BUY']
    sells = df[df['type'] == 'SELL']

    # 승률 계산 (매도 기준)
    if len(sells) > 0 and len(buys) > 0:
        # 간단한 승률 계산 (실제로는 더 정교하게)
        profits = []
        for i, sell in sells.iterrows():
            buy_price = buys.iloc[min(i, len(buys)-1)]['price']
            profit = (sell['price'] - buy_price) / buy_price
            profits.append(profit)

        wins = sum(1 for p in profits if p > 0)
        win_rate = wins / len(profits) if profits else 0
    else:
        win_rate = 0

    return {
        'total_trades': len(trades),
        'buy_trades': len(buys),
        'sell_trades': len(sells),
        'win_rate': f'{win_rate:.2%}'
    }

6. 시각화

import matplotlib.pyplot as plt

def plot_backtest_results(portfolio_df: pd.DataFrame, benchmark: pd.Series = None):
    """백테스트 결과 시각화"""

    fig, axes = plt.subplots(3, 1, figsize=(12, 10))

    # 1. 포트폴리오 가치
    ax1 = axes[0]
    ax1.plot(portfolio_df['date'], portfolio_df['total_value'], label='Strategy')
    if benchmark is not None:
        ax1.plot(portfolio_df['date'], benchmark, label='Benchmark', alpha=0.7)
    ax1.set_title('Portfolio Value')
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # 2. 드로우다운
    ax2 = axes[1]
    cummax = portfolio_df['total_value'].cummax()
    drawdown = (portfolio_df['total_value'] - cummax) / cummax
    ax2.fill_between(portfolio_df['date'], drawdown, 0, color='red', alpha=0.3)
    ax2.set_title('Drawdown')
    ax2.grid(True, alpha=0.3)

    # 3. 일간 수익률
    ax3 = axes[2]
    returns = portfolio_df['total_value'].pct_change()
    ax3.bar(portfolio_df['date'], returns, alpha=0.5)
    ax3.set_title('Daily Returns')
    ax3.grid(True, alpha=0.3)

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

7. 주의사항 ⚠️

7.1 흔한 오류

7.2 검증 방법

def walk_forward_validation(data, strategy_class, n_splits=5, **strategy_params):
    """
    Walk-Forward 검증
    - 데이터를 순차적으로 분할하여 검증
    - 과최적화 방지
    """
    results = []
    split_size = len(data) // n_splits

    for i in range(n_splits - 1):
        # 학습 데이터: 처음 ~ i+1 구간
        train_end = (i + 1) * split_size
        train_data = data.iloc[:train_end]

        # 테스트 데이터: i+1 ~ i+2 구간
        test_start = train_end
        test_end = min((i + 2) * split_size, len(data))
        test_data = data.iloc[test_start:test_end]

        # 전략 실행
        bt = strategy_class(test_data, **strategy_params)
        bt.run()
        metrics = bt.get_metrics()
        metrics['period'] = f'Split {i+1}'
        results.append(metrics)

    return pd.DataFrame(results)

8. 다음 단계

• [ ] 더 복잡한 전략 구현 (팩터 기반)

• [ ] 포트폴리오 최적화 추가

• [ ] 실시간 데이터 연동

• [ ] 자동 매매 시스템 구축

• 실전 전략 코드 템플릿

• 자동매매 시스템 구축

• 특정 전략 상세 설명 등

#백테스팅 #퀀트투자 #Python #알고리즘트레이딩