암호화폐 트레이딩 시스템에서 시장 데이터의 질과 백테스팅의 정밀도는 전략의 생사를 가릅니다. Bybit는 높은 유동성과 다양한 계약 유형으로 퀀트 트레이더들 사이에서 인기가 높지만, 대용량 데이터 수집과 실시간 분석에는 별도의 인프라와 AI 모델 통합이 필요합니다.
저는 HolySheep AI를 통해 단일 API 키로 Bybit 시장 데이터 수집부터 AI 기반 전략 분석까지:end-to-end 파이프라인을 구축한 경험이 있습니다. 이 튜토리얼에서는 2년 넘게 프로덕션 환경에서 검증된 아키텍처와 비용 최적화 전략을 공유합니다.
1. 아키텍처 개요: 데이터 수집 → AI 분석 → 백테스팅 파이프라인
Bybit API를 활용한 완전한 백테스팅 시스템은 다음 네 계층으로 구성됩니다:
- 데이터 수집 계층: WebSocket 실시간 시세 + RESTful Historical Data API
- 데이터 저장 계층: 시계열 데이터베이스 (TimescaleDB 또는 InfluxDB)
- AI 분석 계층: HolySheep AI 게이트웨이를 통한 다중 모델 통합
- 백테스팅 계층: 벡터라이즈 백테스터 + 이벤트 기반 시뮬레이터
# 전체 파이프라인 아키텍처
#
┌─────────────────────────────────────────────────────────┐
│ Bybit Exchange │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ WebSocket │ │ REST API │ │
│ │ (Real-time) │ │ (Historical) │ │
│ └──────┬───────┘ └──────┬───────┘ │
└─────────┼───────────────────┼─────────────────────────┘
│ │
▼ ▼
┌─────────────────────────────────────────────────────────┐
│ Data Collector Service │
│ - Rate Limit Handling (10 req/sec public) │
│ - Auto-retry with exponential backoff │
│ - Data validation & normalization │
└─────────────────────────┬───────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ TimescaleDB (Time-series storage) │
│ - OHLCV candles, orderbook snapshots, trades │
│ - Compression after 7 days │
│ - Retention policy management │
└─────────────────────────┬───────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ HolySheep AI Gateway │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ DeepSeek │ │ GPT-4.1 │ │ Claude │ │
│ │ V3.2 │ │ │ │ Sonnet 4 │ │
│ │ $0.42/MTok │ │ $8/MTok │ │ $15/MTok │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ Single API key for all models │
└─────────────────────────┬───────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Backtesting Engine │
│ - Vectorized execution (pandas) │
│ - Event-driven simulation │
│ - Performance metrics calculation │
└─────────────────────────────────────────────────────────┘
2. Bybit API 데이터 수집实战 구현
2.1 WebSocket 실시간 시세 수집
Bybit WebSocket API는 Public 채널(마켓 데이터)과 Private 채널(계정 정보)으로 나뉩니다. 마켓 데이터는 별도 인증 없이 접근 가능하며, 연결 관리와 재연결 로직이 핵심입니다.
# bybit_websocket_collector.py
import asyncio
import json
import websockets
from datetime import datetime
from typing import Dict, List, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BybitWebSocketCollector:
"""Bybit WebSocket 실시간 데이터 수집기"""
# Public endpoint (인증 불필요)
WS_PUBLIC_URL = "wss://stream.bybit.com/v5/public/linear"
# Rate limit: 10 req/sec per IP for public endpoints
MAX_RECONNECT_ATTEMPTS = 5
RECONNECT_DELAY_BASE = 1 # seconds
def __init__(self, db_writer):
self.db_writer = db_writer
self.websocket = None
self.is_running = False
self.subscription_topics = set()
self.last_ping_time = datetime.utcnow()
async def subscribe(self, symbols: List[str], categories: List[str]):
"""구독할 토픽 등록"""
# category: spot, linear, inverse, option
subscribe_message = {
"op": "subscribe",
"args": []
}
for symbol in symbols:
for category in categories:
topic = f"{category}.candle_1.{symbol}" # 1분 봉
orderbook_topic = f"{category}.orderbook.50.{symbol}" # 50단계 호가창
subscribe_message["args"].append(topic)
subscribe_message["args"].append(orderbook_topic)
self.subscription_topics.add(topic)
if self.websocket:
await self.websocket.send(json.dumps(subscribe_message))
logger.info(f"Subscribed to {len(subscribe_message['args'])} topics")
async def handle_message(self, message: str):
"""수신된 메시지 처리"""
try:
data = json.loads(message)
# 구독 확인 응답
if data.get("op") == "subscribe":
logger.info(f"Subscription confirmed: {data.get('success', [])}")
return
# 에러 응답
if "error" in data:
logger.error(f"WebSocket error: {data['error']}")
return
# 실시간 데이터
topic = data.get("topic", "")
payload = data.get("data", {})
if "candle" in topic:
await self._process_candle(payload)
elif "orderbook" in topic:
await self._process_orderbook(payload, topic)
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e}")
except Exception as e:
logger.error(f"Message handling error: {e}")
async def _process_candle(self, candle: Dict):
"""캔들 데이터 처리 및 저장"""
ohlcv = {
"symbol": candle.get("symbol"),
"timestamp": int(candle.get("ts")),
"open": float(candle.get("open")),
"high": float(candle.get("high")),
"low": float(candle.get("low")),
"close": float(candle.get("close")),
"volume": float(candle.get("volume")),
"turnover": float(candle.get("turnover")),
"interval": candle.get("interval", "1"),
}
await self.db_writer.insert_ohlcv(ohlcv)
async def _process_orderbook(self, orderbook: Dict, topic: str):
"""호가창 데이터 처리"""
# 50단계 호가창 데이터 (메모리 효율적 저장)
bids = [[float(b), float(q)] for b, q in orderbook.get("b", [])]
asks = [[float(a), float(q)] for a, q in orderbook.get("a", [])]
snapshot = {
"symbol": orderbook.get("s"),
"timestamp": int(orderbook.get("ts", 0)),
"bids": bids,
"asks": asks,
"update_type": orderbook.get("u"), # snapshot or delta
}
await self.db_writer.insert_orderbook(snapshot)
async def connect(self):
"""WebSocket 연결 수립"""
attempt = 0
while attempt < self.MAX_RECONNECT_ATTEMPTS:
try:
self.websocket = await websockets.connect(
self.WS_PUBLIC_URL,
ping_interval=20, # 20초마다 ping
ping_timeout=10,
max_size=10 * 1024 * 1024 # 10MB max message
)
self.is_running = True
logger.info("WebSocket connected successfully")
return
except Exception as e:
attempt += 1
delay = self.RECONNECT_DELAY_BASE * (2 ** attempt)
logger.warning(f"Connection failed (attempt {attempt}): {e}. Retrying in {delay}s")
await asyncio.sleep(delay)
raise ConnectionError("Failed to connect after max attempts")
async def run(self, symbols: List[str]):
"""메인 수집 루프"""
await self.connect()
await self.subscribe(symbols, ["linear", "inverse"])
try:
async for message in self.websocket:
await self.handle_message(message)
self.last_ping_time = datetime.utcnow()
except websockets.ConnectionClosed as e:
logger.error(f"Connection closed: {e}")
self.is_running = False
# 자동 재연결
await self._reconnect(symbols)
사용 예시
async def main():
from db_writer import DatabaseWriter
db = DatabaseWriter() # TimescaleDB writer 구현
collector = BybitWebSocketCollector(db)
# 주요 선물 심볼订阅
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT"]
try:
await collector.run(symbols)
except KeyboardInterrupt:
logger.info("Shutting down collector...")
await collector.close()
if __name__ == "__main__":
asyncio.run(main())
2.2 Historical Data 대량 수집 (REST API)
백테스팅에는 수개월에서 수년간의 Historical Data가 필요합니다. REST API로 대량 데이터를 수집할 때 주의할 점은 Rate Limit과 페이지네이션 처리입니다.
# bybit_historical_collector.py
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import time
import logging
logger = logging.getLogger(__name__)
class BybitHistoricalCollector:
"""Bybit REST API Historical Data 수집기"""
# Public API endpoints
BASE_URL = "https://api.bybit.com"
# Rate limits (공식 문서 기준)
# - Public endpoints: 10 req/sec per IP
# - Klines: 10 req/sec, max 1000 items per request
RATE_LIMIT_DELAY = 0.12 # 100ms safety margin
def __init__(self, db_writer):
self.db_writer = db_writer
self.session: Optional[aiohttp.ClientSession] = None
self.request_count = 0
self.last_request_time = 0
async def _rate_limit_wait(self):
"""Rate limit 준수 대기"""
elapsed = time.time() - self.last_request_time
if elapsed < self.RATE_LIMIT_DELAY:
await asyncio.sleep(self.RATE_LIMIT_DELAY - elapsed)
self.last_request_time = time.time()
self.request_count += 1
async def get_klines(
self,
symbol: str,
category: str = "linear",
interval: str = "1",
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 1000
) -> List[Dict]:
"""K线(OHLCV) 데이터 조회"""
await self._rate_limit_wait()
params = {
"category": category,
"symbol": symbol,
"interval": interval,
"limit": limit,
}
if start_time:
params["start"] = start_time
if end_time:
params["end"] = end_time
url = f"{self.BASE_URL}/v5/market/kline"
async with self.session.get(url, params=params) as response:
if response.status != 200:
raise Exception(f"API error: {response.status}")
data = await response.json()
if data.get("retCode") != 0:
raise Exception(f"API error: {data.get('retMsg')}")
return data.get("result", {}).get("list", [])
async def collect_historical_klines(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
interval: str = "1",
category: str = "linear"
) -> int:
"""指定 기간 Historical K线 수집"""
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)
total_records = 0
current_start = start_ts
# Bybit API: 최대 1000개씩 반환, start/end 시간 기준
# 타임스탬프 범위로 페이지네이션
while current_start < end_ts:
records = await self.get_klines(
symbol=symbol,
category=category,
interval=interval,
start_time=current_start,
end_time=end_ts,
limit=1000
)
if not records:
break
# 타임스탬프 기준 정렬 (오래된 것부터)
records = sorted(records, key=lambda x: int(x[0]))
# 데이터 변환 및 저장
ohlcv_records = []
for r in records:
ohlcv_records.append({
"symbol": symbol,
"timestamp": int(r[0]),
"open": float(r[1]),
"high": float(r[2]),
"low": float(r[3]),
"close": float(r[4]),
"volume": float(r[5]),
"turnover": float(r[6]),
"interval": interval,
})
await self.db_writer.bulk_insert_ohlcv(ohlcv_records)
total_records += len(records)
# 마지막 레코드의 타임스탬프 + 1ms부터 재조회
current_start = int(records[-1][0]) + 1
logger.info(
f"{symbol}: {total_records} records collected "
f"({datetime.fromtimestamp(current_start/1000).strftime('%Y-%m-%d %H:%M')})"
)
# 마지막 페이지 체크
if len(records) < 1000:
break
# 타임스탬프 기반 페이지네이션 제한 (Bybit 권장)
# 최대 200개 요청 분량까지만 연속 조회
if total_records >= 200000:
logger.warning(f"Large dataset: pausing at {total_records} records")
await asyncio.sleep(1)
return total_records
async def collect_multiple_symbols(
self,
symbols: List[str],
start_date: datetime,
end_date: datetime,
intervals: List[str] = ["1", "5", "15", "60", "240", "D"]
):
"""여러 심볼, 여러 timeframe 동시 수집"""
async with aiohttp.ClientSession() as session:
self.session = session
tasks = []
for symbol in symbols:
for interval in intervals:
tasks.append(
self.collect_historical_klines(
symbol=symbol,
start_date=start_date,
end_date=end_date,
interval=interval
)
)
results = await asyncio.gather(*tasks, return_exceptions=True)
total = 0
for i, result in enumerate(results):
symbol = symbols[i // len(intervals)]
interval = intervals[i % len(intervals)]
if isinstance(result, Exception):
logger.error(f"{symbol} {interval}: {result}")
else:
total += result
logger.info(f"✓ {symbol} {interval}: {result} records")
return total
사용 예시: 2년치 BTC/USDT 1분봉 수집
async def main():
from db_writer import DatabaseWriter
db = DatabaseWriter()
collector = BybitHistoricalCollector(db)
# 2년치 데이터
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=730) # 약 2년
symbols = ["BTCUSDT", "ETHUSDT"]
logger.info(f"Starting historical data collection...")
logger.info(f"Period: {start_date.date()} ~ {end_date.date()}")
total = await collector.collect_multiple_symbols(
symbols=symbols,
start_date=start_date,
end_date=end_date,
intervals=["1", "5", "60"] # 1분, 5분, 1시간봉
)
logger.info(f"Collection complete: {total:,} total records")
if __name__ == "__main__":
asyncio.run(main())
3. HolySheep AI 게이트웨이 통합: 전략 분석 자동화
수집된 Historical Data를 분석하고 패턴을 발견하려면 AI 모델이 필요합니다. HolySheep AI를 사용하면 단일 API 키로 여러 모델을 조합하여 사용할 수 있습니다. 비용 효율적인 DeepSeek V3.2로 패턴 탐색 → 정밀 분석용 GPT-4.1로 검증하는 2단계 분석 파이프라인을 구축했습니다.
# strategy_analyzer.py
import os
from typing import List, Dict, Tuple, Optional
from openai import AsyncOpenAI
import json
import logging
logger = logging.getLogger(__name__)
HolySheep AI Gateway Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEHEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
class HolySheepStrategyAnalyzer:
"""HolySheep AI를 활용한 전략 분석기"""
def __init__(self, api_key: str):
self.client = AsyncOpenAI(
api_key=api_key,
base_url=HOLYSHEEP_BASE_URL
)
# 모델별 비용 최적화 설정
self.models = {
"deepseek": "deepseek/deepseek-chat-v3",
"gpt4": "gpt-4.1",
"claude": "claude-sonnet-4-5",
"gemini": "gemini-2.5-flash"
}
# 토큰 사용량 추적
self.total_tokens_used = 0
self.total_cost_cents = 0.0
async def analyze_pattern(self, ohlcv_data: List[Dict]) -> Dict:
"""1단계: DeepSeek V3.2로 패턴 탐색 (저비용, 빠른 분석)"""
# 최근 100개 캔들 데이터 요약
recent_data = ohlcv_data[-100:]
prompt = f"""다음은 BTC/USDT 1시간봉 최근 데이터입니다. 주요 기술적 패턴과 이상치를 분석해주세요.
최근 데이터 ({len(recent_data)}개):
- 현재가: ${recent_data[-1]['close']:,.2f}
- 최근 10봉 고점: ${max(d['high'] for d in recent_data[-10:])::,.2f}
- 최근 10봉 저점: ${min(d['low'] for d in recent_data[-10:])::,.2f}
- RSI(14): 계산 필요
- 거래량 추세: {'증가' if recent_data[-1]['volume'] > sum(d['volume'] for d in recent_data[-20:-1])/20 else '감소'}
분석 요청:
1. 현재 추세 방향 (상승/하락/횡보)
2. 주요 지지/저항 구간
3. 주의すべき 기술적 신호
4. 거래량 측면 이상 징후
JSON 형식으로 응답해주세요."""
response = await self.client.chat.completions.create(
model=self.models["deepseek"],
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=800
)
usage = response.usage
cost = (usage.prompt_tokens * 0.042 + usage.completion_tokens * 0.42) / 100
self.total_tokens_used += usage.total_tokens
self.total_cost_cents += cost
return {
"pattern_analysis": response.choices[0].message.content,
"tokens_used": usage.total_tokens,
"cost_usd": cost,
"model": "deepseek-v3.2"
}
async def validate_strategy(
self,
strategy_description: str,
historical_performance: Dict
) -> Dict:
"""2단계: GPT-4.1로 전략 정밀 검증 (고비용, 정확한 분석)"""
prompt = f"""당신은 퀀트 트레이딩 전문가입니다. 다음 전략의 타당성을 검증해주세요.
전략 설명:
{strategy_description}
과거 성과:
- 총 거래 횟수: {historical_performance.get('total_trades', 0)}
- 승률: {historical_performance.get('win_rate', 0):.1f}%
- 평균 수익률: {historical_performance.get('avg_return', 0):.2f}%
- 최대 드로우다운: {historical_performance.get('max_drawdown', 0):.1f}%
- 샤프 비율: {historical_performance.get('sharpe_ratio', 0):.2f}
- 평균 홀딩 기간: {historical_performance.get('avg_holding_hours', 0):.1f}시간
검증 요청:
1. 이 전략의 핵심 강점
2.潜在적 리스크 요인
3. 백테스팅 오버피팅 가능성
4. 개선 제안사항
5. 리스크 조정 수익률 예상
JSON 형식으로 응답해주세요."""
response = await self.client.chat.completions.create(
model=self.models["gpt4"],
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
max_tokens=1200
)
usage = response.usage
# GPT-4.1: $8/MTok = $0.008/1KTok
cost = (usage.prompt_tokens + usage.completion_tokens) * 8 / 100000
self.total_tokens_used += usage.total_tokens
self.total_cost_cents += cost
return {
"validation": response.choices[0].message.content,
"tokens_used": usage.total_tokens,
"cost_usd": cost,
"model": "gpt-4.1"
}
async def explain_trade_signal(
self,
signal: Dict,
market_context: Dict
) -> str:
"""3단계: Claude Sonnet으로 거래 신호 상세 설명 (창작적 분석)"""
prompt = f"""다음 거래 신호의 근거를 상세히 설명해주세요.
신호 정보:
- 신호 유형: {signal.get('type', 'UNKNOWN')}
- 진입 가격: ${signal.get('entry_price', 0):,.2f}
- 손절락: ${signal.get('stop_loss', 0):,.2f}
- 목표가: ${signal.get('take_profit', 0):,.2f}
- 신뢰도: {signal.get('confidence', 0):.0f}%
- 생성 시각: {signal.get('timestamp', 'N/A')}
시장 맥락:
- 현재 추세: {market_context.get('trend', 'N/A')}
- 주요 이벤트: {market_context.get('events', 'None')}
- 시장 심리: {market_context.get('sentiment', 'N/A')}
한국어로 traders易懂하게 설명해주세요. 매수/매도 이유, 위험 요소, 대처 방법을 포함해야 합니다."""
response = await self.client.chat.completions.create(
model=self.models["claude"],
messages=[{"role": "user", "content": prompt}],
temperature=0.4,
max_tokens=1000
)
usage = response.usage
# Claude Sonnet 4.5: $15/MTok = $0.015/1KTok
cost = (usage.prompt_tokens + usage.completion_tokens) * 15 / 100000
self.total_tokens_used += usage.total_tokens
self.total_cost_cents += cost
return response.choices[0].message.content
async def batch_analyze_strategies(
self,
strategies: List[Dict]
) -> List[Dict]:
"""배치 분석: 여러 전략 동시 분석 (Rate Limit 주의)"""
# HolySheep AI는 요청 수준 Rate Limit이 없지만
# HolySheep 프록시 특성상 동시 연결 10개 이하 권장
semaphore = asyncio.Semaphore(5)
async def analyze_with_limit(strategy: Dict) -> Dict:
async with semaphore:
result = await self.validate_strategy(
strategy_description=strategy.get("description", ""),
historical_performance=strategy.get("performance", {})
)
result["strategy_id"] = strategy.get("id")
return result
tasks = [analyze_with_limit(s) for s in strategies]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if not isinstance(r, Exception) else {"error": str(r)}
for r in results
]
def get_cost_summary(self) -> Dict:
"""비용 요약 반환"""
return {
"total_tokens": self.total_tokens_used,
"total_cost_usd": round(self.total_cost_cents, 4),
"total_cost_krw": round(self.total_cost_cents * 1350, 0), # 환율 1350원
"avg_cost_per_request": round(
self.total_cost_cents / max(1, self.total_tokens_used / 500),
4
)
}
사용 예시
async def main():
analyzer = HolySheepStrategyAnalyzer(HOLYSHEEP_API_KEY)
# 샘플 OHLCV 데이터 (실제로는 DB에서 로드)
sample_data = [
{"timestamp": 1704067200, "open": 42000, "high": 42500, "low": 41800, "close": 42350, "volume": 1500},
# ... 실제 데이터
] * 100
# 1단계: 패턴 탐색 (저비용)
pattern_result = await analyzer.analyze_pattern(sample_data)
print(f"패턴 분석 (DeepSeek): ${pattern_result['cost_usd']:.4f}")
# 2단계: 전략 검증 (고비용)
strategy = {
"description": "RSI 오버솔드 구간 매수, 2% 수익시 청산",
"performance": {
"total_trades": 156,
"win_rate": 63.5,
"avg_return": 1.8,
"max_drawdown": 8.2,
"sharpe_ratio": 1.45,
"avg_holding_hours": 4.2
}
}
validation = await analyzer.validate_strategy(
strategy["description"],
strategy["performance"]
)
print(f"전략 검증 (GPT-4.1): ${validation['cost_usd']:.4f}")
# 비용 요약
cost_summary = analyzer.get_cost_summary()
print(f"\n비용 요약:")
print(f"- 총 토큰: {cost_summary['total_tokens']:,}")
print(f"- 총 비용: ${cost_summary['total_cost_usd']:.4f} (₩{cost_summary['total_cost_krw']:,.0f})")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
4. 백테스팅 엔진 구현
AI가 분석한 전략을 Historical Data로 검증하는 벡터라이즈 백테스터를 구현합니다. Pandas의 벡터라이즈 연산으로 수년간의 데이터를 수 초 내 분석합니다.
# backtester.py
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Callable, Optional
from dataclasses import dataclass, field
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
@dataclass
class Trade:
"""거래 기록"""
entry_time: datetime
exit_time: datetime
entry_price: float
exit_price: float
size: float
side: str # "long" or "short"
pnl: float
pnl_pct: float
holding_hours: float
@dataclass
class BacktestResult:
"""백테스트 결과"""
total_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
win_rate: float = 0.0
avg_win: float = 0.0
avg_loss: float = 0.0
profit_factor: float = 0.0
total_return: float = 0.0
max_drawdown: float = 0.0
sharpe_ratio: float = 0.0
sortino_ratio: float = 0.0
avg_holding_hours: float = 0.0
trades: List[Trade] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
"total_trades": self.total_trades,
"win_rate": round(self.win_rate * 100, 1),
"avg_return": round(self.avg_win, 2),
"max_drawdown": round(self.max_drawdown * 100, 1),
"sharpe_ratio": round(self.sharpe_ratio, 2),
"avg_holding_hours": round(self.avg_holding_hours, 1)
}
class VectorizedBacktester:
"""벡터라이즈 백테스터 - Pandas 벡터 연산으로 고속 백테스팅"""
def __init__(self, initial_capital: float = 10000.0):
self.initial_capital = initial_capital
self.current_capital = initial_capital
def load_data(
self,
ohlcv_df: pd.DataFrame,
orderbook_df: Optional[pd.DataFrame] = None
) -> "VectorizedBacktester":
"""데이터 로드 및 전처리"""
# 필수 컬럼 확인
required = ["timestamp", "open", "high", "low", "close", "volume"]
for col in required:
if col not in ohlcv_df.columns:
raise ValueError(f"Missing required column: {col}")
self.df = ohlcv_df.copy()
self.df["timestamp"] = pd.to_datetime(self.df["timestamp"], unit="ms")
self.df = self.df.sort_values("timestamp").reset_index(drop=True)
# 기술적 지표 사전 계산
self._calculate_indicators()
return self
def _calculate_indicators(self):
"""기술적 지표 계산"""
# 이동평균선
for period in [5, 10, 20, 50, 200]:
self.df[f"sma_{period}"] = self.df["close"].rolling(period).mean()
self.df[f"ema_{period}"] = self.df["close"].ewm(span=period).mean()
# RSI
delta = self.df["close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
self.df["rsi"] = 100 - (100 / (1 + rs))
# MACD
exp1 = self.df["close"].ewm(span=12).mean()
exp2 = self.df["close"].ewm(span=26).mean()
self.df["macd"] = exp1 - exp2
self.df["macd_signal"] = self.df["macd"].ewm(span=9).mean()
self.df["macd_hist"] = self.df["macd"] - self.df["macd_signal"]
# ATR (Average True Range)
high_low = self.df["high"] - self.df["low"]
high_close = np.abs(self.df["high"] - self.df["close"].shift())
low_close = np.abs(self.df["low"] - self.df["close"].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
self.df["atr"] = tr.rolling(14).mean()
# 볼린저밴드
self.df["bb_middle"] = self.df["close"].rolling(20).mean()
std = self.df["close"].rolling(20).std()
self.df["bb_upper"] = self.df["bb_middle"] + 2 * std
self.df["bb_lower"] = self.df["bb_middle"] - 2 * std
# 거래량 지표
self.df["volume_sma"] = self.df["volume"].rolling(20).mean()
self.df["volume_ratio"] = self.df["volume"] / self.df["volume_sma"]
def run_strategy(
self,
strategy_func: Callable[[pd.DataFrame, int], Dict],
stop_loss_pct: float = 0.02,
take_profit_pct: float = 0.03,
position_size_pct: float = 0.1
) -> BacktestResult:
"""벡터라이즈 전략 실행"""
df = self.df.dropna().copy()
n = len(df)
# 벡터라이즈된 신호 생성
signals = []
for i in range(n):
if i < 50: # 지표 계산 대기 기간
signals.append({"action": "hold", "confidence": 0})
continue
signal = strategy_func(df, i)
signals.append(signal)
df["signal"] = [s["action"] for s in signals]
df["confidence"] = [s["confidence"] for s in signals]
# 신호 정규화
df["position"] = df["signal"].map({"buy": 1, "sell": -1, "hold": 0})
df["position"] = df["position"].ffill().fillna(0)
# 벡터라이즈 수익률 계산
df["strategy_return"]