"""미국 주식 백테스트 엔진 KIS API의 해외 주식 일봉 데이터를 활용한 백테스트. us_stock_strategy의 매수/매도 신호를 과거 데이터에 적용하여 성과 검증. """ import logging import math import random from datetime import date, datetime, timedelta from typing import Dict, List, Any, Optional from dataclasses import dataclass, asdict import numpy as np from app.services.korea_investment_api import korea_investment_api from app.services.us_stock_strategy import ( check_us_buy_signal, check_us_sell_signal, US_STOP_LOSS_PCT, US_TRAILING_STOP_PCT, ) logger = logging.getLogger(__name__) # ── 상수 ──────────────────────────────────────────────────────────────────────── US_COMMISSION_RATE = 0.001 # 해외 주식 수수료 약 0.1% (매수+매도 각각) RISK_FREE_RATE = 0.04 # 무위험 수익률 4% (연간, USD 기준) TRADING_DAYS_PER_YEAR = 252 US_SLIPPAGE_PCT = 0.0005 # +/-0.05% (미국 주식은 유동성이 높아 슬리피지가 낮음) def _apply_us_buy_slippage(price: float, seed: Optional[int] = None) -> float: """매수 시 슬리피지: 시장가 주문 시 약간 높은 가격에 체결 미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용. """ rng = random.Random(seed) slip = rng.uniform(0, US_SLIPPAGE_PCT) return price * (1 + slip) def _apply_us_sell_slippage(price: float, seed: Optional[int] = None) -> float: """매도 시 슬리피지: 시장가 주문 시 약간 낮은 가격에 체결 미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용. """ rng = random.Random(seed) slip = rng.uniform(0, US_SLIPPAGE_PCT) return price * (1 - slip) @dataclass class USBacktestTrade: """미국 주식 백테스트 거래 기록""" date: str action: str # "BUY" / "SELL" ticker: str price: float quantity: int amount: float commission: float reason: str pnl: float = 0.0 # 매도 시 실현 손익 pnl_pct: float = 0.0 # ── 백테스트 메트릭 계산 ───────────────────────────────────────────────────────── def calculate_backtest_metrics( trades: List[Dict[str, Any]], initial_capital: float, ) -> Dict[str, Any]: """순수 계산 함수 - 거래 리스트로부터 성과 지표 산출 Args: trades: 거래 기록 리스트 (각 항목에 action, amount, pnl, pnl_pct 포함) initial_capital: 초기 자본 (USD) Returns: {"total_return": float, "total_return_pct": float, "win_rate": float, "max_drawdown_pct": float, "sharpe_ratio": float, "trade_count": int, "win_count": int, "lose_count": int, "avg_win_pct": float, "avg_lose_pct": float, "profit_factor": float} """ if not trades: return { "total_return": 0.0, "total_return_pct": 0.0, "win_rate": 0.0, "max_drawdown_pct": 0.0, "sharpe_ratio": 0.0, "trade_count": 0, "win_count": 0, "lose_count": 0, "avg_win_pct": 0.0, "avg_lose_pct": 0.0, "profit_factor": 0.0, } # 매도 거래만 추출 (실현 손익 기준) sell_trades = [t for t in trades if t.get("action") == "SELL"] if not sell_trades: return { "total_return": 0.0, "total_return_pct": 0.0, "win_rate": 0.0, "max_drawdown_pct": 0.0, "sharpe_ratio": 0.0, "trade_count": len(trades), "win_count": 0, "lose_count": 0, "avg_win_pct": 0.0, "avg_lose_pct": 0.0, "profit_factor": 0.0, } # 승/패 분류 wins = [t for t in sell_trades if t.get("pnl", 0) > 0] loses = [t for t in sell_trades if t.get("pnl", 0) <= 0] total_sell = len(sell_trades) win_rate = (len(wins) / total_sell * 100) if total_sell > 0 else 0.0 # 총 수익 total_pnl = sum(t.get("pnl", 0) for t in sell_trades) total_return_pct = (total_pnl / initial_capital * 100) if initial_capital > 0 else 0.0 # 평균 수익/손실 avg_win_pct = ( sum(t.get("pnl_pct", 0) for t in wins) / len(wins) ) if wins else 0.0 avg_lose_pct = ( sum(t.get("pnl_pct", 0) for t in loses) / len(loses) ) if loses else 0.0 # Profit Factor gross_profit = sum(t.get("pnl", 0) for t in wins) gross_loss = abs(sum(t.get("pnl", 0) for t in loses)) profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else float("inf") if gross_profit > 0 else 0.0 # 최대 낙폭 (MDD) 계산 - 누적 손익 기준 equity = initial_capital peak = equity max_dd = 0.0 for t in sell_trades: equity += t.get("pnl", 0) if equity > peak: peak = equity dd = ((equity - peak) / peak) * 100 if peak > 0 else 0.0 if dd < max_dd: max_dd = dd # 샤프 비율 계산 pnl_returns = [t.get("pnl_pct", 0) / 100 for t in sell_trades] sharpe_ratio = _calculate_sharpe(pnl_returns) return { "total_return": round(total_pnl, 2), "total_return_pct": round(total_return_pct, 2), "win_rate": round(win_rate, 2), "max_drawdown_pct": round(max_dd, 2), "sharpe_ratio": round(sharpe_ratio, 2), "trade_count": len(trades), "win_count": len(wins), "lose_count": len(loses), "avg_win_pct": round(avg_win_pct, 2), "avg_lose_pct": round(avg_lose_pct, 2), "profit_factor": round(profit_factor, 2) if profit_factor != float("inf") else 999.99, } def _calculate_sharpe(returns: List[float]) -> float: """샤프 비율 계산 Args: returns: 거래별 수익률 리스트 (소수, 예: 0.03 = 3%) Returns: 연환산 샤프 비율 """ if not returns or len(returns) < 2: return 0.0 arr = np.array(returns, dtype=np.float64) mean_return = np.mean(arr) std_return = np.std(arr, ddof=1) if std_return == 0: return 0.0 # 거래당 초과 수익률 -> 연환산 # 대략적 연환산: 거래 횟수 기반 (정밀하지 않지만 실용적) trades_per_year = min(len(returns), TRADING_DAYS_PER_YEAR) daily_rf = RISK_FREE_RATE / TRADING_DAYS_PER_YEAR sharpe = (mean_return - daily_rf) / std_return * math.sqrt(trades_per_year) return float(sharpe) # ── 메인 백테스트 함수 ──────────────────────────────────────────────────────────── async def run_us_backtest( ticker: str, period_days: int = 180, initial_capital_usd: float = 10000.0, ) -> Dict[str, Any]: """미국 주식 백테스트 실행 Args: ticker: 미국 티커 (예: "AAPL", "NVDA") period_days: 백테스트 기간 (일) initial_capital_usd: 초기 자본 (USD) Returns: {"ticker": str, "period_days": int, "initial_capital": float, "metrics": dict, "trades": list, "equity_curve": list} """ logger.info(f"[US_BACKTEST] {ticker} 백테스트 시작 (기간: {period_days}일, 자본: ${initial_capital_usd:,.0f})") # 1) 과거 데이터 조회 (KIS 해외주식 일봉) try: daily_data = await _fetch_us_daily_data(ticker, period_days) except Exception as e: logger.error(f"[US_BACKTEST] {ticker} 데이터 조회 실패: {e}") return { "ticker": ticker, "period_days": period_days, "initial_capital": initial_capital_usd, "metrics": calculate_backtest_metrics([], initial_capital_usd), "trades": [], "equity_curve": [], "error": str(e), } if not daily_data or len(daily_data) < 30: msg = f"데이터 부족: {len(daily_data) if daily_data else 0}일 (최소 30일 필요)" logger.warning(f"[US_BACKTEST] {ticker} {msg}") return { "ticker": ticker, "period_days": period_days, "initial_capital": initial_capital_usd, "metrics": calculate_backtest_metrics([], initial_capital_usd), "trades": [], "equity_curve": [], "error": msg, } # 2) 기술적 지표 계산 indicators = _calculate_indicators(daily_data) # 3) 시뮬레이션 실행 # # [Look-ahead bias 수정] # T일 close로 신호 생성 -> T+1일 open + 슬리피지로 체결. # 마지막 봉 신호는 다음 봉이 없으므로 무시합니다. cash = initial_capital_usd position_qty = 0 position_avg_price = 0.0 position_high = 0.0 # 트레일링 스탑용 최고가 trades: List[Dict] = [] equity_curve: List[Dict] = [] # 전일 발생한 신호를 저장 (T일 close 신호 -> T+1일 open 체결) pending_buy_signal: Optional[Dict] = None pending_sell_signal: Optional[str] = None for i in range(20, len(daily_data)): # 지표 계산에 최소 20일 필요 bar = daily_data[i] ind = indicators[i] close = bar["close"] bar_date = bar["date"] bar_open = bar["open"] # -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 -- if pending_sell_signal is not None and position_qty > 0: exec_price = _apply_us_sell_slippage(bar_open) amount = position_qty * exec_price commission = amount * US_COMMISSION_RATE pnl = (exec_price - position_avg_price) * position_qty - commission pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 trade = { "date": bar_date, "action": "SELL", "ticker": ticker, "price": round(exec_price, 4), "quantity": position_qty, "amount": round(amount, 2), "commission": round(commission, 2), "reason": pending_sell_signal, "pnl": round(pnl, 2), "pnl_pct": round(pnl_pct, 2), } trades.append(trade) cash += amount - commission position_qty = 0 position_avg_price = 0.0 position_high = 0.0 pending_sell_signal = None if pending_buy_signal is not None and position_qty == 0: exec_price = _apply_us_buy_slippage(bar_open) buy_amount = min(cash * 0.95, cash) if buy_amount > 10: quantity = int(buy_amount / exec_price) if quantity > 0: cost = quantity * exec_price commission = cost * US_COMMISSION_RATE total_cost = cost + commission if total_cost <= cash: cash -= total_cost position_qty = quantity position_avg_price = exec_price position_high = exec_price trade = { "date": bar_date, "action": "BUY", "ticker": ticker, "price": round(exec_price, 4), "quantity": quantity, "amount": round(cost, 2), "commission": round(commission, 2), "reason": pending_buy_signal.get("reason", "buy_signal"), "pnl": 0.0, "pnl_pct": 0.0, } trades.append(trade) pending_buy_signal = None # -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) -- is_last_bar = (i == len(daily_data) - 1) # 트레일링 스탑용 최고가 갱신 (close 기준) if position_qty > 0: if close > position_high: position_high = close if not is_last_bar: # 매도 신호 체크 if position_qty > 0: sell_signal = _check_backtest_sell( close=close, avg_price=position_avg_price, high_since_entry=position_high, rsi=ind.get("rsi", 50), bb_middle=ind.get("bb_middle", close), bb_upper=ind.get("bb_upper", close), ) if sell_signal: pending_sell_signal = sell_signal # 매수 신호 체크 if position_qty == 0 and pending_sell_signal is None: prev_rsi = indicators[i - 3].get("rsi") if i >= 3 else None buy_signal = check_us_buy_signal( rsi=ind.get("rsi", 50), close=close, bb_upper=ind.get("bb_upper", close), bb_middle=ind.get("bb_middle", close), bb_lower=ind.get("bb_lower", close), avg_volume_20d=ind.get("avg_volume_20d", 0), current_volume=bar.get("volume", 0), current_price_usd=close, rsi_prev=prev_rsi, ) if buy_signal: pending_buy_signal = buy_signal # 일별 자산 기록 position_value = position_qty * close if position_qty > 0 else 0.0 total_equity = cash + position_value equity_curve.append({ "date": bar_date, "equity": round(total_equity, 2), "cash": round(cash, 2), "position_value": round(position_value, 2), }) # 4) 미청산 포지션 강제 청산 (백테스트 종료, 슬리피지 적용) if position_qty > 0 and daily_data: last_close = daily_data[-1]["close"] exec_price = _apply_us_sell_slippage(last_close) amount = position_qty * exec_price commission = amount * US_COMMISSION_RATE pnl = (exec_price - position_avg_price) * position_qty - commission pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 trades.append({ "date": daily_data[-1]["date"], "action": "SELL", "ticker": ticker, "price": round(exec_price, 4), "quantity": position_qty, "amount": round(amount, 2), "commission": round(commission, 2), "reason": "backtest_end_force_close", "pnl": round(pnl, 2), "pnl_pct": round(pnl_pct, 2), }) # 5) 메트릭 계산 metrics = calculate_backtest_metrics(trades, initial_capital_usd) logger.info( f"[US_BACKTEST] {ticker} 완료 - " f"수익률: {metrics['total_return_pct']:+.2f}%, " f"승률: {metrics['win_rate']:.1f}%, " f"MDD: {metrics['max_drawdown_pct']:.1f}%, " f"거래수: {metrics['trade_count']}" ) return { "ticker": ticker, "period_days": period_days, "initial_capital": initial_capital_usd, "metrics": metrics, "trades": trades, "equity_curve": equity_curve, } # ── 내부 유틸 함수 ──────────────────────────────────────────────────────────────── async def _fetch_us_daily_data(ticker: str, period_days: int) -> List[Dict]: """KIS API를 통해 미국 주식 일봉 데이터 조회 Returns: [{"date": str, "open": float, "high": float, "low": float, "close": float, "volume": float}, ...] (시간순 정렬) """ try: chart_data = await korea_investment_api.get_overseas_daily_chart( ticker=ticker, period="D", limit=period_days + 10, ) output = chart_data.get("output", chart_data.get("output2", [])) if not output: return [] result = [] for item in output: try: result.append({ "date": item.get("xymd", item.get("stck_bsop_date", "")), "open": float(item.get("open", item.get("ovrs_nmix_oprc", 0))), "high": float(item.get("high", item.get("ovrs_nmix_hgpr", 0))), "low": float(item.get("low", item.get("ovrs_nmix_lwpr", 0))), "close": float(item.get("clos", item.get("ovrs_nmix_prpr", 0))), "volume": float(item.get("tvol", item.get("acml_vol", 0))), }) except (ValueError, TypeError): continue # 시간순 정렬 (오래된 것부터) result.sort(key=lambda x: x["date"]) return result except Exception as e: logger.error(f"[US_BACKTEST_DATA] {ticker} 데이터 조회 실패: {e}") raise def _calculate_indicators(daily_data: List[Dict]) -> List[Dict]: """기술적 지표 계산 (RSI, 볼린저 밴드, 평균 거래량) Returns: daily_data와 동일 길이의 지표 리스트 """ closes = np.array([d["close"] for d in daily_data], dtype=np.float64) volumes = np.array([d["volume"] for d in daily_data], dtype=np.float64) n = len(closes) indicators = [{}] * n # RSI (14) rsi_values = _compute_rsi(closes, period=14) # 볼린저 밴드 (20, 2) bb_middle, bb_upper, bb_lower = _compute_bollinger(closes, period=20, std_mult=2.0) # 20일 평균 거래량 avg_vol_20d = _compute_sma(volumes, period=20) indicators = [] for i in range(n): indicators.append({ "rsi": rsi_values[i] if rsi_values[i] is not None else 50.0, "bb_middle": bb_middle[i] if bb_middle[i] is not None else closes[i], "bb_upper": bb_upper[i] if bb_upper[i] is not None else closes[i], "bb_lower": bb_lower[i] if bb_lower[i] is not None else closes[i], "avg_volume_20d": avg_vol_20d[i] if avg_vol_20d[i] is not None else 0.0, }) return indicators def _compute_rsi(closes: np.ndarray, period: int = 14) -> List[Optional[float]]: """RSI 계산""" n = len(closes) rsi = [None] * n if n <= period: return rsi deltas = np.diff(closes) gains = np.where(deltas > 0, deltas, 0.0) losses = np.where(deltas < 0, -deltas, 0.0) avg_gain = np.mean(gains[:period]) avg_loss = np.mean(losses[:period]) for i in range(period, n - 1): if avg_loss == 0: rsi[i + 1] = 100.0 else: rs = avg_gain / avg_loss rsi[i + 1] = 100.0 - (100.0 / (1.0 + rs)) if i < len(gains): avg_gain = (avg_gain * (period - 1) + gains[i]) / period avg_loss = (avg_loss * (period - 1) + losses[i]) / period return rsi def _compute_bollinger( closes: np.ndarray, period: int = 20, std_mult: float = 2.0, ) -> tuple: """볼린저 밴드 계산 -> (middle, upper, lower)""" n = len(closes) middle = [None] * n upper = [None] * n lower = [None] * n for i in range(period - 1, n): window = closes[i - period + 1:i + 1] m = float(np.mean(window)) s = float(np.std(window, ddof=0)) middle[i] = m upper[i] = m + std_mult * s lower[i] = m - std_mult * s return middle, upper, lower def _compute_sma(data: np.ndarray, period: int = 20) -> List[Optional[float]]: """단순 이동평균""" n = len(data) sma = [None] * n for i in range(period - 1, n): sma[i] = float(np.mean(data[i - period + 1:i + 1])) return sma def _check_backtest_sell( close: float, avg_price: float, high_since_entry: float, rsi: float, bb_middle: float, bb_upper: float, ) -> Optional[str]: """백테스트용 매도 조건 판별 (간소화) Returns: 매도 이유 문자열 또는 None (매도 안 함) """ if avg_price <= 0: return None pnl_pct = ((close - avg_price) / avg_price) * 100 # 손절 if pnl_pct <= US_STOP_LOSS_PCT: return f"stop_loss ({pnl_pct:.1f}%)" # 트레일링 스탑 (고점 대비 -4%) if high_since_entry > 0: drawdown_from_high = ((close - high_since_entry) / high_since_entry) if drawdown_from_high <= -US_TRAILING_STOP_PCT and pnl_pct > 0: return f"trailing_stop (고점 대비 {drawdown_from_high * 100:.1f}%)" # 1차 익절: RSI >= 55 + BB 중앙선 상회 if rsi >= 55 and close > bb_middle: return f"take_profit_1 (RSI={rsi:.0f}, 중앙선 상회)" # 2차 익절: RSI >= 70 또는 BB 상단 근접 if rsi >= 70: return f"take_profit_2 (RSI={rsi:.0f})" if bb_upper > 0 and close >= bb_upper * 0.99: return f"take_profit_bb_upper (BB상단 도달)" return None