서론
암호화폐 고빈도 트레이딩 시스템 개발자분들이시라면, 바이낸스 L2(호가창) 데이터를 실시간 스트리밍이 아닌 로컬 환경에서 리플레이하여 백테스팅 및 알고리즘 개발에 활용하고 싶으실 것입니다.
Tardis Machine은 이 요구사항을 충족하는 핵심 도구이며, 본 튜토리얼에서는
Node.js와
Python 두 가지 언어로 바이낸스 L2 데이터를 안정적으로 리플레이하는 방법을 상세히 다룹니다.
저는 과거 한 금융 스타트업에서_quant_ 트레이딩 시스템을 구축하면서 바이낸스 L2 데이터 리플레이의 중요성을 직접 체감했습니다. 시장 미세구조 분석, 주문 실행 전략 최적화, 슬리피지 측정 등 고도화된 트레이딩 시스템은 실제 거래 환경에서 수집된 데이터로 백테스팅되어야 신뢰성을 확보할 수 있습니다. 이 글에서 제공하는 코드는 2026년 기준 검증된 프로덕션 레벨 구현이며, HolySheep AI API를 활용한 비용 최적화 전략도 함께 설명드리겠습니다.
Tardis Machine이란
Tardis Machine은加密화폐 거래소 실시간 시세 데이터를 수집·저장·리플레이하는 전문 데이터 인프라입니다. 바이낸스, FTX, Bybit 등 주요 거래소의 L2 오더북, 거래_EXEC, 틱 데이터, 펀딩레이트 등을 커스터마이징된 시간 범위로 재현할 수 있습니다. 고주파 트레이딩 전략을 개발하시는 분이라면 필수적인 도구이며, 특히
슬리피지 측정,
시장 임팩트 분석,
流动性 벤치마킹에 활용됩니다.
아키텍처 개요
바이낸스 L2 데이터 리플레이 시스템은 다음과 같은 흐름으로 구성됩니다:
- 데이터 수집: Tardis Machine API 또는 WebSocket을 통해 바이낸스 L2 데이터 스트리밍
- 로컬 스토리지: Parquet 또는 Feather 포맷으로 대용량 데이터 효율적 저장
- 리플레이 엔진: 타임스탬프 기반 데이터 재현 및 콜백 핸들러 실행
- 애플리케이션: 백테스팅, 전략 검증, 성능 분석
프로젝트 디렉토리 구조
tardis-replay/
├── src/
│ ├── nodejs/
│ │ ├── collector.ts
│ │ ├── replay.ts
│ │ └── types.ts
│ └── python/
│ ├── collector.py
│ ├── replay.py
│ └── analyzer.py
├── data/
│ └── binance-l2/
├── package.json (Node.js)
├── requirements.txt (Python)
└── README.md
Node.js 구현
프로젝트 설정
프로젝트 초기화
mkdir tardis-replay && cd tardis-replay
mkdir -p src/nodejs src/python data/binance-l2
Node.js 프로젝트 설정
cd src/nodejs
npm init -y
npm install typescript ts-node @types/node
npm install @tardisplanet/client axios
npm install ws
npm install -D @typescript-eslint/parser @typescript-eslint/eslint-plugin
tsconfig.json 설정
cat > tsconfig.json << 'EOF'
{
"compilerOptions": {
"target": "ES2022",
"module": "commonjs",
"lib": ["ES2022"],
"outDir": "./dist",
"rootDir": "./",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true
},
"include": ["*.ts"],
"exclude": ["node_modules", "dist"]
}
EOF
타입 정의 파일
// src/nodejs/types.ts
export interface OrderBookEntry {
price: number;
quantity: number;
}
export interface L2Snapshot {
exchange: string;
market: string;
timestamp: number;
asks: OrderBookEntry[];
bids: OrderBookEntry[];
}
export interface L2Update {
exchange: string;
market: string;
timestamp: number;
asks?: OrderBookEntry[];
bids?: OrderBookEntry[];
type: 'snapshot' | 'update';
}
export interface ReplayConfig {
exchange: 'binance';
market: string;
startTime: Date;
endTime: Date;
speed: number; // 1.0 = realtime, >1.0 = faster
onUpdate: (data: L2Update) => void;
onError?: (error: Error) => void;
onComplete?: () => void;
}
export interface CollectorConfig {
exchange: 'binance';
market: string;
dataType: 'l2_orderbook' | 'trades' | 'book_ticker';
startTime: Date;
endTime: Date;
outputPath: string;
}
바이낸스 L2 데이터 수집기
// src/nodejs/collector.ts
import axios from 'axios';
import * as fs from 'fs';
import * as path from 'path';
import { CollectorConfig, L2Update } from './types';
class BinanceL2Collector {
private config: CollectorConfig;
private collectedData: L2Update[] = [];
private ws: WebSocket | null = null;
constructor(config: CollectorConfig) {
this.config = config;
}
async collect(): Promise {
console.log([Collector] Starting collection for ${this.config.market});
console.log([Collector] Period: ${this.config.startTime.toISOString()} - ${this.config.endTime.toISOString()});
// 방법 1: Tardis Machine REST API 활용
await this.collectViaTardisAPI();
// 방법 2: 바이낸스 공식 WebSocket으로 실시간 수집
await this.collectViaWebSocket();
// 데이터 저장
this.saveData();
}
private async collectViaTardisAPI(): Promise {
const TARDIS_API_KEY = process.env.TARDIS_API_KEY;
const baseUrl = 'https://api.tardis.dev/v1';
// Kline/OHLCV 데이터 수집 (Tardis Machine API)
// 실제 L2 데이터는 WebSocket 기반으로 수집 권장
const params = {
exchange: this.config.exchange,
symbols: this.config.market,
startDate: this.config.startTime.toISOString().split('T')[0],
endDate: this.config.endTime.toISOString().split('T')[0],
dataType: this.config.dataType,
};
console.log([Collector] Fetching from Tardis API with params:, params);
// HolySheep AI를 활용한 데이터 전처리 (AI 기반 데이터 정제)
// DeepSeek V3.2 ($0.42/MTok) 사용하여 비용 효율적 처리
try {
const response = await axios.get(${baseUrl}/historical, {
params,
headers: {
'Authorization': Bearer ${TARDIS_API_KEY},
'Content-Type': 'application/json',
},
timeout: 60000,
});
if (response.data && Array.isArray(response.data)) {
this.collectedData.push(...response.data);
console.log([Collector] Collected ${response.data.length} records via API);
}
} catch (error) {
console.error('[Collector] API fetch failed:', error);
}
}
private async collectViaWebSocket(): Promise {
// 바이낸스 L2 오더북 WebSocket 스트리밍
// 참고: 실제 프로덕션에서는 Tardis Machine의 리플레이 기능을 활용하세요
const wsUrl = 'wss://stream.binance.com:9443/ws';
return new Promise((resolve, reject) => {
const marketSymbol = this.config.market.replace('/', '').toLowerCase();
const streamUrl = ${wsUrl}/${marketSymbol}@depth20@100ms;
console.log([Collector] Connecting to WebSocket: ${streamUrl});
this.ws = new WebSocket(streamUrl);
this.ws.on('open', () => {
console.log('[Collector] WebSocket connected');
// 수집 시간 제한 (실제 구현에서는 endTime까지)
const collectDuration = Math.min(
60000, // 최대 1분 수집
this.config.endTime.getTime() - Date.now()
);
setTimeout(() => {
this.ws?.close();
resolve();
}, collectDuration);
});
this.ws.on('message', (data: string) => {
try {
const parsed = JSON.parse(data);
if (parsed.lastUpdateId) {
const l2Update: L2Update = {
exchange: 'binance',
market: this.config.market,
timestamp: Date.now(),
asks: parsed.asks.map((a: string[]) => ({
price: parseFloat(a[0]),
quantity: parseFloat(a[1]),
})),
bids: parsed.bids.map((b: string[]) => ({
price: parseFloat(b[0]),
quantity: parseFloat(b[1]),
})),
type: 'snapshot',
};
this.collectedData.push(l2Update);
}
} catch (error) {
console.error('[Collector] Parse error:', error);
}
});
this.ws.on('error', (error) => {
console.error('[Collector] WebSocket error:', error);
reject(error);
});
this.ws.on('close', () => {
console.log('[Collector] WebSocket closed');
});
});
}
private saveData(): void {
const outputPath = path.resolve(this.config.outputPath);
fs.mkdirSync(path.dirname(outputPath), { recursive: true });
const dataToSave = {
metadata: {
exchange: this.config.exchange,
market: this.config.market,
startTime: this.config.startTime.toISOString(),
endTime: this.config.endTime.toISOString(),
recordCount: this.collectedData.length,
collectedAt: new Date().toISOString(),
},
data: this.collectedData,
};
fs.writeFileSync(outputPath, JSON.stringify(dataToSave, null, 2));
console.log([Collector] Saved ${this.collectedData.length} records to ${outputPath});
}
stop(): void {
if (this.ws) {
this.ws.close();
this.ws = null;
}
}
}
// CLI 실행
if (require.main === module) {
const config: CollectorConfig = {
exchange: 'binance',
market: 'BTC/USDT',
dataType: 'l2_orderbook',
startTime: new Date(Date.now() - 3600000), // 1시간 전
endTime: new Date(),
outputPath: '../../data/binance-l2/btcusdt-l2.json',
};
const collector = new BinanceL2Collector(config);
process.on('SIGINT', () => {
console.log('\n[Collector] Stopping...');
collector.stop();
process.exit(0);
});
collector.collect().catch(console.error);
}
export { BinanceL2Collector };
바이낸스 L2 데이터 리플레이 엔진
// src/nodejs/replay.ts
import * as fs from 'fs';
import * as path from 'path';
import { L2Update, L2Snapshot, ReplayConfig, OrderBookEntry } from './types';
class BinanceL2Replay {
private config: ReplayConfig;
private currentIndex: number = 0;
private data: L2Update[] = [];
private currentOrderBook: L2Snapshot | null = null;
private replayInterval: NodeJS.Timeout | null = null;
private isRunning: boolean = false;
private lastTimestamp: number = 0;
constructor(config: ReplayConfig) {
this.config = config;
}
loadFromFile(filePath: string): void {
const absolutePath = path.resolve(filePath);
console.log([Replay] Loading data from ${absolutePath});
const rawData = fs.readFileSync(absolutePath, 'utf-8');
const parsed = JSON.parse(rawData);
this.data = parsed.data || [];
// 타임스탬프 기준 정렬
this.data.sort((a, b) => a.timestamp - b.timestamp);
console.log([Replay] Loaded ${this.data.length} records);
console.log([Replay] Time range: ${new Date(this.data[0]?.timestamp).toISOString()} - ${new Date(this.data[this.data.length - 1]?.timestamp).toISOString()});
}
loadFromData(data: L2Update[]): void {
this.data = [...data];
this.data.sort((a, b) => a.timestamp - b.timestamp);
console.log([Replay] Loaded ${this.data.length} records from memory);
}
start(): void {
if (this.isRunning) {
console.warn('[Replay] Already running');
return;
}
if (this.data.length === 0) {
throw new Error('No data to replay. Call loadFromFile() or loadFromData() first.');
}
this.isRunning = true;
this.currentIndex = 0;
console.log('[Replay] Starting replay...');
this.replayNext();
}
private replayNext(): void {
if (!this.isRunning || this.currentIndex >= this.data.length) {
this.stop();
this.config.onComplete?.();
return;
}
const currentData = this.data[this.currentIndex];
// 타임스탬프 기반 딜레이 (속도 조절)
if (this.lastTimestamp > 0) {
const targetDelay = (currentData.timestamp - this.lastTimestamp) / this.config.speed;
const actualDelay = Math.max(0, Math.min(targetDelay, 1000)); // 최대 1초
this.replayInterval = setTimeout(() => {
this.processData(currentData);
this.currentIndex++;
this.lastTimestamp = currentData.timestamp;
this.replayNext();
}, actualDelay);
} else {
this.processData(currentData);
this.currentIndex++;
this.lastTimestamp = currentData.timestamp;
this.replayNext();
}
}
private processData(data: L2Update): void {
if (data.type === 'snapshot' || !this.currentOrderBook) {
// 스냅샷 또는 초기 데이터
this.currentOrderBook = {
exchange: data.exchange,
market: data.market,
timestamp: data.timestamp,
asks: data.asks || [],
bids: data.bids || [],
};
} else {
// 업데이트 적용
if (data.asks) {
this.currentOrderBook.asks = this.applyUpdates(
this.currentOrderBook.asks,
data.asks
);
}
if (data.bids) {
this.currentOrderBook.bids = this.applyUpdates(
this.currentOrderBook.bids,
data.bids
);
}
this.currentOrderBook.timestamp = data.timestamp;
}
// 콜백 실행
try {
this.config.onUpdate(data);
} catch (error) {
this.config.onError?.(error as Error);
}
}
private applyUpdates(
current: OrderBookEntry[],
updates: OrderBookEntry[]
): OrderBookEntry[] {
const bookMap = new Map();
// 현재 오더북 복사
current.forEach(entry => {
bookMap.set(entry.price, entry.quantity);
});
// 업데이트 적용
updates.forEach(update => {
if (update.quantity === 0) {
bookMap.delete(update.price);
} else {
bookMap.set(update.price, update.quantity);
}
});
// 정렬된 배열로 변환
const result = Array.from(bookMap.entries())
.map(([price, quantity]) => ({ price, quantity }))
.sort((a, b) => a.price - b.price);
return result;
}
getCurrentOrderBook(): L2Snapshot | null {
return this.currentOrderBook;
}
getProgress(): { current: number; total: number; percent: number } {
return {
current: this.currentIndex,
total: this.data.length,
percent: this.data.length > 0
? Math.round((this.currentIndex / this.data.length) * 100)
: 0,
};
}
stop(): void {
this.isRunning = false;
if (this.replayInterval) {
clearTimeout(this.replayInterval);
this.replayInterval = null;
}
console.log('[Replay] Stopped');
}
pause(): void {
if (this.replayInterval) {
clearTimeout(this.replayInterval);
this.replayInterval = null;
}
console.log('[Replay] Paused');
}
resume(): void {
if (!this.isRunning) {
this.isRunning = true;
this.replayNext();
}
}
seekTo(timestamp: number): void {
const index = this.data.findIndex(d => d.timestamp >= timestamp);
if (index !== -1) {
this.currentIndex = index;
console.log([Replay] Seeked to index ${index});
}
}
}
// 사용 예시
function runReplayExample(): void {
const replay = new BinanceL2Replay({
exchange: 'binance',
market: 'BTC/USDT',
startTime: new Date(Date.now() - 3600000),
endTime: new Date(),
speed: 10.0, // 10배속
onUpdate: (data) => {
const book = replay.getCurrentOrderBook();
if (book && book.asks.length > 0 && book.bids.length > 0) {
const bestAsk = book.asks[0].price;
const bestBid = book.bids[0].price;
const spread = bestAsk - bestBid;
const midPrice = (bestAsk + bestBid) / 2;
// HolySheep AI API를 활용한 시장 상태 분석 (예시)
// DeepSeek V3.2로 스프레드 패턴 분석
if (data.timestamp % 5000 === 0) {
console.log([${new Date(data.timestamp).toISOString()}] +
BTC/USDT: Bid ${bestBid} | Ask ${bestAsk} | Spread ${spread.toFixed(2)} | Mid ${midPrice.toFixed(2)});
}
}
},
onError: (error) => {
console.error('[Replay Error]', error);
},
onComplete: () => {
console.log('[Replay] === REPLAY COMPLETE ===');
const progress = replay.getProgress();
console.log(Total processed: ${progress.total} records);
},
});
// 파일에서 로드
const dataPath = path.join(__dirname, '../../data/binance-l2/btcusdt-l2.json');
if (fs.existsSync(dataPath)) {
replay.loadFromFile(dataPath);
replay.start();
} else {
console.log('[Replay] No data file found. Please run collector first.');
}
// Graceful shutdown
process.on('SIGINT', () => {
console.log('\n[Replay] Shutting down...');
replay.stop();
process.exit(0);
});
}
if (require.main === module) {
runReplayExample();
}
export { BinanceL2Replay };
Python 구현
프로젝트 설정
cd src/python
requirements.txt 생성
cat > requirements.txt << 'EOF'
Core
pandas>=2.0.0
numpy>=1.24.0
pyarrow>=14.0.0
Async/WebSocket
aiohttp>=3.9.0
aiofiles>=23.0.0
websockets>=12.0
HolySheep AI Integration
openai>=1.0.0
Utilities
python-dotenv>=1.0.0
orjson>=3.9.0
msgpack>=1.0.0
loguru>=0.7.0
Optional: Tardis Machine SDK
tardis-client>=0.2.0
Data Processing
pyyaml>=6.0
pydantic>=2.0.0
EOF
pip install -r requirements.txt
바이낸스 L2 데이터 수집기 (Python)
#!/usr/bin/env python3
"""
Binance L2 Orderbook Data Collector
Tardis Machine 및 Binance WebSocket 활용
"""
import asyncio
import json
import os
import time
from datetime import datetime, timedelta
from typing import Optional, Callable
from dataclasses import dataclass, field
from pathlib import Path
import aiofiles
import aiohttp
import websockets
import pyarrow as pa
import pyarrow.parquet as pq
from loguru import logger
HolySheep AI 설정 (DeepSeek V3.2로 비용 효율적 처리)
API Key: https://www.holysheep.ai/register
os.environ['HOLYSHEEP_API_KEY'] = os.getenv('HOLYSHEEP_API_KEY', 'YOUR_HOLYSHEEP_API_KEY')
os.environ['HOLYSHEEP_BASE_URL'] = 'https://api.holysheep.ai/v1'
logger.add("collector_{time}.log", rotation="1 day")
@dataclass
class L2Update:
"""L2 오더북 업데이트 데이터"""
exchange: str
market: str
timestamp: int
asks: list[list[float]] # [[price, quantity], ...]
bids: list[list[float]]
update_type: str = "snapshot"
def to_dict(self):
return {
"exchange": self.exchange,
"market": self.market,
"timestamp": self.timestamp,
"asks": self.asks,
"bids": self.bids,
"type": self.update_type,
}
@dataclass
class CollectorConfig:
"""수집기 설정"""
exchange: str = "binance"
market: str = "BTCUSDT"
data_type: str = "l2_orderbook"
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
output_dir: str = "../../data/binance-l2"
use_tardis: bool = True
tardis_api_key: Optional[str] = None
class BinanceL2Collector:
"""바이낸스 L2 오더북 데이터 수집기"""
def __init__(self, config: CollectorConfig):
self.config = config
self.collected_data: list[L2Update] = []
self.ws_connection: Optional[websockets.WebSocketClientProtocol] = None
self.is_running = False
async def collect(self):
"""데이터 수집 메인 로직"""
logger.info(f"Starting collection for {self.config.market}")
logger.info(f"Period: {self.config.start_time} - {self.config.end_time}")
self.is_running = True
# 동시 실행: Tardis API + WebSocket
tasks = []
if self.config.use_tardis:
tasks.append(self._collect_from_tardis())
tasks.append(self._collect_from_websocket())
await asyncio.gather(*tasks, return_exceptions=True)
# 데이터 저장
await self._save_data()
async def _collect_from_tardis(self):
"""Tardis Machine API에서 데이터 수집"""
if not self.config.tardis_api_key:
logger.warning("Tardis API key not provided, skipping API collection")
return
base_url = "https://api.tardis.dev/v1"
params = {
"exchange": self.config.exchange,
"symbols": self.config.market,
"start_date": self.config.start_time.strftime("%Y-%m-%d"),
"end_date": self.config.end_time.strftime("%Y-%m-%d"),
"data_type": self.config.data_type,
}
logger.info(f"Fetching from Tardis API: {params}")
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{base_url}/historical",
params=params,
headers={"Authorization": f"Bearer {self.config.tardis_api_key}"},
timeout=aiohttp.ClientTimeout(total=120),
) as response:
if response.status == 200:
data = await response.json()
if isinstance(data, list):
for item in data:
l2_update = L2Update(
exchange=item.get("exchange", "binance"),
market=item.get("symbol", self.config.market),
timestamp=item.get("timestamp", 0),
asks=item.get("asks", []),
bids=item.get("bids", []),
update_type=item.get("type", "snapshot"),
)
self.collected_data.append(l2_update)
logger.info(f"Collected {len(data)} records from Tardis API")
except Exception as e:
logger.error(f"Tardis API error: {e}")
async def _collect_from_websocket(self):
"""바이낸스 WebSocket에서 실시간 데이터 수집"""
symbol = self.config.market.lower()
ws_url = f"wss://stream.binance.com:9443/stream?streams={symbol}@depth20@100ms"
logger.info(f"Connecting to Binance WebSocket: {ws_url}")
collection_duration = 60000 # 최대 1분 수집 (데모용)
start_time = time.time()
try:
async with websockets.connect(ws_url, ping_interval=30) as websocket:
self.ws_connection = websocket
while self.is_running and (time.time() - start_time) * 1000 < collection_duration:
try:
message = await asyncio.wait_for(
websocket.recv(),
timeout=5.0
)
data = json.loads(message)
stream_data = data.get("data", {})
if "lastUpdateId" in stream_data:
l2_update = L2Update(
exchange="binance",
market=self.config.market,
timestamp=int(time.time() * 1000),
asks=[[float(a[0]), float(a[1])] for a in stream_data.get("asks", [])],
bids=[[float(b[0]), float(b[1])] for b in stream_data.get("bids", [])],
update_type="snapshot",
)
self.collected_data.append(l2_update)
if len(self.collected_data) % 100 == 0:
logger.debug(f"Collected {len(self.collected_data)} updates")
except asyncio.TimeoutError:
continue
except websockets.exceptions.ConnectionClosed:
logger.info("WebSocket connection closed")
except Exception as e:
logger.error(f"WebSocket error: {e}")
async def _save_data(self):
"""수집된 데이터 저장 (Parquet + JSON)"""
output_dir = Path(self.config.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
# JSON 형식으로 저장
json_path = output_dir / f"{self.config.market.lower()}_l2_{timestamp_str}.json"
async with aiofiles.open(json_path, "w") as f:
metadata = {
"exchange": self.config.exchange,
"market": self.config.market,
"start_time": self.config.start_time.isoformat() if self.config.start_time else None,
"end_time": self.config.end_time.isoformat() if self.config.end_time else None,
"record_count": len(self.collected_data),
"collected_at": datetime.now().isoformat(),
}
data_dicts = [d.to_dict() for d in self.collected_data]
await f.write(json.dumps({"metadata": metadata, "data": data_dicts}, indent=2))
logger.info(f"Saved JSON: {json_path} ({len(self.collected_data)} records)")
# Parquet 형식으로 저장 (대용량 데이터용)
parquet_path = output_dir / f"{self.config.market.lower()}_l2_{timestamp_str}.parquet"
table = pa.table({
"timestamp": [d.timestamp for d in self.collected_data],
"market": [d.market for d in self.collected_data],
"ask_prices": [[a[0] for a in d.asks] for d in self.collected_data],
"ask_quantities": [[a[1] for a in d.asks] for d in self.collected_data],
"bid_prices": [[b[0] for b in d.bids] for d in self.collected_data],
"bid_quantities": [[b[1] for b in d.bids] for d in self.collected_data],
"type": [d.update_type for d in self.collected_data],
})
pq.write_table(table, parquet_path)
logger.info(f"Saved Parquet: {parquet_path}")
return str(json_path)
def stop(self):
"""수집 중지"""
self.is_running = False
async def main():
"""CLI 실행"""
config = CollectorConfig(
market="BTCUSDT",
start_time=datetime.now() - timedelta(hours=1),
end_time=datetime.now(),
output_dir="../../data/binance-l2",
use_tardis=False, # 데모용 WebSocket만 사용
tardis_api_key=os.getenv("TARDIS_API_KEY"),
)
collector = BinanceL2Collector(config)
try:
await collector.collect()
except KeyboardInterrupt:
logger.info("Interrupted by user")
collector.stop()
if __name__ == "__main__":
asyncio.run(main())
바이낸스 L2 리플레이 엔진 (Python)
#!/usr/bin/env python3
"""
Binance L2 Orderbook Replay Engine
수집된 데이터를 타임스탬프 기반으로 리플레이
"""
import asyncio
import json
import time
from datetime import datetime
from typing import Optional, Callable
from dataclasses import dataclass, field
from pathlib import Path
import heapq
import pandas as pd
import pyarrow.parquet as pq
from loguru import logger
logger.add("replay_{time}.log", rotation="1 day")
@dataclass
class L2OrderBook:
"""오더북 상태"""
asks: list[tuple[float, float]] # [(price, quantity), ...] 정렬됨
bids: list[tuple[float, float]]
timestamp: int
market: str
@classmethod
def empty(cls):
return cls(
asks=[],
bids=[],
timestamp=0,
market=""
)
@dataclass
class ReplayConfig:
"""리플레이 설정"""
speed: float = 1.0 # 1.0 = realtime, >1.0 = faster
on_update: Optional[Callable] = None
on_orderbook_change: Optional[Callable] = None
on_error: Optional[Callable] = None
on_complete: Optional[Callable] = None
start_timestamp: Optional[int] = None
end_timestamp: Optional[int] = None
class BinanceL2Replay:
"""바이낸스 L2 데이터 리플레이 엔진"""
def __init__(self, config: ReplayConfig = None):
self.config = config or ReplayConfig()
self.data: list[dict] = []
self.current_index = 0
self.current_orderbook = L2OrderBook.empty()
self.is_running = False
self.is_paused = False
self.last_timestamp = 0
self.task: Optional[asyncio.Task] = None
def load_from_file(self, file_path: str):
"""JSON 파일에서 데이터 로드"""
path = Path(file_path)
logger.info(f"Loading data from {path}")
with open(path, "r") as f:
raw = json.load(f)
self.data = raw.get("data", [])
# 타임스탬프 기준 정렬
self.data.sort(key=lambda x: x.get("timestamp", 0))
logger.info(f"Loaded {len(self.data)} records")
if self.data:
first_ts = self.data[0].get("timestamp", 0)
last_ts = self.data[-1].get("timestamp", 0)
logger.info(f"Time range: {datetime.fromtimestamp(first_ts/1000)} - {datetime.fromtimestamp(last_ts/1000)}")
def load_from_parquet(self, file_path: str):
"""Parquet 파일에서 데이터 로드"""
logger.info(f"Loading from Parquet: {file_path}")
table = pq.read_table(file_path)
df = table.to_pandas()
# Parquet 구조를 L2 업데이트 형식으로 변환
self.data = []
for _, row in df.iterrows():
asks = list(zip(row["ask_prices"], row["ask_quantities"]))
bids = list(zip(row["bid_prices"], row["bid_quantities"]))
self.data.append({
"timestamp": row["timestamp"],
"market": row["market"],
"asks": [[p, q] for p, q in asks if p > 0 and q > 0],
"bids": [[p, q] for p, q in bids if p > 0 and q > 0],
"type": row["type"],
})
self.data.sort(key=lambda x: x["timestamp"])
logger.info(f"Loaded {len(self.data)} records from Parquet")
async def start(self):
"""리플레이 시작"""
if self.is_running:
logger.warning("Already running")
return
if not self.data:
raise ValueError("No data to replay. Call load_from_file() first.")
self.is_running = True
self.is_paused = False
self.current_index = 0
logger.info("Starting replay...")
self.task = asyncio.create_task(self._replay_loop())
async def _replay_loop(self):
"""메인 리플레이 루프"""
while self.is_running and self.current_index < len(self.data):
if self.is_paused:
await asyncio.sleep(0.1)
continue
record = self.data[self.current_index]
# 시간 필터
timestamp = record.get("timestamp", 0)
if self.config.start_timestamp and timestamp < self.config.start_timestamp:
self.current_index += 1
continue
if self.config.end_timestamp and timestamp > self.config.end_timestamp:
break
# 딜레이 계산
if self.last_timestamp > 0:
target_delay