shuim
1b9257c4d2
## 배경 "전략이 혼란스럽고 현재와 맞지 않는다"를 조사한 결과, 전략 설계가 아니라 전략을 검증했다는 도구와 파라미터 소스가 어긋나 있었다. ## 파라미터 소스 단일화 - 우선순위가 adaptive > strategy_cfg(DB) > config.py 인데 문서는 맨 마지막인 config.py 를 기술하고 있었다. 11곳이 불일치(BULL RSI 46 vs DB 30, SL -7% vs -6%). - DEFAULT_PARAMS 를 config.py 에 정합 + alembic 0012 로 기존 DB 행 갱신 (운영자가 UI 로 바꾼 값은 보존, 옛 기본값과 같은 행만 갱신, 테이블 가드·멱등) ## 백테스트를 실전 로직에 정합 - calculate_adx_simple 이 정확히 period 배(14.0×) 발산 → Aggressive 게이트 `ADX>25` 가 전 레짐 100% 통과. 전환 빈도 5.6%→2.0%(-65%), 성과 상방 편향 - 백테스트 RSI 는 SMA, 실전은 Wilder → 진입 판정 10~11% 불일치 - us_backtest_engine RSI 1봉 지연 제거 (룩어헤드 없음을 절단 테스트로 증명) - backend/backtest_results/ 25건은 폐기 대상 (§8.4) ## 리스크 가드 복원 - 4% 절대 상한이 try 블록 안에 있어 예외 시 적응형 6~7% 가 통과 → try 밖으로 - 청산을 매수보다 먼저 평가 (보유 종목이 추천에 들면 손절 대신 물타기가 나갔다) - 리스크 가드 4개를 매수 전용화 → MDD -10%·보유한도 도달 시에도 손절 작동 - MDD 갱신 5회 연속 실패 시 신규 진입 차단 fail-safe (청산은 계속 허용) - 스윙 청산이 DT 보유분까지 매도하던 문제, sell_pct 기준 혼재 정정 ## 죽은 코드 제거 (2,695줄) - short_term_strategy, kosdaq_scanner, app/strategy/ 패키지 - app/trading/ + korea_investment_api_adapter (real_market_api 정본은 korea_investment_api.py:1482, adapter 는 import 0건) ## 지표·사이징 단일화 (수치 무변경 증명) - indicators_core.py: 3,274건 비교 최대 오차 0.000e+00 - sizing_core.py: 42,105건 비교 불일치 0 - 미국 전략 상수 13개 config 편입 (hex 까지 동일, 3,168 케이스 전수 비교) ## 테스트 안전 (실 KIS 호출 차단) - 유닛 테스트가 실제 KIS 에 매도 주문을 2회 전송하고 있었다. 리팩터링이 close_stale_positions 의 대상을 _execute_exit → _execute_exit_stale 로 바꿨는데 목이 전자에 걸려 빗나갔다. test_api_endpoints 도 캡처된 싱글톤 참조 누수로 실제 잔고 조회를 타 141초를 소비했다. - backend/tests/conftest.py 신설: 주문·토큰·시세·잔고를 호출 즉시 예외로 차단. 예외는 BaseException 파생(프로덕션 except Exception 폴백에 삼켜지지 않게), 클래스 메서드에 적용(캡처된 참조까지 덮음), DATABASE_URL `_test` 가드 강제 - 전체 유닛 스위트: 365초/6 failed → 11.8초/0 failed (소켓 차단 하 559 passed) ## 신규 (둘 다 플래그 OFF, 백테스트 미검증) - range_trading_strategy.py — SIDEWAYS 담당 전략이 없던 빈 칸. 미배선 - check_bear_buy_signal_v2 — BEAR 4중 AND(발화율 2.3%) → 점수제(12.4%) ## 문서 - CLAUDE.md 를 실제 코드 기준으로 재작성. §0 파라미터 우선순위, §4.1 테스트 규약, §6.4 문서화되지 않았던 진입 차단 게이트 9종, §8.4 백테스트 폐기 근거, §8.5 미해결 19항목 - 루트 md 6개 → 2개(docs/ 이동), 중복 배치 문서 병합, 스테일 worktree 제거 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1666 lines
66 KiB
Python
1666 lines
66 KiB
Python
#!/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
|
||
|
||
# [2026-09-27 §8.5-1 중복 통합] 지표 계산은 실전 경로와 동일한 단일 출처를 사용한다.
|
||
# indicators_core 는 순수 계산 모듈(typing 외 import 없음)이라 DB/설정/네트워크 의존이
|
||
# 없고 순환 import 도 발생하지 않는다. 이 스크립트는 backend/ 를 루트로 실행되므로
|
||
# sys.path 에 스크립트 디렉터리를 명시적으로 넣어 `python backend/run_backtest.py`,
|
||
# `python run_backtest.py` 양쪽에서 동일하게 import 되도록 한다.
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
from app.services.indicators_core import ( # noqa: E402
|
||
adx_wilder as _core_adx_wilder,
|
||
atr_sma as _core_atr_sma,
|
||
bollinger_sma as _core_bollinger_sma,
|
||
rsi_wilder as _core_rsi_wilder,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 백테스트 대상 종목
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# 백테스트 대상 유니버스: {종목코드: (종목명, 시장구분)}
|
||
# 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 (백테스트 적응형 모드용) — 실전과 동일한 Wilder ADX + 20.0 폴백
|
||
|
||
[2026-09-27 §8.5-1 수정] 이전 구현은 마지막 단계에서 DX(0~100)에 **합산형**
|
||
평활(초기값 = sum(dxs[:period]))을 적용해 반환값이 정식 ADX 의 정확히 `period`
|
||
배로 발산했다(측정: 배율 14.0×). 그 결과 호출부 `determine_adaptive_mode()` 의
|
||
Aggressive 게이트 `adx > 25`(§8.4)가 사실상 상시 True 가 되어 백테스트가
|
||
실전보다 공격적인 사이징(포지션 7%·SL -9%)으로 돌았다 = 성과 상방 편향.
|
||
|
||
이제 `calculate_adx`(= indicators_core.adx_wilder = 실전 경로
|
||
mean_reversion_strategy.calculate_adx)와 **동일 알고리즘**이다. 두 함수는
|
||
폴백 계약만 다르다(아래 참조) — 호출부 호환을 위해 이름·시그니처는 유지한다.
|
||
|
||
폴백 계약(기존 유지):
|
||
- `len(closes) < period + 1` → 20.0 (중립값)
|
||
- adx_wilder 가 데이터 부족으로 None (`len < period*2 + 1` 또는 DX 부족) → 20.0
|
||
"""
|
||
if len(closes) < period + 1:
|
||
return 20.0
|
||
adx = calculate_adx(highs, lows, closes, period)
|
||
return adx if adx is not None 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) — Wilder 평활 (실전 경로와 동일)
|
||
|
||
[2026-09-27 §8.5-1 수정] 이전에는 `indicators_core.rsi_sma`(평활 없는 SMA 방식)
|
||
를 썼으나 실전 경로(mean_reversion_strategy.calculate_rsi = rsi_wilder)와
|
||
알고리즘이 달라 진입/청산 게이트 판정이 약 10% 의 봉에서 갈렸다
|
||
(측정 n=78,000: mean|Δ|=4.68, p95=14.51, RSI<46 진입게이트 불일치 9.7%,
|
||
RSI>=70 전량청산 불일치 5.7%). 이제 `rsi_wilder` 로 통일한다.
|
||
→ 이 변경으로 과거 백테스트 수치와 결과가 달라지는 것은 **의도된 것**이다.
|
||
|
||
시그니처·`None` 계약(데이터 부족 시 None)은 그대로 유지한다.
|
||
"""
|
||
return _core_rsi_wilder(closes, period)
|
||
|
||
|
||
def calculate_bollinger(closes: List[float], period: int = 20) -> Optional[Tuple[float, float, float]]:
|
||
"""볼린저 밴드 (2sigma) -> (upper, middle, lower)
|
||
|
||
[2026-09-27 §8.5-1 중복 통합] indicators_core.bollinger_sma(num_std=2.0) 위임.
|
||
모표준편차(ddof=0) 사용 — 실전 경로와 동일.
|
||
"""
|
||
return _core_bollinger_sma(closes, period, 2.0)
|
||
|
||
|
||
def calculate_atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]:
|
||
"""ATR(Average True Range) — TR 의 단순평균
|
||
|
||
[2026-09-27 §8.5-1 중복 통합] indicators_core.atr_sma() 위임. 실전 경로와 동일.
|
||
"""
|
||
return _core_atr_sma(highs, lows, closes, 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
|
||
|
||
[2026-09-27 §8.5-1 중복 통합] indicators_core.adx_wilder() 위임. 실전 경로와 동일.
|
||
(DX 평활은 평균형 — 합산형을 쓰면 ADX 가 100을 넘어 발산하고 SIDEWAYS 게이트가
|
||
무력화된다. 2026-09-27 부터 calculate_adx_simple 도 이 함수에 위임한다.)
|
||
"""
|
||
return _core_adx_wilder(highs, lows, closes, period)
|
||
|
||
|
||
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()
|