shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e0913d9ccbd62c17d49fe79fbb29424d8f39b661
auto-stock/backend/run_backtest.py
shuim 8a8e3700db feat: 안전형 스윙 개선(서킷브레이커·own_trend) + 매수 현금해결 방어 
퀀트 리서치(공정 유니버스+walk-forward)로 검증된 안전 게이트 2종과
매수 경로 현금 해결 방어 코드 추가. 모두 기존 시그니처 유지(§4).

- 지수 서킷브레이커: KODEX200 종가 < MA20이면 신규 스윙 진입 전면 중단
  (하락장 상관붕괴=2024H2형 방어). execute_swing_entries가 기존
  check_market_regime의 drop_from_ma20을 재사용(추가 API 없음). 백테스트
  루프에도 동일 적용. 검증: 최근1년 Sharpe 0.82→1.13, CAGR 7.2→8.6%, MDD 개선.
- own_trend 게이트: 평균회귀·눌림목 매수 시 종목 MA20 >= MA60×0.97일 때만
  진입(하락추세 물타기 방지). 실서비스는 recommendations.is_mean_reversion_candidate
  에서 StockAnalysis.ma20/ma60로, check_buy_signal엔 ma20/ma60 선택 파라미터
  (기본 None=하위호환) 추가. 백테스트(kr_backtest_engine, run_backtest)도 평균회귀 한정.
- 매수 현금 해결 방어(trade_executor._execute_buy): 0-현금 캐시를 신뢰하지 않고
  현금<=0이면 자산과 무관하게 KIS 예수금 재조회, 실현금(>0)일 때만 캐시.
  0-현금 고착으로 수량 0이 되는 잠재 무한루프 방어.

검증: code-verifier 통과(CONFIRMED 버그 없음, None-safe, §3/§4/§5 준수),
pytest 442 통과, 6파일 컨테이너 컴파일·리로드 정상.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HBmzf1HDnvFHLraJ8eaaKJ
2026-09-16 13:17:17 +09:00

1721 lines
67 KiB
Python
Raw Blame History

#!/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 종가<MA20 발동일수: {circuit_breaker_days}/{_sim_trading_days}일 "
f"({circuit_breaker_days / _sim_trading_days * 100:.1f}%)")
return metrics, all_trades, equity_curve, drawdown_curve, open_positions, mode_changes
# ---------------------------------------------------------------------------
# 종목별 성과 집계
# ---------------------------------------------------------------------------
def per_stock_summary(trades: List[Trade], universe: Dict[str, Tuple[str, str]]) -> 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()
Reference in New Issue View Git Blame Copy Permalink