Jongheon Kim
b64a76a8b9
- Introduced a `backtest_strategy` endpoint to enable strategy backtesting with user-specified parameters. - Implemented a generic backtesting engine allowing rebalancing, equity curve tracking, and performance metric calculations. - Added `BacktestMixin` for strategies to support backtesting-related operations. - Extended BAA strategy to support backtesting with ticker data download and portfolio simulation. - Updated `urls.py` to include the new backtesting endpoint. - Enhanced logging and error handling throughout the backtesting process.
293 lines
9.7 KiB
Python
293 lines
9.7 KiB
Python
"""
|
|
백테스트 엔진 모듈
|
|
|
|
전략에 무관한 범용 백테스팅 로직을 제공합니다.
|
|
BacktestMixin을 구현한 전략이면 어떤 전략이든 백테스트할 수 있습니다.
|
|
"""
|
|
|
|
import math
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from ..base import BacktestMixin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def run_backtest(strategy_instance, parameters):
|
|
"""범용 백테스트 실행
|
|
|
|
Args:
|
|
strategy_instance: BacktestMixin을 구현한 전략 인스턴스
|
|
parameters: 백테스트 파라미터 (backtest_start_date, backtest_end_date, initial_capital 등)
|
|
|
|
Returns:
|
|
백테스트 결과 딕셔너리
|
|
"""
|
|
if not isinstance(strategy_instance, BacktestMixin):
|
|
raise ValueError(f'{strategy_instance.__class__.__name__} does not implement BacktestMixin')
|
|
|
|
# 파라미터 추출
|
|
start_date_str = parameters['backtest_start_date']
|
|
end_date_str = parameters['backtest_end_date']
|
|
initial_capital = parameters.get('initial_capital', 100000)
|
|
transaction_cost_rate = parameters.get('transaction_cost', 0.001)
|
|
|
|
start_date = datetime.strptime(start_date_str, '%Y-%m-%d')
|
|
end_date = datetime.strptime(end_date_str, '%Y-%m-%d')
|
|
|
|
# 1. 티커 수집
|
|
tickers = strategy_instance.get_backtest_tickers(parameters)
|
|
|
|
# 2. 전체 기간 데이터 일괄 다운로드
|
|
data_cache = strategy_instance.bulk_download_data(tickers, start_date, end_date)
|
|
|
|
if not data_cache:
|
|
raise ValueError('Failed to download market data')
|
|
|
|
# 3. 리밸런싱 날짜 생성 (월말 영업일)
|
|
rebalance_dates = pd.date_range(
|
|
start=start_date, end=end_date, freq='BMS'
|
|
)
|
|
# BMS = Business Month Start -> 첫 영업일이지만, 실제로는 월말에 리밸런싱하므로 BMonthEnd 사용
|
|
rebalance_dates = pd.date_range(
|
|
start=start_date, end=end_date, freq='BME'
|
|
)
|
|
|
|
if len(rebalance_dates) == 0:
|
|
raise ValueError('No rebalancing dates in the given period')
|
|
|
|
# 4. 모든 티커의 공통 거래일 인덱스 구축
|
|
all_dates = set()
|
|
for series in data_cache.values():
|
|
all_dates.update(series.index)
|
|
all_dates = sorted(all_dates)
|
|
all_trading_dates = pd.DatetimeIndex(all_dates)
|
|
# start_date 이후의 거래일만
|
|
all_trading_dates = all_trading_dates[all_trading_dates >= pd.Timestamp(start_date)]
|
|
all_trading_dates = all_trading_dates[all_trading_dates <= pd.Timestamp(end_date)]
|
|
|
|
if len(all_trading_dates) == 0:
|
|
raise ValueError('No trading dates in the given period')
|
|
|
|
# 5. 백테스트 루프
|
|
portfolio_value = float(initial_capital)
|
|
current_weights = {} # ticker -> weight
|
|
prev_weights = {}
|
|
|
|
daily_equity_curve = []
|
|
monthly_returns_list = []
|
|
allocation_history = []
|
|
|
|
# 리밸런싱 날짜를 set으로 변환하여 빠른 조회
|
|
rebal_date_set = set()
|
|
for rd in rebalance_dates:
|
|
# 가장 가까운 거래일 찾기 (rd 이하)
|
|
candidates = all_trading_dates[all_trading_dates <= rd]
|
|
if len(candidates) > 0:
|
|
rebal_date_set.add(candidates[-1])
|
|
|
|
rebal_dates_sorted = sorted(rebal_date_set)
|
|
|
|
month_start_value = portfolio_value
|
|
last_rebal_idx = -1
|
|
|
|
for i, date in enumerate(all_trading_dates):
|
|
# 일별 포트폴리오 가치 업데이트 (리밸런싱 사이)
|
|
if current_weights and i > 0:
|
|
prev_date = all_trading_dates[i - 1]
|
|
daily_return = 0.0
|
|
for ticker, weight in current_weights.items():
|
|
if ticker in data_cache:
|
|
series = data_cache[ticker]
|
|
if date in series.index and prev_date in series.index:
|
|
prev_price = float(series[prev_date])
|
|
curr_price = float(series[date])
|
|
if prev_price > 0:
|
|
ticker_return = (curr_price / prev_price) - 1
|
|
daily_return += weight * ticker_return
|
|
portfolio_value *= (1 + daily_return)
|
|
|
|
# 리밸런싱 체크
|
|
if date in rebal_date_set:
|
|
result = strategy_instance.calculate_portfolio_for_date(
|
|
parameters, date, data_cache
|
|
)
|
|
new_weights = result['portfolio_weights']
|
|
mode = result['mode']
|
|
|
|
# 턴오버 계산
|
|
turnover = _calculate_turnover(current_weights, new_weights)
|
|
cost = portfolio_value * turnover * transaction_cost_rate
|
|
portfolio_value -= cost
|
|
|
|
prev_weights = current_weights.copy()
|
|
current_weights = new_weights.copy()
|
|
|
|
allocation_history.append({
|
|
'date': date.strftime('%Y-%m-%d'),
|
|
'mode': mode,
|
|
'weights': {k: round(v, 4) for k, v in new_weights.items()},
|
|
'turnover': round(turnover, 4),
|
|
'cost': round(cost, 2)
|
|
})
|
|
|
|
# 일별 equity curve 기록
|
|
daily_equity_curve.append({
|
|
'date': date.strftime('%Y-%m-%d'),
|
|
'value': round(portfolio_value, 2)
|
|
})
|
|
|
|
# 월말 수익률 기록
|
|
if date in rebal_date_set:
|
|
monthly_return = (portfolio_value / month_start_value) - 1 if month_start_value > 0 else 0
|
|
monthly_returns_list.append({
|
|
'date': date.strftime('%Y-%m-%d'),
|
|
'return': round(monthly_return, 6)
|
|
})
|
|
month_start_value = portfolio_value
|
|
|
|
# 6. 성과 지표 계산
|
|
daily_values = [d['value'] for d in daily_equity_curve]
|
|
monthly_rets = [m['return'] for m in monthly_returns_list]
|
|
|
|
metrics = calculate_performance_metrics(daily_values, monthly_rets, initial_capital)
|
|
|
|
# drawdown 시리즈 계산
|
|
drawdown_series = _calculate_drawdown_series(daily_equity_curve)
|
|
|
|
return {
|
|
'strategy': strategy_instance.name,
|
|
'variant': parameters.get('variant', ''),
|
|
'backtest_start_date': start_date_str,
|
|
'backtest_end_date': end_date_str,
|
|
'initial_capital': initial_capital,
|
|
'final_capital': round(portfolio_value, 2),
|
|
'metrics': metrics,
|
|
'daily_equity_curve': daily_equity_curve,
|
|
'monthly_returns': monthly_returns_list,
|
|
'allocation_history': allocation_history,
|
|
'drawdown_series': drawdown_series,
|
|
}
|
|
|
|
|
|
def calculate_performance_metrics(daily_values, monthly_returns, initial_capital):
|
|
"""백테스트 성과 지표 계산
|
|
|
|
Args:
|
|
daily_values: 일별 포트폴리오 가치 리스트
|
|
monthly_returns: 월별 수익률 리스트
|
|
initial_capital: 초기 자본
|
|
|
|
Returns:
|
|
성과 지표 딕셔너리
|
|
"""
|
|
if not daily_values or len(daily_values) < 2:
|
|
return {}
|
|
|
|
final_value = daily_values[-1]
|
|
total_return = (final_value / initial_capital - 1) * 100
|
|
|
|
# 일별 수익률
|
|
daily_returns = []
|
|
for i in range(1, len(daily_values)):
|
|
if daily_values[i - 1] > 0:
|
|
daily_returns.append(daily_values[i] / daily_values[i - 1] - 1)
|
|
|
|
if not daily_returns:
|
|
return {}
|
|
|
|
daily_returns_arr = np.array(daily_returns)
|
|
trading_days = len(daily_returns)
|
|
years = trading_days / 252
|
|
|
|
# CAGR
|
|
if years > 0 and initial_capital > 0:
|
|
cagr = (final_value / initial_capital) ** (1 / years) - 1
|
|
else:
|
|
cagr = 0
|
|
|
|
# 연율화 변동성
|
|
annual_volatility = float(np.std(daily_returns_arr) * np.sqrt(252))
|
|
|
|
# 최대 낙폭 (MDD)
|
|
peak = daily_values[0]
|
|
max_dd = 0
|
|
for val in daily_values:
|
|
if val > peak:
|
|
peak = val
|
|
dd = (peak - val) / peak
|
|
if dd > max_dd:
|
|
max_dd = dd
|
|
|
|
# Sharpe ratio (무위험 수익률 0 가정)
|
|
if annual_volatility > 0:
|
|
sharpe_ratio = cagr / annual_volatility
|
|
else:
|
|
sharpe_ratio = 0
|
|
|
|
# Sortino ratio
|
|
downside_returns = daily_returns_arr[daily_returns_arr < 0]
|
|
if len(downside_returns) > 0:
|
|
downside_vol = float(np.std(downside_returns) * np.sqrt(252))
|
|
sortino_ratio = cagr / downside_vol if downside_vol > 0 else 0
|
|
else:
|
|
sortino_ratio = 0
|
|
|
|
# Calmar ratio
|
|
calmar_ratio = cagr / max_dd if max_dd > 0 else 0
|
|
|
|
# 월별 통계
|
|
if monthly_returns:
|
|
win_months = sum(1 for r in monthly_returns if r > 0)
|
|
win_rate = (win_months / len(monthly_returns)) * 100
|
|
best_month = max(monthly_returns) * 100
|
|
worst_month = min(monthly_returns) * 100
|
|
else:
|
|
win_rate = 0
|
|
best_month = 0
|
|
worst_month = 0
|
|
|
|
return {
|
|
'cagr': round(cagr * 100, 2),
|
|
'total_return': round(total_return, 2),
|
|
'annual_volatility': round(annual_volatility * 100, 2),
|
|
'max_drawdown': round(max_dd * 100, 2),
|
|
'sharpe_ratio': round(sharpe_ratio, 2),
|
|
'sortino_ratio': round(sortino_ratio, 2),
|
|
'calmar_ratio': round(calmar_ratio, 2),
|
|
'win_rate': round(win_rate, 2),
|
|
'best_month': round(best_month, 2),
|
|
'worst_month': round(worst_month, 2),
|
|
}
|
|
|
|
|
|
def _calculate_turnover(old_weights, new_weights):
|
|
"""포트폴리오 턴오버 계산 (0~2 범위)"""
|
|
all_tickers = set(list(old_weights.keys()) + list(new_weights.keys()))
|
|
turnover = 0.0
|
|
for ticker in all_tickers:
|
|
old_w = old_weights.get(ticker, 0)
|
|
new_w = new_weights.get(ticker, 0)
|
|
turnover += abs(new_w - old_w)
|
|
return turnover
|
|
|
|
|
|
def _calculate_drawdown_series(daily_equity_curve):
|
|
"""drawdown 시리즈 계산"""
|
|
drawdowns = []
|
|
peak = 0
|
|
for entry in daily_equity_curve:
|
|
val = entry['value']
|
|
if val > peak:
|
|
peak = val
|
|
dd = (val - peak) / peak if peak > 0 else 0
|
|
drawdowns.append({
|
|
'date': entry['date'],
|
|
'drawdown': round(dd * 100, 2)
|
|
})
|
|
return drawdowns
|