트레이딩 봇을 개발할 때 가장 큰 도전 중 하나는 과거 데이터를 실시간 스트림처럼 시뮬레이션하는 것입니다. Tardis Machine은 마치 타임머신처럼 과거의 시장 데이터를 미래의 거래 봇에게 "실시간"으로 흘려보내는 기술입니다. 이 튜토리얼에서는 Python 기반 로컬 WebSocket 서버를 구축하여 캔들스틱, 오더북, 체결 데이터를 과거에서 재생하는 방법을 단계별로 설명합니다.
비교표: HolySheep vs 공식 API vs 다른 릴레이 서비스
| 특징 | HolySheep AI | 공식 Binance API | 기타 릴레이 서비스 |
|---|---|---|---|
| 로컬 결제 지원 | ✅ 해외 신용카드 불필요 | ❌ 해외 신용카드 필수 | ⚠️ 제한적 |
| 다중 모델 통합 | ✅ GPT-4.1, Claude, Gemini, DeepSeek 단일 키 | ❌ 각 서비스별 별도 키 | ⚠️ 2-3개 모델만 지원 |
| WebSocket 스트리밍 | ✅ 네이티브 지원 | ✅ 실시간 거래 데이터 | ⚠️ 지연 발생 |
| 트레이딩 봇 백테스트 지원 | ✅ AI 분석 + 시뮬레이션 | ❌ 별도 시뮬레이터 필요 | ⚠️ 기본 히스토리만 |
| 가격 (GPT-4.1) | $8/MTok | $15/MTok | $10-12/MTok |
| 무료 크레딧 | ✅ 가입 시 제공 | ❌ 없음 | ⚠️ 제한적 |
| API Gateway 통합 | ✅ 단일 엔드포인트 | ❌ 다중 서비스 관리 | ⚠️ 수동 전환 |
Tardis Machine이란?
Tardis Machine은 과거의 시장 데이터를 시간 여행시켜 마치 실시간으로 흘러오는 것처럼 시뮬레이션하는 시스템입니다. 핵심 개념은 다음과 같습니다:
- 데이터 캡처: 과거 Binance, Bybit, OKX 등의 websocket 스트림에서 캔들스틱, 오더북, 체결 데이터를 저장
- 시간 왜곡: 저장된 데이터를 지정된 속도(실시간, 10배속, 100배속)로 재생
- 봇 호환성: 기존 실시간 websocket 클라이언트 코드를 수정 없이 사용 가능
- 시점 제어: 특정 시간으로 되돌아가거나, 일시 정지, 특정 구간 반복 재생
아키텍처 개요
┌─────────────────────────────────────────────────────────────────┐
│ Tardis Machine 아키텍처 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Historical │ │ Local │ │ Trading │ │
│ │ Data │─────▶│ WebSocket │─────▶│ Bot │ │
│ │ Store │ │ Server │ │ (기존 코드) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ SQLite/ │ │ asyncio │ │ Binance │ │
│ │ Parquet │ │ Playback │ │ Compatible │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ 시간 흐름: 2024-01-15 09:30 ───────────▶ 2024-01-15 10:00 │
│ (과거 데이터) (현재 시뮬레이션 시점) │
└─────────────────────────────────────────────────────────────────┘필수 설치 패키지
# 핵심 의존성 설치
pip install fastapi uvicorn websockets pandas numpy asyncio aiofiles
선택적 의존성 (성능 최적화)
pip install uvloop orjson # 고성능 이벤트 루프 및 JSON 처리
데이터 저장용
pip install sqlalchemy aiosqlite # 비동기 데이터베이스1단계: 과거 데이터 캡처 및 저장
먼저 Binance websocket에서 실시간 데이터를 캡처하여 저장하는 스크립트를 작성합니다. HolySheep AI의 다중 모델 지원을 활용하면 캡처된 데이터를 AI로 분석하여 패턴을 식별할 수 있습니다.
# data_capture.py - 과거 데이터 캡처 스크립트
import asyncio
import json
import aiofiles
from datetime import datetime
from pathlib import Path
import websockets
class MarketDataCapture:
def __init__(self, symbol: str = "btcusdt", output_dir: str = "./historical_data"):
self.symbol = symbol.lower()
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.buffer = []
self.capture_active = False
async def connect_binance_websocket(self):
"""Binance WebSocket 스트림 연결"""
# 캔들스틱 (1분), 오더북, 체결 데이터 스트림
streams = [
f"{self.symbol}@kline_1m",
f"{self.symbol}@depth20@100ms",
f"{self.symbol}@trade"
]
uri = f"wss://stream.binance.com:9443/stream?streams={'/'.join(streams)}"
return uri
async def save_to_file(self, data_type: str, data: dict):
"""데이터를 날짜별 파일로 저장"""
timestamp = datetime.fromisoformat(data.get('event_time', datetime.now().isoformat()))
date_str = timestamp.strftime("%Y-%m-%d")
filename = self.output_dir / f"{self.symbol}_{data_type}_{date_str}.jsonl"
async with aiofiles.open(filename, mode='a') as f:
await f.write(json.dumps(data) + '\n')
async def process_message(self, message: dict):
"""WebSocket 메시지 처리 및 저장"""
try:
stream = message.get('stream', '')
data = message.get('data', {})
if 'kline' in stream:
await self.save_to_file('kline', {
'event_time': data['k']['t'],
'symbol': data['s'],
'open': float(data['k']['o']),
'high': float(data['k']['h']),
'low': float(data['k']['l']),
'close': float(data['k']['c']),
'volume': float(data['k']['v']),
'is_closed': data['k']['x']
})
elif 'depth' in stream:
await self.save_to_file('orderbook', {
'event_time': data.get('E'),
'symbol': data['s'],
'bids': [[float(p), float(q)] for p, q in data.get('b', [])],
'asks': [[float(p), float(q)] for p, q in data.get('a', [])]
})
elif 'trade' in stream:
await self.save_to_file('trade', {
'event_time': data['T'],
'symbol': data['s'],
'price': float(data['p']),
'quantity': float(data['q']),
'is_buyer_maker': data['m']
})
except Exception as e:
print(f"처리 오류: {e}")
async def start_capture(self, duration_minutes: int = 60):
"""지정된 시간 동안 데이터 캡처 시작"""
uri = await self.connect_binance_websocket()
print(f"📡 {uri}에 연결 중...")
self.capture_active = True
capture_start = datetime.now()
async with websockets.connect(uri) as ws:
print(f"✅ 캡처 시작: {self.symbol} ({duration_minutes}분간)")
while self.capture_active:
elapsed = (datetime.now() - capture_start).total_seconds() / 60
if elapsed >= duration_minutes:
print(f"⏰ 지정된 시간({duration_minutes}분) 완료")
break
try:
message = await asyncio.wait_for(ws.recv(), timeout=30)
data = json.loads(message)
await self.process_message(data)
if len(self.buffer) % 100 == 0 and self.buffer:
print(f"📊 캡처됨: {len(self.buffer)}건")
except asyncio.TimeoutError:
print("⚠️ 30초 타임아웃, 재연결 시도...")
except Exception as e:
print(f"❌ 오류 발생: {e}")
await asyncio.sleep(5)
def stop_capture(self):
"""캡처 중지"""
self.capture_active = False
async def main():
capture = MarketDataCapture(symbol="ethusdt", output_dir="./data")
# 1시간 동안 데이터 캡처 (테스트용으로 5분으로 줄임)
await capture.start_capture(duration_minutes=5)
print("📁 저장된 파일 목록:")
for f in capture.output_dir.glob("*.jsonl"):
print(f" - {f.name}")
if __name__ == "__main__":
asyncio.run(main())2단계: Tardis Machine WebSocket 서버 구현
이제 캡처한 과거 데이터를 실시간 스트림처럼 재생하는 WebSocket 서버를 구현합니다. HolySheep AI의 스트리밍 지원을 활용하면 AI 기반 시장 분석과 백테스트 결과를 실시간으로 받을 수 있습니다.
# tardis_server.py - Tardis Machine WebSocket 서버
import asyncio
import json
import heapq
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Set, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import websockets
from websockets.server import WebSocketServerProtocol
@dataclass(order=True)
class TimedEvent:
event_time: float # Unix timestamp (sort key)
event_type: str = field(compare=False)
symbol: str = field(compare=False)
data: dict = field(compare=False)
class TardisMachine:
"""시계 방향/time-warp market data replay engine"""
def __init__(self, data_dir: str = "./historical_data", playback_speed: float = 1.0):
self.data_dir = Path(data_dir)
self.playback_speed = playback_speed # 1.0 = realtime, 10.0 = 10x speed
self.clients: Set[WebSocketServerProtocol] = set()
self.subscriptions: Dict[str, Set[WebSocketServerProtocol]] = defaultdict(set)
self.event_queue: list[TimedEvent] = []
self.is_playing = False
self.current_time = 0.0
self.start_time = 0.0
self.end_time = 0.0
def load_historical_data(self, symbol: str = None, date: str = None):
"""저장된 과거 데이터 로드 및 인덱싱"""
print(f"📂 데이터 로딩 중: {self.data_dir}")
files = list(self.data_dir.glob("*.jsonl"))
all_events = []
for file_path in files:
if symbol and symbol not in file_path.name:
continue
if date and date not in file_path.name:
continue
print(f" 로딩: {file_path.name}")
with open(file_path, 'r') as f:
for line in f:
try:
data = json.loads(line.strip())
event_time = data.get('event_time', 0) / 1000 # ms to seconds
# 이벤트 타입 결정
if 'open' in data: # kline
event_type = 'kline'
elif 'bids' in data: # orderbook
event_type = 'depth'
elif 'price' in data: # trade
event_type = 'trade'
else:
continue
heapq.heappush(all_events, TimedEvent(
event_time=event_time,
event_type=event_type,
symbol=data.get('symbol', ''),
data=data
))
except json.JSONDecodeError:
continue
self.event_queue = all_events
if self.event_queue:
self.start_time = self.event_queue[0].event_time
self.end_time = self.event_queue[-1].event_time
print(f"✅ {len(self.event_queue):,}개 이벤트 로드됨")
print(f" 시간 범위: {datetime.fromtimestamp(self.start_time)} ~ {datetime.fromtimestamp(self.end_time)}")
else:
print("⚠️ 로드된 데이터 없음")
async def broadcast_to_subscribers(self, event: TimedEvent):
"""구독자에게 이벤트 브로드캐스트"""
subscribers = self.subscriptions.get(event.event_type, set())
subscribers |= self.subscriptions.get('all', set())
if not subscribers:
return
# Binance 호환 형식으로 변환
message = self._format_binance_message(event)
# 모든 구독자에게 전송
disconnected = set()
for client in subscribers:
try:
await client.send(json.dumps(message))
except websockets.exceptions.ConnectionClosed:
disconnected.add(client)
# 연결 끊긴 클라이언트 제거
for client in disconnected:
await self.remove_client(client)
def _format_binance_message(self, event: TimedEvent) -> dict:
"""Binance WebSocket 형식으로 변환"""
data = event.data
if event.event_type == 'kline':
return {
"stream": f"{data['symbol'].lower()}@kline_1m",
"data": {
"e": "kline",
"E": int(event.event_time * 1000),
"s": data['symbol'],
"k": {
"t": int(event.event_time * 1000),
"o": str(data['open']),
"h": str(data['high']),
"l": str(data['low']),
"c": str(data['close']),
"v": str(data['volume']),
"x": data.get('is_closed', False)
}
}
}
elif event.event_type == 'depth':
return {
"stream": f"{data['symbol'].lower()}@depth20@100ms",
"data": {
"e": "depthUpdate",
"E": int(event.event_time * 1000),
"s": data['symbol'],
"b": [[str(p), str(q)] for p, q in data.get('bids', [])],
"a": [[str(p), str(q)] for p, q in data.get('asks', [])]
}
}
elif event.event_type == 'trade':
return {
"stream": f"{data['symbol'].lower()}@trade",
"data": {
"e": "trade",
"E": int(event.event_time * 1000),
"s": data['symbol'],
"p": str(data['price']),
"q": str(data['quantity']),
"m": data.get('is_buyer_maker', False)
}
}
return {}
async def playback_loop(self):
"""이벤트 재생 루프"""
if not self.event_queue:
print("⚠️ 재생할 이벤트가 없음")
return
print(f"▶️ 재생 시작 (속도: {self.playback_speed}x)")
self.is_playing = True
self.current_time = self.start_time
processed_count = 0
last_print_time = asyncio.get_event_loop().time()
while self.event_queue and self.is_playing:
# 가장 빠른 시간의 이벤트 가져오기
next_event = heapq.heappop(self.event_queue)
# 재생 속도에 따른 대기 시간 계산
event_timestamp = next_event.event_time
if event_timestamp > self.current_time:
real_wait = (event_timestamp - self.current_time) / self.playback_speed
if real_wait > 0:
await asyncio.sleep(min(real_wait, 1.0)) # 최대 1초 대기
self.current_time = event_timestamp
await self.broadcast_to_subscribers(next_event)
processed_count += 1
# 1초마다 상태 출력
now = asyncio.get_event_loop().time()
if now - last_print_time >= 1.0:
elapsed = datetime.fromtimestamp(self.current_time)
print(f" 📊 {elapsed} | 처리: {processed_count:,} | 남은: {len(self.event_queue):,}")
last_print_time = now
print(f"✅ 재생 완료: {processed_count:,}개 이벤트")
self.is_playing = False
async def handle_client(self, websocket: WebSocketServerProtocol, path: str):
"""WebSocket 클라이언트 핸들러"""
client_id = id(websocket)
print(f"🔗 클라이언트 연결: {client_id}")
self.clients.add(websocket)
try:
async for message in websocket:
try:
data = json.loads(message)
await self.handle_client_message(websocket, data)
except json.JSONDecodeError:
await websocket.send(json.dumps({"error": "Invalid JSON"}))
except websockets.exceptions.ConnectionClosed:
print(f"🔌 클라이언트断开: {client_id}")
finally:
await self.remove_client(websocket)
async def handle_client_message(self, websocket: WebSocketServerProtocol, data: dict):
"""클라이언트 메시지 처리"""
msg_type = data.get('type', '')
if msg_type == 'subscribe':
streams = data.get('streams', [])
for stream in streams:
self.subscriptions[stream].add(websocket)
print(f"📥 구독: {stream} (총 {len(self.subscriptions[stream])}명)")
await websocket.send(json.dumps({
"type": "subscribed",
"streams": streams
}))
elif msg_type == 'unsubscribe':
streams = data.get('streams', [])
for stream in streams:
self.subscriptions[stream].discard(websocket)
await websocket.send(json.dumps({
"type": "unsubscribed",
"streams": streams
}))
elif msg_type == 'control':
action = data.get('action', '')
if action == 'play':
if not self.is_playing:
asyncio.create_task(self.playback_loop())
elif action == 'pause':
self.is_playing = False
elif action == 'speed':
self.playback_speed = data.get('speed', 1.0)
print(f"⚡ 재생 속도: {self.playback_speed}x")
async def remove_client(self, websocket: WebSocketServerProtocol):
"""클라이언트 제거"""
self.clients.discard(websocket)
for stream in self.subscriptions:
self.subscriptions[stream].discard(websocket)
async def start_server(self, host: str = "localhost", port: int = 8765):
"""서버 시작"""
print(f"🚀 Tardis Machine 서버 시작: ws://{host}:{port}")
self.load_historical_data(symbol="ethusdt") # 기본값: ETH/USDT
async with websockets.serve(self.handle_client, host, port):
print("✅ WebSocket 서버 실행 중...")
print(" 클라이언트가 연결되기를 기다리는 중...")
# 무한 대기 (Ctrl+C로 종료)
await asyncio.Future()
async def main():
tardis = TardisMachine(
data_dir="./data",
playback_speed=10.0 # 10배속 재생
)
await tardis.start_server(host="0.0.0.0", port=8765)
if __name__ == "__main__":
asyncio.run(main())3단계: 거래 봇 연동
기존 Binance WebSocket 클라이언트 코드를 그대로 사용하여 Tardis Machine에 연결할 수 있습니다. HolySheep AI의 다중 모델 통합 기능을 활용하면 AI 기반 거래 시그널 생성도 가능합니다.
# trading_bot.py - Tardis Machine에 연결하는 거래 봇
import asyncio
import json
import websockets
from datetime import datetime
from typing import Dict, List, Optional
import numpy as np
class SimpleTradingBot:
"""단순 이동평균 교차 거래 봇"""
def __init__(self, symbol: str = "ethusdt", short_window: int = 5, long_window: int = 20):
self.symbol = symbol
self.short_window = short_window
self.long_window = long_window
self.prices: List[float] = []
self.position = 0 # 0: 없음, 1: 매수, -1: 매도
self.trades: List[dict] = []
# HolySheep AI API 설정
self.holysheep_api_key = "YOUR_HOLYSHEEP_API_KEY"
self.holysheep_base_url = "https://api.holysheep.ai/v1"
async def analyze_with_ai(self, price_data: dict) -> Optional[str]:
"""HolySheep AI를 활용한 시장 분석"""
try:
import openai
client = openai.AsyncOpenAI(
api_key=self.holysheep_api_key,
base_url=self.holysheep_base_url
)
response = await client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": "당신은 전문 트레이딩 분석가입니다. 현재 가격 데이터 기반 매매 신호를 'BUY', 'SELL', 'HOLD'로만 응답하세요."},
{"role": "user", "content": f"현재 가격: {price_data.get('close', 0)}, 이전 종가: {price_data.get('prev_close', 0)}, 거래량: {price_data.get('volume', 0)}"}
],
max_tokens=10,
temperature=0.1
)
return response.choices[0].message.content.strip()
except Exception as e:
print(f"AI 분석 오류: {e}")
return None
def calculate_sma(self, prices: List[float], window: int) -> Optional[float]:
"""단순 이동평균 계산"""
if len(prices) < window:
return None
return np.mean(prices[-window:])
def generate_signal(self) -> str:
"""매매 신호 생성 (이동평균 교차)"""
if len(self.prices) < self.long_window:
return "HOLD"
short_ma = self.calculate_sma(self.prices, self.short_window)
long_ma = self.calculate_sma(self.prices, self.long_window)
if short_ma is None or long_ma is None:
return "HOLD"
# 골든크로스 / 데드크로스 감지
prev_short = self.prices[-2] if len(self.prices) > 1 else short_ma
prev_long = self.prices[-len(self.prices)+1] if len(self.prices) > 1 else long_ma
if prev_short <= prev_long and short_ma > long_ma:
return "BUY"
elif prev_short >= prev_long and short_ma < long_ma:
return "SELL"
return "HOLD"
def execute_trade(self, signal: str, price: float):
"""거래 실행 시뮬레이션"""
if signal == "BUY" and self.position <= 0:
self.position = 1
self.trades.append({
"time": datetime.now().isoformat(),
"type": "BUY",
"price": price,
"position": self.position
})
print(f"🟢 매수 실행: ${price:.2f}")
elif signal == "SELL" and self.position >= 0:
self.position = -1
self.trades.append({
"time": datetime.now().isoformat(),
"type": "SELL",
"price": price,
"position": self.position
})
print(f"🔴 매도 실행: ${price:.2f}")
async def handle_kline(self, data: dict):
"""캔들스틱 데이터 처리"""
kline = data.get('k', {})
close_price = float(kline.get('c', 0))
volume = float(kline.get('v', 0))
if close_price > 0:
self.prices.append(close_price)
# 이동평균 계산을 위해 최근 50개만 유지
if len(self.prices) > 50:
self.prices.pop(0)
# 매매 신호 생성
signal = self.generate_signal()
if signal != "HOLD":
self.execute_trade(signal, close_price)
# HolySheep AI 분석 (10개 캔들마다)
if len(self.prices) % 10 == 0:
ai_signal = await self.analyze_with_ai({
"close": close_price,
"prev_close": self.prices[-2] if len(self.prices) > 1 else close_price,
"volume": volume
})
if ai_signal:
print(f"🤖 AI 신호: {ai_signal}")
async def handle_depth(self, data: dict):
"""오더북 데이터 처리"""
bids = data.get('b', [])
asks = data.get('a', [])
if bids and asks:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread = (best_ask - best_bid) / best_bid * 100
if len(self.prices) > 0:
spread_indicator = "📊 스프레드: {:.4f}%".format(spread)
if spread > 0.5:
spread_indicator = "⚠️ 스프레드 확대: {:.4f}%".format(spread)
async def connect_to_tardis(self, url: str = "ws://localhost:8765"):
"""Tardis Machine에 연결"""
print(f"🔗 Tardis Machine 연결 중: {url}")
subscribe_msg = {
"type": "subscribe",
"streams": [
f"{self.symbol.lower()}@kline_1m",
f"{self.symbol.lower()}@depth20@100ms"
]
}
async with websockets.connect(url) as ws:
await ws.send(json.dumps(subscribe_msg))
print(f"✅ 구독 완료: {subscribe_msg['streams']}")
# 재생 시작 요청
await ws.send(json.dumps({
"type": "control",
"action": "play",
"speed": 10.0
}))
print("▶️ 데이터 재생 시작 (10x 속도)")
async for message in ws:
try:
data = json.loads(message)
# Binance 스트림 타입 확인
if 'stream' in data:
stream = data.get('stream', '')
payload = data.get('data', {})
if 'kline' in stream:
await self.handle_kline(payload)
elif 'depth' in stream:
await self.handle_depth(payload)
else:
# 제어 메시지 처리
print(f"📨 서버 메시지: {data}")
except json.JSONDecodeError:
print(f"❌ JSON 파싱 오류: {message[:100]}")
def print_summary(self):
"""거래 요약 출력"""
print("\n" + "="*50)
print("📊 거래 봇 백테스트 결과")
print("="*50)
print(f"총 거래 횟수: {len(self.trades)}")
if self.trades:
total_pnl = 0
for i in range(0, len(self.trades)-1, 2):
if i+1 < len(self.trades):
buy_trade = self.trades[i]
sell_trade = self.trades[i+1]
pnl = sell_trade['price'] - buy_trade['price']
total_pnl += pnl
print(f" 거래 {i//2+1}: 매수가 ${buy_trade['price']:.2f} → 매도가 ${sell_trade['price']:.2f} (P&L: ${pnl:.2f})")
print(f"\n총 손익: ${total_pnl:.2f}")
print(f"최종 포지션: {'매수' if self.position == 1 else '매도' if self.position == -1 else '중립'}")
async def main():
bot = SimpleTradingBot(
symbol="ETHUSDT",
short_window=5,
long_window=20
)
try:
await bot.connect_to_tardis("ws://localhost:8765")
except KeyboardInterrupt:
print("\n⏹️ 봇 종료")
bot.print_summary()
if __name__ == "__main__":
asyncio.run(main())고급 기능: 오더북 리플레이 및 체결 데이터]
# advanced_replay.py - 고급 오더북/체결 데이터 리플레이
import asyncio
import json
import aiofiles
from pathlib import Path
from typing import List, Dict
from dataclasses import dataclass
from collections import deque
@dataclass
class OrderBookLevel:
price: float
quantity: float
class OrderBookReplayer:
"""오더북 상태 추적 및 리플레이"""
def __init__(self, depth_levels: int = 20):
self.depth_levels = depth_levels
self.bids: deque = deque(maxlen=depth_levels) # 내림차순 정렬
self.asks: deque = deque(maxlen=depth_levels) # 오름차순 정렬
self.last_update_id = 0
def apply_delta(self, bids: List[List[float]], asks: List[List[float]]):
"""오더북 델타 적용"""
# 기존 레벨 매핑
bid_map = {level.price: level for level in self.bids}
ask_map = {level.price: level for level in self.asks}
# bids 업데이트
for price, qty in bids:
if qty == 0:
bid_map.pop(price, None)
else:
bid_map[price] = OrderBookLevel(price, qty)
# asks 업데이트
for price, qty in asks:
if qty == 0:
ask_map.pop(price, None)
else:
ask_map[price] = OrderBookLevel(price, qty)
# 정렬된 deque 재생성
self.bids = deque(
sorted(bid_map.values(), key=lambda x: -x.price),
maxlen=self.depth_levels
)
self.asks = deque(
sorted(ask_map.values(), key=lambda x: x.price),
maxlen=self.depth_levels
)
def get_mid_price(self) -> float:
"""중간 가격 계산"""
if self.bids and self.asks:
return (self.bids[0].price + self.asks[0].price) / 2
return 0.0
def get_spread(self) -> float:
"""스프레드 계산 (bps)"""
if self.bids and self.asks:
mid = self.get_mid_price()
if mid > 0:
return (self.asks[0].price - self.bids[0].price) / mid * 10000
return 0.0
def to_dict(self) -> dict:
"""딕셔너리 변환"""
return {
"bids": [[level.price, level.quantity] for level in self.bids],
"asks": [[level.price, level.quantity] for level in self.asks],
"mid_price": self.get_mid_price(),
"spread_bps": self.get_spread()
}
class TradeReplayer:
"""체결 데이터 리플레이 및 VWAP 계산"""
def __init__(self, window_size: int = 100):
self.window_size = window_size
self.recent_trades: deque = deque(maxlen=window_size)
self.vwap = 0.0
def add_trade(self, price: float, quantity: float):
"""체결 추가"""
self.recent_trades.append({
"price": price,
"quantity": quantity,
"value": price * quantity
})
self.calculate_vwap()
def calculate_vwap(self):
"""VWAP (표면加权平均価格) 계산"""
if not self.recent_trades:
self.vwap = 0.0
return
total_value = sum(t['value'] for t in self.recent_trades)
total_volume = sum(t['quantity'] for t in self.recent_trades)
self.vwap = total_value / total_volume if total_volume > 0 else 0.0