""" 한국 주식 백테스트 엔진 stock_daily_prices DB 테이블에서 일봉 OHLCV를 로드하여 날짜별 순차 시뮬레이션으로 전략 성과를 평가합니다. [설계 원칙] - 동기 함수 (배치 작업이므로 async 불필요, DB 조회만 async) - numpy 최소 사용 (표준 라이브러리 우선) - 기존 mean_reversion_strategy.py의 RSI/BB 함수 재활용 - CLAUDE.md 리스크 규칙 준수 (수수료, 세금, 슬리피지, 손절/익절) [수수료 모델] - 매수 수수료: 0.015% - 매도 수수료: 0.015% + 증권거래세 (KOSPI 0.18%, KOSDAQ 0.23%) [슬리피지 모델] - 시장가 기준 +/-0.1% 랜덤 슬리피지 """ import logging import math import random from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import date, datetime from typing import Any, Dict, List, Optional, Tuple from app.core.config import settings from app.services.mean_reversion_strategy import ( MarketRegime, calculate_atr, calculate_bollinger, calculate_rsi, check_buy_signal, check_exit_signal, detect_market_regime, is_reversal_candle, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 데이터 구조 # --------------------------------------------------------------------------- @dataclass class BarData: """일봉 OHLCV 데이터 (1행)""" trade_date: date open_price: int high_price: int low_price: int close_price: int volume: int stock_code: str = "" stock_name: str = "" @dataclass class BacktestTrade: """체결된 거래 1건""" stock_code: str side: str # "BUY" | "SELL" entry_date: Optional[date] = None exit_date: Optional[date] = None entry_price: float = 0.0 exit_price: float = 0.0 quantity: int = 0 commission: float = 0.0 tax: float = 0.0 slippage: float = 0.0 pnl: float = 0.0 pnl_pct: float = 0.0 reason: str = "" @dataclass class BacktestMetrics: """백테스트 성과 지표""" total_return_pct: float = 0.0 cagr_pct: float = 0.0 sharpe_ratio: float = 0.0 sortino_ratio: float = 0.0 max_drawdown_pct: float = 0.0 win_rate: float = 0.0 profit_factor: float = 0.0 avg_win_pct: float = 0.0 avg_loss_pct: float = 0.0 total_trades: int = 0 winning_trades: int = 0 losing_trades: int = 0 max_consecutive_wins: int = 0 max_consecutive_losses: int = 0 avg_holding_days: float = 0.0 @dataclass class BacktestResult: """백테스트 전체 결과""" metrics: BacktestMetrics equity_curve: List[Dict[str, Any]] = field(default_factory=list) trades: List[Dict[str, Any]] = field(default_factory=list) drawdown_curve: List[Dict[str, Any]] = field(default_factory=list) config: Dict[str, Any] = field(default_factory=dict) @dataclass class Position: """현재 보유 포지션""" stock_code: str entry_date: date entry_price: float quantity: int highest_since_entry: float # 트레일링 스탑용 최고가 remaining_pct: float = 1.0 # 분할매도 잔여 비율 # --------------------------------------------------------------------------- # 수수료/세금/슬리피지 모델 # --------------------------------------------------------------------------- # 수수료율 (config.py 참조) BUY_COMMISSION_RATE = settings.KR_COMMISSION_RATE # 0.00015 (0.015%) SELL_COMMISSION_RATE = settings.KR_COMMISSION_RATE # 0.00015 (0.015%) SELL_TAX_KOSPI = settings.KR_SELL_TAX_RATE_KOSPI # 0.0018 (0.18%) SELL_TAX_KOSDAQ = settings.KR_SELL_TAX_RATE_KOSDAQ # 0.0023 (0.23%) SLIPPAGE_PCT = 0.001 # +/-0.1% def _apply_buy_slippage(price: float, seed: Optional[int] = None) -> float: """매수 시 슬리피지: 약간 높은 가격에 체결""" rng = random.Random(seed) slip = rng.uniform(0, SLIPPAGE_PCT) return price * (1 + slip) def _apply_sell_slippage(price: float, seed: Optional[int] = None) -> float: """매도 시 슬리피지: 약간 낮은 가격에 체결""" rng = random.Random(seed) slip = rng.uniform(0, SLIPPAGE_PCT) return price * (1 - slip) def _calc_buy_cost(price: float, quantity: int) -> Tuple[float, float]: """매수 비용 계산 -> (총비용, 수수료)""" amount = price * quantity commission = amount * BUY_COMMISSION_RATE return amount + commission, commission def _calc_sell_proceeds( price: float, quantity: int, market: str = "KOSPI", ) -> Tuple[float, float, float]: """매도 수입 계산 -> (순수입, 수수료, 세금)""" amount = price * quantity commission = amount * SELL_COMMISSION_RATE tax_rate = SELL_TAX_KOSDAQ if market.upper() == "KOSDAQ" else SELL_TAX_KOSPI tax = amount * tax_rate return amount - commission - tax, commission, tax # --------------------------------------------------------------------------- # 성과 지표 계산 # --------------------------------------------------------------------------- def _compute_metrics( equity_curve: List[float], trades: List[BacktestTrade], initial_capital: float, trading_days: int, ) -> BacktestMetrics: """equity_curve와 trades로부터 성과 지표 산출""" m = BacktestMetrics() if not equity_curve or len(equity_curve) < 2: return m final_equity = equity_curve[-1] m.total_return_pct = ((final_equity - initial_capital) / initial_capital) * 100 # CAGR years = trading_days / 252.0 if years > 0 and final_equity > 0: m.cagr_pct = ((final_equity / initial_capital) ** (1 / years) - 1) * 100 else: m.cagr_pct = 0.0 # 일별 수익률 daily_returns: List[float] = [] for i in range(1, len(equity_curve)): prev = equity_curve[i - 1] if prev > 0: daily_returns.append((equity_curve[i] - prev) / prev) # Sharpe Ratio (무위험 이자율 3% 가정) if daily_returns: rf_daily = 0.03 / 252 excess = [r - rf_daily for r in daily_returns] avg_excess = sum(excess) / len(excess) variance = sum((r - avg_excess) ** 2 for r in excess) / max(len(excess) - 1, 1) std = math.sqrt(variance) if variance > 0 else 0 m.sharpe_ratio = round((avg_excess / std) * math.sqrt(252), 4) if std > 0 else 0.0 else: m.sharpe_ratio = 0.0 # Sortino Ratio (하방 편차만 사용) if daily_returns: rf_daily = 0.03 / 252 excess = [r - rf_daily for r in daily_returns] avg_excess = sum(excess) / len(excess) downside = [min(r, 0) ** 2 for r in excess] downside_var = sum(downside) / max(len(downside) - 1, 1) downside_std = math.sqrt(downside_var) if downside_var > 0 else 0 m.sortino_ratio = round((avg_excess / downside_std) * math.sqrt(252), 4) if downside_std > 0 else 0.0 else: m.sortino_ratio = 0.0 # Max Drawdown peak = equity_curve[0] max_dd = 0.0 for eq in equity_curve: if eq > peak: peak = eq dd = (eq - peak) / peak if peak > 0 else 0 if dd < max_dd: max_dd = dd m.max_drawdown_pct = round(max_dd * 100, 4) # 거래 통계 closed_trades = [t for t in trades if t.side == "SELL"] m.total_trades = len(closed_trades) if m.total_trades == 0: return m wins = [t for t in closed_trades if t.pnl > 0] losses = [t for t in closed_trades if t.pnl <= 0] m.winning_trades = len(wins) m.losing_trades = len(losses) m.win_rate = round(m.winning_trades / m.total_trades * 100, 2) # Profit Factor gross_profit = sum(t.pnl for t in wins) if wins else 0 gross_loss = abs(sum(t.pnl for t in losses)) if losses else 0 m.profit_factor = round(gross_profit / gross_loss, 4) if gross_loss > 0 else float("inf") if gross_profit > 0 else 0.0 # Average Win/Loss % m.avg_win_pct = round(sum(t.pnl_pct for t in wins) / len(wins), 4) if wins else 0.0 m.avg_loss_pct = round(sum(t.pnl_pct for t in losses) / len(losses), 4) if losses else 0.0 # 연속 승/패 consecutive_w = 0 consecutive_l = 0 max_cw = 0 max_cl = 0 for t in closed_trades: if t.pnl > 0: consecutive_w += 1 consecutive_l = 0 else: consecutive_l += 1 consecutive_w = 0 max_cw = max(max_cw, consecutive_w) max_cl = max(max_cl, consecutive_l) m.max_consecutive_wins = max_cw m.max_consecutive_losses = max_cl # 평균 보유일 holding_days = [] for t in closed_trades: if t.entry_date and t.exit_date: delta = (t.exit_date - t.entry_date).days holding_days.append(delta) m.avg_holding_days = round(sum(holding_days) / len(holding_days), 1) if holding_days else 0.0 return m # --------------------------------------------------------------------------- # 전략 추상 클래스 # --------------------------------------------------------------------------- class BacktestStrategy(ABC): """백테스트 전략 인터페이스 on_bar()는 매 일봉마다 호출되며, 매수/매도 주문 리스트를 반환합니다. """ @abstractmethod def name(self) -> str: """전략 이름""" ... @abstractmethod def description(self) -> str: """전략 설명""" ... @abstractmethod def default_params(self) -> Dict[str, Any]: """기본 파라미터""" ... @abstractmethod def on_bar( self, bar_idx: int, bars: Dict[str, List[BarData]], positions: Dict[str, Position], cash: float, total_equity: float, params: Dict[str, Any], ) -> List[Dict[str, Any]]: """일봉 콜백 Args: bar_idx: 현재 봉 인덱스 (0-based, 모든 종목 공통 날짜 기준) bars: {stock_code: [BarData, ...]} 전체 일봉 (bar_idx까지 접근 가능) positions: 현재 보유 포지션 {stock_code: Position} cash: 현재 현금 total_equity: 현재 총 자산 (현금 + 평가금) params: 전략 파라미터 Returns: 주문 리스트 [{"action": "BUY"|"SELL", "stock_code": str, "price": float, "quantity": int, "reason": str, "sell_pct": float (SELL 시 매도 비율)}] """ ... # --------------------------------------------------------------------------- # 내장 전략: MeanReversionBacktest (RSI + BB 평균회귀) # --------------------------------------------------------------------------- class MeanReversionBacktest(BacktestStrategy): """CLAUDE.md 섹션 6 RSI+BB 평균회귀 전략 백테스트 버전 기존 mean_reversion_strategy.py의 check_buy_signal / check_exit_signal을 그대로 재활용합니다. 시장 국면(regime)은 코스피200 데이터가 없으므로 전체 종목 평균으로 대체하거나, 파라미터로 고정 지정 가능합니다. """ def name(self) -> str: return "mean_reversion" def description(self) -> str: return "RSI + Bollinger Band 평균회귀 전략 (CLAUDE.md 섹션 6)" def default_params(self) -> Dict[str, Any]: return { "rsi_period": settings.RSI_PERIOD, "bb_period": settings.BOLLINGER_PERIOD, "atr_period": settings.ATR_PERIOD, "max_positions": settings.MAX_HOLDINGS_COUNT, "position_size_pct": settings.MAX_POSITION_SIZE, "stop_loss_pct": settings.STOP_LOSS_PERCENT, "atr_stop_multiplier": settings.ATR_STOP_MULTIPLIER, "atr_trailing_multiplier": settings.ATR_TRAILING_MULTIPLIER, "rsi_exit_threshold": settings.MEAN_REVERSION_RSI_EXIT_THRESHOLD, "rsi_full_exit": settings.MEAN_REVERSION_RSI_FULL_EXIT, "regime": "BULL", # "BULL", "BEAR", "AUTO" "market": "KOSPI", # 세금 계산용 "min_price": settings.MIN_PRICE_KRW, "min_data_bars": 60, # RSI/BB 계산에 필요한 최소 봉 수 } def on_bar( self, bar_idx: int, bars: Dict[str, List[BarData]], positions: Dict[str, Position], cash: float, total_equity: float, params: Dict[str, Any], ) -> List[Dict[str, Any]]: orders: List[Dict[str, Any]] = [] min_bars = params.get("min_data_bars", 60) # --- 매도 시그널 먼저 (기존 포지션) --- for code, pos in list(positions.items()): stock_bars = bars.get(code, []) if bar_idx >= len(stock_bars): continue current_bar = stock_bars[bar_idx] closes = [b.close_price for b in stock_bars[: bar_idx + 1]] highs = [b.high_price for b in stock_bars[: bar_idx + 1]] lows = [b.low_price for b in stock_bars[: bar_idx + 1]] if len(closes) < min_bars: continue rsi = calculate_rsi(closes, params.get("rsi_period", 14)) bb = calculate_bollinger(closes, params.get("bb_period", 20)) atr = calculate_atr(highs, lows, closes, params.get("atr_period", 14)) if rsi is None or bb is None or atr is None: continue bb_upper, bb_middle, bb_lower = bb # 최고가 업데이트 pos.highest_since_entry = max(pos.highest_since_entry, current_bar.high_price) exit_signal = check_exit_signal( regime=self._get_regime(params), rsi=rsi, close=float(current_bar.close_price), bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, entry_price=pos.entry_price, atr=atr, lowest_since_entry=float(current_bar.low_price), remaining_pct=pos.remaining_pct, trailing_high=pos.highest_since_entry, ) if exit_signal: sell_qty = int(pos.quantity * exit_signal.sell_pct) if sell_qty <= 0: sell_qty = pos.quantity orders.append({ "action": "SELL", "stock_code": code, "price": float(current_bar.close_price), "quantity": sell_qty, "reason": exit_signal.reason, "sell_pct": exit_signal.sell_pct, }) # --- 매수 시그널 --- max_pos = params.get("max_positions", 20) if len(positions) >= max_pos: return orders for code, stock_bars in bars.items(): if code in positions: continue if len(positions) + sum(1 for o in orders if o["action"] == "BUY") >= max_pos: break if bar_idx >= len(stock_bars): continue current_bar = stock_bars[bar_idx] closes = [b.close_price for b in stock_bars[: bar_idx + 1]] highs = [b.high_price for b in stock_bars[: bar_idx + 1]] lows = [b.low_price for b in stock_bars[: bar_idx + 1]] opens = [b.open_price for b in stock_bars[: bar_idx + 1]] volumes = [b.volume for b in stock_bars[: bar_idx + 1]] if len(closes) < min_bars: continue # 가격 필터 if current_bar.close_price < params.get("min_price", 5000): continue rsi = calculate_rsi(closes, params.get("rsi_period", 14)) bb = calculate_bollinger(closes, params.get("bb_period", 20)) if rsi is None or bb is None: continue bb_upper, bb_middle, bb_lower = bb # 거래량 20일 평균 vol_window = volumes[-20:] if len(volumes) >= 20 else volumes avg_vol = sum(vol_window) / len(vol_window) if vol_window else 0 # 이전 RSI (반등 확인) rsi_prev = None if len(closes) > min_bars + 3: rsi_prev = calculate_rsi(closes[:-3], params.get("rsi_period", 14)) # 되돌림 캔들 확인 (BEAR 진입 조건) reversal = is_reversal_candle( [float(x) for x in opens[-3:]], [float(x) for x in highs[-3:]], [float(x) for x in lows[-3:]], [float(x) for x in closes[-3:]], ) prev_vol = volumes[-2] if len(volumes) >= 2 else 0 # [안전형 개선①] own_trend 게이트용 — 종목 자체 MA20/MA60 (데이터 있으면 재사용) ma20_val = sum(closes[-20:]) / 20 if len(closes) >= 20 else None ma60_val = sum(closes[-60:]) / 60 if len(closes) >= 60 else None signal = check_buy_signal( regime=self._get_regime(params), rsi=rsi, close=float(current_bar.close_price), bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, avg_volume_20d=avg_vol, current_volume=current_bar.volume, prev_volume=prev_vol, is_reversal=reversal, rsi_prev=rsi_prev, ma20=ma20_val, ma60=ma60_val, ) if signal: # 포지션 사이징: total_equity * position_size_pct budget = total_equity * params.get("position_size_pct", 0.04) budget = min(budget, cash) if budget <= 0: continue price = float(current_bar.close_price) quantity = int(budget / price) if quantity <= 0: continue orders.append({ "action": "BUY", "stock_code": code, "price": price, "quantity": quantity, "reason": signal.get("description", ""), "sell_pct": 0.0, }) return orders @staticmethod def _get_regime(params: Dict[str, Any]) -> MarketRegime: regime_str = params.get("regime", "BULL").upper() if regime_str == "BEAR": return MarketRegime.BEAR if regime_str == "SIDEWAYS": return MarketRegime.SIDEWAYS return MarketRegime.BULL # --------------------------------------------------------------------------- # 전략 레지스트리 # --------------------------------------------------------------------------- _STRATEGY_REGISTRY: Dict[str, type] = { "mean_reversion": MeanReversionBacktest, } def get_available_strategies() -> List[Dict[str, Any]]: """사용 가능한 전략 목록 반환""" result = [] for key, cls in _STRATEGY_REGISTRY.items(): instance = cls() result.append({ "key": key, "name": instance.name(), "description": instance.description(), "default_params": instance.default_params(), }) return result def get_strategy(name: str) -> BacktestStrategy: """이름으로 전략 인스턴스 반환""" cls = _STRATEGY_REGISTRY.get(name) if cls is None: raise ValueError(f"Unknown strategy: {name}. Available: {list(_STRATEGY_REGISTRY.keys())}") return cls() # --------------------------------------------------------------------------- # 데이터 로더 (async — DB 조회) # --------------------------------------------------------------------------- async def load_daily_bars( stock_codes: List[str], start_date: date, end_date: date, ) -> Tuple[Dict[str, List[BarData]], List[date]]: """stock_daily_prices 테이블에서 일봉 데이터 로드 Returns: (bars_by_code, all_dates) - bars_by_code: {stock_code: [BarData...]} 날짜 오름차순 - all_dates: 전체 거래일 리스트 (유니온) """ from sqlalchemy import select from app.core.database import AsyncSessionLocal from app.models.models import StockDailyPrice bars_by_code: Dict[str, List[BarData]] = {} all_dates_set: set = set() async with AsyncSessionLocal() as session: for code in stock_codes: stmt = ( select(StockDailyPrice) .where( StockDailyPrice.stock_code == code, StockDailyPrice.trade_date >= start_date, StockDailyPrice.trade_date <= end_date, ) .order_by(StockDailyPrice.trade_date.asc()) ) result = await session.execute(stmt) rows = result.scalars().all() if not rows: logger.debug("종목 %s: 기간 내 데이터 없음 (%s ~ %s)", code, start_date, end_date) continue bar_list: List[BarData] = [] for row in rows: bar_list.append(BarData( trade_date=row.trade_date, open_price=row.open_price or 0, high_price=row.high_price or 0, low_price=row.low_price or 0, close_price=row.close_price or 0, volume=row.volume or 0, stock_code=row.stock_code, stock_name=row.stock_name or "", )) all_dates_set.add(row.trade_date) bars_by_code[code] = bar_list all_dates = sorted(all_dates_set) return bars_by_code, all_dates def _align_bars_to_dates( bars_by_code: Dict[str, List[BarData]], all_dates: List[date], ) -> Dict[str, List[BarData]]: """모든 종목을 동일 날짜 인덱스에 맞춤 (결측일은 이전 종가 복제) bar_idx로 접근할 때 모든 종목이 같은 날짜를 가리키도록 합니다. """ aligned: Dict[str, List[BarData]] = {} date_to_idx = {d: i for i, d in enumerate(all_dates)} for code, bar_list in bars_by_code.items(): # 원본 데이터를 날짜 인덱스로 매핑 date_map = {b.trade_date: b for b in bar_list} result: List[BarData] = [] last_bar: Optional[BarData] = None for d in all_dates: if d in date_map: last_bar = date_map[d] result.append(last_bar) elif last_bar is not None: # 이전 종가를 OHLC 모두 같게 복제 (거래량 0) filler = BarData( trade_date=d, open_price=last_bar.close_price, high_price=last_bar.close_price, low_price=last_bar.close_price, close_price=last_bar.close_price, volume=0, stock_code=code, stock_name=last_bar.stock_name, ) result.append(filler) # 첫 날짜 이전 데이터가 없으면 건너뜀 (나중에 min_bars로 필터) # 빈 봉이 앞에 있을 수 있으므로 첫 실제 데이터부터 시작 if result: aligned[code] = result return aligned # --------------------------------------------------------------------------- # 백테스트 엔진 # --------------------------------------------------------------------------- class KrBacktestEngine: """한국 주식 백테스트 엔진 (메인 클래스) 사용법: engine = KrBacktestEngine() result = await engine.run( stock_codes=["005930", "035720"], start_date=date(2025, 1, 1), end_date=date(2025, 12, 31), strategy_name="mean_reversion", params={...}, initial_capital=10_000_000, market="KOSPI", random_seed=42, ) """ def __init__(self): pass async def run( self, stock_codes: List[str], start_date: date, end_date: date, strategy_name: str = "mean_reversion", params: Optional[Dict[str, Any]] = None, initial_capital: float = 10_000_000, market: str = "KOSPI", random_seed: Optional[int] = 42, ) -> BacktestResult: """백테스트 실행 Args: stock_codes: 대상 종목 코드 리스트 start_date: 시작일 end_date: 종료일 strategy_name: 전략 키 ("mean_reversion" 등) params: 전략 파라미터 (None이면 기본값 사용) initial_capital: 초기 자본금 market: 시장 구분 (KOSPI/KOSDAQ) - 세금 계산용 random_seed: 슬리피지 재현성을 위한 랜덤 시드 Returns: BacktestResult """ strategy = get_strategy(strategy_name) effective_params = strategy.default_params() if params: effective_params.update(params) effective_params["market"] = market # 1. 데이터 로드 logger.info( "백테스트 데이터 로드: %d 종목, %s ~ %s", len(stock_codes), start_date, end_date, ) bars_by_code, all_dates = await load_daily_bars(stock_codes, start_date, end_date) if not bars_by_code or not all_dates: logger.warning("백테스트 데이터 없음") return BacktestResult( metrics=BacktestMetrics(), config=self._build_config_dict( stock_codes, start_date, end_date, strategy_name, effective_params, initial_capital, ), ) # 2. 날짜 정렬 aligned = _align_bars_to_dates(bars_by_code, all_dates) # 3. 시뮬레이션 실행 result = self._simulate( aligned, all_dates, strategy, effective_params, initial_capital, market, random_seed, ) result.config = self._build_config_dict( stock_codes, start_date, end_date, strategy_name, effective_params, initial_capital, ) return result def _simulate( self, bars: Dict[str, List[BarData]], all_dates: List[date], strategy: BacktestStrategy, params: Dict[str, Any], initial_capital: float, market: str, random_seed: Optional[int], ) -> BacktestResult: """핵심 시뮬레이션 루프 (동기) [Look-ahead bias 수정] T일 close로 신호(on_bar) 생성 -> T+1일 open + 슬리피지로 체결. 마지막 봉 신호는 다음 봉이 없으므로 무시합니다. """ cash = initial_capital positions: Dict[str, Position] = {} all_trades: List[BacktestTrade] = [] # 포지션별 진입 거래 추적 (매도 시 PnL 계산용) entry_trades: Dict[str, BacktestTrade] = {} equity_values: List[float] = [initial_capital] equity_curve: List[Dict[str, Any]] = [{ "date": str(all_dates[0]) if all_dates else "", "equity": initial_capital, }] drawdown_curve: List[Dict[str, Any]] = [] peak_equity = initial_capital seed_counter = random_seed if random_seed is not None else None # 전일 신호를 저장 (T일 close 신호 -> T+1일 open 체결) pending_orders: List[Dict[str, Any]] = [] for bar_idx in range(len(all_dates)): current_date = all_dates[bar_idx] # 현재 평가 자산 position_value = 0.0 for code, pos in positions.items(): stock_bars = bars.get(code, []) if bar_idx < len(stock_bars): position_value += stock_bars[bar_idx].close_price * pos.quantity total_equity = cash + position_value # -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 -- for order in pending_orders: action = order["action"] code = order["stock_code"] quantity = order["quantity"] # T+1일 open 가격을 base price로 사용 stock_bars = bars.get(code, []) if bar_idx >= len(stock_bars): continue next_open = float(stock_bars[bar_idx].open_price) # 슬리피지용 시드 if seed_counter is not None: seed_counter += 1 slip_seed = seed_counter else: slip_seed = None if action == "BUY": exec_price = _apply_buy_slippage(next_open, slip_seed) total_cost, commission = _calc_buy_cost(exec_price, quantity) if total_cost > cash: # 자금 부족 시 수량 축소 quantity = int(cash / (exec_price * (1 + BUY_COMMISSION_RATE))) if quantity <= 0: continue total_cost, commission = _calc_buy_cost(exec_price, quantity) cash -= total_cost slippage_amount = abs(exec_price - next_open) * quantity trade = BacktestTrade( stock_code=code, side="BUY", entry_date=current_date, entry_price=exec_price, quantity=quantity, commission=commission, slippage=slippage_amount, reason=order.get("reason", ""), ) all_trades.append(trade) entry_trades[code] = trade positions[code] = Position( stock_code=code, entry_date=current_date, entry_price=exec_price, quantity=quantity, highest_since_entry=exec_price, ) elif action == "SELL": pos = positions.get(code) if pos is None: continue sell_qty = min(quantity, pos.quantity) exec_price = _apply_sell_slippage(next_open, slip_seed) proceeds, commission, tax = _calc_sell_proceeds(exec_price, sell_qty, market) cash += proceeds slippage_amount = abs(next_open - exec_price) * sell_qty # PnL 계산 entry_cost = pos.entry_price * sell_qty * (1 + BUY_COMMISSION_RATE) pnl = proceeds - entry_cost pnl_pct = (pnl / entry_cost) * 100 if entry_cost > 0 else 0.0 trade = BacktestTrade( stock_code=code, side="SELL", entry_date=pos.entry_date, exit_date=current_date, entry_price=pos.entry_price, exit_price=exec_price, quantity=sell_qty, commission=commission, tax=tax, slippage=slippage_amount, pnl=round(pnl, 2), pnl_pct=round(pnl_pct, 4), reason=order.get("reason", ""), ) all_trades.append(trade) # 포지션 업데이트 remaining_qty = pos.quantity - sell_qty if remaining_qty <= 0: del positions[code] entry_trades.pop(code, None) else: pos.quantity = remaining_qty pos.remaining_pct = remaining_qty / (pos.quantity + sell_qty) pending_orders.clear() # -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) -- is_last_bar = (bar_idx == len(all_dates) - 1) if not is_last_bar: # 체결 이후의 상태로 평가 자산 재계산 (전략에 전달) position_value = 0.0 for code, pos in positions.items(): stock_bars_eval = bars.get(code, []) if bar_idx < len(stock_bars_eval): position_value += stock_bars_eval[bar_idx].close_price * pos.quantity total_equity = cash + position_value orders = strategy.on_bar( bar_idx=bar_idx, bars=bars, positions=positions, cash=cash, total_equity=total_equity, params=params, ) pending_orders = list(orders) # 일말 평가 position_value = 0.0 for code, pos in positions.items(): stock_bars = bars.get(code, []) if bar_idx < len(stock_bars): position_value += stock_bars[bar_idx].close_price * pos.quantity total_equity = cash + position_value equity_values.append(total_equity) equity_curve.append({ "date": str(current_date), "equity": round(total_equity, 2), }) # Drawdown if total_equity > peak_equity: peak_equity = total_equity dd = ((total_equity - peak_equity) / peak_equity * 100) if peak_equity > 0 else 0 drawdown_curve.append({ "date": str(current_date), "drawdown_pct": round(dd, 4), }) # 성과 지표 계산 metrics = _compute_metrics( equity_curve=equity_values, trades=all_trades, initial_capital=initial_capital, trading_days=len(all_dates), ) # 거래 내역 직렬화 trades_dicts = [] for t in all_trades: trades_dicts.append({ "stock_code": t.stock_code, "side": t.side, "entry_date": str(t.entry_date) if t.entry_date else None, "exit_date": str(t.exit_date) if t.exit_date else None, "entry_price": t.entry_price, "exit_price": t.exit_price, "quantity": t.quantity, "commission": round(t.commission, 2), "tax": round(t.tax, 2), "slippage": round(t.slippage, 2), "pnl": t.pnl, "pnl_pct": t.pnl_pct, "reason": t.reason, }) return BacktestResult( metrics=metrics, equity_curve=equity_curve, trades=trades_dicts, drawdown_curve=drawdown_curve, ) @staticmethod def _build_config_dict( stock_codes: List[str], start_date: date, end_date: date, strategy_name: str, params: Dict[str, Any], initial_capital: float, ) -> Dict[str, Any]: return { "stock_codes": stock_codes, "start_date": str(start_date), "end_date": str(end_date), "strategy": strategy_name, "params": params, "initial_capital": initial_capital, "commission": { "buy_rate": BUY_COMMISSION_RATE, "sell_rate": SELL_COMMISSION_RATE, "tax_kospi": SELL_TAX_KOSPI, "tax_kosdaq": SELL_TAX_KOSDAQ, }, "slippage_pct": SLIPPAGE_PCT, }