Python의 Pandas와 yfinance를 활용한 주식 데이터 분석에 이어, 오늘은 이렇게 수집한 데이터를 바탕으로 투자 전략을 백테스트하고 평가하는 방법에 대해 알아보겠습니다.
백테스트란?
백테스트(Backtest)는 과거 데이터를 사용하여 투자 전략의 성능을 시뮬레이션하는 방법입니다. 이는 실제 시장에 투자하기 전에 전략의 효과를 검증할 수 있는 중요한 단계입니다.
백테스트를 위한 환경 구성
먼저 필요한 라이브러리를 설치하고 불러옵니다.
# 필요한 라이브러리 설치
!pip install pandas numpy yfinance matplotlib backtesting
# 라이브러리 불러오기
import pandas as pd
import numpy as np
import yfinance as yf
import matplotlib.pyplot as plt
from datetime import datetime, timedelta간단한 이동평균 교차 전략 구현 및 백테스트
이동평균선 교차 전략은 가장 기본적인 기술적 분석 전략 중 하나입니다. 단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수하고, 하향 돌파할 때 매도하는 전략입니다.
def simple_ma_strategy(data, short_window=20, long_window=50):
# 데이터 복사본 생성
signals = data.copy()
# 이동평균 계산
signals['short_ma'] = signals['Close'].rolling(window=short_window, min_periods=1).mean()
signals['long_ma'] = signals['Close'].rolling(window=long_window, min_periods=1).mean()
# 신호 생성 (1: 매수, -1: 매도, 0: 관망)
signals['signal'] = 0
signals['signal'][short_window:] = np.where(signals['short_ma'][short_window:] > signals['long_ma'][short_window:], 1, 0)
# 포지션 변화 탐지
signals['position'] = signals['signal'].diff()
# 초기 자본금 설정
initial_capital = 100000
positions = pd.DataFrame(index=signals.index).fillna(0.0)
# 주식 보유량과 현금 계산
positions['asset'] = 100 * signals['signal'] # 1주당 100달러 투자
positions['holdings'] = positions['asset'] * signals['Close']
# 현금 흐름 계산
portfolio = positions.copy()
pos_diff = positions['asset'].diff()
portfolio['cash_flow'] = -pos_diff * signals['Close']
portfolio['cash_flow'].iloc[0] = initial_capital - positions['holdings'].iloc[0]
portfolio['cash'] = portfolio['cash_flow'].cumsum()
# 총 가치 계산
portfolio['total'] = portfolio['cash'] + positions['holdings']
portfolio['returns'] = portfolio['total'].pct_change()
return signals, portfolio
# 데이터 불러오기
start_date = datetime.now() - timedelta(days=365*5) # 5년 데이터
end_date = datetime.now()
data = yf.download('AAPL', start=start_date, end=end_date)
# 전략 적용
signals, portfolio = simple_ma_strategy(data)
# 결과 시각화
plt.figure(figsize=(12, 8))
# 가격 및 이동평균선 시각화
plt.subplot(2, 1, 1)
plt.plot(data.index, data['Close'], label='AAPL')
plt.plot(signals.index, signals['short_ma'], label='Short MA')
plt.plot(signals.index, signals['long_ma'], label='Long MA')
plt.plot(signals.loc[signals['position'] == 1].index,
signals['short_ma'][signals['position'] == 1],
'^', markersize=10, color='g', label='Buy')
plt.plot(signals.loc[signals['position'] == -1].index,
signals['short_ma'][signals['position'] == -1],
'v', markersize=10, color='r', label='Sell')
plt.legend()
plt.title('이동평균 교차 전략 (AAPL)')
# 포트폴리오 가치 시각화
plt.subplot(2, 1, 2)
plt.plot(portfolio.index, portfolio['total'], label='Portfolio Value')
plt.plot(data.index, initial_capital * (1 + data['Close'].pct_change().cumsum()),
label='Buy & Hold Strategy')
plt.legend()
plt.title('포트폴리오 가치 변화')
plt.tight_layout()
plt.show()백테스트 평가 지표
투자 전략의 성과를 평가하기 위한 주요 지표들입니다.
def evaluate_strategy(portfolio, risk_free_rate=0.02):
# 연간 수익률
annual_return = portfolio['returns'].mean() * 252
# 연간 변동성
annual_volatility = portfolio['returns'].std() * np.sqrt(252)
# 샤프 비율
sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility
# 최대 낙폭 (MDD)
cumulative_returns = (1 + portfolio['returns']).cumprod()
peak = cumulative_returns.expanding().max()
drawdown = (cumulative_returns / peak) - 1
max_drawdown = drawdown.min()
# 승률
winning_days = portfolio['returns'][portfolio['returns'] > 0].count()
total_days = portfolio['returns'].count()
win_rate = winning_days / total_days
# 결과 출력
evaluation = {
'Annual Return': annual_return,
'Annual Volatility': annual_volatility,
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown': max_drawdown,
'Win Rate': win_rate
}
return evaluation
# 전략 평가
evaluation = evaluate_strategy(portfolio)
print("전략 평가 결과:")
for key, value in evaluation.items():
print(f"{key}: {value:.4f}")여러 주식에 대한 전략 테스트
동일한 전략을 여러 종목에 적용하여 전략의 일반적인 성능을 평가해봅니다.
def test_strategy_on_multiple_stocks(stocks, start_date, end_date, strategy_func, **strategy_params):
results = {}
for stock in stocks:
try:
# 데이터 불러오기
data = yf.download(stock, start=start_date, end=end_date)
# 전략 적용
signals, portfolio = strategy_func(data, **strategy_params)
# 평가
evaluation = evaluate_strategy(portfolio)
results[stock] = evaluation
except Exception as e:
print(f"Error processing {stock}: {e}")
# 결과를 데이터프레임으로 변환
results_df = pd.DataFrame(results).T
return results_df
# 테스트할 주식 목록
stocks = ['AAPL', 'MSFT', 'GOOG', 'AMZN', 'META', 'TSLA', 'NVDA', 'JPM', 'V', 'WMT']
# 여러 주식에 전략 적용
start_date = datetime.now() - timedelta(days=365*3) # 3년 데이터
end_date = datetime.now()
results = test_strategy_on_multiple_stocks(stocks, start_date, end_date,
simple_ma_strategy,
short_window=20, long_window=50)
# 결과 출력
print("여러 주식에 대한 전략 평가 결과:")
print(results)
# 결과 시각화
plt.figure(figsize=(14, 8))
# 연간 수익률 비교
plt.subplot(2, 2, 1)
results['Annual Return'].sort_values().plot(kind='bar')
plt.title('연간 수익률')
plt.axhline(y=0, color='r', linestyle='-')
# 샤프 비율 비교
plt.subplot(2, 2, 2)
results['Sharpe Ratio'].sort_values().plot(kind='bar')
plt.title('샤프 비율')
plt.axhline(y=0, color='r', linestyle='-')
# 최대 낙폭 비교
plt.subplot(2, 2, 3)
results['Max Drawdown'].sort_values().plot(kind='bar')
plt.title('최대 낙폭')
# 변동성 비교
plt.subplot(2, 2, 4)
results['Annual Volatility'].sort_values().plot(kind='bar')
plt.title('연간 변동성')
plt.tight_layout()
plt.show()전략 최적화
그리드 서치를 통해 최적의 파라미터를 찾아봅니다.
def optimize_strategy(data, strategy_func, param_grid):
best_sharpe = -np.inf
best_params = None
results = []
# 모든 파라미터 조합 시도
for short_window in param_grid['short_window']:
for long_window in param_grid['long_window']:
if short_window >= long_window:
continue
# 전략 적용
signals, portfolio = strategy_func(data, short_window=short_window, long_window=long_window)
# 평가
evaluation = evaluate_strategy(portfolio)
evaluation['short_window'] = short_window
evaluation['long_window'] = long_window
results.append(evaluation)
# 최고 샤프 비율 업데이트
if evaluation['Sharpe Ratio'] > best_sharpe:
best_sharpe = evaluation['Sharpe Ratio']
best_params = {'short_window': short_window, 'long_window': long_window}
# 결과를 데이터프레임으로 변환
results_df = pd.DataFrame(results)
return best_params, results_df
# 최적화를 위한 파라미터 그리드
param_grid = {
'short_window': range(5, 30, 5),
'long_window': range(30, 100, 10)
}
# 데이터 불러오기
data = yf.download('SPY', start=start_date, end=end_date)
# 전략 최적화
best_params, optimization_results = optimize_strategy(data, simple_ma_strategy, param_grid)
print(f"최적 파라미터: {best_params}")
print("최적화 결과 상위 5개:")
print(optimization_results.sort_values('Sharpe Ratio', ascending=False).head())
# 히트맵으로 시각화
heatmap_data = optimization_results.pivot(index='short_window',
columns='long_window',
values='Sharpe Ratio')
plt.figure(figsize=(10, 8))
plt.pcolor(heatmap_data.columns, heatmap_data.index, heatmap_data)
plt.colorbar(label='Sharpe Ratio')
plt.xlabel('Long Window')
plt.ylabel('Short Window')
plt.title('이동평균 파라미터 최적화 (Sharpe Ratio)')
plt.show()워크포워드 분석
과거 최적화된 파라미터를 미래 기간에 적용하여 전략의 견고성을 테스트합니다.
def walk_forward_analysis(data, strategy_func, train_size=0.6, step=60):
results = []
# 데이터 분할
total_days = len(data)
train_days = int(total_days * train_size)
for i in range(0, total_days - train_days, step):
if i + train_days + step > total_days:
break
# 훈련 및 테스트 데이터 분할
train_data = data.iloc[i:i+train_days]
test_data = data.iloc[i+train_days:i+train_days+step]
# 훈련 데이터로 최적화
param_grid = {'short_window': range(5, 30, 5), 'long_window': range(30, 100, 10)}
best_params, _ = optimize_strategy(train_data, strategy_func, param_grid)
# 테스트 데이터에 최적화된 파라미터 적용
signals, portfolio = strategy_func(test_data, **best_params)
evaluation = evaluate_strategy(portfolio)
# 결과 저장
result = {
'period_start': test_data.index[0],
'period_end': test_data.index[-1],
'best_short_window': best_params['short_window'],
'best_long_window': best_params['long_window']
}
result.update(evaluation)
results.append(result)
# 결과를 데이터프레임으로 변환
results_df = pd.DataFrame(results)
return results_df
# 5년 데이터 불러오기
data = yf.download('SPY', start=datetime.now()-timedelta(days=365*5), end=datetime.now())
# 워크포워드 분석 실행
walk_forward_results = walk_forward_analysis(data, simple_ma_strategy)
print("워크포워드 분석 결과:")
print(walk_forward_results)
# 결과 시각화
plt.figure(figsize=(14, 10))
# 기간별 수익률
plt.subplot(3, 1, 1)
plt.bar(range(len(walk_forward_results)), walk_forward_results['Annual Return'])
plt.axhline(y=0, color='r', linestyle='-')
plt.title('기간별 연간 수익률')
# 최적 파라미터 변화
plt.subplot(3, 1, 2)
plt.plot(walk_forward_results['period_start'], walk_forward_results['best_short_window'],
label='Short Window')
plt.plot(walk_forward_results['period_start'], walk_forward_results['best_long_window'],
label='Long Window')
plt.legend()
plt.title('최적 파라미터 변화')
# 샤프 비율 변화
plt.subplot(3, 1, 3)
plt.plot(walk_forward_results['period_start'], walk_forward_results['Sharpe Ratio'])
plt.axhline(y=0, color='r', linestyle='-')
plt.title('샤프 비율 변화')
plt.tight_layout()
plt.show()고급 백테스트: Backtesting.py 라이브러리 활용
더 복잡한 백테스트를 위해 Backtesting.py 라이브러리를 활용합니다.
from backtesting import Backtest, Strategy
from backtesting.lib import crossover
class SmaCrossStrategy(Strategy):
# 전략 파라미터 정의
n1 = 20 # 단기 이동평균
n2 = 50 # 장기 이동평균
def init(self):
# 이동평균 계산
self.sma1 = self.I(lambda: self.data.Close.rolling(self.n1).mean())
self.sma2 = self.I(lambda: self.data.Close.rolling(self.n2).mean())
def next(self):
# 매수 신호: 단기 이동평균이 장기 이동평균을 상향 돌파
if crossover(self.sma1, self.sma2):
self.buy()
# 매도 신호: 단기 이동평균이 장기 이동평균을 하향 돌파
elif crossover(self.sma2, self.sma1):
self.sell()
# 데이터 준비
data = yf.download('AAPL', start=start_date, end=end_date)
# 백테스트 실행
bt = Backtest(data, SmaCrossStrategy,
cash=100000, commission=.002,
exclusive_orders=True)
stats = bt.run()
# 결과 출력
print(stats)
# 백테스트 결과 시각화
bt.plot()백테스트 결과 해석 및 주의사항
def interpret_backtest_results(stats):
print("=== 백테스트 결과 해석 ===")
# 기본 성능 평가
print(f"총 수익률: {stats['Return [%]']:.2f}%")
print(f"연간 수익률: {stats['Annual Return [%]']:.2f}%")
print(f"샤프 비율: {stats['Sharpe Ratio']:.2f}")
print(f"최대 낙폭: {stats['Max. Drawdown [%]']:.2f}%")
print(f"승률: {stats['Win Rate [%]']:.2f}%")
# 해석
if stats['Sharpe Ratio'] > 1:
print("✅ 샤프 비율이 1 이상으로 위험 대비 수익이 양호합니다.")
else:
print("⚠️ 샤프 비율이 1 미만으로 위험 대비 수익이 낮습니다.")
if stats['Max. Drawdown [%]'] > 20:
print("⚠️ 최대 낙폭이 20%를 초과하여 리스크가 큽니다.")
else:
print("✅ 최대 낙폭이 20% 이하로 리스크가 관리 가능한 수준입니다.")
if stats['Win Rate [%]'] < 40:
print("⚠️ 승률이 40% 미만으로 전략의 일관성이 낮을 수 있습니다.")
else:
print("✅ 승률이 40% 이상으로 전략의 일관성이 양호합니다.")
# 주의사항
print("\n=== 백테스트 주의사항 ===")
print("1. 과적합(Overfitting) 위험: 특정 기간에 최적화된 전략이 미래에도 잘 작동한다는 보장이 없습니다.")
print("2. 룩어헤드 바이어스(Look-ahead bias): 미래 정보를 실수로 사용하지 않도록 주의해야 합니다.")
print("3. 생존 바이어스(Survivorship bias): 현재 살아남은 기업만 분석하면 결과가 왜곡될 수 있습니다.")
print("4. 거래 비용: 수수료, 슬리피지, 세금 등의 거래 비용을 고려해야 합니다.")
print("5. 유동성: 실제 시장에서는 원하는 가격에 거래가 불가능할 수 있습니다.")
# 결과 해석
interpret_backtest_results(stats)결론
투자 전략의 백테스트는 실제 자금을 투입하기 전에 전략의 성능을 평가하는 중요한 단계입니다. 하지만 과거의 성과가 미래의 성과를 보장하지는 않는다는 것을 항상 명심해야 합니다. 백테스트 결과를 해석할 때는 다양한 시장 조건에서의 성능, 최대 낙폭, 위험 조정 수익률 등을 종합적으로 고려해야 합니다.
효과적인 백테스트를 위해서는 다음과 같은 접근 방식이 중요합니다:
- 충분한 기간의 데이터 사용 (최소 3-5년)
- 다양한 시장 환경 포함 (상승장, 하락장, 횡보장)
- 거래 비용 및 슬리피지 고려
- 워크포워드 분석을 통한 과적합 방지
- 다양한 자산군에 대한 테스트
이러한 방법을 통해 보다 신뢰할 수 있는 투자 전략을 개발하고 검증할 수 있습니다. 그러나 백테스트는 참고 자료일 뿐, 실전 투자에서는 시장 상황의 변화에 따라 전략을 유연하게 조정하는 것이 중요합니다.