gahusb/ai-trade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: web-ai V1 assets → signal_v1/ (graduation prep)

Browse Source 
Atomic mv of root V1 assets (main_server.py + modules/ + data/ +
tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory.
load_dotenv() updated to load web-ai/.env explicitly via Path.

Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat
(signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2.

Tests: signal_v1/tests/unit baseline preserved (no regression).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gahusb
2026-05-16 03:00:11 +09:00
parent 42b91d03cf
commit 7ea1a21487
39 changed files with 722 additions and 691 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
signal_v1/modules/analysis/backtest.py Normal file
View File

@@ -0,0 +1,274 @@
"""
백테스팅 프레임워크 (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