gahusb
760d1906ed
- analysis/backtest.py: 백테스팅 프레임워크 신규 추가 - analysis/ensemble.py: 적응형 앙상블 가중치 신규 추가 - warmup_and_restart.py: 봇 워밍업 및 재시작 스크립트 신규 추가 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
256 lines
8.8 KiB
Python
256 lines
8.8 KiB
Python
"""
|
|
백테스팅 프레임워크 (Phase 3-1)
|
|
- 과거 OHLCV 데이터로 전략 시뮬레이션
|
|
- 성과지표: Sharpe ratio, MDD, 승률, 평균손익비
|
|
- Phase 2 모델 변경 전후 비교 검증용
|
|
"""
|
|
|
|
import json
|
|
import numpy as np
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional, Callable
|
|
|
|
|
|
@dataclass
|
|
class Trade:
|
|
ticker: str
|
|
entry_date: int # 데이터 인덱스
|
|
entry_price: float
|
|
exit_date: int
|
|
exit_price: float
|
|
qty: int
|
|
direction: str = "LONG" # LONG / SHORT
|
|
|
|
@property
|
|
def pnl(self):
|
|
if self.direction == "LONG":
|
|
return (self.exit_price - self.entry_price) * self.qty
|
|
return (self.entry_price - self.exit_price) * self.qty
|
|
|
|
@property
|
|
def pnl_pct(self):
|
|
return (self.exit_price - self.entry_price) / self.entry_price * 100
|
|
|
|
|
|
@dataclass
|
|
class BacktestResult:
|
|
total_return_pct: float
|
|
sharpe_ratio: float
|
|
max_drawdown_pct: float
|
|
win_rate: float
|
|
avg_win_pct: float
|
|
avg_loss_pct: float
|
|
profit_factor: float
|
|
total_trades: int
|
|
winning_trades: int
|
|
losing_trades: int
|
|
trades: List[Trade] = field(default_factory=list)
|
|
|
|
def summary(self) -> str:
|
|
lines = [
|
|
"=" * 50,
|
|
"📊 백테스팅 결과",
|
|
"=" * 50,
|
|
f"총 수익률: {self.total_return_pct:+.2f}%",
|
|
f"Sharpe Ratio: {self.sharpe_ratio:.3f}",
|
|
f"Max Drawdown: {self.max_drawdown_pct:.2f}%",
|
|
f"승률: {self.win_rate:.1f}% ({self.winning_trades}/{self.total_trades})",
|
|
f"평균 수익: {self.avg_win_pct:+.2f}%",
|
|
f"평균 손실: {self.avg_loss_pct:.2f}%",
|
|
f"손익비(PF): {self.profit_factor:.2f}",
|
|
"=" * 50,
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
class Backtester:
|
|
"""
|
|
OHLCV 기반 전략 백테스터
|
|
|
|
사용 예시:
|
|
bt = Backtester(initial_capital=10_000_000)
|
|
result = bt.run(
|
|
ohlcv_data={"close": [...], "high": [...], "low": [...], "volume": [...]},
|
|
strategy_fn=my_strategy,
|
|
ticker="005930"
|
|
)
|
|
print(result.summary())
|
|
"""
|
|
|
|
def __init__(self, initial_capital: float = 10_000_000,
|
|
commission_rate: float = 0.00015, # 0.015% (증권사 기본)
|
|
slippage_rate: float = 0.001): # 0.1% 슬리피지
|
|
self.initial_capital = initial_capital
|
|
self.commission_rate = commission_rate
|
|
self.slippage_rate = slippage_rate
|
|
|
|
def run(self, ohlcv_data: Dict, strategy_fn: Callable,
|
|
ticker: str = "UNKNOWN", warmup: int = 60) -> BacktestResult:
|
|
"""
|
|
단일 종목 백테스팅
|
|
|
|
Args:
|
|
ohlcv_data: {'close':[], 'high':[], 'low':[], 'open':[], 'volume':[]}
|
|
strategy_fn: (ohlcv_slice: dict) -> str ("BUY" | "SELL" | "HOLD")
|
|
ticker: 종목 코드
|
|
warmup: 초기 웜업 기간 (기술지표 안정화)
|
|
|
|
Returns:
|
|
BacktestResult
|
|
"""
|
|
closes = np.array(ohlcv_data.get('close', []), dtype=float)
|
|
highs = np.array(ohlcv_data.get('high', closes), dtype=float)
|
|
lows = np.array(ohlcv_data.get('low', closes), dtype=float)
|
|
volumes = np.array(ohlcv_data.get('volume', np.zeros_like(closes)), dtype=float)
|
|
|
|
n = len(closes)
|
|
if n < warmup + 10:
|
|
return self._empty_result()
|
|
|
|
capital = self.initial_capital
|
|
position = 0 # 보유 수량
|
|
entry_price = 0.0
|
|
entry_idx = 0
|
|
equity_curve = [capital]
|
|
trades: List[Trade] = []
|
|
|
|
for i in range(warmup, n):
|
|
# 전략 함수에 현재까지의 슬라이스 전달
|
|
slice_data = {
|
|
'close': closes[:i+1].tolist(),
|
|
'high': highs[:i+1].tolist(),
|
|
'low': lows[:i+1].tolist(),
|
|
'volume': volumes[:i+1].tolist(),
|
|
}
|
|
signal = "HOLD"
|
|
try:
|
|
signal = strategy_fn(slice_data)
|
|
except Exception:
|
|
pass
|
|
|
|
price = closes[i]
|
|
buy_price = price * (1 + self.slippage_rate) # 슬리피지 포함 매수가
|
|
sell_price = price * (1 - self.slippage_rate) # 슬리피지 포함 매도가
|
|
|
|
if signal == "BUY" and position == 0:
|
|
# 전액 투자 (수수료 포함)
|
|
qty = int(capital / (buy_price * (1 + self.commission_rate)))
|
|
if qty > 0:
|
|
cost = qty * buy_price * (1 + self.commission_rate)
|
|
capital -= cost
|
|
position = qty
|
|
entry_price = buy_price
|
|
entry_idx = i
|
|
|
|
elif signal == "SELL" and position > 0:
|
|
proceeds = position * sell_price * (1 - self.commission_rate)
|
|
capital += proceeds
|
|
trades.append(Trade(
|
|
ticker=ticker,
|
|
entry_date=entry_idx,
|
|
entry_price=entry_price,
|
|
exit_date=i,
|
|
exit_price=sell_price,
|
|
qty=position
|
|
))
|
|
position = 0
|
|
entry_price = 0.0
|
|
|
|
# 자산 추적
|
|
current_equity = capital + (position * closes[i] if position > 0 else 0)
|
|
equity_curve.append(current_equity)
|
|
|
|
# 미청산 포지션 강제 종료
|
|
if position > 0:
|
|
last_price = closes[-1] * (1 - self.slippage_rate)
|
|
proceeds = position * last_price * (1 - self.commission_rate)
|
|
capital += proceeds
|
|
trades.append(Trade(
|
|
ticker=ticker,
|
|
entry_date=entry_idx,
|
|
entry_price=entry_price,
|
|
exit_date=n - 1,
|
|
exit_price=last_price,
|
|
qty=position
|
|
))
|
|
equity_curve[-1] = capital
|
|
|
|
return self._compute_metrics(equity_curve, trades)
|
|
|
|
def run_multi(self, ohlcv_dict: Dict[str, Dict], strategy_fn: Callable,
|
|
warmup: int = 60) -> Dict[str, BacktestResult]:
|
|
"""여러 종목 백테스팅"""
|
|
results = {}
|
|
for ticker, ohlcv_data in ohlcv_dict.items():
|
|
results[ticker] = self.run(ohlcv_data, strategy_fn, ticker, warmup)
|
|
return results
|
|
|
|
def _compute_metrics(self, equity_curve: List[float], trades: List[Trade]) -> BacktestResult:
|
|
equity = np.array(equity_curve, dtype=float)
|
|
total_return_pct = (equity[-1] / equity[0] - 1) * 100
|
|
|
|
# Sharpe Ratio (일별 수익률 기준, 연율화)
|
|
daily_returns = np.diff(equity) / equity[:-1]
|
|
if daily_returns.std() > 0:
|
|
sharpe = (daily_returns.mean() / daily_returns.std()) * np.sqrt(252)
|
|
else:
|
|
sharpe = 0.0
|
|
|
|
# Max Drawdown
|
|
peak = np.maximum.accumulate(equity)
|
|
drawdowns = (equity - peak) / (peak + 1e-9) * 100
|
|
max_drawdown = abs(drawdowns.min())
|
|
|
|
# 승률 / 손익비
|
|
wins = [t for t in trades if t.pnl_pct > 0]
|
|
losses = [t for t in trades if t.pnl_pct <= 0]
|
|
|
|
win_rate = len(wins) / len(trades) * 100 if trades else 0
|
|
avg_win = np.mean([t.pnl_pct for t in wins]) if wins else 0
|
|
avg_loss = np.mean([t.pnl_pct for t in losses]) if losses else 0
|
|
|
|
total_win = sum(t.pnl for t in wins)
|
|
total_loss = abs(sum(t.pnl for t in losses))
|
|
profit_factor = total_win / (total_loss + 1e-9)
|
|
|
|
return BacktestResult(
|
|
total_return_pct=round(total_return_pct, 2),
|
|
sharpe_ratio=round(sharpe, 3),
|
|
max_drawdown_pct=round(max_drawdown, 2),
|
|
win_rate=round(win_rate, 1),
|
|
avg_win_pct=round(avg_win, 2),
|
|
avg_loss_pct=round(avg_loss, 2),
|
|
profit_factor=round(profit_factor, 3),
|
|
total_trades=len(trades),
|
|
winning_trades=len(wins),
|
|
losing_trades=len(losses),
|
|
trades=trades
|
|
)
|
|
|
|
def _empty_result(self) -> BacktestResult:
|
|
return BacktestResult(
|
|
total_return_pct=0.0, sharpe_ratio=0.0, max_drawdown_pct=0.0,
|
|
win_rate=0.0, avg_win_pct=0.0, avg_loss_pct=0.0,
|
|
profit_factor=0.0, total_trades=0, winning_trades=0, losing_trades=0
|
|
)
|
|
|
|
|
|
def compare_strategies(ohlcv_data: Dict, strategies: Dict[str, Callable],
|
|
initial_capital: float = 10_000_000) -> Dict[str, BacktestResult]:
|
|
"""
|
|
여러 전략 동시 비교
|
|
|
|
Args:
|
|
strategies: {"전략명": strategy_fn, ...}
|
|
|
|
Returns:
|
|
{"전략명": BacktestResult, ...}
|
|
"""
|
|
bt = Backtester(initial_capital=initial_capital)
|
|
results = {}
|
|
for name, fn in strategies.items():
|
|
results[name] = bt.run(ohlcv_data, fn)
|
|
print(f"\n[{name}]")
|
|
print(results[name].summary())
|
|
return results
|