""" 백테스트 엔진 모듈 전략에 무관한 범용 백테스팅 로직을 제공합니다. 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