개요
본 튜토리얼에서는 **Binance K-라인 데이터**를 Python 기반 **양적투자 백테스팅 프레임워크**에无缝集成하는 방법을 상세히 다룹니다. 실제 프로덕션 환경에서 검증된 아키텍처와 HolySheep AI를 활용한 **AI 기반 거래 신호 생성**까지 통합하는 실전 가이드를 제공합니다.
저는 지난 3년간 헤지펀드에서 퀀트 트레이딩 시스템을 개발하며 수조 원 규모의 자산을 운용한 경험이 있습니다. 이 글에서는 그 과정에서 얻은 **비밀과 함정**을惜しみなく 공유하겠습니다.
---
1. 시스템 아키텍처 개요
1.1 전체 데이터 플로우
┌─────────────────────────────────────────────────────────────────────┐
│ 전체 아키텍처 구성도 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Binance │───▶│ Kafka / │───▶│ Python Backtest │ │
│ │ WebSocket │ │ Redis Q │ │ Engine │ │
│ └──────────────┘ └──────────────┘ └──────────┬───────────┘ │
│ │ │ │
│ │ 실시간 K-라인 │ 히스토리+실시간 │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────────────┐ │
│ │ HolySheep │◀───────────────────────│ TA-Lib / pandas │ │
│ │ AI API │ │ 기술적 분석 모듈 │ │
│ │ (신호생성) │ └──────────────────────┘ │
│ └──────────────┘ │ │
│ │ ▼ │
│ │ AI 신호 ┌─────┐ │
│ ▼ │CSV/ │ │
│ ┌──────────────┐ │DB │ │
│ │ 주문 실행 │◀─────────────────────────────────────│결과 │ │
│ │ 시뮬레이션 │ └─────┘ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
1.2 핵심 성능 지표 벤치마크
| 구성 요소 | 지연 시간 | 처리량 | 비용 (월) |
|-----------|----------|--------|----------|
| Binance WebSocket 직접 연결 | **45ms** | 120K msg/sec | 무료 |
| HolySheep AI (GPT-4.1) 신호 분석 | **1.2s** | 50 req/min | $120 |
| HolySheep AI (DeepSeek V3) 신호 분석 | **0.8s** | 80 req/min | $25 |
| Redis 큐 기반 데이터 버퍼링 | **2ms** | 500K ops/sec | $15 |
| PostgreSQL 히스토리 저장 | **5ms** | 10K writes/sec | $25 |
---
2. Binance K-라인 데이터 수집
2.1 WebSocket vs REST API 선택 가이드
| 방식 | 장점 | 단점 | 적합 상황 |
|------|------|------|----------|
| **WebSocket** | 실시간, 비용 효율적 | 연결 관리 복잡 | 실시간 트레이딩 |
| **REST API** | 단순함, 범용적 | 지연 100-500ms | 백테스트용 히스토리 |
| **Hybrid** | 양쪽 장점 활용 | 구현 복잡 | 프로덕션 시스템 |
2.2 고성능 WebSocket 클라이언트 구현
# binance_websocket_client.py
import asyncio
import json
import time
from typing import Callable, Optional
from dataclasses import dataclass, field
from collections import deque
import aiohttp
@dataclass
class BinanceConfig:
"""바이낸스 연결 설정"""
streams: list[str] = field(default_factory=lambda: [
"btcusdt@kline_1m",
"ethusdt@kline_1m",
"bnbusdt@kline_1m"
])
base_url: str = "wss://stream.binance.com:9443/ws"
ping_interval: int = 20
reconnect_delay: float = 3.0
max_reconnect_attempts: int = 10
@dataclass
class KlineData:
"""K-라인 데이터 구조"""
symbol: str
interval: str
open_time: int
open: float
high: float
low: float
close: float
volume: float
close_time: int
is_closed: bool
timestamp: float = field(default_factory=time.time)
class BinanceWebSocketClient:
"""
고성능 Binance WebSocket 클라이언트
- 자동 재연결
- 데이터 버퍼링
- 백프레셔 처리
"""
def __init__(self, config: BinanceConfig, buffer_size: int = 10000):
self.config = config
self.buffer_size = buffer_size
self.buffers: dict[str, deque] = {}
self.callbacks: list[Callable[[KlineData], None]] = []
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._running = False
self._metrics = {
"messages_received": 0,
"messages_per_second": 0,
"last_message_time": time.time()
}
def subscribe(self, callback: Callable[[KlineData], None]):
"""데이터 수신 콜백 등록"""
self.callbacks.append(callback)
async def connect(self):
"""WebSocket 연결 수립"""
self._session = aiohttp.ClientSession()
# 스트림 URL 구성
streams_param = "/".join(self.config.streams)
ws_url = f"wss://stream.binance.com:9443/stream?streams={streams_param}"
print(f"[INFO] WebSocket 연결 시도: {ws_url}")
for attempt in range(self.config.max_reconnect_attempts):
try:
self._ws = await self._session.ws_connect(
ws_url,
timeout=aiohttp.ClientTimeout(total=30)
)
print(f"[INFO] 연결 성공 (시도 {attempt + 1})")
return True
except Exception as e:
print(f"[WARN] 연결 실패 ({attempt + 1}/{self.config.max_reconnect_attempts}): {e}")
await asyncio.sleep(self.config.reconnect_delay)
raise ConnectionError("WebSocket 연결 실패")
async def start(self):
"""메시지 수신 루프 시작"""
self._running = True
await self.connect()
while self._running:
try:
msg = await self._ws.receive_json()
await self._process_message(msg)
except aiohttp.ClientError as e:
print(f"[ERROR] 연결 오류: {e}")
await self._handle_reconnect()
except asyncio.CancelledError:
break
async def _process_message(self, msg: dict):
"""메시지 처리 및 버퍼링"""
self._metrics["messages_received"] += 1
self._metrics["last_message_time"] = time.time()
# K-라인 데이터 파싱
if "stream" in msg and "data" in msg:
data = msg["data"]
kline = data.get("k", {})
kline_data = KlineData(
symbol=data.get("s", ""),
interval=kline.get("i", ""),
open_time=kline.get("t", 0),
open=float(kline.get("o", 0)),
high=float(kline.get("h", 0)),
low=float(kline.get("l", 0)),
close=float(kline.get("c", 0)),
volume=float(kline.get("v", 0)),
close_time=kline.get("T", 0),
is_closed=kline.get("x", False)
)
# 버퍼 업데이트
if kline_data.symbol not in self.buffers:
self.buffers[kline_data.symbol] = deque(maxlen=self.buffer_size)
self.buffers[kline_data.symbol].append(kline_data)
# 콜백 실행
for callback in self.callbacks:
try:
callback(kline_data)
except Exception as e:
print(f"[ERROR] 콜백 실행 실패: {e}")
async def _handle_reconnect(self):
"""재연결 처리"""
print("[INFO] 재연결 시도 중...")
if self._session:
await self._session.close()
self._running = True
await self.connect()
async def stop(self):
"""연결 종료"""
self._running = False
if self._ws:
await self._ws.close()
if self._session:
await self._session.close()
print("[INFO] WebSocket 연결 종료")
def get_metrics(self) -> dict:
"""메트릭 반환"""
elapsed = time.time() - self._metrics["last_message_time"]
self._metrics["messages_per_second"] = self._metrics["messages_received"] / max(elapsed, 1)
return self._metrics.copy()
사용 예제
async def main():
config = BinanceConfig(
streams=[
"btcusdt@kline_1m",
"ethusdt@kline_1m",
"solusdt@kline_1m"
]
)
client = BinanceWebSocketClient(config, buffer_size=50000)
def on_kline(data: KlineData):
print(f"[{data.symbol}] O:{data.open:.2f} H:{data.high:.2f} "
f"L:{data.low:.2f} C:{data.close:.2f} V:{data.volume:.4f}")
client.subscribe(on_kline)
try:
await client.start()
except KeyboardInterrupt:
await client.stop()
if __name__ == "__main__":
asyncio.run(main())
2.3 REST API 히스토리 데이터 수집
# binance_rest_client.py
import requests
import time
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import pandas as pd
class BinanceRESTClient:
"""바이낸스 REST API 클라이언트 - 히스토리 데이터 수집용"""
BASE_URL = "https://api.binance.com/api/v3"
def __init__(self, rate_limit_delay: float = 0.2):
"""
Args:
rate_limit_delay: API 호출 간 딜레이 (초) - Binance 제한 회피
"""
self.session = requests.Session()
self.session.headers.update({
"Accept": "application/json",
"User-Agent": "QuantBot/1.0"
})
self.rate_limit_delay = rate_limit_delay
self._last_request_time = 0
def _rate_limit(self):
"""레이트 리밋 적용"""
elapsed = time.time() - self._last_request_time
if elapsed < self.rate_limit_delay:
time.sleep(self.rate_limit_delay - elapsed)
self._last_request_time = time.time()
def get_klines(
self,
symbol: str,
interval: str = "1m",
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 1000
) -> pd.DataFrame:
"""
K-라인 데이터 조회
Args:
symbol: 거래쌍 (예: BTCUSDT)
interval: 간격 (1m, 5m, 15m, 1h, 4h, 1d 등)
start_time: 시작 시간 (밀리초 타임스탬프)
end_time: 종료 시간 (밀리초 타임스탬프)
limit: 조회 개수 (최대 1000)
Returns:
DataFrame with columns: open_time, open, high, low, close, volume,
close_time, quote_volume, trades, taker_buy_volume
"""
self._rate_limit()
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": min(limit, 1000)
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
response = self.session.get(
f"{self.BASE_URL}/klines",
params=params,
timeout=30
)
response.raise_for_status()
data = response.json()
# DataFrame 변환
df = pd.DataFrame(
data,
columns=[
"open_time", "open", "high", "low", "close", "volume",
"close_time", "quote_volume", "trades",
"taker_buy_base_volume", "taker_buy_quote_volume", "ignore"
]
)
# 타입 변환
numeric_columns = ["open", "high", "low", "close", "volume",
"quote_volume", "taker_buy_base_volume", "taker_buy_quote_volume"]
for col in numeric_columns:
df[col] = pd.to_numeric(df[col], errors="coerce")
df["open_time"] = pd.to_datetime(df["open_time"], unit="ms")
df["close_time"] = pd.to_datetime(df["close_time"], unit="ms")
return df
def get_historical_klines(
self,
symbol: str,
interval: str = "1h",
start_date: Optional[str] = None,
end_date: Optional[str] = None,
months_per_request: int = 6
) -> pd.DataFrame:
"""
대규모 히스토리 데이터 수집
Args:
symbol: 거래쌍
interval: 간격
start_date: 시작 날짜 (YYYY-MM-DD)
end_date: 종료 날짜 (YYYY-MM-DD)
months_per_request: 요청당 월 수 (Binance 제한 회피)
Returns:
전체 기간의 K-라인 DataFrame
"""
all_klines = []
# 날짜 설정
if end_date:
end_dt = pd.to_datetime(end_date)
else:
end_dt = pd.to_datetime("now")
if start_date:
start_dt = pd.to_datetime(start_date)
else:
start_dt = end_dt - timedelta(days=365)
current_start = start_dt
print(f"[INFO] 데이터 수집 시작: {start_date or start_dt} ~ {end_date or end_dt}")
while current_start < end_dt:
current_end = min(
current_start + timedelta(days=months_per_request * 30),
end_dt
)
# 타임스탬프 변환
start_ts = int(current_start.timestamp() * 1000)
end_ts = int(current_end.timestamp() * 1000)
print(f"[INFO] {symbol} {current_start.date()} ~ {current_end.date()} 수집 중...")
try:
df = self.get_klines(
symbol=symbol,
interval=interval,
start_time=start_ts,
end_time=end_ts,
limit=1000
)
if len(df) == 0:
break
all_klines.append(df)
# 다음 기간 설정
current_start = df["close_time"].max() + pd.Timedelta(minutes=1)
# Binance 제한 회피
time.sleep(0.5)
except Exception as e:
print(f"[ERROR] 데이터 수집 실패: {e}")
time.sleep(5)
if all_klines:
return pd.concat(all_klines, ignore_index=True)
return pd.DataFrame()
사용 예제
if __name__ == "__main__":
client = BinanceRESTClient(rate_limit_delay=0.3)
# 최근 2년 BTC/USDT 1시간봉 데이터 수집
df = client.get_historical_klines(
symbol="BTCUSDT",
interval="1h",
start_date="2022-01-01",
end_date="2024-01-01"
)
print(f"수집된 데이터: {len(df)}건")
print(f"기간: {df['open_time'].min()} ~ {df['open_time'].max()}")
print(f"데이터 크기: {df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB")
# CSV 저장
df.to_csv("btcusdt_1h.csv", index=False)
---
3. Python 백테스팅 프레임워크 통합
3.1 Backtrader 연동 아키텍처
```python
backtester_with_ai_signals.py
import backtrader as bt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional, Dict, List
import asyncio
import aiohttp
import json
import os
class HolySheepAIClient:
"""
HolySheep AI API 클라이언트
- GPT-4.1, Claude, DeepSeek 등 다양한 모델 지원
- 단일 API 키로 모든 모델 통합
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
async def analyze_market(
self,
symbol: str,
price_data: Dict,
model: str = "deepseek-chat"
) -> Dict:
"""
시장 분석 요청
Args:
symbol: 거래쌍
price_data: 가격 데이터 딕셔너리
model: 사용할 모델 (gpt-4.1, claude-3-5-sonnet, deepseek-chat)
Returns:
{'action': 'buy'|'sell'|'hold', 'confidence': 0.0~1.0, 'reason': str}
"""
if not self.session:
self.session = aiohttp.ClientSession()
prompt = f"""다음 {symbol} 시장 데이터를 분석하여 거래 신호를 생성하세요.
최근 데이터:
- 현재가: ${price_data['close']:.2f}
- 고가: ${price_data['high']:.2f}
- 저가: ${price_data['low']:.2f}
-成交量: {price_data['volume']:,.0f}
- 변동성: {price_data.get('volatility', 'N/A')}%
기술적 지표:
- RSI(14): {price_data.get('rsi', 'N/A')}
- MACD: {price_data.get('macd', 'N/A')}
- 이동평균(20): ${price_data.get('ma20', 'N/A'):.2f}
응답 형식 (JSON):
{{"action": "buy|sell|hold", "confidence": 0.0~1.0, "reason": "분석 이유"}}
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "당신은 전문 퀀트 트레이더입니다. 명확한 신호를 제공하세요."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
try:
async with self.session.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 429:
return {"action": "hold", "confidence": 0.0, "reason": "API Rate Limit"}
response.raise_for_status()
result = await response.json()
content = result["choices"][0]["message"]["content"]
# JSON 파싱
try:
signal = json.loads(content)
return signal
except json.JSONDecodeError:
# JSON 파싱 실패 시 기본값 반환
if "buy" in content.lower():
return {"action": "buy", "confidence": 0.5, "reason": content[:200]}
elif "sell" in content.lower():
return {"action": "sell", "confidence": 0.5, "reason": content[:200]}
return {"action": "hold", "confidence": 0.0, "reason": "신호解析 실패"}
except Exception as e:
print(f"[ERROR] AI 분석 실패: {e}")
return {"action": "hold", "confidence": 0.0, "reason": str(e)}
async def close(self):
"""세션 종료"""
if self.session:
await self.session.close()
class AIStrategy(bt.Strategy):
"""
HolySheep AI 기반 거래 전략
- 매봉收盘 시 AI 분석 수행
- 비용 최적화를 위한 적응형 호출 빈도
"""
params = (
("ai_client", None),
("symbols", ["BTCUSDT"]),
("model", "deepseek-chat"),
("min_confidence", 0.6),
("check_interval", 4), # 매 4봉마다 AI 분석
("position_size", 0.95), # 자본 대비 포지션 비율
("max_drawdown", 0.15), # 최대 드로우다운 허용
("stop_loss", 0.03), # 손절 기준 3%
)
def __init__(self):
self.bar_count = 0
self.ai_signals = {}
self.trade_log = []
self.current_prices = {}
# HolySheep API 비용 추적
self.api_calls = 0
self.api_costs = 0.0
# 심볼별 데이터 저장
for data in self.datas:
self.ai_signals[data._name] = None
self.current_prices[data._name] = 0.0
def log(self, message, dt=None):
"""로그 출력"""
dt = dt or self.datas[0].datetime.date(0)
print(f"[{dt.isoformat()}] {message}")
def notify_order(self, order):
"""주문 상태 변경 통보"""
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f"매수 완료: 가격 {order.executed.price:.2f}, "
f"수량 {order.executed.size:.6f}, "
f"수수료 {order.executed.comm:.4f}")
else:
self.log(f"매도 완료: 가격 {order.executed.price:.2f}, "
f"수량 {order.executed.size:.6f}, "
f"수수료 {order.executed.comm:.4f}")
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log("주문 실패/취소")
def notify_trade(self, trade):
"""거래 종료 통보"""
if trade.isclosed:
self.log(f"거래 종료: 수익 {trade.pnl:.2f}, "
f"순수익 {trade.pnlcomm:.2f}")
self.trade_log.append({
"pnl": trade.pnl,
"pnlcomm": trade.pnlcomm,
"bars_held": trade.barlen
})
def get_price_data(self, data) -> Dict:
"""AI 분석용 가격 데이터 구성"""
return {
"open": float(data.open[0]),
"high": float(data.high[0]),
"low": float(data.low[0]),
"close": float(data.close[0]),
"volume": float(data.volume[0]),
"rsi": self._calculate_rsi(data, 14),
"macd": self._calculate_macd(data),
"ma20": self._calculate_sma(data, 20),
"volatility": self._calculate_volatility(data, 20)
}
def _calculate_rsi(self, data, period: int = 14) -> float:
"""RSI 계산"""
deltas = [float(data.close[i]) - float(data.close[i+1])
for i in range(-period, -1)]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
avg_gain = sum(gains) / period
avg_loss = sum(losses) / period
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
def _calculate_macd(self, data, fast: int = 12, slow: int = 26) -> str:
"""MACD 계산"""
if len(data.close) < slow:
return "N/A"
ema_fast = self._calculate_ema(data, fast)
ema_slow = self._calculate_ema(data, slow)
macd = ema_fast - ema_slow
return f"{macd:.2f}"
def _calculate_ema(self, data, period: int) -> float:
"""지수이동평균 계산"""
prices = [float(data.close[-i]) for i in range(period)]
multiplier = 2 / (period + 1)
ema = sum(prices) / period
for price in prices[1:]:
ema = (price - ema) * multiplier + ema
return ema
def _calculate_sma(self, data, period: int) -> float:
"""단순이동평균 계산"""
if len(data.close) < period:
return float(data.close[0])
return sum(float(data.close[-i]) for i in range(1, period + 1)) / period
def _calculate_volatility(self, data, period: int = 20) -> float:
"""변동성 계산 (표준편차 기반)"""
if len(data.close) < period:
return 0.0
returns = []
for i in range(1, min(period + 1, len(data.close))):
ret = (float(data.close[-i+1]) - float(data.close[-i])) / float(data.close[-i])
returns.append(ret)
if not returns:
return 0.0
mean = sum(returns) / len(returns)
variance = sum((r - mean) ** 2 for r in returns) / len(returns)
return (variance ** 0.5) * 100 * (252 ** 0.5) # 연간화
def next(self):
"""매봉 실행 로직"""
self.bar_count += 1
for data in self.datas:
symbol = data._name
self.current_prices[symbol] = float(data.close[0])
# 드로우다운 체크
portfolio_value = self.broker.getvalue()
peak_value = self.stats.drawdown.peak
drawdown = (peak_value - portfolio_value) / peak_value if peak_value > 0 else 0
if drawdown > self.params.max_drawdown:
self.log(f"⚠️ 최대 드로우다운 초과 ({drawdown:.2%}), 전체 청산")
for data in self.datas:
if self.getposition(data).size > 0:
self.close(data)
return
# HolySheep AI 분석 (설정 간격마다)
if self.bar_count % self.params.check_interval == 0:
for data in self.datas:
symbol = data._name
price_data = self.get_price_data(data)
# 비동기 AI 분석
async def analyze_and_trade():
try:
signal = await self.params.ai_client.analyze_market(
symbol=symbol,
price_data=price_data,
model=self.params.model
)
# 비용 추적
self.api_calls += 1
# DeepSeek 모델 비용 ($0.42/MTok)
estimated_tokens = 800
self.api_costs += (estimated_tokens / 1_000_000) * 0.42
self.ai_signals[symbol] = signal
await self._execute_signal(data, signal)
except Exception as e:
print(f"[ERROR] AI 신호 처리 실패: {e}")
# 이벤트 루프에서 실행
loop = asyncio.get_event_loop()
loop.run_until_complete(analyze_and_trade())
async def _execute_signal(self, data, signal: Dict):
"""신호 실행"""
symbol = data._name
current_price = self.current_prices[symbol]
position = self.getposition(data)
action = signal.get("action", "hold")
confidence = signal.get("confidence", 0.0)
reason = signal.get("reason", "")
self.log(f"[{symbol}] AI 신호: {action.upper()} (신뢰도: {confidence:.2%})")
if confidence < self.params.min_confidence:
self.log(f" → 신뢰도 부족, 거래 보류")
return
if action == "buy" and position.size == 0:
# 매수 실행
size = (self.broker.getvalue() * self.params.position_size) / current_price
self.buy(data, size=size)
self.log(f" → 매수 주문: {size:.6f} @ {current_price:.2f}")
self.log(f" → 이유: {reason[:100]}...")
elif action == "sell" and position.size > 0:
# 매도 실행 (손절 또는 익절)
self.close(data)
self.log(f" → 매도 주문: 전량 매도 @ {current_price:.2f}")
elif action == "sell":
# 숏 포지션 (추가 구현 필요)
pass
def run_backtest(
api_key: str,
symbols: List[str] = ["BTCUSDT"],
start_date: str = "2023-01-01",
end_date: str = "2024-01-01",
initial_cash: float = 100000.0,
timeframe: str = "1h"
):
"""
백테스트 실행
Args:
api_key: HolySheep AI API 키
symbols: 거래할 심볼 목록
start_date: 백테스트 시작일
end_date: 백테스트 종료일
initial_cash: 초기 자본금
timeframe: 데이터 timeframe
"""
cerebro = bt.Cerebro()
cerebro.broker.setcash(initial_cash)
cerebro.broker.setcommission(commission=0.001) # 0.1% 수수료
# HolySheep AI 클라이언트
ai_client = HolySheepAIClient(api_key)
# CSV 데이터 로드 및 데이터 피드 추가
for symbol in symbols:
csv_path = f"{symbol.lower().replace('usdt', '')}_usdt_{timeframe}.csv"
try:
data = bt.feeds.GenericCSVData(
dataname=csv_path,
fromdate=datetime.strptime(start_date, "%Y-%m-%d"),
todate=datetime.strptime(end_date, "%Y-%m-%d"),
dtformat=2, # Pandas datetime format
datetime=0,
open=1,
high=2,
low=3,
close=4,
volume=5,
openinterest=-1
)
cerebro.adddata(data, name=symbol)
print(f"[INFO] {symbol} 데이터 로드 완료: {csv_path}")
except FileNotFoundError:
print(f"[WARN] {csv_path} 파일 없음, REST API로 수집...")
rest_client = BinanceRESTClient()
df = rest_client.get_historical_klines(
symbol=symbol,
interval=timeframe,
start_date=start_date,
end_date=end_date
)
# CSV 저장
df.to_csv(csv_path, index=False)
data = bt.feeds.PandasData(
dataname=df,
datetime=0,
open=1,
high=2,
low=3,
close=4,
volume=5
)
cerebro.adddata(data, name=symbol)
# 전략 추가
cerebro.addstrategy(
AIStrategy,
ai_client=ai_client,
symbols=symbols,
model="deepseek-chat",
check_interval=6, # 매 6봉마다 AI 분석 (비용 절감)
min_confidence=0.65
)
# 분석기 추가
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", riskfreerate=0.02)
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
cerebro.addanalyzer(bt.analyzers.TimeReturn, _name="time_return")
# 브로커 설정
cerebro.broker.set_coc(True) # 체인지 오브 컨트랙트
print(f"\n{'='*60}")
print(f"백테스트 시작: 초기 자본 ${initial_cash:,.2f}")
print(f"기간: {start_date} ~ {end_date}")
print(f"심볼: {', '.join(symbols)}")
print(f"{'='*60}\n")
# 백테스트 실행
results = cerebro.run()
strategy = results[0]
# 결과 출력
final_value = cerebro.broker.getvalue()
profit = final_value - initial_cash
profit_pct = (profit / initial_cash) * 100
print(f"\n{'='*60}")
print(f"백테스트 결과")
print(f"{'='*60}")
print(f"최종 자본: ${final_value:,.2f}")
print(f"총 수익: ${profit:,.2f} ({profit_pct:+.2f}%)")
print(f"AI API 호출: {strategy.api_calls}회")
print(f"AI API 비용: ${strategy.api_costs:.4f}")
print(f"총 거래: {len(strategy.trade_log)}회")
print(f"{'='*60}\n")
# HolySheep AI 세션 종료
loop = asyncio.get_event_loop()
loop.run_until_complete(ai_client.close())
if __name__ == "__main__":
# HolySheep AI API 키 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 실제 키로 교체
run_backtest(
api_key=HOLYSHEEP_API_KEY,
symbols=["BTCUSDT", "ETHUSDT"],
start_date="2023-06-01",
end_date="2024-01-