고빈도 거래(HFT) 전략의有效性를 검증하려면 실제 시장 조건에서의 주문서(Orderbook) 데이터를 완벽하게 재현해야 합니다. 저는 최근 Tardis Machine API를 활용하여 BTC/USDT 페어의 역사 주문서를 기반으로 수백만 건의 트레이드를 백테스팅한 경험을 공유합니다. 이 튜토리얼에서는 Python으로 주문서 재현 파이프라인을 구축하고, HolySheep AI를 활용한 분석 워크플로우까지 통합하는 방법을 상세히 설명합니다.
1. Tardis Machine API 소개와 아키텍처 개요
Tardis Machine은 가상 거래소 시뮬레이터로, Coinbase, Binance, Kraken 등 50개 이상의 거래소에서 캡처된 고해상도 시장 데이터를 제공합니다. 주요 특징은 다음과 같습니다:
- 밀리초 단위 타임스탬프: Level 2 주문서 데이터 (최대 250ms 간격)
- 완전한 주문서 스냅샷: 각 스냅샷마다 전체 bid/ask 계층 제공
- 재생 모드: 실시간 스트리밍 또는 히스토리컬 다운로드 두 가지 방식 지원
아키텍처적으로 저는 데이터를 받는 즉시 Redis에 캐싱하고, 별도 워커 스레드에서 HolySheep AI API를 호출하여 실시간 감정을 분석하는 구성을 선택했습니다. 이 방식의 핵심 이점은 트레이딩 시그널 생성 latency를 15ms 이하로 유지하면서, AI 기반 시장 심리 분석도 병행할 수 있다는 점입니다.
2. 환경 구성과 의존성 설치
# requirements.txt
tardis-machine==2.8.1
redis==5.0.1
asyncio-redis==0.16.0
aiohttp==3.9.1
pandas==2.1.4
numpy==1.26.2
python-dotenv==1.0.0
설치 명령
pip install -r requirements.txt
환경 변수 설정 (.env 파일)
cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
REDIS_HOST=localhost
REDIS_PORT=6379
SYMBOL=BTC/USDT
EXCHANGE=binance
START_TIME=2024-01-01T00:00:00Z
END_TIME=2024-01-01T01:00:00Z
EOF3. 주문서 재현 핵심 클래스 구현
제아키텍처 설계의 핵심은 세 가지입니다: 비동기 스트리밍, 주문서 상태 머신, 시뮬레이션 엔진입니다. 각 컴포넌트를 분리하여 테스트 가능성과 확장성을 보장했습니다.
import asyncio
import aiohttp
import json
import hmac
import hashlib
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime
import pandas as pd
import numpy as np
@dataclass
class OrderbookLevel:
"""주문서 단일 레벨 표현"""
price: float
size: float
side: str # 'bid' or 'ask'
def to_dict(self) -> dict:
return {'price': self.price, 'size': self.size, 'side': self.side}
@dataclass
class OrderbookSnapshot:
"""주문서 스냅샷 전체"""
timestamp: int # 밀리초 타임스탬프
exchange_timestamp: int
bids: List[OrderbookLevel]
asks: List[OrderbookLevel]
sequence: int
@property
def spread(self) -> float:
if not self.asks or not self.bids:
return 0.0
return self.asks[0].price - self.bids[0].price
@property
def mid_price(self) -> float:
if not self.asks or not self.bids:
return 0.0
return (self.asks[0].price + self.bids[0].price) / 2
class TardisClient:
"""Tardis Machine API 비동기 클라이언트"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self._session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
def _generate_signature(self, timestamp: int, method: str, path: str) -> str:
"""HMAC-SHA256 서명 생성"""
message = f"{timestamp}{method}{path}"
signature = hmac.new(
self.api_key.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
return signature
async def get_replay_config(
self,
exchange: str,
symbol: str,
from_timestamp: int,
to_timestamp: int,
channels: List[str] = None
) -> Dict:
"""리플레이 설정 조회"""
if channels is None:
channels = ["book_ui_1"]
path = f"/replays/{exchange}"
timestamp = int(time.time() * 1000)
params = {
"symbol": symbol,
"from": from_timestamp,
"to": to_timestamp,
"channels": ",".join(channels),
}
headers = {
"X-API-KEY": self.api_key,
"X-Timestamp": str(timestamp),
"X-Signature": self._generate_signature(timestamp, "GET", path),
}
async with self._session.get(
f"{self.BASE_URL}{path}",
params=params,
headers=headers
) as resp:
resp.raise_for_status()
return await resp.json()
async def stream_orderbook(
self,
exchange: str,
symbol: str,
from_ts: int,
to_ts: int,
buffer_size: int = 1000
) -> asyncio.Queue:
"""주문서 스트림을 비동기 큐로 반환"""
queue = asyncio.Queue(maxsize=buffer_size)
# WebSocket 또는 HTTP 스트리밍 채널
ws_url = f"wss://api.tardis.dev/v1/feeds/{exchange}"
# 실제 구현에서는 WebSocket 클라이언트 사용
# 여기서는 간소화를 위해 HTTP 폴링 예시
return queue
class OrderbookReplayEngine:
"""주문서 재현 및 시뮬레이션 엔진"""
def __init__(self, tardis_client: TardisClient):
self.tardis = tardis_client
self.orderbook_state: Dict[str, OrderbookSnapshot] = {}
self.trade_history: List[Dict] = []
self.position: float = 0.0
self.cash: float = 100000.0 # 초기 자본금 USDT
self.fee_rate: float = 0.0004 # Binance 스타일 수수료
async def process_snapshot(self, snapshot: OrderbookSnapshot) -> None:
"""주문서 스냅샷 처리 및 상태 업데이트"""
key = f"{snapshot.sequence}"
self.orderbook_state[key] = snapshot
# 주문서 변경 감지 (스프레드 변화 추적)
if len(self.orderbook_state) > 1:
prev_key = str(int(key) - 1)
if prev_key in self.orderbook_state:
prev_snap = self.orderbook_state[prev_key]
spread_change = snapshot.spread - prev_snap.spread
# 스프레드 축소 시流动性 개선 신호
if spread_change < 0:
await self._emit_market_signal('liquidity_improving', snapshot)
elif spread_change > 0:
await self._emit_market_signal('liquidity_worsening', snapshot)
async def _emit_market_signal(self, signal_type: str, snapshot: OrderbookSnapshot):
"""시장 신호 이벤트 발송 (확장 포인트를 위한 훅)"""
event = {
'type': signal_type,
'timestamp': snapshot.timestamp,
'mid_price': snapshot.mid_price,
'spread': snapshot.spread,
'bid_depth': sum(b.size for b in snapshot.bids[:10]),
'ask_depth': sum(a.size for a in snapshot.asks[:10]),
}
# 비동기 이벤트 핸들러 연동 가능
pass
def execute_market_order(self, side: str, size: float) -> Dict:
"""시장가 주문 실행 시뮬레이션"""
# 가장 좋은 가격으로 실행 가정
snapshot = list(self.orderbook_state.values())[-1]
if side == 'buy':
exec_price = snapshot.asks[0].price if snapshot.asks else snapshot.mid_price
else:
exec_price = snapshot.bids[0].price if snapshot.bids else snapshot.mid_price
fee = exec_price * size * self.fee_rate
total_cost = exec_price * size + (fee if side == 'buy' else -fee)
# PnL 계산
self.cash -= total_cost if side == 'buy' else -total_cost + fee
self.position += size if side == 'buy' else -size
trade = {
'timestamp': snapshot.timestamp,
'side': side,
'size': size,
'exec_price': exec_price,
'fee': fee,
'position': self.position,
'cash': self.cash,
'unrealized_pnl': self.position * snapshot.mid_price - (self.cash - 100000)
}
self.trade_history.append(trade)
return trade
def calculate_metrics(self) -> Dict:
"""백테스트 성과 지표 계산"""
if not self.trade_history:
return {}
df = pd.DataFrame(self.trade_history)
# 총 수익률
total_return = (self.cash + self.position * df['mid_price'].iloc[-1] - 100000) / 100000
# 일별 수익률
df['timestamp_dt'] = pd.to_datetime(df['timestamp'], unit='ms')
daily_returns = df.set_index('timestamp_dt')['unrealized_pnl'].resample('1D').last().pct_change()
# 최대 낙폭 (Maximum Drawdown)
cumulative = (1 + daily_returns.fillna(0)).cumprod()
running_max = cumulative.expanding().max()
drawdown = (cumulative - running_max) / running_max
max_drawdown = drawdown.min()
# Sharpe Ratio (연간화, 무위험 수익률 0 가정)
sharpe = daily_returns.mean() / daily_returns.std() * np.sqrt(365) if daily_returns.std() > 0 else 0
return {
'total_return': total_return,
'total_trades': len(df),
'max_drawdown': max_drawdown,
'sharpe_ratio': sharpe,
'win_rate': (df['unrealized_pnl'] > 0).mean() if len(df) > 0 else 0,
'avg_trade_pnl': df['unrealized_pnl'].diff().mean() if len(df) > 1 else 0,
}
print("✓ Orderbook Replay Engine 모듈 로드 완료")4. HolySheep AI 통합: 실시간 시장 심리 분석
백테스트 과정에서 생성된 시장 데이터를 HolySheep AI에 전달하여 정성적 분석을 수행하는 모듈을 구현했습니다. HolySheep AI의 低成本 고성능 API 덕분에 수백만 건의 데이터 포인트도 효율적으로 처리할 수 있습니다.
import aiohttp
import json
from typing import List, Dict, Optional
from datetime import datetime
class HolySheepMarketAnalyzer:
"""HolySheep AI를 활용한 시장 심리 및 정성 분석"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self._session: Optional[aiohttp.ClientSession] = None
# HolySheep AI 가격: GPT-4.1 $8/MTok, Gemini 2.5 Flash $2.50/MTok
self.model = "gpt-4.1" # 비용 효율적 모델 선택
async def __aenter__(self):
self._session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
async def analyze_market_sentiment(
self,
price_data: List[Dict],
orderbook_summary: Dict,
timeframe: str = "1h"
) -> Dict:
"""
시장 심리 분석 요청
- price_data: 가격 히스토리 (timestamp, open, high, low, close, volume)
- orderbook_summary: 주문서 요약 (spread, bid_depth, ask_depth, volatility)
"""
# 프롬프트 구성
prompt = self._build_sentiment_prompt(price_data, orderbook_summary, timeframe)
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "당신은 암호화폐 시장 분석 전문가입니다. 데이터 기반의 객관적 분석을 제공합니다."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3, # 일관된 분석을 위한 낮은 temperature
"max_tokens": 500
}
start_time = asyncio.get_event_loop().time()
async with self._session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as resp:
response = await resp.json()
end_time = asyncio.get_event_loop().time()
latency_ms = (end_time - start_time) * 1000
# 토큰 사용량 파싱 (HolySheep는 usage 필드 제공)
usage = response.get('usage', {})
input_tokens = usage.get('prompt_tokens', 0)
output_tokens = usage.get('completion_tokens', 0)
# 비용 계산 ($8/MTok = $0.008/1K tokens)
input_cost = input_tokens / 1_000_000 * 8 # USD
output_cost = output_tokens / 1_000_000 * 8 # USD
total_cost = input_cost + output_cost
return {
'sentiment_analysis': response['choices'][0]['message']['content'],
'latency_ms': round(latency_ms, 2),
'input_tokens': input_tokens,
'output_tokens': output_tokens,
'cost_usd': round(total_cost, 6),
'model': self.model
}
def _build_sentiment_prompt(
self,
price_data: List[Dict],
orderbook_summary: Dict,
timeframe: str
) -> str:
"""분석 프롬프트 생성"""
# 최근 10개 데이터 포인트 요약
recent = price_data[-10:] if len(price_data) >= 10 else price_data
price_summary = "\n".join([
f"- {datetime.fromtimestamp(p['timestamp']/1000).strftime('%H:%M:%S')}: "
f"${p['close']:.2f} (변화: {p.get('change_pct', 0):+.2f}%)"
for p in recent
])
return f"""다음 BTC/USDT 시장 데이터를 분석하여 시장 심리를 평가해주세요.
[시간대] {timeframe}
[가격 추이]
{price_summary}
[주문서 상태]
- 스프레드: {orderbook_summary.get('spread', 0):.4f}%
-Bid 심도 (상위 10): {orderbook_summary.get('bid_depth', 0):.4f} BTC
-Ask 심도 (상위 10): {orderbook_summary.get('ask_depth', 0):.4f} BTC
-변동성: {orderbook_summary.get('volatility', 0):.4f}%
다음 항목을 포함하여 분석해주세요:
1. 전반적 시장 심리 (Bullish/Bearish/Neutral)
2. 주요 관찰 사항
3. 잠재적 리스크 요인
4. 단기 거래 전략 제안"""
class BacktestAnalyzer:
"""백테스트 결과 종합 분석기 (HolySheep AI 통합)"""
def __init__(self, holy_sheep_key: str):
self.analyzer = HolySheepMarketAnalyzer(holy_sheep_key)
async def generate_insights(
self,
metrics: Dict,
trade_history: List[Dict],
market_context: Dict
) -> Dict:
"""
백테스트 결과에 대한 AI 기반 인사이트 생성
Args:
metrics: calculate_metrics() 결과
trade_history: execute_market_order()로 기록된 거래 내역
market_context: 시장 컨텍스트 정보
"""
# 거래 요약
if trade_history:
recent_trades = trade_history[-20:] # 최근 20건
avg_pnl = sum(t['unrealized_pnl'] for t in recent_trades) / len(recent_trades)
total_volume = sum(abs(t['size']) for t in recent_trades)
else:
recent_trades = []
avg_pnl = 0
total_volume = 0
# HolySheep AI 분석 요청
analysis_result = await self.analyzer.analyze_market_sentiment(
price_data=market_context.get('prices', []),
orderbook_summary=market_context.get('orderbook', {}),
timeframe=market_context.get('timeframe', '1h')
)
# 전략 개선 제안 요청
improvement_prompt = f"""백테스트 결과를 분석하여 전략 개선점을 제안해주세요.
[성과 지표]
- 총 수익률: {metrics.get('total_return', 0)*100:.2f}%
- 최대 낙폭: {metrics.get('max_drawdown', 0)*100:.2f}%
- Sharpe Ratio: {metrics.get('sharpe_ratio', 0):.2f}
- 승률: {metrics.get('win_rate', 0)*100:.1f}%
- 총 거래 수: {metrics.get('total_trades', 0)}
- 평균 거래 손익: ${avg_pnl:.2f}
[거래 패턴 요약]
- 매수 거래: {sum(1 for t in recent_trades if t['side']=='buy')}
- 매도 거래: {sum(1 for t in recent_trades if t['side']=='sell')}
- 총 거래량: {total_volume:.4f} BTC
개선 가능한 영역과 구체적인 권장사항을 제시해주세요."""
# 두 번째 API 호출 (전략 최적화)
strategy_payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 고빈도 트레이딩 전략 전문가입니다."},
{"role": "user", "content": improvement_prompt}
],
"temperature": 0.2,
"max_tokens": 600
}
async with self.analyzer._session.post(
f"{self.analyzer.BASE_URL}/chat/completions",
json=strategy_payload
) as resp:
strategy_response = await resp.json()
return {
'backtest_metrics': metrics,
'market_sentiment': analysis_result['sentiment_analysis'],
'strategy_recommendations': strategy_response['choices'][0]['message']['content'],
'analysis_cost': analysis_result['cost_usd'],
'analysis_latency_ms': analysis_result['latency_ms']
}
async def run_complete_backtest():
"""완전한 백테스트 워크플로우 실행"""
HOLYSHEEP_KEY = "YOUR_HOLYSHEEP_API_KEY"
# 1. Tardis API로 주문서 데이터 수신
async with TardisClient("YOUR_TARDIS_KEY") as tardis:
# Binance BTC/USDT 2024년 1월 1일 1시간 데이터
start_ts = int(datetime(2024, 1, 1, 0, 0).timestamp() * 1000)
end_ts = int(datetime(2024, 1, 1, 1, 0).timestamp() * 1000)
# 설정 조회 (실제 스트리밍 대신 설정 정보 획득)
config = await tardis.get_replay_config(
exchange="binance",
symbol="BTC/USDT",
from_timestamp=start_ts,
to_timestamp=end_ts
)
print(f"리플레이 설정: {config.get('id', 'N/A')}")
# 2. HolySheep AI 분석기 초기화
async with HolySheepMarketAnalyzer(HOLYSHEEP_KEY) as analyzer:
# 샘플 데이터로 분석 테스트
sample_prices = [
{'timestamp': 1704067200000, 'close': 42250.50, 'change_pct': 0.0},
{'timestamp': 1704067260000, 'close': 42280.25, 'change_pct': 0.07},
{'timestamp': 1704067320000, 'close': 42245.00, 'change_pct': -0.08},
]
sample_orderbook = {
'spread': 0.015,
'bid_depth': 12.5,
'ask_depth': 11.8,
'volatility': 0.0023
}
result = await analyzer.analyze_market_sentiment(
price_data=sample_prices,
orderbook_summary=sample_orderbook,
timeframe="1h"
)
print(f"분석 Latency: {result['latency_ms']}ms")
print(f"토큰 사용량: {result['input_tokens']} → {result['output_tokens']}")
print(f"API 비용: ${result['cost_usd']}")
print(f"\n분석 결과:\n{result['sentiment_analysis']}")
실행
if __name__ == "__main__":
asyncio.run(run_complete_backtest())5. 고빈도 전략 시뮬레이션 실행기
이제 실제 주문서를 기반으로 고빈도 전략을 시뮬레이션하는 실행기를 구현합니다. 저는 세 가지 전형적인 전략을 구현했습니다: 마켓 메이커 힓지, 스프레드 역전, 유동성 인사이트.
import asyncio
from typing import List, Callable, Dict
from dataclasses import dataclass
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class StrategyType(Enum):
"""지원되는 전략 유형"""
MARKET_MAKER = "market_maker"
SPREAD_REVERSAL = "spread_reversal"
LIQUIDITY_GRAB = "liquidity_grab"
MOMENTUM = "momentum"
@dataclass
class StrategyConfig:
"""전략 설정"""
name: StrategyType
max_position: float = 1.0 # 최대 포지션 (BTC)
order_size: float = 0.01 # 기본 주문 크기
spread_threshold: float = 0.0005 # 스프레드 임계값 (0.05%)
reversion_threshold: float = 0.002 # 역전 임계값
lookback_periods: int = 10 # 이동 평균 기간
class HighFrequencyStrategy:
"""고빈도 거래 전략 실행기"""
def __init__(self, config: StrategyConfig, engine: OrderbookReplayEngine):
self.config = config
self.engine = engine
self.price_history: List[float] = []
self.signal_history: List[Dict] = []
def _calculate_mid_price_ma(self, periods: int) -> float:
"""중간 가격 이동 평균"""
if len(self.price_history) < periods:
return self.price_history[-1] if self.price_history else 0
return sum(self.price_history[-periods:]) / periods
def _calculate_volatility(self, periods: int = 10) -> float:
"""변동성 계산 (표준편차 기반)"""
if len(self.price_history) < periods:
return 0
recent = self.price_history[-periods:]
mean = sum(recent) / len(recent)
variance = sum((x - mean) ** 2 for x in recent) / len(recent)
return variance ** 0.5
async def on_orderbook_update(self, snapshot: OrderbookSnapshot):
"""주문서 업데이트 시 전략 로직 실행"""
# 1. 가격 히스토리 업데이트
self.price_history.append(snapshot.mid_price)
if len(self.price_history) > 100:
self.price_history.pop(0)
# 2. 현재 상태 확인
current_position = self.engine.position
current_spread = snapshot.spread
volatility = self._calculate_volatility()
signal = None
# 3. 전략별 로직
if self.config.name == StrategyType.MARKET_MAKER:
signal = await self._market_maker_strategy(
snapshot, current_position, current_spread
)
elif self.config.name == StrategyType.SPREAD_REVERSAL:
signal = await self._spread_reversal_strategy(
snapshot, volatility
)
elif self.config.name == StrategyType.LIQUIDITY_GRAB:
signal = await self._liquidity_grab_strategy(snapshot)
elif self.config.name == StrategyType.MOMENTUM:
signal = await self._momentum_strategy(snapshot, volatility)
# 4. 신호 실행
if signal:
await self._execute_signal(signal, snapshot)
async def _market_maker_strategy(
self,
snapshot: OrderbookSnapshot,
position: float,
spread: float
) -> Optional[Dict]:
"""
마켓 메이커 전략: 스프레드에서 수익 확보
- Bid/Ask 양쪽에 리밋 오더 배치 시뮬레이션
-库存 관리로 directional bias 최소화
"""
# 스프레드가 충분한가?
if spread < self.config.spread_threshold:
return None
# 포지션 한도 체크
if abs(position) >= self.config.max_position:
return None
# 시장이 균형 상태이면 시장 중립 포지션
bid_size = sum(b.size for b in snapshot.bids[:5])
ask_size = sum(a.size for a in snapshot.asks[:5])
if bid_size > ask_size * 1.5:
# 강한 매수 압력 → 숏 포지션 증가
return {'action': 'sell', 'size': self.config.order_size, 'reason': 'imbalance_short'}
elif ask_size > bid_size * 1.5:
# 강한 매도 압력 → 롱 포지션 증가
return {'action': 'buy', 'size': self.config.order_size, 'reason': 'imbalance_long'}
return None
async def _spread_reversal_strategy(
self,
snapshot: OrderbookSnapshot,
volatility: float
) -> Optional[Dict]:
"""
스프레드 역전 전략: 비정상적 스프레드 확대 시 수익
- 스프레드가 historically 높을 때 반대 방향 거래
- 변동성 정상화 시 이익 실현
"""
ma_price = self._calculate_mid_price_ma(self.config.lookback_periods)
current_price = snapshot.mid_price
deviation_pct = (current_price - ma_price) / ma_price
# 가격이 이동 평균 대비 너무偏离 → 역전 신호
if deviation_pct > self.config.reversion_threshold:
return {'action': 'sell', 'size': self.config.order_size, 'reason': 'overvalued_reversion'}
elif deviation_pct < -self.config.reversion_threshold:
return {'action': 'buy', 'size': self.config.order_size, 'reason': 'undervalued_reversion'}
return None
async def _liquidity_grab_strategy(
self,
snapshot: OrderbookSnapshot
) -> Optional[Dict]:
"""
流动性 인사이트 전략: 대규모 주문 급등 시 반응
- 특정 가격 레벨의流动性 고갈 감지
-price 급변 후 반등 노림
"""
# Bid-Ask 사이즈 급변 감지
bid_depth = sum(b.size for b in snapshot.bids[:10])
ask_depth = sum(a.size for a in snapshot.asks[:10])
#Ask 깊이 감소 (流动性 제거) → 가격 상승 가능성
if ask_depth < 1.0 and snapshot.asks[0].size > 0.5:
return {'action': 'buy', 'size': self.config.order_size * 2, 'reason': 'liquidity_vacuum'}
return None
async def _momentum_strategy(
self,
snapshot: OrderbookSnapshot,
volatility: float
) -> Optional[Dict]:
"""
모멘텀 전략: 강세 추세 편승
- 연속적인 가격 상승 시 매수
- 방향 전환 시 청산
"""
if len(self.price_history) < 5:
return None
recent = self.price_history[-5:]
momentum = (recent[-1] - recent[0]) / recent[0]
if momentum > 0.001: # 0.1% 이상 상승
return {'action': 'buy', 'size': self.config.order_size, 'reason': 'momentum_up'}
elif momentum < -0.001:
return {'action': 'sell', 'size': self.config.order_size, 'reason': 'momentum_down'}
return None
async def _execute_signal(
self,
signal: Dict,
snapshot: OrderbookSnapshot
) -> None:
"""신호 실행 및 기록"""
try:
trade = self.engine.execute_market_order(
side=signal['action'],
size=signal['size']
)
self.signal_history.append({
'timestamp': snapshot.timestamp,
'signal': signal,
'execution': trade,
'mid_price': snapshot.mid_price,
'spread': snapshot.spread
})
logger.info(
f"신호 실행: {signal['action']} {signal['size']} BTC @ "
f"${trade['exec_price']:.2f} (이유: {signal['reason']})"
)
except Exception as e:
logger.error(f"신호 실행 실패: {e}")
async def run_simulation():
"""백테스트 시뮬레이션 실행"""
# 엔진 초기화
engine = OrderbookReplayEngine(TardisClient("YOUR_TARDIS_KEY"))
# 전략 설정
strategies = [
StrategyConfig(
name=StrategyType.MARKET_MAKER,
max_position=0.5,
order_size=0.01,
spread_threshold=0.0003
),
StrategyConfig(
name=StrategyType.SPREAD_REVERSAL,
max_position=1.0,
order_size=0.02,
reversion_threshold=0.0015
),
StrategyConfig(
name=StrategyType.LIQUIDITY_GRAB,
max_position=0.3,
order_size=0.05
)
]
# 각 전략별 인스턴스 생성
strategy_instances = [
HighFrequencyStrategy(config, engine) for config in strategies
]
# 샘플 주문서 스냅샷으로 시뮬레이션
sample_snapshots = [
OrderbookSnapshot(
timestamp=1704067200000 + i * 1000,
exchange_timestamp=1704067200000 + i * 1000,
bids=[
OrderbookLevel(price=42250.00 - i * 0.5, size=0.5 + i * 0.1, side='bid')
for i in range(5)
],
asks=[
OrderbookLevel(price=42251.00 + i * 0.5, size=0.4 + i * 0.1, side='ask')
for i in range(5)
],
sequence=i
)
for i in range(50)
]
# 시뮬레이션 실행
for snapshot in sample_snapshots:
await engine.process_snapshot(snapshot)
# 모든 전략에 주문서 업데이트 전파
tasks = [
strategy.on_orderbook_update(snapshot)
for strategy in strategy_instances
]
await asyncio.gather(*tasks)
# 결과 분석
metrics = engine.calculate_metrics()
print("\n" + "="*60)
print("백테스트 결과 요약")
print("="*60)
print(f"총 수익률: {metrics['total_return']*100:.2f}%")
print(f"총 거래 수: {metrics['total_trades']}")
print(f"최대 낙폭: {metrics['max_drawdown']*100:.2f}%")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"승률: {metrics['win_rate']*100:.1f}%")
print("="*60)
return metrics, engine.trade_history