#!/usr/bin/env python3 """ Auto-Stock 독립 3년 백테스트 스크립트 DB 연결 없이 FinanceDataReader로 일봉 데이터를 직접 가져와 RSI + BB 평균회귀 + 모멘텀 돌파 + 눌림목 매수의 듀얼 전략 성과를 시뮬레이션합니다. 적응형 모드(Aggressive/Normal/Defensive/Crisis)가 시장 변동성에 따라 자동 전환됩니다. 실행 방법: python backend/run_backtest.py python backend/run_backtest.py --start 2022-01-01 --end 2025-01-01 python backend/run_backtest.py --capital 50000000 CLAUDE.md 섹션 5(리스크), 6(평균회귀), 8.4(듀얼전략), 8.5(적응형모드) 규칙 준수. 출력: - 터미널: 성과 요약 (수익률, CAGR, MDD, 승률, Sharpe 등) - JSON: backtest_results/ 디렉토리에 상세 결과 저장 시뮬레이션 특징: - KODEX200 기반 시장 국면 판단 (BULL/BEAR/SIDEWAYS) - 적응형 모드: ATR 백분위 + ADX + 5일 수익률로 4단계 자동 전환 - 수수료/세금/슬리피지 반영 (KOSPI: 0.18%, KOSDAQ: 0.23%) - 분할 매도 지원 (50%/30% 단계별) - 섹터 집중 방지 (동일 섹터 최대 2종목) """ import json import math import os import random import sys import time from dataclasses import dataclass, field, asdict from datetime import date, datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import argparse try: import FinanceDataReader as fdr except ImportError: print("[ERROR] FinanceDataReader가 설치되어 있지 않습니다.") print(" pip install finance-datareader") sys.exit(1) import pandas as pd # --------------------------------------------------------------------------- # 백테스트 대상 종목 # --------------------------------------------------------------------------- # 백테스트 대상 유니버스: {종목코드: (종목명, 시장구분)} # KOSPI 대형주 + KOSDAQ 변동성 종목 + 중소형주 조합 # 총 32종목으로 다양한 시장 환경 시뮬레이션 BACKTEST_UNIVERSE: Dict[str, Tuple[str, str]] = { # ── KOSPI 초대형주 (추세 필터로 보호) ── "005930": ("삼성전자", "KOSPI"), "000660": ("SK하이닉스", "KOSPI"), "005380": ("현대차", "KOSPI"), "000270": ("기아", "KOSPI"), # ── KOSPI 대형주 ── "035420": ("NAVER", "KOSPI"), "035720": ("카카오", "KOSPI"), "051910": ("LG화학", "KOSPI"), "006400": ("삼성SDI", "KOSPI"), "068270": ("셀트리온", "KOSPI"), "055550": ("신한지주", "KOSPI"), "036570": ("엔씨소프트", "KOSPI"), "323410": ("카카오뱅크", "KOSPI"), "352820": ("하이브", "KOSPI"), "003670": ("포스코퓨처엠", "KOSPI"), # ── KOSDAQ 변동성 종목 (평균회귀 적합) ── "247540": ("에코프로비엠", "KOSDAQ"), "196170": ("알테오젠", "KOSDAQ"), "403870": ("HPSP", "KOSDAQ"), "293490": ("카카오게임즈", "KOSDAQ"), "263750": ("펄어비스", "KOSDAQ"), "086520": ("에코프로", "KOSDAQ"), # ── 추가: 변동성 높은 중소형주 ── "028260": ("삼성물산", "KOSPI"), "009150": ("삼성전기", "KOSPI"), "034730": ("SK", "KOSPI"), "066570": ("LG전자", "KOSPI"), "012330": ("현대모비스", "KOSPI"), "028050": ("삼성엔지니어링", "KOSPI"), "011170": ("롯데케미칼", "KOSPI"), "010130": ("고려아연", "KOSPI"), "034020": ("두산에너빌리티", "KOSPI"), "047050": ("포스코인터내셔널", "KOSPI"), "005490": ("POSCO홀딩스", "KOSPI"), "018260": ("삼성에스디에스", "KOSPI"), "032830": ("삼성생명", "KOSPI"), "316140": ("우리금융지주", "KOSPI"), } # KODEX 200 ETF 코드 — 시장 국면(BULL/BEAR/SIDEWAYS) 판단의 기준 지수 KODEX200_CODE = "069500" # --------------------------------------------------------------------------- # 기본 설정 # --------------------------------------------------------------------------- # 초기 자본금 (원) — CLI --capital 옵션으로 변경 가능 INITIAL_CAPITAL = 100_000_000 # 1억원 # 수수료/세금/슬리피지 (실제 한국투자증권 기준) BUY_COMMISSION_RATE = 0.00015 # 매수 수수료 0.015% SELL_COMMISSION_RATE = 0.00015 # 매도 수수료 0.015% SELL_TAX_KOSPI = 0.0018 # KOSPI 거래세 0.18% SELL_TAX_KOSDAQ = 0.0023 # KOSDAQ 거래세 0.23% SLIPPAGE_PCT = 0.001 # 슬리피지 +/-0.1% (시장가 주문 시뮬레이션) # 기술적 지표 기간 (고정 — 실서비스 config.py와 동일) RSI_PERIOD = 14 # RSI 계산 기간 BB_PERIOD = 20 # 볼린저 밴드 이동평균 기간 ATR_PERIOD = 14 # ATR 계산 기간 # 시장 국면 판단 파라미터 (CLAUDE.md 6.1) MA200_PERIOD = 200 # 200일 이동평균선 MA200_SLOPE_LOOKBACK = 20 # MA200 기울기 계산용 과거 기간 ADX_SIDEWAYS_THRESHOLD = 15 # ADX < 15이면 SIDEWAYS 판정 # 볼린저 밴드 상단 청산 기준 (99% 도달 시 전량 매도) BB_UPPER_THRESHOLD = 0.99 # 난수 시드 — 슬리피지 재현성 보장 RANDOM_SEED = 42 # --------------------------------------------------------------------------- # 적응형 전략 모드 파라미터 (시장 상황에 따라 자동 전환) # determine_adaptive_mode()에 의해 ATR 백분위 + ADX + 5일 수익률로 모드 결정 # run_backtest_sweep.py에서 이 블록을 동적으로 교체하여 파라미터 스윕 실행 # --------------------------------------------------------------------------- ADAPTIVE_MODES = { "aggressive": { "rsi_entry": 48.0, "bear_rsi_entry": 38.0, "bb_pct_b_scale": 0.45, "position_size": 0.07, "max_holdings": 15, "stop_loss": -8.0, "stop_loss_vol": -12.0, "atr_stop_mult": 3.0, "atr_trail_mult": 1.8, "rsi_partial": 58.0, "rsi_full": 72.0, }, "normal": { "rsi_entry": 43.0, "bear_rsi_entry": 32.0, "bb_pct_b_scale": 0.35, "position_size": 0.05, "max_holdings": 12, "stop_loss": -6.0, "stop_loss_vol": -10.0, "atr_stop_mult": 2.5, "atr_trail_mult": 1.5, "rsi_partial": 52.0, "rsi_full": 67.0, }, "defensive": { "rsi_entry": 35.0, "bear_rsi_entry": 25.0, "bb_pct_b_scale": 0.20, "position_size": 0.03, "max_holdings": 8, "stop_loss": -5.0, "stop_loss_vol": -8.0, "atr_stop_mult": 2.0, "atr_trail_mult": 1.2, "rsi_partial": 48.0, "rsi_full": 60.0, }, "crisis": { "rsi_entry": 28.0, "bear_rsi_entry": 20.0, "bb_pct_b_scale": 0.10, "position_size": 0.0, # 신규 진입 차단 "max_holdings": 3, "stop_loss": -4.0, "stop_loss_vol": -6.0, "atr_stop_mult": 1.5, "atr_trail_mult": 1.0, "rsi_partial": 45.0, "rsi_full": 55.0, }, } def determine_adaptive_mode( idx_closes: List[float], idx_highs: List[float], idx_lows: List[float], ) -> str: """시장 상황에 따라 적응형 모드 결정. - Aggressive: 변동성 낮고 + 추세 강하고 + 최근 상승 - Defensive: 변동성 높거나 + 최근 하락 - Crisis: 변동성 극단 또는 급락 - Normal: 기본 """ if len(idx_closes) < 60: return "normal" # ATR 백분위 (20일 ATR의 60일 롤링 백분위) atrs = [] for i in range(max(20, len(idx_closes) - 60), len(idx_closes)): window_h = idx_highs[max(0, i - 14):i] window_l = idx_lows[max(0, i - 14):i] window_c = idx_closes[max(0, i - 14):i] if len(window_c) >= 2: trs = [] for j in range(1, len(window_c)): tr = max( window_h[j] - window_l[j], abs(window_h[j] - window_c[j - 1]), abs(window_l[j] - window_c[j - 1]), ) trs.append(tr) if trs: atrs.append(sum(trs) / len(trs)) current_atr = atrs[-1] if atrs else 0 atr_pctile = sum(1 for a in atrs if a <= current_atr) / max(len(atrs), 1) * 100 # 5일 수익률 ret_5d = (idx_closes[-1] - idx_closes[-6]) / idx_closes[-6] * 100 if len(idx_closes) >= 6 else 0 # ADX (간이 계산) adx = calculate_adx_simple(idx_highs[-30:], idx_lows[-30:], idx_closes[-30:]) if len(idx_closes) >= 30 else 20 # Crisis 판정 (최우선) if atr_pctile > 90 or ret_5d < -5.0: return "crisis" # Defensive 판정 if atr_pctile > 70 or ret_5d < -3.0: return "defensive" # Aggressive 판정 (모든 조건 AND) if atr_pctile < 30 and adx > 25 and ret_5d > 2.0: return "aggressive" return "normal" def calculate_adx_simple(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> float: """간이 ADX 계산 (백테스트용).""" if len(closes) < period + 1: return 20.0 plus_dm, minus_dm, tr_list = [], [], [] for i in range(1, len(closes)): h, l, pc = highs[i], lows[i], closes[i - 1] up = h - highs[i - 1] down = lows[i - 1] - l plus_dm.append(up if up > down and up > 0 else 0) minus_dm.append(down if down > up and down > 0 else 0) tr_list.append(max(h - l, abs(h - pc), abs(l - pc))) def _smooth(data, p): if len(data) < p: return [sum(data) / max(len(data), 1)] s = [sum(data[:p])] for v in data[p:]: s.append(s[-1] - s[-1] / p + v) return s s_tr = _smooth(tr_list, period) s_pdm = _smooth(plus_dm, period) s_mdm = _smooth(minus_dm, period) dxs = [] for i in range(min(len(s_tr), len(s_pdm), len(s_mdm))): if s_tr[i] == 0: continue pdi = s_pdm[i] / s_tr[i] * 100 mdi = s_mdm[i] / s_tr[i] * 100 denom = pdi + mdi if denom > 0: dxs.append(abs(pdi - mdi) / denom * 100) if not dxs: return 20.0 adx_smooth = _smooth(dxs, period) return adx_smooth[-1] if adx_smooth else 20.0 # --------------------------------------------------------------------------- # 데이터 구조 # --------------------------------------------------------------------------- @dataclass class BarData: """일봉 OHLCV 데이터""" trade_date: date open_price: float high_price: float low_price: float close_price: float volume: int stock_code: str = "" stock_name: str = "" @dataclass class Position: """보유 포지션""" stock_code: str entry_date: date entry_price: float quantity: int highest_since_entry: float remaining_pct: float = 1.0 strategy: str = "MEAN_REVERSION" # 진입 전략 confidence: float = 0.5 # 진입 신뢰도 original_qty: int = 0 # 최초 진입 수량 (분할 매도 remaining_pct 계산용) @dataclass class Trade: """체결 거래""" stock_code: str stock_name: 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 = "" market: str = "KOSPI" @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 # --------------------------------------------------------------------------- # 기술적 지표 계산 (순수 Python, 외부 의존성 없음) # --------------------------------------------------------------------------- def calculate_rsi(closes: List[float], period: int = 14) -> Optional[float]: """RSI(Relative Strength Index)""" if len(closes) < period + 1: return None gains, losses = [], [] for i in range(-period, 0): diff = closes[i] - closes[i - 1] gains.append(max(diff, 0)) losses.append(max(-diff, 0)) avg_gain = sum(gains) / period avg_loss = sum(losses) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) def calculate_bollinger(closes: List[float], period: int = 20) -> Optional[Tuple[float, float, float]]: """볼린저 밴드 (2sigma) -> (upper, middle, lower)""" if len(closes) < period: return None window = closes[-period:] middle = sum(window) / period variance = sum((p - middle) ** 2 for p in window) / period std = variance ** 0.5 return (middle + 2 * std, middle, middle - 2 * std) def calculate_atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]: """ATR(Average True Range)""" if len(highs) < period + 1 or len(lows) < period + 1 or len(closes) < period + 1: return None true_ranges = [] for i in range(1, len(highs)): h, l, c_prev = highs[i], lows[i], closes[i - 1] tr = max(h - l, abs(h - c_prev), abs(l - c_prev)) true_ranges.append(tr) if len(true_ranges) < period: return None return sum(true_ranges[-period:]) / period def calculate_adx(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]: """ADX(Average Directional Index) - Wilder's smoothing""" n = len(highs) if n < period * 2 + 1: return None trs, plus_dms, minus_dms = [], [], [] for i in range(1, n): h, l, c_prev = highs[i], lows[i], closes[i - 1] h_prev, l_prev = highs[i - 1], lows[i - 1] tr = max(h - l, abs(h - c_prev), abs(l - c_prev)) up_move = h - h_prev down_move = l_prev - l plus_dm = up_move if up_move > down_move and up_move > 0 else 0.0 minus_dm = down_move if down_move > up_move and down_move > 0 else 0.0 trs.append(tr) plus_dms.append(plus_dm) minus_dms.append(minus_dm) def _wilder_smooth(values): smoothed = [sum(values[:period])] for v in values[period:]: smoothed.append(smoothed[-1] - smoothed[-1] / period + v) return smoothed atr_s = _wilder_smooth(trs) plus_dm_s = _wilder_smooth(plus_dms) minus_dm_s = _wilder_smooth(minus_dms) dxs = [] for atr_v, p_dm, m_dm in zip(atr_s, plus_dm_s, minus_dm_s): if atr_v == 0: dxs.append(0.0) continue plus_di = 100.0 * p_dm / atr_v minus_di = 100.0 * m_dm / atr_v di_sum = plus_di + minus_di dxs.append(100.0 * abs(plus_di - minus_di) / di_sum if di_sum else 0.0) if len(dxs) < period: return None # [FIX] ADX는 합산형 _wilder_smooth(초기값=sum)가 아니라 평균형 Wilder 평활을 # 써야 한다. 기존엔 DX(0~100)에 sum-accumulation을 적용해 ADX가 100을 넘어 # 발산했고, SIDEWAYS 게이트(ADX<20)가 사실상 무력화됐다. mean_reversion_strategy와 동일 수정. adx_val = sum(dxs[:period]) / period # 초기값 = 평균 for dx_v in dxs[period:]: adx_val = adx_val - adx_val / period + dx_v / period return round(adx_val, 2) def is_reversal_candle(opens: List[float], highs: List[float], lows: List[float], closes: List[float]) -> bool: """되돌림 캔들 감지 (양봉 또는 긴 하단 꼬리)""" if not closes: return False o, h, l, c = opens[-1], highs[-1], lows[-1], closes[-1] body = abs(c - o) lower_shadow = min(o, c) - l return (c > o) or (lower_shadow > 2 * max(body, 0.01)) # --------------------------------------------------------------------------- # 시장 국면 판단 (CLAUDE.md 6.1) # --------------------------------------------------------------------------- class MarketRegime: BULL = "BULL" BEAR = "BEAR" SIDEWAYS = "SIDEWAYS" def detect_market_regime( index_closes: List[float], index_highs: Optional[List[float]] = None, index_lows: Optional[List[float]] = None, ) -> str: """KODEX200 기반 시장 국면 판단 (CLAUDE.md 6.1) BULL: 종가 > MA200 AND 기울기 > 0 BEAR: 종가 < MA200 OR 기울기 < 0 SIDEWAYS: ADX < 20 OR 기울기 ~= 0 """ if len(index_closes) < MA200_PERIOD + MA200_SLOPE_LOOKBACK: return MarketRegime.BEAR # 데이터 부족 시 보수적 # ADX 체크 (SIDEWAYS) if index_highs and index_lows: adx = calculate_adx(index_highs, index_lows, index_closes) if adx is not None and adx < ADX_SIDEWAYS_THRESHOLD: return MarketRegime.SIDEWAYS # MA200 현재 ma200_current = sum(index_closes[-MA200_PERIOD:]) / MA200_PERIOD # MA200 20일 전 offset = MA200_PERIOD + MA200_SLOPE_LOOKBACK if len(index_closes) >= offset: ma200_prev = sum(index_closes[-offset:-MA200_SLOPE_LOOKBACK]) / MA200_PERIOD else: ma200_prev = sum(index_closes[-(MA200_PERIOD + 5):-5]) / MA200_PERIOD slope_positive = ma200_current > ma200_prev # 기울기 거의 0이면 SIDEWAYS slope_ratio = abs(ma200_current - ma200_prev) / ma200_prev if ma200_prev > 0 else 0 if slope_ratio < 0.001: return MarketRegime.SIDEWAYS if index_closes[-1] > ma200_current and slope_positive: return MarketRegime.BULL return MarketRegime.BEAR # --------------------------------------------------------------------------- # 매매 시그널 (CLAUDE.md 섹션 6.2, 6.3) # --------------------------------------------------------------------------- def check_buy_signal( regime: str, rsi: float, close: float, bb_upper: float, bb_middle: float, bb_lower: float, avg_volume_20d: float, current_volume: int, is_reversal: bool = False, rsi_prev: Optional[float] = None, mode_params: Optional[Dict] = None, closes: Optional[List[float]] = None, highs: Optional[List[float]] = None, lows: Optional[List[float]] = None, atr: Optional[float] = None, ) -> Optional[Dict]: """듀얼 전략 매수 신호 판단 (CLAUDE.md 8.4). 3가지 전략을 순서대로 체크하여 첫 번째 발동 신호 반환: 1. 평균회귀: RSI 과매도 + BB 하단 (BULL/BEAR 조건 상이) 2. 모멘텀 돌파: 20일 신고가 + 거래량 2x (BULL에서만) 3. 눌림목 매수: MA20>MA60 + MA20 근처 조정 (BULL에서만) Args: regime: 시장 국면 (BULL/BEAR/SIDEWAYS) rsi: RSI(14) 값 close: 현재 종가 bb_upper, bb_middle, bb_lower: 볼린저 밴드 상/중/하 avg_volume_20d: 20일 평균 거래량 current_volume: 당일 거래량 is_reversal: 반전 캔들 여부 (BEAR 진입 필수 조건) rsi_prev: 3일 전 RSI (반등 확인용) mode_params: 적응형 모드 파라미터 closes, highs, lows: 히스토리 가격 (모멘텀/눌림목 계산용) atr: ATR 값 (미사용, 확장 예비) Returns: dict | None: 매수 신호 {signal, strategy, confidence, description} """ mp = mode_params or ADAPTIVE_MODES["normal"] if regime == MarketRegime.SIDEWAYS: return None if mp.get("position_size", 0) <= 0: return None # [안전형 개선①] own_trend 게이트 (검증 완료: Sharpe 0.27→0.57) # 종목 자체 MA20 < MA60*0.97 (하락추세)이면 평균회귀(물타기) 진입 차단. # 모멘텀 돌파(추세추종)에는 미적용 — 전략 2 블록 진입 전에만 반영. _own_trend_ok = True if closes and len(closes) >= 60: _mr_ma20 = sum(closes[-20:]) / 20 _mr_ma60 = sum(closes[-60:]) / 60 if _mr_ma60 > 0 and _mr_ma20 < _mr_ma60 * 0.97: _own_trend_ok = False # ═══════════════════════════════════════════════════════════════ # 전략 1: 평균회귀 (RSI 과매도 + BB 하단) # ═══════════════════════════════════════════════════════════════ if current_volume >= avg_volume_20d and _own_trend_ok: if regime == MarketRegime.BULL: rsi_thresh = mp["rsi_entry"] if rsi < rsi_thresh: if rsi_prev is None or rsi > rsi_prev: # RSI 반등 확인 bb_width = bb_upper - bb_lower pct_b = (close - bb_lower) / bb_width if bb_width > 0 else 0.5 base_pct_b = mp["bb_pct_b_scale"] pct_b_max = base_pct_b + 0.10 if rsi < 30 else base_pct_b if pct_b <= pct_b_max: # 신뢰도: RSI가 낮을수록, 거래량이 클수록 높음 conf = min(1.0, 0.5 + (rsi_thresh - rsi) / 30 + (current_volume / avg_volume_20d - 1) * 0.1) return { "signal": "BUY", "strategy": "MEAN_REVERSION", "confidence": conf, "description": f"평균회귀: RSI({rsi:.1f}) %B({pct_b:.2f}) vol({current_volume/avg_volume_20d:.1f}x)", } elif regime == MarketRegime.BEAR: bear_thresh = mp["bear_rsi_entry"] if rsi < bear_thresh and is_reversal and current_volume > 1.5 * avg_volume_20d: bb_width = bb_upper - bb_lower pct_b = (close - bb_lower) / bb_width if bb_width > 0 else 0.5 if pct_b <= 0.30: conf = min(1.0, 0.4 + (bear_thresh - rsi) / 20) return { "signal": "BUY", "strategy": "MEAN_REVERSION_BEAR", "confidence": conf, "description": f"약세반등: RSI({rsi:.1f}) reversal vol({current_volume/avg_volume_20d:.1f}x)", } # ═══════════════════════════════════════════════════════════════ # 전략 2: 모멘텀 돌파 (BULL에서만) — 20일 신고가 돌파 + 거래량 급증 # ═══════════════════════════════════════════════════════════════ if regime == MarketRegime.BULL and closes and highs and len(closes) >= 20: high_20d = max(highs[-20:]) # 20일 고점 돌파 + 거래량 2배 이상 + RSI 50~70 (과매수 아닌 강한 구간) if close > high_20d and current_volume > avg_volume_20d * 2.0 and 50 < rsi < 70: conf = min(1.0, 0.6 + (current_volume / avg_volume_20d - 2) * 0.05) return { "signal": "BUY", "strategy": "MOMENTUM_BREAKOUT", "confidence": conf, "description": f"돌파: 20일고점({high_20d:.0f}) 돌파 vol({current_volume/avg_volume_20d:.1f}x) RSI({rsi:.1f})", } # ═══════════════════════════════════════════════════════════════ # 전략 3: 눌림목 매수 (BULL에서) — 상승추세 중 조정 후 반등 # ═══════════════════════════════════════════════════════════════ if regime == MarketRegime.BULL and closes and len(closes) >= 60: ma20 = sum(closes[-20:]) / 20 ma60 = sum(closes[-60:]) / 60 # 상승추세(MA20 > MA60) 중 MA20 근처까지 눌림 후 반등 if ma20 > ma60 and close > ma20 * 0.97 and close < ma20 * 1.02: if rsi > 40 and rsi < 55 and current_volume > avg_volume_20d * 1.2: if rsi_prev is not None and rsi > rsi_prev: # RSI 반등 중 conf = 0.55 return { "signal": "BUY", "strategy": "PULLBACK_BUY", "confidence": conf, "description": f"눌림목: MA20({ma20:.0f}) 근처 반등 RSI({rsi:.1f})", } return None def check_exit_signal( rsi: float, close: float, bb_upper: float, bb_middle: float, bb_lower: float, entry_price: float, atr: float, remaining_pct: float, trailing_high: float, mode_params: Optional[Dict] = None, strategy: str = "MEAN_REVERSION", holding_days: int = 0, ) -> Optional[Tuple[str, float, str]]: """전략별 차별화된 청산 신호 판단. 공통 청산: - 손절: ATR 기반 vs 하드스탑 중 타이트한 쪽 (max로 비교 = 가격 기준) - 시간 손절: 20일 보유 + 수익 < 3% -> 기회비용 청산 평균회귀/눌림목: - 1차 익절(50%): RSI >= rsi_partial OR BB 중앙선 OR 2xATR 수익 - 전량 청산: RSI >= rsi_full OR BB 상단(99%) OR 트레일링스탑 모멘텀 돌파: - 1차 익절(30%): +10% 이상 - 2차 익절(30%): +20% 이상 - 전량 청산: RSI >= 80 OR 트레일링스탑 (넓은 트레일링) Args: rsi: RSI(14) 값 close: 현재 종가 bb_upper, bb_middle, bb_lower: 볼린저 밴드 entry_price: 진입가 atr: ATR(14) 값 remaining_pct: 잔여 비율 (1.0=전량) trailing_high: 진입 이후 최고가 mode_params: 적응형 모드 파라미터 strategy: 진입 전략명 (MEAN_REVERSION/MOMENTUM_BREAKOUT/PULLBACK_BUY) holding_days: 보유 일수 Returns: (action, sell_pct, reason) | None """ mp = mode_params or ADAPTIVE_MODES["normal"] profit_pct = (close - entry_price) / entry_price * 100 if entry_price > 0 else 0 # ══ 공통: 손절 ══ # ATR이 종가의 3% 이상이면 변동성 높은 종목 → 넓은 하드스탑 적용 atr_pct = (atr / close * 100) if close > 0 else 0 stop_loss_pct = mp["stop_loss_vol"] if atr_pct > 3.0 else mp["stop_loss"] # ATR 기반 손절가 vs 하드스탑 가격 비교 # max() 사용: 가격 비교에서 높은 값 = 진입가에 가까운 = 타이트한 손절 atr_stop = entry_price - mp["atr_stop_mult"] * atr pct_stop = entry_price * (1 + stop_loss_pct / 100) stop_price = max(atr_stop, pct_stop) if close <= stop_price: return ("SELL_ALL", 1.0, f"손절 ({profit_pct:+.1f}%)") # ══ 시간 손절: 20일 보유해도 +3% 미만이면 기회비용 청산 ══ if holding_days >= 20 and profit_pct < 3.0: return ("SELL_ALL", 1.0, f"시간손절 ({holding_days}일, {profit_pct:+.1f}%)") # ══ 전략별 익절 ══ if strategy in ("MOMENTUM_BREAKOUT", "PULLBACK_BUY"): # 모멘텀/눌림목: 이익을 최대한 키움 (트레일링 위주) # 1차 익절: +10% 이상에서 30%만 매도 if remaining_pct > 0.70 and profit_pct >= 10.0: return ("SELL_PARTIAL", 0.30, f"모멘텀 1차 익절(30%): +{profit_pct:.1f}%") # 2차 익절: +20% 이상에서 추가 30% 매도 if remaining_pct > 0.40 and profit_pct >= 20.0: return ("SELL_PARTIAL", 0.30, f"모멘텀 2차 익절(30%): +{profit_pct:.1f}%") # 트레일링: 고점 대비 -ATR*2.0 하회 시 전량 (넉넉한 트레일링) if trailing_high > entry_price * 1.05: trail_stop = trailing_high - mp["atr_trail_mult"] * 1.3 * atr if close < trail_stop: return ("SELL_ALL", 1.0, f"모멘텀 트레일링 ({profit_pct:+.1f}%)") # RSI 극단 과매수 if rsi >= 80: return ("SELL_ALL", 1.0, f"과매수 청산 RSI({rsi:.0f})") else: # 평균회귀: 기존 로직 (중앙값 복귀 시 익절) min_profit_price = entry_price + 2.0 * atr if remaining_pct > 0.50: rsi_exit = rsi >= mp["rsi_partial"] bb_exit = close > bb_middle atr_exit = close >= min_profit_price if rsi_exit or bb_exit or atr_exit: reasons = [] if rsi_exit: reasons.append(f"RSI {rsi:.0f}") if bb_exit: reasons.append("BB중앙") if atr_exit: reasons.append("2xATR") return ("SELL_PARTIAL", 0.50, f"익절(50%): {' | '.join(reasons)}") # 전량 청산 full_reasons = [] if rsi >= mp["rsi_full"]: full_reasons.append(f"RSI {rsi:.0f}") if close >= bb_upper * BB_UPPER_THRESHOLD: full_reasons.append("BB상단") if trailing_high > entry_price: trail_stop = trailing_high - mp["atr_trail_mult"] * atr if close < trail_stop: full_reasons.append(f"트레일링") if full_reasons: return ("SELL_ALL", 1.0, f"청산: {' | '.join(full_reasons)} ({profit_pct:+.1f}%)") return None # --------------------------------------------------------------------------- # 수수료 / 슬리피지 # --------------------------------------------------------------------------- def apply_buy_slippage(price: float, rng: random.Random) -> float: slip = rng.uniform(0, SLIPPAGE_PCT) return price * (1 + slip) def apply_sell_slippage(price: float, rng: random.Random) -> float: 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) -> 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[Trade], initial_capital: float, trading_days: int, ) -> BacktestMetrics: """백테스트 성과 지표 종합 계산. 산출 지표: - 총 수익률, CAGR (연복리수익률) - Sharpe Ratio (무위험 3% 기준) - Sortino Ratio (하방 편차만) - Max Drawdown - 승률, Profit Factor - 평균 수익/손실, 연승/연패, 평균 보유일 Args: equity_curve: 일별 자산 가치 리스트 trades: 전체 거래 내역 initial_capital: 초기 자본금 trading_days: 총 거래일 수 Returns: BacktestMetrics: 성과 지표 데이터클래스 """ m = BacktestMetrics() if not equity_curve or len(equity_curve) < 2: return m final_equity = equity_curve[-1] m.total_return_pct = round(((final_equity - initial_capital) / initial_capital) * 100, 4) # CAGR years = trading_days / 252.0 if years > 0 and final_equity > 0: m.cagr_pct = round(((final_equity / initial_capital) ** (1 / years) - 1) * 100, 4) # 일별 수익률 daily_returns = [] 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 (무위험 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 # Sortino (하방 편차만) 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 # 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) # 거래 통계 (SELL만) closed = [t for t in trades if t.side == "SELL"] m.total_trades = len(closed) if m.total_trades == 0: return m wins = [t for t in closed if t.pnl > 0] losses = [t for t in closed 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) 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 ) 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 # 연속 승/패 cw, cl, max_cw, max_cl = 0, 0, 0, 0 for t in closed: if t.pnl > 0: cw += 1; cl = 0 else: cl += 1; cw = 0 max_cw = max(max_cw, cw) max_cl = max(max_cl, cl) m.max_consecutive_wins = max_cw m.max_consecutive_losses = max_cl # 평균 보유일 hold_days = [] for t in closed: if t.entry_date and t.exit_date: hold_days.append((t.exit_date - t.entry_date).days) m.avg_holding_days = round(sum(hold_days) / len(hold_days), 1) if hold_days else 0.0 return m # --------------------------------------------------------------------------- # 데이터 로더 (FinanceDataReader) # --------------------------------------------------------------------------- def load_data_fdr( codes: Dict[str, Tuple[str, str]], start_date: str, end_date: str, index_code: str = KODEX200_CODE, ) -> Tuple[Dict[str, List[BarData]], List[BarData], List[date]]: """FinanceDataReader로 일봉 데이터 로드 Returns: (bars_by_code, index_bars, all_dates) """ print(f"\n 데이터 수집 중 ({len(codes)}종목 + KODEX200)...") # KODEX 200 (시장 국면용) print(f" [{index_code}] KODEX 200 ...", end=" ", flush=True) idx_df = fdr.DataReader(index_code, start_date, end_date) if idx_df is None or idx_df.empty: print("FAIL (데이터 없음)") sys.exit(1) print(f"OK ({len(idx_df)}일)") index_bars: List[BarData] = [] for dt, row in idx_df.iterrows(): d = dt.date() if hasattr(dt, 'date') else dt index_bars.append(BarData( trade_date=d, open_price=float(row.get("Open", 0)), high_price=float(row.get("High", 0)), low_price=float(row.get("Low", 0)), close_price=float(row.get("Close", 0)), volume=int(row.get("Volume", 0)), stock_code=index_code, stock_name="KODEX 200", )) # 개별 종목 bars_by_code: Dict[str, List[BarData]] = {} all_dates_set: set = set() for i, (code, (name, market)) in enumerate(codes.items(), 1): print(f" [{code}] {name} ({i}/{len(codes)}) ...", end=" ", flush=True) try: df = fdr.DataReader(code, start_date, end_date) if df is None or df.empty: print("SKIP (데이터 없음)") continue print(f"OK ({len(df)}일)") except Exception as e: print(f"FAIL ({e})") continue bar_list: List[BarData] = [] for dt, row in df.iterrows(): d = dt.date() if hasattr(dt, 'date') else dt bar_list.append(BarData( trade_date=d, open_price=float(row.get("Open", 0)), high_price=float(row.get("High", 0)), low_price=float(row.get("Low", 0)), close_price=float(row.get("Close", 0)), volume=int(row.get("Volume", 0)), stock_code=code, stock_name=name, )) all_dates_set.add(d) if bar_list: bars_by_code[code] = bar_list all_dates = sorted(all_dates_set) return bars_by_code, index_bars, all_dates def align_bars_to_dates( bars_by_code: Dict[str, List[BarData]], all_dates: List[date], ) -> Dict[str, List[BarData]]: """모든 종목을 동일 날짜 인덱스에 정렬""" aligned: Dict[str, List[BarData]] = {} 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: 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) if result: aligned[code] = result return aligned # --------------------------------------------------------------------------- # 메인 시뮬레이션 루프 # --------------------------------------------------------------------------- def run_simulation( bars: Dict[str, List[BarData]], index_bars: List[BarData], all_dates: List[date], initial_capital: float, universe: Dict[str, Tuple[str, str]], sim_start_date: Optional[date] = None, ) -> Tuple[BacktestMetrics, List[Trade], List[Dict], List[Dict]]: """핵심 백테스트 시뮬레이션 -- 일봉 단위로 매매 결정 실행. 시뮬레이션 루프: 각 거래일마다: 1. 인덱스 데이터 누적 → 시장 국면 + 적응형 모드 갱신 (5일마다) 2. 보유 포지션 청산 체크 (전략별 분기) 3. 매수 후보 스캔 + confidence/RSI 순 정렬 → 매수 실행 4. 일말 평가 → equity curve / drawdown curve 기록 Args: bars: 종목별 일봉 데이터 {코드: [BarData, ...]} index_bars: KODEX200 일봉 데이터 (시장 국면용) all_dates: 전체 거래일 목록 (정렬됨) initial_capital: 초기 자본금 universe: 종목 유니버스 {코드: (이름, 시장)} Returns: (metrics, all_trades, equity_curve, drawdown_curve, open_positions, mode_changes) """ cash = initial_capital positions: Dict[str, Position] = {} all_trades: List[Trade] = [] equity_values: List[float] = [initial_capital] equity_curve: List[Dict] = [{"date": str(all_dates[0]), "equity": initial_capital}] drawdown_curve: List[Dict] = [] peak_equity = initial_capital rng = random.Random(RANDOM_SEED) # 인덱스 데이터를 날짜별로 매핑 idx_date_map = {b.trade_date: b for b in index_bars} # 인덱스 누적 리스트 (시장 국면 계산용) idx_closes_acc: List[float] = [] idx_highs_acc: List[float] = [] idx_lows_acc: List[float] = [] min_bars = 60 # RSI/BB 계산에 필요한 최소 봉 수 regime_cache = MarketRegime.BEAR regime_update_counter = 0 adaptive_mode = "normal" mode_params = ADAPTIVE_MODES["normal"] mode_changes: List[Dict] = [] # 모드 전환 이력 circuit_breaker_days = 0 # [안전형 개선②] 서킷브레이커 발동일 수 (검증용 통계) total_days = len(all_dates) last_progress = -1 for bar_idx in range(total_days): current_date = all_dates[bar_idx] # 진행률 표시 progress = int(bar_idx / total_days * 100) if progress >= last_progress + 10: last_progress = progress print(f" 시뮬레이션 진행: {progress}% ({current_date})") # 인덱스 데이터 누적 idx_bar = idx_date_map.get(current_date) if idx_bar: idx_closes_acc.append(idx_bar.close_price) idx_highs_acc.append(idx_bar.high_price) idx_lows_acc.append(idx_bar.low_price) # [안전형 개선②] 지수 서킷브레이커 (검증 완료: 2024H2 CAGR -9.9%→+1.0%, MDD -7→-1.9) # KODEX200 당일 종가 < MA20(20일) → 신규 진입 전면 중단 (기존 보유 청산/관리는 정상). # 매일 갱신(적응형 모드처럼 5일 지연시키지 않음) — 하락장 상관붕괴에 즉시 반응해야 함. idx_circuit_breaker = False if len(idx_closes_acc) >= 20: _idx_ma20_cb = sum(idx_closes_acc[-20:]) / 20 if idx_closes_acc[-1] < _idx_ma20_cb: idx_circuit_breaker = True # 시장 국면 + 적응형 모드 (매 5일마다 업데이트) regime_update_counter += 1 if regime_update_counter >= 5 and len(idx_closes_acc) >= MA200_PERIOD + MA200_SLOPE_LOOKBACK: regime_cache = detect_market_regime(idx_closes_acc, idx_highs_acc, idx_lows_acc) # 적응형 모드 결정 new_mode = determine_adaptive_mode(idx_closes_acc, idx_highs_acc, idx_lows_acc) if new_mode != adaptive_mode: mode_changes.append({ "date": str(current_date), "from": adaptive_mode, "to": new_mode, }) adaptive_mode = new_mode mode_params = ADAPTIVE_MODES[adaptive_mode] regime_update_counter = 0 # [FIX] 워밍업 구간(sim_start_date 이전): 국면·지표 누적만 하고 # 매매·평가·성과기록은 건너뛴다. 이렇게 해야 국면판정 220봉 워밍업이 # sim_start_date 시점에 이미 완료되어, 짧은 기간 백테스트도 유효해지고 # 구간 분할 시 거래수·수익이 가산적으로 일치한다. if sim_start_date is not None and current_date < sim_start_date: continue if idx_circuit_breaker: circuit_breaker_days += 1 # 현재 평가 자산 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 # ---- 매도 시그널 (기존 포지션) ---- sell_orders = [] 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_val = calculate_rsi(closes, RSI_PERIOD) bb = calculate_bollinger(closes, BB_PERIOD) atr_val = calculate_atr(highs, lows, closes, ATR_PERIOD) if rsi_val is None or bb is None or atr_val is None: continue bb_upper, bb_middle, bb_lower = bb # 최고가 업데이트 pos.highest_since_entry = max(pos.highest_since_entry, current_bar.high_price) _holding_days = (current_date - pos.entry_date).days if pos.entry_date else 0 exit_sig = check_exit_signal( rsi=rsi_val, close=current_bar.close_price, bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, entry_price=pos.entry_price, atr=atr_val, remaining_pct=pos.remaining_pct, trailing_high=pos.highest_since_entry, mode_params=mode_params, strategy=pos.strategy, holding_days=_holding_days, ) if exit_sig: action, sell_pct, reason = exit_sig sell_qty = int(pos.quantity * sell_pct) if sell_qty <= 0: sell_qty = pos.quantity sell_orders.append((code, current_bar, sell_qty, sell_pct, reason)) # 매도 실행 for code, bar, sell_qty, sell_pct, reason in sell_orders: pos = positions.get(code) if pos is None: continue sell_qty = min(sell_qty, pos.quantity) exec_price = apply_sell_slippage(bar.close_price, rng) market = universe.get(code, ("", "KOSPI"))[1] proceeds, commission, tax = calc_sell_proceeds(exec_price, sell_qty, market) cash += proceeds slippage_amount = abs(bar.close_price - exec_price) * sell_qty 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 = Trade( stock_code=code, stock_name=universe.get(code, ("", ""))[0], 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=reason, market=market, ) all_trades.append(trade) remaining_qty = pos.quantity - sell_qty if remaining_qty <= 0: del positions[code] else: pos.quantity = remaining_qty # remaining_pct: 원래 총수량 기준으로 계산 (연속 분할 매도 정확성 보장) if pos.original_qty > 0: pos.remaining_pct = remaining_qty / pos.original_qty else: pos.remaining_pct = remaining_qty / (remaining_qty + sell_qty) # ---- 매수 시그널 ---- # [안전형 개선②] 서킷브레이커 발동 중이면 신규 진입 전면 스킵 (기존 보유 관리는 위에서 정상 수행) _max_holdings = mode_params.get("max_holdings", 12) if len(positions) < _max_holdings and not idx_circuit_breaker: buy_candidates = [] for code, stock_bars in bars.items(): if code in positions: continue 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 # 가격 필터 (CLAUDE.md 5.4) if current_bar.close_price < 5000: continue rsi_val = calculate_rsi(closes, RSI_PERIOD) bb = calculate_bollinger(closes, BB_PERIOD) if rsi_val is None or bb is None: continue bb_upper, bb_middle, bb_lower = bb vol_window = volumes[-20:] if len(volumes) >= 20 else volumes avg_vol = sum(vol_window) / len(vol_window) if vol_window else 0 rsi_prev = None if len(closes) > min_bars + 3: rsi_prev = calculate_rsi(closes[:-3], RSI_PERIOD) reversal = is_reversal_candle( opens[-3:], highs[-3:], lows[-3:], closes[-3:], ) # ── [v3 필터 1] 급락 방어 — 최근 5일 하락률 > 10%면 진입 대기 ── if len(closes) >= 6: recent_5d_chg = (closes[-1] - closes[-6]) / closes[-6] if recent_5d_chg < -0.10: # 5일간 -10% 이상 급락 continue # 패닉 구간 — 바닥 확인 전 진입 금지 # ── [v3 필터 2] 극단적 추세만 차단 ── if len(closes) >= 60: ma20 = sum(closes[-20:]) / 20 ma60 = sum(closes[-60:]) / 60 if ma20 > ma60 * 1.08: continue if ma20 < ma60 * 0.88: continue # ── [v3 필터 3] 동일 섹터 집중 방지 (간이 상관관계 필터) ── # 이미 포지션이 있는 종목과 같은 "그룹"이면 스킵 _SECTOR_GROUPS = { "반도체": {"005930", "000660", "009150"}, "자동차": {"005380", "000270", "012330"}, "2차전지": {"006400", "051910", "003670", "247540", "086520"}, "플랫폼": {"035420", "035720", "323410", "293490"}, "게임": {"036570", "263750", "352820"}, } _sector_of = {} for _sec, _codes in _SECTOR_GROUPS.items(): for _c in _codes: _sector_of[_c] = _sec my_sector = _sector_of.get(code) if my_sector: same_sector_count = sum( 1 for pc in positions if _sector_of.get(pc) == my_sector ) if same_sector_count >= 2: # 동일 섹터 최대 2종목 continue # ── [v3 필터 4] 변동성 필터 ── atr_val_buy = calculate_atr(highs, lows, closes, ATR_PERIOD) if atr_val_buy and current_bar.close_price > 0: atr_ratio = atr_val_buy / current_bar.close_price if atr_ratio < 0.01: continue atr_val_entry = calculate_atr(highs, lows, closes, ATR_PERIOD) signal = check_buy_signal( regime=regime_cache, rsi=rsi_val, close=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, is_reversal=reversal, rsi_prev=rsi_prev, mode_params=mode_params, closes=closes, highs=highs, lows=lows, atr=atr_val_entry, ) if signal: buy_candidates.append((code, current_bar, signal, rsi_val)) # 신뢰도 높은 순 → RSI 낮은 순으로 우선 매수 buy_candidates.sort(key=lambda x: (-x[2].get("confidence", 0.5), x[3])) for code, bar, signal, _ in buy_candidates: if len(positions) >= _max_holdings: break _pos_size = mode_params.get("position_size", 0.05) # 신뢰도 기반 동적 포지션: 신뢰도 0.7+ → 최대 1.5배 _conf = signal.get("confidence", 0.5) _conf_mult = 0.7 + _conf * 0.8 # conf 0.5→1.1x, 0.8→1.34x, 1.0→1.5x budget = total_equity * _pos_size * _conf_mult budget = min(budget, cash) if budget <= 0: break exec_price = apply_buy_slippage(bar.close_price, rng) quantity = int(budget / (exec_price * (1 + BUY_COMMISSION_RATE))) if quantity <= 0: continue 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 - bar.close_price) * quantity trade = Trade( stock_code=code, stock_name=universe.get(code, ("", ""))[0], side="BUY", entry_date=current_date, entry_price=exec_price, quantity=quantity, commission=commission, slippage=slippage_amount, reason=signal.get("description", ""), market=universe.get(code, ("", "KOSPI"))[1], ) all_trades.append(trade) positions[code] = Position( stock_code=code, entry_date=current_date, entry_price=exec_price, quantity=quantity, highest_since_entry=exec_price, strategy=signal.get("strategy", "MEAN_REVERSION"), confidence=_conf, original_qty=quantity, ) # 일말 평가 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), }) 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), }) # 미청산 포지션 정보 (최종 평가) open_positions = [] for code, pos in positions.items(): stock_bars = bars.get(code, []) if stock_bars: last_close = stock_bars[-1].close_price unrealized_pnl = (last_close - pos.entry_price) * pos.quantity open_positions.append({ "code": code, "name": universe.get(code, ("", ""))[0], "entry_price": pos.entry_price, "last_close": last_close, "quantity": pos.quantity, "unrealized_pnl": round(unrealized_pnl, 2), "unrealized_pnl_pct": round((last_close / pos.entry_price - 1) * 100, 2), }) # [FIX] CAGR 연율화 분모는 워밍업(sim_start_date 이전)을 제외한 실제 거래일 수를 # 써야 한다. len(all_dates)는 워밍업 340일을 포함해 years를 부풀려 CAGR을 과소평가했다. _sim_trading_days = sum( 1 for _d in all_dates if sim_start_date is None or _d >= sim_start_date ) metrics = compute_metrics(equity_values, all_trades, initial_capital, _sim_trading_days) # [안전형 개선②] 서킷브레이커 발동 통계 출력 (검증용) if _sim_trading_days: print(f" [서킷브레이커] KODEX200 종가 List[Dict]: """종목별 매매 성과 집계""" stock_stats: Dict[str, Dict] = {} for t in trades: if t.side != "SELL": continue code = t.stock_code if code not in stock_stats: stock_stats[code] = { "code": code, "name": universe.get(code, ("?", "?"))[0], "market": universe.get(code, ("?", "?"))[1], "wins": 0, "losses": 0, "total_pnl": 0.0, "trades": 0, } s = stock_stats[code] s["trades"] += 1 s["total_pnl"] += t.pnl if t.pnl > 0: s["wins"] += 1 else: s["losses"] += 1 result = [] for s in stock_stats.values(): s["win_rate"] = round(s["wins"] / s["trades"] * 100, 1) if s["trades"] > 0 else 0 s["total_pnl"] = round(s["total_pnl"], 0) result.append(s) result.sort(key=lambda x: x["total_pnl"], reverse=True) return result # --------------------------------------------------------------------------- # 출력 # --------------------------------------------------------------------------- def format_currency(val: float) -> str: """통화 형식 (1,000,000원)""" if val >= 0: return f"{val:>15,.0f}원" return f"{val:>15,.0f}원" def print_results( metrics: BacktestMetrics, stock_summary: List[Dict], start_date: str, end_date: str, initial_capital: float, final_equity: float, open_positions: List[Dict], ): w = 60 print() print("=" * w) print(f" Auto-Stock 3년 백테스트 결과 ({start_date} ~ {end_date})") print("=" * w) print(f" 초기 자본: {format_currency(initial_capital)}") print(f" 최종 자본: {format_currency(final_equity)}") print(f" 총 수익률: {metrics.total_return_pct:>10.2f}%") print(f" CAGR: {metrics.cagr_pct:>10.2f}%") print(f" Sharpe Ratio: {metrics.sharpe_ratio:>10.4f}") print(f" Sortino Ratio: {metrics.sortino_ratio:>10.4f}") print(f" Max Drawdown: {metrics.max_drawdown_pct:>10.2f}%") print(f" 승률: {metrics.win_rate:>7.1f}% ({metrics.winning_trades}승 / {metrics.losing_trades}패)") print(f" Profit Factor: {metrics.profit_factor:>10.4f}") pf_str = f"{metrics.profit_factor:.4f}" if metrics.profit_factor != float("inf") else "INF" print(f" 평균 수익(승): {metrics.avg_win_pct:>10.2f}%") print(f" 평균 손실(패): {metrics.avg_loss_pct:>10.2f}%") print(f" 총 매매 횟수: {metrics.total_trades:>10d}회") print(f" 평균 보유 기간: {metrics.avg_holding_days:>10.1f}일") print(f" 최대 연승: {metrics.max_consecutive_wins:>10d}회") print(f" 최대 연패: {metrics.max_consecutive_losses:>10d}회") print("-" * w) print(" 종목별 성과 Top 5 (수익 기준):") for s in stock_summary[:5]: pnl_str = f"+{s['total_pnl']:,.0f}" if s["total_pnl"] >= 0 else f"{s['total_pnl']:,.0f}" print(f" {s['name']:>10s}({s['code']}): {pnl_str}원 ({s['wins']}승 {s['losses']}패, 승률 {s['win_rate']}%)") if len(stock_summary) > 5: print() print(" 종목별 성과 Bottom 5 (손실 기준):") for s in stock_summary[-5:]: pnl_str = f"+{s['total_pnl']:,.0f}" if s["total_pnl"] >= 0 else f"{s['total_pnl']:,.0f}" print(f" {s['name']:>10s}({s['code']}): {pnl_str}원 ({s['wins']}승 {s['losses']}패, 승률 {s['win_rate']}%)") if open_positions: print("-" * w) print(f" 미청산 포지션 ({len(open_positions)}종목):") for p in open_positions: sign = "+" if p["unrealized_pnl"] >= 0 else "" print(f" {p['name']:>10s}({p['code']}): {sign}{p['unrealized_pnl']:,.0f}원 ({sign}{p['unrealized_pnl_pct']:.1f}%)") print("=" * w) # --------------------------------------------------------------------------- # JSON 저장 # --------------------------------------------------------------------------- def save_results_json( metrics: BacktestMetrics, all_trades: List[Trade], equity_curve: List[Dict], drawdown_curve: List[Dict], stock_summary: List[Dict], open_positions: List[Dict], start_date: str, end_date: str, initial_capital: float, output_dir: str, ): ts = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"backtest_{start_date}_{end_date}_{ts}.json" filepath = os.path.join(output_dir, filename) # Trade -> dict 변환 trades_list = [] for t in all_trades: trades_list.append({ "stock_code": t.stock_code, "stock_name": t.stock_name, "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, "market": t.market, }) # metrics -> dict metrics_dict = { "total_return_pct": metrics.total_return_pct, "cagr_pct": metrics.cagr_pct, "sharpe_ratio": metrics.sharpe_ratio, "sortino_ratio": metrics.sortino_ratio, "max_drawdown_pct": metrics.max_drawdown_pct, "win_rate": metrics.win_rate, "profit_factor": metrics.profit_factor if metrics.profit_factor != float("inf") else 999999.0, "avg_win_pct": metrics.avg_win_pct, "avg_loss_pct": metrics.avg_loss_pct, "total_trades": metrics.total_trades, "winning_trades": metrics.winning_trades, "losing_trades": metrics.losing_trades, "max_consecutive_wins": metrics.max_consecutive_wins, "max_consecutive_losses": metrics.max_consecutive_losses, "avg_holding_days": metrics.avg_holding_days, } result = { "config": { "start_date": start_date, "end_date": end_date, "initial_capital": initial_capital, "strategy": "RSI+BB Mean Reversion (CLAUDE.md Section 6)", "universe": {k: {"name": v[0], "market": v[1]} for k, v in BACKTEST_UNIVERSE.items()}, "params": { "rsi_period": RSI_PERIOD, "bb_period": BB_PERIOD, "atr_period": ATR_PERIOD, "adaptive_modes": ADAPTIVE_MODES, }, "costs": { "buy_commission": BUY_COMMISSION_RATE, "sell_commission": SELL_COMMISSION_RATE, "tax_kospi": SELL_TAX_KOSPI, "tax_kosdaq": SELL_TAX_KOSDAQ, "slippage_pct": SLIPPAGE_PCT, }, }, "metrics": metrics_dict, "stock_summary": stock_summary, "open_positions": open_positions, "trades": trades_list, "equity_curve_sample": equity_curve[::5], # 5일 간격으로 샘플링 (파일 크기 제한) "drawdown_curve_sample": drawdown_curve[::5], } with open(filepath, "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) size_kb = os.path.getsize(filepath) / 1024 print(f"\n 결과 저장: {filepath} ({size_kb:.1f} KB)") return filepath # --------------------------------------------------------------------------- # 메인 # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Auto-Stock 독립 3년 백테스트") parser.add_argument("--start", type=str, default="2023-03-25", help="시작일 (YYYY-MM-DD)") parser.add_argument("--end", type=str, default="2026-03-25", help="종료일 (YYYY-MM-DD)") parser.add_argument("--capital", type=int, default=INITIAL_CAPITAL, help="초기 자본금 (원)") args = parser.parse_args() start_date = args.start end_date = args.end initial_capital = args.capital print() print("=" * 60) print(" Auto-Stock RSI+BB 평균회귀 백테스트") print(f" 기간: {start_date} ~ {end_date}") print(f" 자본: {initial_capital:,}원") print(f" 종목: {len(BACKTEST_UNIVERSE)}종목") print("=" * 60) t0 = time.time() # 1. 데이터 수집 # [FIX] 국면판정(MA200+기울기=220거래일) 및 지표 워밍업을 위해 start_date보다 # 약 340일(≈230거래일) 앞선 데이터를 추가 로드한다. 실제 매매·성과 집계는 # run_simulation의 sim_start_date=start_date 부터만 이뤄진다(워밍업 아티팩트 제거). from datetime import datetime as _dt from datetime import timedelta as _td _sim_start = _dt.strptime(start_date, "%Y-%m-%d").date() _warmup_start = (_sim_start - _td(days=340)).strftime("%Y-%m-%d") bars_by_code, index_bars, all_dates = load_data_fdr( BACKTEST_UNIVERSE, _warmup_start, end_date, ) if not bars_by_code: print("\n [ERROR] 데이터 수집 실패. 종목 데이터를 가져올 수 없습니다.") sys.exit(1) print(f"\n 수집 완료: {len(bars_by_code)}종목, {len(all_dates)}거래일") print(f" 인덱스(KODEX200): {len(index_bars)}일") # 2. 날짜 정렬 aligned = align_bars_to_dates(bars_by_code, all_dates) # 3. 시뮬레이션 print(f"\n 시뮬레이션 시작...") metrics, all_trades, equity_curve, drawdown_curve, open_positions, mode_changes = run_simulation( aligned, index_bars, all_dates, initial_capital, BACKTEST_UNIVERSE, sim_start_date=_sim_start, ) elapsed = time.time() - t0 # 4. 종목별 집계 stock_summary = per_stock_summary(all_trades, BACKTEST_UNIVERSE) # 적응형 모드 전환 이력 출력 if mode_changes: print(f"\n 적응형 모드 전환 ({len(mode_changes)}회):") for mc in mode_changes[-15:]: # 최근 15건 print(f" {mc['date']}: {mc['from']:>10s} → {mc['to']}") if len(mode_changes) > 15: print(f" ... (이전 {len(mode_changes)-15}건 생략)") # 5. 최종 자본 (equity_curve 마지막 값) final_equity = equity_curve[-1]["equity"] if equity_curve else initial_capital # 6. 결과 출력 print_results( metrics, stock_summary, start_date, end_date, initial_capital, final_equity, open_positions, ) print(f"\n 실행 시간: {elapsed:.1f}초") # 7. JSON 저장 output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backtest_results") os.makedirs(output_dir, exist_ok=True) save_results_json( metrics, all_trades, equity_curve, drawdown_curve, stock_summary, open_positions, start_date, end_date, initial_capital, output_dir, ) if __name__ == "__main__": main()