"""Backtest engine core implementation.""" from typing import Dict, List, Any from decimal import Decimal from datetime import datetime, timedelta from dateutil.relativedelta import relativedelta from sqlalchemy.orm import Session from app.backtest.portfolio import BacktestPortfolio from app.backtest.rebalancer import Rebalancer from app.backtest.metrics import ( calculate_total_return, calculate_cagr, calculate_max_drawdown, calculate_sharpe_ratio, calculate_sortino_ratio, calculate_win_rate, calculate_volatility, calculate_calmar_ratio ) class BacktestEngine: """백테스트 엔진.""" def __init__( self, initial_capital: float = 10000000.0, commission_rate: float = 0.0015, rebalance_frequency: str = 'monthly' ): """ 초기화. Args: initial_capital: 초기 자본금 (기본 1천만원) commission_rate: 수수료율 (기본 0.15%) rebalance_frequency: 리밸런싱 주기 ('monthly', 'quarterly', 'yearly') """ self.initial_capital = Decimal(str(initial_capital)) self.commission_rate = Decimal(str(commission_rate)) self.rebalance_frequency = rebalance_frequency self.portfolio = BacktestPortfolio( initial_capital=self.initial_capital, commission_rate=self.commission_rate ) self.rebalancer = Rebalancer(self.portfolio) self.equity_curve: List[Dict] = [] self.all_trades: List[Dict] = [] def run( self, strategy, start_date: datetime, end_date: datetime, db_session: Session ) -> Dict[str, Any]: """ 백테스트 실행. Args: strategy: 전략 객체 (BaseStrategy 인터페이스 구현) start_date: 시작일 end_date: 종료일 db_session: 데이터베이스 세션 Returns: 백테스트 결과 딕셔너리 """ # 리밸런싱 날짜 생성 rebalance_dates = self._generate_rebalance_dates(start_date, end_date) print(f"백테스트 시작: {start_date.date()} ~ {end_date.date()}") print(f"리밸런싱 주기: {self.rebalance_frequency} ({len(rebalance_dates)}회)") # 각 리밸런싱 날짜에 전략 실행 for i, rebal_date in enumerate(rebalance_dates): print(f"\n[{i+1}/{len(rebalance_dates)}] 리밸런싱: {rebal_date.date()}") # 전략 실행 → 종목 선정 selected_stocks = strategy.select_stocks( rebal_date=rebal_date, db_session=db_session ) if not selected_stocks: print(" 선정된 종목 없음") continue # 현재 가격 조회 current_prices = strategy.get_prices( tickers=selected_stocks, date=rebal_date, db_session=db_session ) if not current_prices: print(" 가격 정보 없음") continue # 리밸런싱 sell_trades, buy_trades = self.rebalancer.rebalance( target_tickers=selected_stocks, current_prices=current_prices, trade_date=rebal_date ) print(f" 매도: {len(sell_trades)}건, 매수: {len(buy_trades)}건") # 거래 기록 self.all_trades.extend(sell_trades) self.all_trades.extend(buy_trades) # 스냅샷 저장 snapshot = self.portfolio.take_snapshot(rebal_date) self.equity_curve.append({ 'date': rebal_date, 'value': float(snapshot.total_value), 'cash': float(snapshot.cash), 'positions_value': float(snapshot.positions_value) }) # 성과 분석 results = self._calculate_results() print(f"\n{'='*50}") print(f"백테스트 완료") print(f"총 수익률: {results['total_return_pct']:.2f}%") print(f"CAGR: {results['cagr']:.2f}%") print(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}") print(f"MDD: {results['max_drawdown_pct']:.2f}%") print(f"승률: {results['win_rate_pct']:.2f}%") print(f"{'='*50}") return results def _generate_rebalance_dates( self, start_date: datetime, end_date: datetime ) -> List[datetime]: """ 리밸런싱 날짜 생성. Args: start_date: 시작일 end_date: 종료일 Returns: 리밸런싱 날짜 리스트 """ dates = [] current = start_date while current <= end_date: dates.append(current) if self.rebalance_frequency == 'monthly': current += relativedelta(months=1) elif self.rebalance_frequency == 'quarterly': current += relativedelta(months=3) elif self.rebalance_frequency == 'yearly': current += relativedelta(years=1) else: # 기본값: 월간 current += relativedelta(months=1) return dates def _calculate_results(self) -> Dict[str, Any]: """ 성과 지표 계산. Returns: 성과 지표 딕셔너리 """ if not self.equity_curve: return self._empty_results() # 최종 자산 final_value = Decimal(str(self.equity_curve[-1]['value'])) # 총 수익률 total_return_pct = calculate_total_return(self.initial_capital, final_value) # CAGR (연평균 복리 수익률) years = (self.equity_curve[-1]['date'] - self.equity_curve[0]['date']).days / 365.25 cagr = calculate_cagr(self.initial_capital, final_value, years) if years > 0 else 0.0 # MDD (최대 낙폭) equity_values = [Decimal(str(eq['value'])) for eq in self.equity_curve] max_drawdown_pct = calculate_max_drawdown(equity_values) # 일별 수익률 계산 daily_returns = [] for i in range(1, len(equity_values)): prev_value = equity_values[i - 1] curr_value = equity_values[i] if prev_value > 0: daily_return = float((curr_value - prev_value) / prev_value * 100) daily_returns.append(daily_return) # Sharpe Ratio sharpe_ratio = calculate_sharpe_ratio(daily_returns) if daily_returns else 0.0 # Sortino Ratio sortino_ratio = calculate_sortino_ratio(daily_returns) if daily_returns else 0.0 # Volatility (변동성) volatility = calculate_volatility(daily_returns) if daily_returns else 0.0 # 승률 win_rate_pct = calculate_win_rate(self.all_trades) if self.all_trades else 0.0 # Calmar Ratio calmar_ratio = calculate_calmar_ratio(total_return_pct, max_drawdown_pct, years) if years > 0 else 0.0 # 총 거래 수 total_trades = len(self.all_trades) return { 'initial_capital': float(self.initial_capital), 'final_value': float(final_value), 'total_return_pct': round(total_return_pct, 2), 'cagr': round(cagr, 2), 'max_drawdown_pct': round(max_drawdown_pct, 2), 'sharpe_ratio': round(sharpe_ratio, 2), 'sortino_ratio': round(sortino_ratio, 2), 'volatility': round(volatility, 2), 'win_rate_pct': round(win_rate_pct, 2), 'calmar_ratio': round(calmar_ratio, 2), 'total_trades': total_trades, 'equity_curve': self.equity_curve, 'trades': self.all_trades } def _empty_results(self) -> Dict[str, Any]: """빈 결과 반환.""" return { 'initial_capital': float(self.initial_capital), 'final_value': float(self.initial_capital), 'total_return_pct': 0.0, 'cagr': 0.0, 'max_drawdown_pct': 0.0, 'sharpe_ratio': 0.0, 'sortino_ratio': 0.0, 'volatility': 0.0, 'win_rate_pct': 0.0, 'calmar_ratio': 0.0, 'total_trades': 0, 'equity_curve': [], 'trades': [] }