gahusb
42b91d03cf
- DailyLedger: 당일 매수 회계 + 연속 손절 카운터 + 매수 신호 점수 한 객체로 집약 (bot.py 정리) - RiskGate: 테마당 동시 보유 + 노출 비율 상한 검증 (포트폴리오 레벨) - news_snapshot: 뉴스 SQLite 영구 저장 + 사후 감성 재검증 인프라 - backtest_runner: 전 종목 KIS 일봉 기반 백테스트 (Sharpe/MDD/Calmar) - bot.py 274 line 정리 (DailyLedger 분리) - backtest.py 173 line 재작성 (v3.2 next-bar 체결 + 거래세) - daily_launcher.py 폐기 (warmup_and_restart 통합) - .gitignore: .claude/ 제외 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""
|
|
백테스팅 프레임워크 (v3.2 — Realism 보강)
|
|
|
|
개선 사항 (v3.2):
|
|
1. 다음 봉 시가 체결 옵션 (look-ahead bias 제거)
|
|
2. 증권거래세 (매도 시 0.2%, 수수료와 별개 부과)
|
|
3. 거래량 기반 부분 체결 (한 봉 거래량의 N% 상한)
|
|
4. Calmar, Payoff, Turnover 지표 추가
|
|
"""
|
|
|
|
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"
|
|
|
|
@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
|
|
calmar_ratio: float = 0.0
|
|
payoff_ratio: float = 0.0 # 평균수익 / |평균손실|
|
|
turnover_ratio: float = 0.0 # 총 매매대금 / 초기자본
|
|
trades: List[Trade] = field(default_factory=list)
|
|
|
|
def summary(self) -> str:
|
|
lines = [
|
|
"=" * 50,
|
|
"📊 백테스팅 결과 (v3.2)",
|
|
"=" * 50,
|
|
f"총 수익률: {self.total_return_pct:+.2f}%",
|
|
f"Sharpe Ratio: {self.sharpe_ratio:.3f}",
|
|
f"Calmar Ratio: {self.calmar_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}",
|
|
f"Payoff Ratio: {self.payoff_ratio:.2f}",
|
|
f"Turnover: {self.turnover_ratio:.2f}x",
|
|
"=" * 50,
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
class Backtester:
|
|
"""
|
|
OHLCV 기반 전략 백테스터.
|
|
|
|
체결 모델 (v3.2):
|
|
- next_bar_open=True: 신호 발생 다음 봉 시가로 체결 (look-ahead 제거)
|
|
- slippage: 체결가에 ±slippage_rate 적용
|
|
- commission_rate: 매수/매도 양쪽에 부과 (증권사 수수료)
|
|
- sell_tax_rate: 매도 시에만 부과 (증권거래세 0.2%)
|
|
- max_volume_participation: 봉 거래량의 N% 이하로 체결 제한
|
|
"""
|
|
|
|
def __init__(self,
|
|
initial_capital: float = 10_000_000,
|
|
commission_rate: float = 0.00015,
|
|
slippage_rate: float = 0.001,
|
|
sell_tax_rate: float = 0.002,
|
|
next_bar_open: bool = True,
|
|
max_volume_participation: float = 0.01):
|
|
self.initial_capital = initial_capital
|
|
self.commission_rate = commission_rate
|
|
self.slippage_rate = slippage_rate
|
|
self.sell_tax_rate = sell_tax_rate
|
|
self.next_bar_open = next_bar_open
|
|
self.max_volume_participation = max_volume_participation
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 단일 종목
|
|
# ──────────────────────────────────────────────
|
|
def run(self, ohlcv_data: Dict, strategy_fn: Callable,
|
|
ticker: str = "UNKNOWN", warmup: int = 60) -> BacktestResult:
|
|
closes = np.array(ohlcv_data.get('close', []), dtype=float)
|
|
opens = np.array(ohlcv_data.get('open', closes), 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] = []
|
|
total_turnover = 0.0 # 누적 매매대금
|
|
|
|
# 마지막 인덱스는 next-bar 체결 시 여유 필요
|
|
last_signal_idx = n - 2 if self.next_bar_open else n - 1
|
|
|
|
for i in range(warmup, last_signal_idx + 1):
|
|
slice_data = {
|
|
'close': closes[:i+1].tolist(),
|
|
'open': opens[: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
|
|
|
|
# 체결가 산출 — next_bar_open이면 i+1 시가, 아니면 i 종가
|
|
fill_idx = i + 1 if self.next_bar_open and i + 1 < n else i
|
|
base_price = opens[fill_idx] if self.next_bar_open else closes[fill_idx]
|
|
fill_volume = volumes[fill_idx]
|
|
|
|
buy_price = base_price * (1 + self.slippage_rate)
|
|
sell_price = base_price * (1 - self.slippage_rate)
|
|
|
|
if signal == "BUY" and position == 0:
|
|
# 전액 투자 (수수료 포함 총비용 기준)
|
|
raw_qty = int(capital / (buy_price * (1 + self.commission_rate)))
|
|
# 거래량 상한 — 봉 거래량의 N%까지만 체결
|
|
vol_cap = int(fill_volume * self.max_volume_participation)
|
|
qty = min(raw_qty, vol_cap) if vol_cap > 0 else raw_qty
|
|
if qty > 0:
|
|
cost = qty * buy_price * (1 + self.commission_rate)
|
|
capital -= cost
|
|
position = qty
|
|
entry_price = buy_price
|
|
entry_idx = fill_idx
|
|
total_turnover += qty * buy_price
|
|
|
|
elif signal == "SELL" and position > 0:
|
|
# 매도: 수수료 + 증권거래세
|
|
sell_cost_rate = self.commission_rate + self.sell_tax_rate
|
|
vol_cap = int(fill_volume * self.max_volume_participation) if fill_volume > 0 else position
|
|
exec_qty = min(position, vol_cap) if vol_cap > 0 else position
|
|
proceeds = exec_qty * sell_price * (1 - sell_cost_rate)
|
|
capital += proceeds
|
|
total_turnover += exec_qty * sell_price
|
|
trades.append(Trade(
|
|
ticker=ticker,
|
|
entry_date=entry_idx,
|
|
entry_price=entry_price,
|
|
exit_date=fill_idx,
|
|
exit_price=sell_price,
|
|
qty=exec_qty
|
|
))
|
|
position -= exec_qty
|
|
if 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)
|
|
sell_cost_rate = self.commission_rate + self.sell_tax_rate
|
|
proceeds = position * last_price * (1 - sell_cost_rate)
|
|
capital += proceeds
|
|
total_turnover += position * last_price
|
|
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
|
|
position = 0
|
|
|
|
return self._compute_metrics(equity_curve, trades, total_turnover)
|
|
|
|
def run_multi(self, ohlcv_dict: Dict[str, Dict], strategy_fn: Callable,
|
|
warmup: int = 60) -> Dict[str, BacktestResult]:
|
|
return {t: self.run(d, strategy_fn, t, warmup) for t, d in ohlcv_dict.items()}
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 지표 계산
|
|
# ──────────────────────────────────────────────
|
|
def _compute_metrics(self, equity_curve: List[float], trades: List[Trade],
|
|
total_turnover: float) -> BacktestResult:
|
|
equity = np.array(equity_curve, dtype=float)
|
|
total_return_pct = (equity[-1] / equity[0] - 1) * 100
|
|
|
|
daily_returns = np.diff(equity) / (equity[:-1] + 1e-9)
|
|
sharpe = (daily_returns.mean() / daily_returns.std()) * np.sqrt(252) \
|
|
if daily_returns.std() > 0 else 0.0
|
|
|
|
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 = float(np.mean([t.pnl_pct for t in wins])) if wins else 0.0
|
|
avg_loss = float(np.mean([t.pnl_pct for t in losses])) if losses else 0.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)
|
|
|
|
# 신규 지표
|
|
calmar = (total_return_pct / max_drawdown) if max_drawdown > 0 else 0.0
|
|
payoff = (avg_win / abs(avg_loss)) if avg_loss != 0 else 0.0
|
|
turnover_ratio = total_turnover / (self.initial_capital + 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),
|
|
calmar_ratio=round(calmar, 3),
|
|
payoff_ratio=round(payoff, 3),
|
|
turnover_ratio=round(turnover_ratio, 3),
|
|
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]:
|
|
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
|