1008 lines
35 KiB
Python
1008 lines
35 KiB
Python
"""
|
|
한국 주식 백테스트 엔진
|
|
|
|
stock_daily_prices DB 테이블에서 일봉 OHLCV를 로드하여
|
|
날짜별 순차 시뮬레이션으로 전략 성과를 평가합니다.
|
|
|
|
[설계 원칙]
|
|
- 동기 함수 (배치 작업이므로 async 불필요, DB 조회만 async)
|
|
- numpy 최소 사용 (표준 라이브러리 우선)
|
|
- 기존 mean_reversion_strategy.py의 RSI/BB 함수 재활용
|
|
- CLAUDE.md 리스크 규칙 준수 (수수료, 세금, 슬리피지, 손절/익절)
|
|
|
|
[수수료 모델]
|
|
- 매수 수수료: 0.015%
|
|
- 매도 수수료: 0.015% + 증권거래세 (KOSPI 0.18%, KOSDAQ 0.23%)
|
|
|
|
[슬리피지 모델]
|
|
- 시장가 기준 +/-0.1% 랜덤 슬리피지
|
|
"""
|
|
import logging
|
|
import math
|
|
import random
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, datetime
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from app.core.config import settings
|
|
from app.services.mean_reversion_strategy import (
|
|
MarketRegime,
|
|
calculate_atr,
|
|
calculate_bollinger,
|
|
calculate_rsi,
|
|
check_buy_signal,
|
|
check_exit_signal,
|
|
detect_market_regime,
|
|
is_reversal_candle,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 데이터 구조
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class BarData:
|
|
"""일봉 OHLCV 데이터 (1행)"""
|
|
trade_date: date
|
|
open_price: int
|
|
high_price: int
|
|
low_price: int
|
|
close_price: int
|
|
volume: int
|
|
stock_code: str = ""
|
|
stock_name: str = ""
|
|
|
|
|
|
@dataclass
|
|
class BacktestTrade:
|
|
"""체결된 거래 1건"""
|
|
stock_code: 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 = ""
|
|
|
|
|
|
@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
|
|
|
|
|
|
@dataclass
|
|
class BacktestResult:
|
|
"""백테스트 전체 결과"""
|
|
metrics: BacktestMetrics
|
|
equity_curve: List[Dict[str, Any]] = field(default_factory=list)
|
|
trades: List[Dict[str, Any]] = field(default_factory=list)
|
|
drawdown_curve: List[Dict[str, Any]] = field(default_factory=list)
|
|
config: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class Position:
|
|
"""현재 보유 포지션"""
|
|
stock_code: str
|
|
entry_date: date
|
|
entry_price: float
|
|
quantity: int
|
|
highest_since_entry: float # 트레일링 스탑용 최고가
|
|
remaining_pct: float = 1.0 # 분할매도 잔여 비율
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 수수료/세금/슬리피지 모델
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# 수수료율 (config.py 참조)
|
|
BUY_COMMISSION_RATE = settings.KR_COMMISSION_RATE # 0.00015 (0.015%)
|
|
SELL_COMMISSION_RATE = settings.KR_COMMISSION_RATE # 0.00015 (0.015%)
|
|
SELL_TAX_KOSPI = settings.KR_SELL_TAX_RATE_KOSPI # 0.0018 (0.18%)
|
|
SELL_TAX_KOSDAQ = settings.KR_SELL_TAX_RATE_KOSDAQ # 0.0023 (0.23%)
|
|
SLIPPAGE_PCT = 0.001 # +/-0.1%
|
|
|
|
|
|
def _apply_buy_slippage(price: float, seed: Optional[int] = None) -> float:
|
|
"""매수 시 슬리피지: 약간 높은 가격에 체결"""
|
|
rng = random.Random(seed)
|
|
slip = rng.uniform(0, SLIPPAGE_PCT)
|
|
return price * (1 + slip)
|
|
|
|
|
|
def _apply_sell_slippage(price: float, seed: Optional[int] = None) -> float:
|
|
"""매도 시 슬리피지: 약간 낮은 가격에 체결"""
|
|
rng = random.Random(seed)
|
|
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 = "KOSPI",
|
|
) -> 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[BacktestTrade],
|
|
initial_capital: float,
|
|
trading_days: int,
|
|
) -> BacktestMetrics:
|
|
"""equity_curve와 trades로부터 성과 지표 산출"""
|
|
m = BacktestMetrics()
|
|
|
|
if not equity_curve or len(equity_curve) < 2:
|
|
return m
|
|
|
|
final_equity = equity_curve[-1]
|
|
m.total_return_pct = ((final_equity - initial_capital) / initial_capital) * 100
|
|
|
|
# CAGR
|
|
years = trading_days / 252.0
|
|
if years > 0 and final_equity > 0:
|
|
m.cagr_pct = ((final_equity / initial_capital) ** (1 / years) - 1) * 100
|
|
else:
|
|
m.cagr_pct = 0.0
|
|
|
|
# 일별 수익률
|
|
daily_returns: List[float] = []
|
|
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 Ratio (무위험 이자율 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
|
|
else:
|
|
m.sharpe_ratio = 0.0
|
|
|
|
# Sortino Ratio (하방 편차만 사용)
|
|
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
|
|
else:
|
|
m.sortino_ratio = 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)
|
|
|
|
# 거래 통계
|
|
closed_trades = [t for t in trades if t.side == "SELL"]
|
|
m.total_trades = len(closed_trades)
|
|
if m.total_trades == 0:
|
|
return m
|
|
|
|
wins = [t for t in closed_trades if t.pnl > 0]
|
|
losses = [t for t in closed_trades 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)
|
|
|
|
# Profit Factor
|
|
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
|
|
|
|
# Average Win/Loss %
|
|
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
|
|
|
|
# 연속 승/패
|
|
consecutive_w = 0
|
|
consecutive_l = 0
|
|
max_cw = 0
|
|
max_cl = 0
|
|
for t in closed_trades:
|
|
if t.pnl > 0:
|
|
consecutive_w += 1
|
|
consecutive_l = 0
|
|
else:
|
|
consecutive_l += 1
|
|
consecutive_w = 0
|
|
max_cw = max(max_cw, consecutive_w)
|
|
max_cl = max(max_cl, consecutive_l)
|
|
m.max_consecutive_wins = max_cw
|
|
m.max_consecutive_losses = max_cl
|
|
|
|
# 평균 보유일
|
|
holding_days = []
|
|
for t in closed_trades:
|
|
if t.entry_date and t.exit_date:
|
|
delta = (t.exit_date - t.entry_date).days
|
|
holding_days.append(delta)
|
|
m.avg_holding_days = round(sum(holding_days) / len(holding_days), 1) if holding_days else 0.0
|
|
|
|
return m
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 전략 추상 클래스
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BacktestStrategy(ABC):
|
|
"""백테스트 전략 인터페이스
|
|
|
|
on_bar()는 매 일봉마다 호출되며, 매수/매도 주문 리스트를 반환합니다.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""전략 이름"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def description(self) -> str:
|
|
"""전략 설명"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def default_params(self) -> Dict[str, Any]:
|
|
"""기본 파라미터"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def on_bar(
|
|
self,
|
|
bar_idx: int,
|
|
bars: Dict[str, List[BarData]],
|
|
positions: Dict[str, Position],
|
|
cash: float,
|
|
total_equity: float,
|
|
params: Dict[str, Any],
|
|
) -> List[Dict[str, Any]]:
|
|
"""일봉 콜백
|
|
|
|
Args:
|
|
bar_idx: 현재 봉 인덱스 (0-based, 모든 종목 공통 날짜 기준)
|
|
bars: {stock_code: [BarData, ...]} 전체 일봉 (bar_idx까지 접근 가능)
|
|
positions: 현재 보유 포지션 {stock_code: Position}
|
|
cash: 현재 현금
|
|
total_equity: 현재 총 자산 (현금 + 평가금)
|
|
params: 전략 파라미터
|
|
|
|
Returns:
|
|
주문 리스트 [{"action": "BUY"|"SELL", "stock_code": str,
|
|
"price": float, "quantity": int, "reason": str,
|
|
"sell_pct": float (SELL 시 매도 비율)}]
|
|
"""
|
|
...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 내장 전략: MeanReversionBacktest (RSI + BB 평균회귀)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class MeanReversionBacktest(BacktestStrategy):
|
|
"""CLAUDE.md 섹션 6 RSI+BB 평균회귀 전략 백테스트 버전
|
|
|
|
기존 mean_reversion_strategy.py의 check_buy_signal / check_exit_signal을
|
|
그대로 재활용합니다. 시장 국면(regime)은 코스피200 데이터가 없으므로
|
|
전체 종목 평균으로 대체하거나, 파라미터로 고정 지정 가능합니다.
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "mean_reversion"
|
|
|
|
def description(self) -> str:
|
|
return "RSI + Bollinger Band 평균회귀 전략 (CLAUDE.md 섹션 6)"
|
|
|
|
def default_params(self) -> Dict[str, Any]:
|
|
return {
|
|
"rsi_period": settings.RSI_PERIOD,
|
|
"bb_period": settings.BOLLINGER_PERIOD,
|
|
"atr_period": settings.ATR_PERIOD,
|
|
"max_positions": settings.MAX_HOLDINGS_COUNT,
|
|
"position_size_pct": settings.MAX_POSITION_SIZE,
|
|
"stop_loss_pct": settings.STOP_LOSS_PERCENT,
|
|
"atr_stop_multiplier": settings.ATR_STOP_MULTIPLIER,
|
|
"atr_trailing_multiplier": settings.ATR_TRAILING_MULTIPLIER,
|
|
"rsi_exit_threshold": settings.MEAN_REVERSION_RSI_EXIT_THRESHOLD,
|
|
"rsi_full_exit": settings.MEAN_REVERSION_RSI_FULL_EXIT,
|
|
"regime": "BULL", # "BULL", "BEAR", "AUTO"
|
|
"market": "KOSPI", # 세금 계산용
|
|
"min_price": settings.MIN_PRICE_KRW,
|
|
"min_data_bars": 60, # RSI/BB 계산에 필요한 최소 봉 수
|
|
}
|
|
|
|
def on_bar(
|
|
self,
|
|
bar_idx: int,
|
|
bars: Dict[str, List[BarData]],
|
|
positions: Dict[str, Position],
|
|
cash: float,
|
|
total_equity: float,
|
|
params: Dict[str, Any],
|
|
) -> List[Dict[str, Any]]:
|
|
orders: List[Dict[str, Any]] = []
|
|
min_bars = params.get("min_data_bars", 60)
|
|
|
|
# --- 매도 시그널 먼저 (기존 포지션) ---
|
|
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 = calculate_rsi(closes, params.get("rsi_period", 14))
|
|
bb = calculate_bollinger(closes, params.get("bb_period", 20))
|
|
atr = calculate_atr(highs, lows, closes, params.get("atr_period", 14))
|
|
|
|
if rsi is None or bb is None or atr is None:
|
|
continue
|
|
|
|
bb_upper, bb_middle, bb_lower = bb
|
|
|
|
# 최고가 업데이트
|
|
pos.highest_since_entry = max(pos.highest_since_entry, current_bar.high_price)
|
|
|
|
exit_signal = check_exit_signal(
|
|
regime=self._get_regime(params),
|
|
rsi=rsi,
|
|
close=float(current_bar.close_price),
|
|
bb_upper=bb_upper,
|
|
bb_middle=bb_middle,
|
|
bb_lower=bb_lower,
|
|
entry_price=pos.entry_price,
|
|
atr=atr,
|
|
lowest_since_entry=float(current_bar.low_price),
|
|
remaining_pct=pos.remaining_pct,
|
|
trailing_high=pos.highest_since_entry,
|
|
)
|
|
|
|
if exit_signal:
|
|
sell_qty = int(pos.quantity * exit_signal.sell_pct)
|
|
if sell_qty <= 0:
|
|
sell_qty = pos.quantity
|
|
|
|
orders.append({
|
|
"action": "SELL",
|
|
"stock_code": code,
|
|
"price": float(current_bar.close_price),
|
|
"quantity": sell_qty,
|
|
"reason": exit_signal.reason,
|
|
"sell_pct": exit_signal.sell_pct,
|
|
})
|
|
|
|
# --- 매수 시그널 ---
|
|
max_pos = params.get("max_positions", 20)
|
|
if len(positions) >= max_pos:
|
|
return orders
|
|
|
|
for code, stock_bars in bars.items():
|
|
if code in positions:
|
|
continue
|
|
if len(positions) + sum(1 for o in orders if o["action"] == "BUY") >= max_pos:
|
|
break
|
|
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
|
|
|
|
# 가격 필터
|
|
if current_bar.close_price < params.get("min_price", 5000):
|
|
continue
|
|
|
|
rsi = calculate_rsi(closes, params.get("rsi_period", 14))
|
|
bb = calculate_bollinger(closes, params.get("bb_period", 20))
|
|
|
|
if rsi is None or bb is None:
|
|
continue
|
|
|
|
bb_upper, bb_middle, bb_lower = bb
|
|
|
|
# 거래량 20일 평균
|
|
vol_window = volumes[-20:] if len(volumes) >= 20 else volumes
|
|
avg_vol = sum(vol_window) / len(vol_window) if vol_window else 0
|
|
|
|
# 이전 RSI (반등 확인)
|
|
rsi_prev = None
|
|
if len(closes) > min_bars + 3:
|
|
rsi_prev = calculate_rsi(closes[:-3], params.get("rsi_period", 14))
|
|
|
|
# 되돌림 캔들 확인 (BEAR 진입 조건)
|
|
reversal = is_reversal_candle(
|
|
[float(x) for x in opens[-3:]],
|
|
[float(x) for x in highs[-3:]],
|
|
[float(x) for x in lows[-3:]],
|
|
[float(x) for x in closes[-3:]],
|
|
)
|
|
|
|
prev_vol = volumes[-2] if len(volumes) >= 2 else 0
|
|
|
|
signal = check_buy_signal(
|
|
regime=self._get_regime(params),
|
|
rsi=rsi,
|
|
close=float(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,
|
|
prev_volume=prev_vol,
|
|
is_reversal=reversal,
|
|
rsi_prev=rsi_prev,
|
|
)
|
|
|
|
if signal:
|
|
# 포지션 사이징: total_equity * position_size_pct
|
|
budget = total_equity * params.get("position_size_pct", 0.04)
|
|
budget = min(budget, cash)
|
|
if budget <= 0:
|
|
continue
|
|
|
|
price = float(current_bar.close_price)
|
|
quantity = int(budget / price)
|
|
if quantity <= 0:
|
|
continue
|
|
|
|
orders.append({
|
|
"action": "BUY",
|
|
"stock_code": code,
|
|
"price": price,
|
|
"quantity": quantity,
|
|
"reason": signal.get("description", ""),
|
|
"sell_pct": 0.0,
|
|
})
|
|
|
|
return orders
|
|
|
|
@staticmethod
|
|
def _get_regime(params: Dict[str, Any]) -> MarketRegime:
|
|
regime_str = params.get("regime", "BULL").upper()
|
|
if regime_str == "BEAR":
|
|
return MarketRegime.BEAR
|
|
if regime_str == "SIDEWAYS":
|
|
return MarketRegime.SIDEWAYS
|
|
return MarketRegime.BULL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 전략 레지스트리
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_STRATEGY_REGISTRY: Dict[str, type] = {
|
|
"mean_reversion": MeanReversionBacktest,
|
|
}
|
|
|
|
|
|
def get_available_strategies() -> List[Dict[str, Any]]:
|
|
"""사용 가능한 전략 목록 반환"""
|
|
result = []
|
|
for key, cls in _STRATEGY_REGISTRY.items():
|
|
instance = cls()
|
|
result.append({
|
|
"key": key,
|
|
"name": instance.name(),
|
|
"description": instance.description(),
|
|
"default_params": instance.default_params(),
|
|
})
|
|
return result
|
|
|
|
|
|
def get_strategy(name: str) -> BacktestStrategy:
|
|
"""이름으로 전략 인스턴스 반환"""
|
|
cls = _STRATEGY_REGISTRY.get(name)
|
|
if cls is None:
|
|
raise ValueError(f"Unknown strategy: {name}. Available: {list(_STRATEGY_REGISTRY.keys())}")
|
|
return cls()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 데이터 로더 (async — DB 조회)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def load_daily_bars(
|
|
stock_codes: List[str],
|
|
start_date: date,
|
|
end_date: date,
|
|
) -> Tuple[Dict[str, List[BarData]], List[date]]:
|
|
"""stock_daily_prices 테이블에서 일봉 데이터 로드
|
|
|
|
Returns:
|
|
(bars_by_code, all_dates)
|
|
- bars_by_code: {stock_code: [BarData...]} 날짜 오름차순
|
|
- all_dates: 전체 거래일 리스트 (유니온)
|
|
"""
|
|
from sqlalchemy import select
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.models.models import StockDailyPrice
|
|
|
|
bars_by_code: Dict[str, List[BarData]] = {}
|
|
all_dates_set: set = set()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
for code in stock_codes:
|
|
stmt = (
|
|
select(StockDailyPrice)
|
|
.where(
|
|
StockDailyPrice.stock_code == code,
|
|
StockDailyPrice.trade_date >= start_date,
|
|
StockDailyPrice.trade_date <= end_date,
|
|
)
|
|
.order_by(StockDailyPrice.trade_date.asc())
|
|
)
|
|
result = await session.execute(stmt)
|
|
rows = result.scalars().all()
|
|
|
|
if not rows:
|
|
logger.debug("종목 %s: 기간 내 데이터 없음 (%s ~ %s)", code, start_date, end_date)
|
|
continue
|
|
|
|
bar_list: List[BarData] = []
|
|
for row in rows:
|
|
bar_list.append(BarData(
|
|
trade_date=row.trade_date,
|
|
open_price=row.open_price or 0,
|
|
high_price=row.high_price or 0,
|
|
low_price=row.low_price or 0,
|
|
close_price=row.close_price or 0,
|
|
volume=row.volume or 0,
|
|
stock_code=row.stock_code,
|
|
stock_name=row.stock_name or "",
|
|
))
|
|
all_dates_set.add(row.trade_date)
|
|
|
|
bars_by_code[code] = bar_list
|
|
|
|
all_dates = sorted(all_dates_set)
|
|
return bars_by_code, all_dates
|
|
|
|
|
|
def _align_bars_to_dates(
|
|
bars_by_code: Dict[str, List[BarData]],
|
|
all_dates: List[date],
|
|
) -> Dict[str, List[BarData]]:
|
|
"""모든 종목을 동일 날짜 인덱스에 맞춤 (결측일은 이전 종가 복제)
|
|
|
|
bar_idx로 접근할 때 모든 종목이 같은 날짜를 가리키도록 합니다.
|
|
"""
|
|
aligned: Dict[str, List[BarData]] = {}
|
|
date_to_idx = {d: i for i, d in enumerate(all_dates)}
|
|
|
|
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:
|
|
# 이전 종가를 OHLC 모두 같게 복제 (거래량 0)
|
|
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)
|
|
# 첫 날짜 이전 데이터가 없으면 건너뜀 (나중에 min_bars로 필터)
|
|
|
|
# 빈 봉이 앞에 있을 수 있으므로 첫 실제 데이터부터 시작
|
|
if result:
|
|
aligned[code] = result
|
|
|
|
return aligned
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 백테스트 엔진
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class KrBacktestEngine:
|
|
"""한국 주식 백테스트 엔진 (메인 클래스)
|
|
|
|
사용법:
|
|
engine = KrBacktestEngine()
|
|
result = await engine.run(
|
|
stock_codes=["005930", "035720"],
|
|
start_date=date(2025, 1, 1),
|
|
end_date=date(2025, 12, 31),
|
|
strategy_name="mean_reversion",
|
|
params={...},
|
|
initial_capital=10_000_000,
|
|
market="KOSPI",
|
|
random_seed=42,
|
|
)
|
|
"""
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
async def run(
|
|
self,
|
|
stock_codes: List[str],
|
|
start_date: date,
|
|
end_date: date,
|
|
strategy_name: str = "mean_reversion",
|
|
params: Optional[Dict[str, Any]] = None,
|
|
initial_capital: float = 10_000_000,
|
|
market: str = "KOSPI",
|
|
random_seed: Optional[int] = 42,
|
|
) -> BacktestResult:
|
|
"""백테스트 실행
|
|
|
|
Args:
|
|
stock_codes: 대상 종목 코드 리스트
|
|
start_date: 시작일
|
|
end_date: 종료일
|
|
strategy_name: 전략 키 ("mean_reversion" 등)
|
|
params: 전략 파라미터 (None이면 기본값 사용)
|
|
initial_capital: 초기 자본금
|
|
market: 시장 구분 (KOSPI/KOSDAQ) - 세금 계산용
|
|
random_seed: 슬리피지 재현성을 위한 랜덤 시드
|
|
|
|
Returns:
|
|
BacktestResult
|
|
"""
|
|
strategy = get_strategy(strategy_name)
|
|
effective_params = strategy.default_params()
|
|
if params:
|
|
effective_params.update(params)
|
|
effective_params["market"] = market
|
|
|
|
# 1. 데이터 로드
|
|
logger.info(
|
|
"백테스트 데이터 로드: %d 종목, %s ~ %s",
|
|
len(stock_codes), start_date, end_date,
|
|
)
|
|
bars_by_code, all_dates = await load_daily_bars(stock_codes, start_date, end_date)
|
|
|
|
if not bars_by_code or not all_dates:
|
|
logger.warning("백테스트 데이터 없음")
|
|
return BacktestResult(
|
|
metrics=BacktestMetrics(),
|
|
config=self._build_config_dict(
|
|
stock_codes, start_date, end_date, strategy_name, effective_params, initial_capital,
|
|
),
|
|
)
|
|
|
|
# 2. 날짜 정렬
|
|
aligned = _align_bars_to_dates(bars_by_code, all_dates)
|
|
|
|
# 3. 시뮬레이션 실행
|
|
result = self._simulate(
|
|
aligned, all_dates, strategy, effective_params,
|
|
initial_capital, market, random_seed,
|
|
)
|
|
|
|
result.config = self._build_config_dict(
|
|
stock_codes, start_date, end_date, strategy_name, effective_params, initial_capital,
|
|
)
|
|
|
|
return result
|
|
|
|
def _simulate(
|
|
self,
|
|
bars: Dict[str, List[BarData]],
|
|
all_dates: List[date],
|
|
strategy: BacktestStrategy,
|
|
params: Dict[str, Any],
|
|
initial_capital: float,
|
|
market: str,
|
|
random_seed: Optional[int],
|
|
) -> BacktestResult:
|
|
"""핵심 시뮬레이션 루프 (동기)
|
|
|
|
[Look-ahead bias 수정]
|
|
T일 close로 신호(on_bar) 생성 -> T+1일 open + 슬리피지로 체결.
|
|
마지막 봉 신호는 다음 봉이 없으므로 무시합니다.
|
|
"""
|
|
cash = initial_capital
|
|
positions: Dict[str, Position] = {}
|
|
all_trades: List[BacktestTrade] = []
|
|
|
|
# 포지션별 진입 거래 추적 (매도 시 PnL 계산용)
|
|
entry_trades: Dict[str, BacktestTrade] = {}
|
|
|
|
equity_values: List[float] = [initial_capital]
|
|
equity_curve: List[Dict[str, Any]] = [{
|
|
"date": str(all_dates[0]) if all_dates else "",
|
|
"equity": initial_capital,
|
|
}]
|
|
drawdown_curve: List[Dict[str, Any]] = []
|
|
|
|
peak_equity = initial_capital
|
|
seed_counter = random_seed if random_seed is not None else None
|
|
|
|
# 전일 신호를 저장 (T일 close 신호 -> T+1일 open 체결)
|
|
pending_orders: List[Dict[str, Any]] = []
|
|
|
|
for bar_idx in range(len(all_dates)):
|
|
current_date = all_dates[bar_idx]
|
|
|
|
# 현재 평가 자산
|
|
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
|
|
|
|
# -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 --
|
|
for order in pending_orders:
|
|
action = order["action"]
|
|
code = order["stock_code"]
|
|
quantity = order["quantity"]
|
|
|
|
# T+1일 open 가격을 base price로 사용
|
|
stock_bars = bars.get(code, [])
|
|
if bar_idx >= len(stock_bars):
|
|
continue
|
|
next_open = float(stock_bars[bar_idx].open_price)
|
|
|
|
# 슬리피지용 시드
|
|
if seed_counter is not None:
|
|
seed_counter += 1
|
|
slip_seed = seed_counter
|
|
else:
|
|
slip_seed = None
|
|
|
|
if action == "BUY":
|
|
exec_price = _apply_buy_slippage(next_open, slip_seed)
|
|
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 - next_open) * quantity
|
|
|
|
trade = BacktestTrade(
|
|
stock_code=code,
|
|
side="BUY",
|
|
entry_date=current_date,
|
|
entry_price=exec_price,
|
|
quantity=quantity,
|
|
commission=commission,
|
|
slippage=slippage_amount,
|
|
reason=order.get("reason", ""),
|
|
)
|
|
all_trades.append(trade)
|
|
entry_trades[code] = trade
|
|
|
|
positions[code] = Position(
|
|
stock_code=code,
|
|
entry_date=current_date,
|
|
entry_price=exec_price,
|
|
quantity=quantity,
|
|
highest_since_entry=exec_price,
|
|
)
|
|
|
|
elif action == "SELL":
|
|
pos = positions.get(code)
|
|
if pos is None:
|
|
continue
|
|
|
|
sell_qty = min(quantity, pos.quantity)
|
|
|
|
exec_price = _apply_sell_slippage(next_open, slip_seed)
|
|
proceeds, commission, tax = _calc_sell_proceeds(exec_price, sell_qty, market)
|
|
|
|
cash += proceeds
|
|
slippage_amount = abs(next_open - exec_price) * sell_qty
|
|
|
|
# PnL 계산
|
|
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 = BacktestTrade(
|
|
stock_code=code,
|
|
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=order.get("reason", ""),
|
|
)
|
|
all_trades.append(trade)
|
|
|
|
# 포지션 업데이트
|
|
remaining_qty = pos.quantity - sell_qty
|
|
if remaining_qty <= 0:
|
|
del positions[code]
|
|
entry_trades.pop(code, None)
|
|
else:
|
|
pos.quantity = remaining_qty
|
|
pos.remaining_pct = remaining_qty / (pos.quantity + sell_qty)
|
|
|
|
pending_orders.clear()
|
|
|
|
# -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) --
|
|
is_last_bar = (bar_idx == len(all_dates) - 1)
|
|
if not is_last_bar:
|
|
# 체결 이후의 상태로 평가 자산 재계산 (전략에 전달)
|
|
position_value = 0.0
|
|
for code, pos in positions.items():
|
|
stock_bars_eval = bars.get(code, [])
|
|
if bar_idx < len(stock_bars_eval):
|
|
position_value += stock_bars_eval[bar_idx].close_price * pos.quantity
|
|
total_equity = cash + position_value
|
|
|
|
orders = strategy.on_bar(
|
|
bar_idx=bar_idx,
|
|
bars=bars,
|
|
positions=positions,
|
|
cash=cash,
|
|
total_equity=total_equity,
|
|
params=params,
|
|
)
|
|
pending_orders = list(orders)
|
|
|
|
# 일말 평가
|
|
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),
|
|
})
|
|
|
|
# Drawdown
|
|
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),
|
|
})
|
|
|
|
# 성과 지표 계산
|
|
metrics = _compute_metrics(
|
|
equity_curve=equity_values,
|
|
trades=all_trades,
|
|
initial_capital=initial_capital,
|
|
trading_days=len(all_dates),
|
|
)
|
|
|
|
# 거래 내역 직렬화
|
|
trades_dicts = []
|
|
for t in all_trades:
|
|
trades_dicts.append({
|
|
"stock_code": t.stock_code,
|
|
"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,
|
|
})
|
|
|
|
return BacktestResult(
|
|
metrics=metrics,
|
|
equity_curve=equity_curve,
|
|
trades=trades_dicts,
|
|
drawdown_curve=drawdown_curve,
|
|
)
|
|
|
|
@staticmethod
|
|
def _build_config_dict(
|
|
stock_codes: List[str],
|
|
start_date: date,
|
|
end_date: date,
|
|
strategy_name: str,
|
|
params: Dict[str, Any],
|
|
initial_capital: float,
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"stock_codes": stock_codes,
|
|
"start_date": str(start_date),
|
|
"end_date": str(end_date),
|
|
"strategy": strategy_name,
|
|
"params": params,
|
|
"initial_capital": initial_capital,
|
|
"commission": {
|
|
"buy_rate": BUY_COMMISSION_RATE,
|
|
"sell_rate": SELL_COMMISSION_RATE,
|
|
"tax_kospi": SELL_TAX_KOSPI,
|
|
"tax_kosdaq": SELL_TAX_KOSDAQ,
|
|
},
|
|
"slippage_pct": SLIPPAGE_PCT,
|
|
}
|