저는 3년 넘게 알트코인 시그널 봇을 개발하며 수천만 건의 오더북 데이터를 다뤄왔습니다. 최근 클레이머스 거래소 롤백 사건 이후 많은 개발자들이 자체 백테스팅 환경의 중요성을 절감하고 계실 겁니다. 오늘은 Binance와 OKX의 히스토리컬 L2(오더북) 데이터를 안정적으로 확보하고, HolySheep AI 게이트웨이를 통해 AI 모델과 연계하여 고빈도 백테스팅 파이프라인을 구축하는 방법을 상세히 안내드리겠습니다.
왜 L2 데이터인가: 시그널 봇 개발자의 관점
틱 데이터(Trade)와 호가창 데이터(Level 2)는 완전히 다른 세계입니다. 제가 처음 시그널 봇을 개발했을 때, OHLCV 데이터만으로 충분하다고 생각했습니다. 하지만 실제 거래에서는:
- 슬리피지 계산: 매수호가와 매도호가 사이 간격(Bid-Ask Spread)이 0.1%인지 0.5%인지에 따라 수익률이 완전히 달라집니다
- 유동성 분석: 대형 호가창 뒤에 숨겨진 주문량을 파악해야 정확한进场 시점을 잡을 수 있습니다
- 시장 미세 구조: 주문 유입 속도와 취소 패턴을 분석하면 대형 참여자의 의도를 파악할 수 있습니다
특히 저와 같이 1분 이하 단위로 거래하는 고빈도 시그널 봇 개발자에게, L2 데이터 없이는 백테스팅 결과가 실전과 40% 이상 차이나는 경우가 빈번했습니다.
주요 데이터 소스 비교
| 공급자 | 데이터 해상도 | 보관 기간 | API 지연 | 월간 비용 | 한국어 지원 |
|---|---|---|---|---|---|
| Binance 직접 API | 100ms 스냅샷 | 최대 7일 | ~50ms | 무료(제한) | 미흡 |
| OKX 공식 | 50ms 스냅샷 | 최대 3일 | ~80ms | 무료(제한) | 없음 |
| Nexus Protocol | 10ms 스냅샷 | 5년 | ~20ms | $299/월 | 없음 |
| Ludvig Data | 25ms 스냅샷 | 3년 | ~30ms | $199/월 | 없음 |
| HolySheep AI 게이트웨이 | 다중 모델 통합 | 비용 최적화 | 다중 프로바이더 | $8~15/MTok | 완벽 |
데이터 확보 아키텍처
1단계: Binance L2 데이터 수집
#!/usr/bin/env python3
"""
Binance 스냅샷 기반 L2 데이터 수집기
HolySheep AI 게이트웨이 연동을 위한 기초 데이터 파이프라인
"""
import asyncio
import aiohttp
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import hmac
import hashlib
class BinanceL2Collector:
"""Binance 오더북 L2 데이터 수집기"""
BASE_URL = "https://api.binance.com"
def __init__(self, api_key: str = None, api_secret: str = None):
self.api_key = api_key
self.api_secret = api_secret
self.session = None
async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict:
"""
Binance 호가창 스냅샷 조회
Rate Limit: 1200 요청/분 (무인증), 120 요청/분 (인증)
"""
endpoint = "/api/v3/depth"
params = {
"symbol": symbol.upper(),
"limit": limit
}
url = f"{self.BASE_URL}{endpoint}"
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return {
"exchange": "binance",
"symbol": symbol.upper(),
"timestamp": int(time.time() * 1000),
"datetime": datetime.utcnow().isoformat(),
"bids": [[float(p), float(q)] for p, q in data.get("bids", [])],
"asks": [[float(p), float(q)] for p, q in data.get("asks", [])],
"lastUpdateId": data.get("lastUpdateId")
}
else:
raise Exception(f"Binance API Error: {response.status}")
async def stream_order_book(self, symbol: str, duration_seconds: int = 60):
"""
Binance WebSocket 실시간 L2 스트리밍
실제 백테스팅용 히스토리컬 데이터는 별도 수집 필요
"""
stream_name = f"{symbol.lower()}@depth@100ms"
ws_url = f"wss://stream.binance.com:9443/ws/{stream_name}"
print(f"연결 중: {ws_url}")
async with aiohttp.ClientSession() as session:
async with session.ws_connect(ws_url) as ws:
start_time = time.time()
messages = []
while time.time() - start_time < duration_seconds:
msg = await ws.receive_json()
messages.append({
"timestamp": msg.get("E"),
"symbol": msg.get("s"),
"bid_depth": msg.get("b"), # [price, quantity]
"ask_depth": msg.get("a")
})
return messages
사용 예제
async def main():
collector = BinanceL2Collector()
# BTCUSDT 스냅샷 조회
snapshot = await collector.get_order_book_snapshot("BTCUSDT", limit=20)
print(f"거래소: {snapshot['exchange']}")
print(f"심볼: {snapshot['symbol']}")
print(f"호가 시간: {snapshot['datetime']}")
print(f"매수호가 (상위 3건):")
for bid in snapshot['bids'][:3]:
print(f" {bid[0]} USDT | 수량: {bid[1]}")
print(f"매도호가 (상위 3건):")
for ask in snapshot['asks'][:3]:
print(f" {ask[0]} USDT | 수량: {ask[1]}")
if __name__ == "__main__":
asyncio.run(main())
2단계: OKX L2 데이터 수집
#!/usr/bin/env python3
"""
OKX 오더북 L2 데이터 수집기
Binance과 병행 사용 시 시장 전체 유동성 분석 가능
"""
import requests
import time
import hmac
import base64
import hashlib
from typing import Dict, List
from datetime import datetime
class OKXL2Collector:
"""OKX 거래소 호가창 데이터 수집기"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = None, api_secret: str = None, passphrase: str = None):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""OKX HMAC-SHA256 서명 생성"""
message = timestamp + method + path + body
mac = hmac.new(
self.api_secret.encode("utf-8"),
message.encode("utf-8"),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode("utf-8")
def get_order_book(self, inst_id: str, sz: int = 20) -> Dict:
"""
OKX 호가창 조회
inst_id 예시: BTC-USDT-SWAP ( Perpetual Swap )
"""
endpoint = "/api/v5/market/books"
params = {
"instId": inst_id,
"sz": sz # 최대 400
}
headers = {}
if self.api_key:
timestamp = datetime.utcnow().isoformat() + "Z"
headers = {
"OKX-API-KEY": self.api_key,
"OKX-TIMESTAMP": timestamp,
"OKX-SIGN": self._sign(timestamp, "GET", f"{endpoint}?instId={inst_id}&sz={sz}"),
"OKX-PASSPHRASE": self.passphrase
}
url = f"{self.BASE_URL}{endpoint}"
response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
data = response.json()
if data.get("code") == "0":
books = data.get("data", [{}])[0]
return {
"exchange": "okx",
"symbol": inst_id,
"timestamp": int(time.time() * 1000),
"datetime": datetime.utcnow().isoformat(),
"bids": [[float(books["bids"][i]), float(books["bids"][i+1])]
for i in range(0, min(len(books["bids"]), 40), 4)],
"asks": [[float(books["asks"][i]), float(books["asks"][i+1])]
for i in range(0, min(len(books["asks"]), 40), 4)],
"ts": books.get("ts"),
"checksum": books.get("checksum")
}
else:
raise Exception(f"OKX API Error: {data.get('msg')}")
else:
raise Exception(f"HTTP Error: {response.status_code}")
def get_historical_candles(self, inst_id: str, bar: str = "1m",
after: int = None, before: int = None, limit: int = 100) -> List[Dict]:
"""
OKX Historical Candlestick (과거 데이터용)
1m, 3m, 5m, 15m, 1H, 2H, 4H, 6H, 12H, 1D, 2D, 3D, 1W, 1M, 2M, 3M, 6M, 1Y
"""
endpoint = "/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": min(limit, 100)
}
if after:
params["after"] = after
if before:
params["before"] = before
url = f"{self.BASE_URL}{endpoint}"
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
if data.get("code") == "0":
return [
{
"timestamp": int(candle[0]),
"open": float(candle[1]),
"high": float(candle[2]),
"low": float(candle[3]),
"close": float(candle[4]),
"volume": float(candle[5]),
"quote_volume": float(candle[7]) if len(candle) > 7 else 0
}
for candle in data.get("data", [])
]
return []
사용 예제
def main():
collector = OKXL2Collector()
# 현물 호가창
snapshot = collector.get_order_book("BTC-USDT", sz=20)
print(f"OKX BTC-USDT 호가창")
print(f"매수호가: {snapshot['bids'][:3]}")
print(f"매도호가: {snapshot['asks'][:3]}")
# Perpetual Swap 호가창 (고빈도 거래에 주로 사용)
perp_snapshot = collector.get_order_book("BTC-USDT-SWAP", sz=100)
print(f"OKX BTC-USDT-SWAP 호가창 (최상위 5건):")
for ask in perp_snapshot['asks'][:5]:
print(f" 매도 {ask[0]} | 수량: {ask[1]:.4f}")
if __name__ == "__main__":
main()
3단계: AI 기반 시그널 분석 파이프라인
수집된 L2 데이터를 HolySheep AI 게이트웨이를 통해 AI 모델로 분석하면, 전통적인 기술적 지표만으로는 포착하기 어려운 시장 패턴을 감지할 수 있습니다. 저는 최근 Gemini 2.5 Flash를 활용하여 유동성 쇼크 패턴을 자동 탐지하는 시스템을 구축했습니다.
#!/usr/bin/env python3
"""
HolySheep AI 게이트웨이 연동을 통한 L2 데이터 AI 분석
GPT-4.1, Claude Sonnet 4, Gemini 2.5 Flash 다중 모델 활용
"""
import os
import json
import httpx
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
HolySheep AI 설정 (반드시 공식 엔드포인트 사용)
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
@dataclass
class L2AnalysisResult:
"""AI 분석 결과 데이터 클래스"""
symbol: str
exchange: str
signal_type: str # bullish, bearish, neutral
confidence: float
reasoning: str
recommended_action: str
risk_level: str
timestamp: str
class L2SignalAnalyzer:
"""HolySheep AI 기반 L2 시그널 분석기"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def calculate_l2_features(self, order_book: Dict) -> Dict:
"""L2 데이터에서 특징값 계산"""
bids = order_book.get("bids", [])
asks = order_book.get("asks", [])
if not bids or not asks:
return {}
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price * 100
bid_volume = sum(float(b[1]) for b in bids[:10])
ask_volume = sum(float(a[1]) for a in asks[:10])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume) * 100
# VWAP 근접도
total_volume = bid_volume + ask_volume
weighted_price = (
sum(float(b[0]) * float(b[1]) for b in bids[:10]) +
sum(float(a[0]) * float(a[1]) for a in asks[:10])
) / (total_volume if total_volume > 0 else 1)
return {
"spread_bps": round(spread * 100, 2), # basis points
"bid_volume_10": round(bid_volume, 4),
"ask_volume_10": round(ask_volume, 4),
"volume_imbalance": round(imbalance, 2),
"mid_price": round(mid_price, 4),
"vwap": round(weighted_price, 4),
"bid_ask_ratio": round(bid_volume / ask_volume if ask_volume > 0 else 1, 4)
}
async def analyze_with_gpt41(self, order_book: Dict, features: Dict) -> L2AnalysisResult:
"""
GPT-4.1을 사용한 상세 분석
비용: $8/MTok (HolySheep 공식 가격)
"""
prompt = f"""당신은 고빈도 거래 전문가입니다. 아래 L2 호가창 데이터를 분석해주세요.
거래소: {order_book.get('exchange')}
심볼: {order_book.get('symbol')}
시간: {order_book.get('datetime')}
호가창 특징:
- 스프레드: {features['spread_bps']}bps
- 매수호가량 (상위10): {features['bid_volume_10']}
- 매도호가량 (상위10): {features['ask_volume_10']}
- 거래량 불균형: {features['volume_imbalance']}%
- 중가격: ${features['mid_price']}
- VWAP: ${features['vwap']}
- Bid/Ask 비율: {features['bid_ask_ratio']}
상위 매수호가 5건:
{chr(10).join([f'{i+1}. ${b[0]} | 수량: {b[1]}' for i, b in enumerate(order_book.get('bids', [])[:5])])}
상위 매도호가 5건:
{chr(10).join([f'{i+1}. ${a[0]} | 수량: {a[1]}' for i, a in enumerate(order_book.get('asks', [])[:5])])}
JSON으로 응답해주세요:
{{"signal_type": "bullish/bearish/neutral", "confidence": 0.0~1.0,
"reasoning": "분석 근거", "recommended_action": "진입/관망/청산",
"risk_level": "low/medium/high"}}
"""
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 500
}
)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
try:
analysis = json.loads(content)
return L2AnalysisResult(
symbol=order_book.get("symbol"),
exchange=order_book.get("exchange"),
**analysis,
timestamp=order_book.get("datetime")
)
except json.JSONDecodeError:
return L2AnalysisResult(
symbol=order_book.get("symbol"),
exchange=order_book.get("exchange"),
signal_type="neutral",
confidence=0.5,
reasoning="파싱 오류",
recommended_action="관망",
risk_level="medium",
timestamp=order_book.get("datetime")
)
else:
raise Exception(f"API Error: {response.status_code}")
async def analyze_with_gemini_flash(self, order_book: Dict, features: Dict) -> L2AnalysisResult:
"""
Gemini 2.5 Flash를 사용한 고속 분석
비용: $2.50/MTok (HolySheep 공식 가격) - GPT-4.1 대비 3배 저렴
"""
prompt = f"""L2 오더북 분석:
{order_book.get('exchange')} {order_book.get('symbol')}
스프레드: {features['spread_bps']}bps | 불균형: {features['volume_imbalance']}%
VWAP 대비 중가격: {'매수우위' if features['mid_price'] > features['vwap'] else '매도우위'}
신호 분석 결과를 JSON으로:
{{"signal_type": "short", "confidence": 0.7, "reasoning": "VWAP 하회 및 매도호가량 우세",
"recommended_action": "공격적 숏", "risk_level": "high"}}
"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 300
}
)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
try:
analysis = json.loads(content)
return L2AnalysisResult(
symbol=order_book.get("symbol"),
exchange=order_book.get("exchange"),
**analysis,
timestamp=order_book.get("datetime")
)
except json.JSONDecodeError:
return L2AnalysisResult(
symbol=order_book.get("symbol"),
exchange=order_book.get("exchange"),
signal_type="neutral",
confidence=0.5,
reasoning="파싱 오류",
recommended_action="관망",
risk_level="medium",
timestamp=order_book.get("datetime")
)
else:
raise Exception(f"API Error: {response.status_code}")
통합 분석 시스템
class MultiExchangeSignalSystem:
"""Binance + OKX 다중 거래소 통합 시그널 시스템"""
def __init__(self, holysheep_api_key: str):
self.binance_collector = None # 앞서 정의한 BinanceL2Collector
self.okx_collector = None # 앞서 정의한 OKXL2Collector
self.analyzer = L2SignalAnalyzer(holysheep_api_key)
async def get_cross_exchange_signals(self, symbol: str) -> Dict:
"""
Binance와 OKX 양쪽에서 L2 데이터를 가져와
크로스 익스체인지 시그널 생성
"""
# 양쪽 거래소에서 동시 수집
binance_snapshot = await self.binance_collector.get_order_book_snapshot(
symbol=f"{symbol}USDT", limit=20
)
okx_snapshot = self.okx_collector.get_order_book(
inst_id=f"{symbol}-USDT", sz=20
)
# 특징값 계산
binance_features = self.analyzer.calculate_l2_features(binance_snapshot)
okx_features = self.analyzer.calculate_l2_features(okx_snapshot)
# HolySheep AI로 병렬 분석
gpt_task = self.analyzer.analyze_with_gpt41(binance_snapshot, binance_features)
gemini_task = self.analyzer.analyze_with_gemini_flash(okx_snapshot, okx_features)
gpt_result, gemini_result = await asyncio.gather(gpt_task, gemini_task)
# 앙상블 시그널 (가중 평균)
ensemble_confidence = (
gpt_result.confidence * 0.6 + # GPT-4.1에 높은 가중치
gemini_result.confidence * 0.4
)
# 신호 방향 정렬 확인
signal_alignment = (
gpt_result.signal_type == gemini_result.signal_type
)
return {
"symbol": symbol,
"timestamp": datetime.utcnow().isoformat(),
"binance_analysis": gpt_result,
"okx_analysis": gemini_result,
"ensemble": {
"confidence": round(ensemble_confidence, 3),
"signal_alignment": signal_alignment,
"final_signal": gpt_result.signal_type if signal_alignment else "neutral",
"risk_boost": 1.2 if signal_alignment else 0.8
}
}
사용 예제
async def demo():
# HolySheep API 키 설정
analyzer = L2SignalAnalyzer()
# 샘플 L2 데이터
sample_order_book = {
"exchange": "binance",
"symbol": "BTCUSDT",
"datetime": datetime.utcnow().isoformat(),
"bids": [
[67450.00, 2.5],
[67448.50, 1.8],
[67447.00, 3.2],
[67445.50, 5.0],
[67444.00, 8.5]
],
"asks": [
[67451.00, 3.1],
[67452.50, 2.2],
[67454.00, 4.0],
[67456.00, 6.5],
[67458.50, 10.2]
]
}
# 특징값 계산
features = analyzer.calculate_l2_features(sample_order_book)
print(f"L2 특징값: {features}")
# GPT-4.1 분석 (상세)
# gpt_result = await analyzer.analyze_with_gpt41(sample_order_book, features)
# print(f"GPT-4.1 신호: {gpt_result.signal_type} ({gpt_result.confidence})")
# Gemini 2.5 Flash 분석 (고속)
# gemini_result = await analyzer.analyze_with_gemini_flash(sample_order_book, features)
# print(f"Gemini 신호: {gemini_result.signal_type} ({gemini_result.confidence})")
if __name__ == "__main__":
import asyncio
asyncio.run(demo())
고빈도 백테스팅 시스템 구축
#!/usr/bin/env python3
"""
고빈도 백테스팅 엔진
Binance/OKX L2 데이터를 활용한 시뮬레이션 거래
"""
import sqlite3
import json
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
from dataclasses import dataclass, field
from collections import deque
@dataclass
class BacktestTrade:
"""백테스트 거래 기록"""
timestamp: int
signal: str
entry_price: float
exit_price: float
size: float
pnl: float
pnl_pct: float
holding_ms: int
@dataclass
class BacktestConfig:
"""백테스트 설정"""
symbol: str
start_date: datetime
end_date: datetime
initial_balance: float = 10000.0
commission_rate: float = 0.0004 # Binance Taker Fee
slippage_bps: float = 1.0 # 1 basis point
position_size_pct: float = 0.1 # 잔고의 10%
@dataclass
class BacktestResult:
"""백테스트 결과"""
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
total_pnl_pct: float
max_drawdown: float
sharpe_ratio: float
avg_trade_duration_ms: int
trades: List[BacktestTrade] = field(default_factory=list)
class HighFrequencyBacktester:
"""고빈도 백테스팅 엔진"""
def __init__(self, config: BacktestConfig):
self.config = config
self.balance = config.initial_balance
self.position = None
self.trades: List[BacktestTrade] = []
self.equity_curve = []
self.max_equity = config.initial_balance
def execute_signal(self, timestamp: int, price: float, signal: str, l2_data: Dict = None):
"""
시그널에 따른 거래 실행
signal: "long", "short", "close"
"""
slippage = price * (self.config.slippage_bps / 10000)
if signal == "long" and self.position is None:
# 롱 진입
size = (self.balance * self.config.position_size_pct) / (price + slippage)
cost = size * (price + slippage)
commission = cost * self.config.commission_rate
self.balance -= (cost + commission)
self.position = {
"type": "long",
"entry_price": price + slippage,
"size": size,
"entry_time": timestamp,
"entry_cost": cost + commission
}
elif signal == "short" and self.position is None:
# 숏 진입
size = (self.balance * self.config.position_size_pct) / (price - slippage)
cost = size * (price - slippage)
commission = cost * self.config.commission_rate
self.balance += (cost - commission)
self.position = {
"type": "short",
"entry_price": price - slippage,
"size": size,
"entry_time": timestamp,
"entry_cost": cost - commission
}
elif signal == "close" and self.position is not None:
# 포지션 청산
pos = self.position
if pos["type"] == "long":
pnl = (price - slippage - pos["entry_price"]) * pos["size"]
pnl -= pnl * self.config.commission_rate
else:
pnl = (pos["entry_price"] - price + slippage) * pos["size"]
pnl -= pnl * self.config.commission_rate
self.balance += pos["entry_cost"] + pnl
holding_ms = timestamp - pos["entry_time"]
trade = BacktestTrade(
timestamp=timestamp,
signal=pos["type"],
entry_price=pos["entry_price"],
exit_price=price,
size=pos["size"],
pnl=round(pnl, 2),
pnl_pct=round(pnl / pos["entry_cost"] * 100, 4),
holding_ms=holding_ms
)
self.trades.append(trade)
self.position = None
def calculate_metrics(self) -> BacktestResult:
"""성과 지표 계산"""
if not self.trades:
return BacktestResult(
total_trades=0, winning_trades=0, losing_trades=0,
win_rate=0, total_pnl=0, total_pnl_pct=0,
max_drawdown=0, sharpe_ratio=0, avg_trade_duration_ms=0
)
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
total_pnl = sum(t.pnl for t in self.trades)
total_pnl_pct = (self.balance - self.config.initial_balance) / self.config.initial_balance * 100
#Equity Curve 기반 MDD 계산
equity = self.config.initial_balance
peak = equity
max_dd = 0
for trade in self.trades:
equity += trade.pnl
if equity > peak:
peak = equity
dd = (peak - equity) / peak * 100
max_dd = max(max_dd, dd)
#샤프 비율 (단순화)
if len(winning_trades) > 1:
returns = [t.pnl_pct / 100 for t in self.trades]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252 * 24 * 60) if np.std(returns) > 0 else 0
else:
sharpe = 0
avg_duration = int(np.mean([t.holding_ms for t in self.trades]))
return BacktestResult(
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
win_rate=len(winning_trades) / len(self.trades) * 100,
total_pnl=round(total_pnl, 2),
total_pnl_pct=round(total_pnl_pct, 2),
max_drawdown=round(max_dd, 2),
sharpe_ratio=round(sharpe, 3),
avg_trade_duration_ms=avg_duration,
trades=self.trades
)
def run_from_sqlite(self, db_path: str, strategy_func) -> BacktestResult:
"""
SQLite에 저장된 L2 데이터로 백테스트 실행
strategy_func: (l2_data, features) -> signal
"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# L2 데이터 조회 (테이블 구조에 맞게 조정)
cursor.execute("""
SELECT timestamp, bids, asks, mid_price
FROM l2_data
WHERE symbol = ? AND timestamp BETWEEN ? AND ?
ORDER BY timestamp ASC
""", (
self.config.symbol,
int(self.config.start_date.timestamp() * 1000),
int(self.config.end_date.timestamp() * 1000)
))
rows = cursor.fetchall()
conn.close()
for row in rows:
timestamp, bids_json, asks_json, mid_price = row
bids = json.loads(bids_json)
asks = json.loads(asks_json)
l2_data = {
"timestamp": timestamp,
"bids": bids,
"asks": asks,
"mid_price": mid_price
}
# 시그널 생성
signal = strategy_func(l2_data)
# 거래 실행
self.execute_signal(timestamp, mid_price, signal, l2_data)
# 에쿼티 갱신
if self.position:
equity = self.balance - self.position["entry_cost"] + \
(mid_price - self.position["entry_price"]) * self.position["size"]
else:
equity = self.balance
self.equity_curve.append(equity)
return self.calculate_metrics()
샘플 전략: 유동성 불균형 기반
def liquidity_imbalance_strategy(l2_data: Dict) -> str:
"""호가창 유동성 불균형을 활용한 전략"""
bids = l2_data.get("bids", [])
asks = l2_data.get("asks", [])
bid_vol = sum(float(b[1]) for b in bids[:5])
ask_vol = sum(float(a[1]) for a in asks[:5])
imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol)
if imbalance > 0.2:
return "long"
elif imbalance < -0.2:
return "short"
else:
return "close"
실행 예제
def run_sample_backtest():
config = BacktestConfig(
symbol="BTCUSDT",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 31),
initial_balance=10000.0,
position_size_pct=0.1
)
backtester = HighFrequencyBacktester(config)
# db_path에 실제 L2 데이터 저장소 경로 입력
# result = backtester.run