"""미국 주식 백테스트 엔진 KIS API의 해외 주식 일봉 데이터를 활용한 백테스트. us_stock_strategy의 매수/매도 신호를 과거 데이터에 적용하여 성과 검증. """ import logging import math 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 @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) 시뮬레이션 실행 cash = initial_capital_usd position_qty = 0 position_avg_price = 0.0 position_high = 0.0 # 트레일링 스탑용 최고가 trades: List[Dict] = [] equity_curve: List[Dict] = [] for i in range(20, len(daily_data)): # 지표 계산에 최소 20일 필요 bar = daily_data[i] ind = indicators[i] close = bar["close"] bar_date = bar["date"] # 포지션 보유 시: 매도 조건 체크 if position_qty > 0: # 트레일링 스탑용 최고가 갱신 if close > position_high: position_high = close 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: amount = position_qty * close commission = amount * US_COMMISSION_RATE pnl = (close - position_avg_price) * position_qty - commission pnl_pct = ((close - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 trade = { "date": bar_date, "action": "SELL", "ticker": ticker, "price": close, "quantity": position_qty, "amount": round(amount, 2), "commission": round(commission, 2), "reason": 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 # 포지션 없을 때: 매수 조건 체크 if position_qty == 0: 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: # 전체 자본의 최대 95% 투입 (여유 현금 확보) buy_amount = min(cash * 0.95, cash) if buy_amount > 10: # 최소 $10 이상 quantity = int(buy_amount / close) if quantity > 0: cost = quantity * close commission = cost * US_COMMISSION_RATE total_cost = cost + commission if total_cost <= cash: cash -= total_cost position_qty = quantity position_avg_price = close position_high = close trade = { "date": bar_date, "action": "BUY", "ticker": ticker, "price": close, "quantity": quantity, "amount": round(cost, 2), "commission": round(commission, 2), "reason": buy_signal.get("reason", "buy_signal"), "pnl": 0.0, "pnl_pct": 0.0, } trades.append(trade) # 일별 자산 기록 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"] amount = position_qty * last_close commission = amount * US_COMMISSION_RATE pnl = (last_close - position_avg_price) * position_qty - commission pnl_pct = ((last_close - 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": last_close, "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