""" 백테스팅 프레임워크 (Phase 3-1) - 과거 OHLCV 데이터로 전략 시뮬레이션 - 성과지표: Sharpe ratio, MDD, 승률, 평균손익비 - Phase 2 모델 변경 전후 비교 검증용 """ import json import numpy as np from dataclasses import dataclass, field from typing import Dict, List, Optional, Callable @dataclass class Trade: ticker: str entry_date: int # 데이터 인덱스 entry_price: float exit_date: int exit_price: float qty: int direction: str = "LONG" # LONG / SHORT @property def pnl(self): if self.direction == "LONG": return (self.exit_price - self.entry_price) * self.qty return (self.entry_price - self.exit_price) * self.qty @property def pnl_pct(self): return (self.exit_price - self.entry_price) / self.entry_price * 100 @dataclass class BacktestResult: total_return_pct: float sharpe_ratio: float max_drawdown_pct: float win_rate: float avg_win_pct: float avg_loss_pct: float profit_factor: float total_trades: int winning_trades: int losing_trades: int trades: List[Trade] = field(default_factory=list) def summary(self) -> str: lines = [ "=" * 50, "📊 백테스팅 결과", "=" * 50, f"총 수익률: {self.total_return_pct:+.2f}%", f"Sharpe Ratio: {self.sharpe_ratio:.3f}", f"Max Drawdown: {self.max_drawdown_pct:.2f}%", f"승률: {self.win_rate:.1f}% ({self.winning_trades}/{self.total_trades})", f"평균 수익: {self.avg_win_pct:+.2f}%", f"평균 손실: {self.avg_loss_pct:.2f}%", f"손익비(PF): {self.profit_factor:.2f}", "=" * 50, ] return "\n".join(lines) class Backtester: """ OHLCV 기반 전략 백테스터 사용 예시: bt = Backtester(initial_capital=10_000_000) result = bt.run( ohlcv_data={"close": [...], "high": [...], "low": [...], "volume": [...]}, strategy_fn=my_strategy, ticker="005930" ) print(result.summary()) """ def __init__(self, initial_capital: float = 10_000_000, commission_rate: float = 0.00015, # 0.015% (증권사 기본) slippage_rate: float = 0.001): # 0.1% 슬리피지 self.initial_capital = initial_capital self.commission_rate = commission_rate self.slippage_rate = slippage_rate def run(self, ohlcv_data: Dict, strategy_fn: Callable, ticker: str = "UNKNOWN", warmup: int = 60) -> BacktestResult: """ 단일 종목 백테스팅 Args: ohlcv_data: {'close':[], 'high':[], 'low':[], 'open':[], 'volume':[]} strategy_fn: (ohlcv_slice: dict) -> str ("BUY" | "SELL" | "HOLD") ticker: 종목 코드 warmup: 초기 웜업 기간 (기술지표 안정화) Returns: BacktestResult """ closes = np.array(ohlcv_data.get('close', []), dtype=float) highs = np.array(ohlcv_data.get('high', closes), dtype=float) lows = np.array(ohlcv_data.get('low', closes), dtype=float) volumes = np.array(ohlcv_data.get('volume', np.zeros_like(closes)), dtype=float) n = len(closes) if n < warmup + 10: return self._empty_result() capital = self.initial_capital position = 0 # 보유 수량 entry_price = 0.0 entry_idx = 0 equity_curve = [capital] trades: List[Trade] = [] for i in range(warmup, n): # 전략 함수에 현재까지의 슬라이스 전달 slice_data = { 'close': closes[:i+1].tolist(), 'high': highs[:i+1].tolist(), 'low': lows[:i+1].tolist(), 'volume': volumes[:i+1].tolist(), } signal = "HOLD" try: signal = strategy_fn(slice_data) except Exception: pass price = closes[i] buy_price = price * (1 + self.slippage_rate) # 슬리피지 포함 매수가 sell_price = price * (1 - self.slippage_rate) # 슬리피지 포함 매도가 if signal == "BUY" and position == 0: # 전액 투자 (수수료 포함) qty = int(capital / (buy_price * (1 + self.commission_rate))) if qty > 0: cost = qty * buy_price * (1 + self.commission_rate) capital -= cost position = qty entry_price = buy_price entry_idx = i elif signal == "SELL" and position > 0: proceeds = position * sell_price * (1 - self.commission_rate) capital += proceeds trades.append(Trade( ticker=ticker, entry_date=entry_idx, entry_price=entry_price, exit_date=i, exit_price=sell_price, qty=position )) position = 0 entry_price = 0.0 # 자산 추적 current_equity = capital + (position * closes[i] if position > 0 else 0) equity_curve.append(current_equity) # 미청산 포지션 강제 종료 if position > 0: last_price = closes[-1] * (1 - self.slippage_rate) proceeds = position * last_price * (1 - self.commission_rate) capital += proceeds trades.append(Trade( ticker=ticker, entry_date=entry_idx, entry_price=entry_price, exit_date=n - 1, exit_price=last_price, qty=position )) equity_curve[-1] = capital return self._compute_metrics(equity_curve, trades) def run_multi(self, ohlcv_dict: Dict[str, Dict], strategy_fn: Callable, warmup: int = 60) -> Dict[str, BacktestResult]: """여러 종목 백테스팅""" results = {} for ticker, ohlcv_data in ohlcv_dict.items(): results[ticker] = self.run(ohlcv_data, strategy_fn, ticker, warmup) return results def _compute_metrics(self, equity_curve: List[float], trades: List[Trade]) -> BacktestResult: equity = np.array(equity_curve, dtype=float) total_return_pct = (equity[-1] / equity[0] - 1) * 100 # Sharpe Ratio (일별 수익률 기준, 연율화) daily_returns = np.diff(equity) / equity[:-1] if daily_returns.std() > 0: sharpe = (daily_returns.mean() / daily_returns.std()) * np.sqrt(252) else: sharpe = 0.0 # Max Drawdown peak = np.maximum.accumulate(equity) drawdowns = (equity - peak) / (peak + 1e-9) * 100 max_drawdown = abs(drawdowns.min()) # 승률 / 손익비 wins = [t for t in trades if t.pnl_pct > 0] losses = [t for t in trades if t.pnl_pct <= 0] win_rate = len(wins) / len(trades) * 100 if trades else 0 avg_win = np.mean([t.pnl_pct for t in wins]) if wins else 0 avg_loss = np.mean([t.pnl_pct for t in losses]) if losses else 0 total_win = sum(t.pnl for t in wins) total_loss = abs(sum(t.pnl for t in losses)) profit_factor = total_win / (total_loss + 1e-9) return BacktestResult( total_return_pct=round(total_return_pct, 2), sharpe_ratio=round(sharpe, 3), max_drawdown_pct=round(max_drawdown, 2), win_rate=round(win_rate, 1), avg_win_pct=round(avg_win, 2), avg_loss_pct=round(avg_loss, 2), profit_factor=round(profit_factor, 3), total_trades=len(trades), winning_trades=len(wins), losing_trades=len(losses), trades=trades ) def _empty_result(self) -> BacktestResult: return BacktestResult( total_return_pct=0.0, sharpe_ratio=0.0, max_drawdown_pct=0.0, win_rate=0.0, avg_win_pct=0.0, avg_loss_pct=0.0, profit_factor=0.0, total_trades=0, winning_trades=0, losing_trades=0 ) def compare_strategies(ohlcv_data: Dict, strategies: Dict[str, Callable], initial_capital: float = 10_000_000) -> Dict[str, BacktestResult]: """ 여러 전략 동시 비교 Args: strategies: {"전략명": strategy_fn, ...} Returns: {"전략명": BacktestResult, ...} """ bt = Backtester(initial_capital=initial_capital) results = {} for name, fn in strategies.items(): results[name] = bt.run(ohlcv_data, fn) print(f"\n[{name}]") print(results[name].summary()) return results