高頻度取引（HFT）システムにおいて、過去の市場データを正確に再現し
Tardis とは： Tick データインフラの地位
Tardisは暗号資産および伝統的金融市場向けのティックレベルデータ提供商として知られています。彼らの提供する низкоуровневая API は、米国の CME、NASDAQ，欧洲の Euronext、アジアのJPXを含む 主要取引所のリアルタイム・歴史的データを統一スキーマで 提供します。
私が以前担当していたプロジェクトでは、日次 約 50GB の Tick データを処理し、複数の
Tick データリプレイのアーキテクチャ設計
リプレイエンジンの核心原理
Tick データリプレイの本质は、タイムスタンプ順に排列された 市场 событий を忠実に再現し、 市场微型構造（Market Microstructure）を模擬することです。Tardis はこのために2つの主要モードを提供します：
- Live Mode：リアルタイムストリーミング、生データ即刻�
- Historical Replay Mode：指定時間範囲のデータを историческая фазе で再生
import httpx
import asyncio
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Optional
import json
HolySheep AI API 設定 - Tardis データ統合用
BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class TickData:
exchange: str
symbol: str
timestamp: datetime
price: float
volume: float
side: str # 'bid' or 'ask'
order_id: Optional[str] = None
@dataclass
class OrderbookSnapshot:
exchange: str
symbol: str
timestamp: datetime
bids: List[tuple] # [(price, size), ...]
asks: List[tuple] # [(price, size), ...]
class TardisReplayer:
"""
Tardis исторических данных リプレイクライアント
リアルタイム/歴史的両モード対応
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=60.0)
async def fetch_historical_ticks(
self,
exchange: str,
symbol: str,
start: datetime,
end: datetime,
filters: Optional[dict] = None
) -> List[TickData]:
"""
指定期間の Tick データをフェッチ
性能要件：<50ms API レイテンシ (HolySheep 利用時)
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"exchange": exchange,
"symbol": symbol,
"start": start.isoformat(),
"end": end.isoformat(),
"filters": filters or {}
}
response = await self.client.post(
f"{BASE_URL}/tardis/historical/ticks",
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
return [
TickData(
exchange=t['exchange'],
symbol=t['symbol'],
timestamp=datetime.fromisoformat(t['timestamp']),
price=t['price'],
volume=t['volume'],
side=t['side'],
order_id=t.get('order_id')
)
for t in data['ticks']
]
async def replay_ticks(
self,
ticks: List[TickData],
playback_speed: float = 1.0,
on_tick_callback=None
):
"""
Tick リプレイエンジン
playback_speed: 1.0 = リアルタイム、10.0 = 10倍速
"""
if not ticks:
return
# 最初と最後の Tick から再生時間を計算
start_time = ticks[0].timestamp
end_time = ticks[-1].timestamp
real_duration = (end_time - start_time).total_seconds()
playback_duration = real_duration / playback_speed
tick_interval = playback_duration / len(ticks) if len(ticks) > 1 else 0
for i, tick in enumerate(ticks):
# Playback速度に応じた待機
if i > 0 and tick_interval > 0:
await asyncio.sleep(tick_interval)
if on_tick_callback:
await on_tick_callback(tick, i, len(ticks))
async def close(self):
await self.client.aclose()
使用例
async def main():
replayer = TardisReplayer(HOLYSHEEP_API_KEY)
try:
# 2024年 Q4 の BTC/USD ティックデータを取得
ticks = await replayer.fetch_historical_ticks(
exchange="binance",
symbol="BTC-USDT",
start=datetime(2024, 10, 1),
end=datetime(2024, 10, 2),
filters={"side": "both"} # 約定と板情報双方
)
print(f"取得 Tick 数: {len(ticks):,}")
print(f"期間: {ticks[0].timestamp} - {ticks[-1].timestamp}")
# 10倍速でリプレイ
await replayer.replay_ticks(
ticks,
playback_speed=10.0,
on_tick_callback=process_tick
)
finally:
await replayer.close()
async def process_tick(tick: TickData, index: int, total: int):
if index % 10000 == 0:
print(f"Progress: {index/total*100:.1f}% - {tick.price}")
if __name__ == "__main__":
asyncio.run(main())
Orderbook リconstruction アルゴリズム
Tick データから
レベル別価格 aggregation
from collections import defaultdict
from sortedcontainers import SortedDict
from typing import Dict, Tuple, List
import threading
from dataclasses import dataclass, field
@dataclass
class OrderbookLevel:
price: float
size: float
order_count: int
class OrderbookReconstructor:
"""
Tick データから Orderbook を再構築
性能要件: 100万Tick/秒処理能力
"""
def __init__(self, price_precision: int = 2):
self.price_precision = price_precision
self.bids = SortedDict() # price -> OrderbookLevel
self.asks = SortedDict()
self.last_update_id: int = 0
self.sequence_gaps: List[Tuple[int, int]] = []
def _round_price(self, price: float) -> float:
"""価格精度的统一处理"""
return round(price, self.price_precision)
def apply_snapshot(self, snapshot: OrderbookSnapshot):
"""フルスナップショットで板を初期化"""
self.bids.clear()
self.asks.clear()
for price, size in snapshot.bids:
rounded_price = self._round_price(price)
self.bids[rounded_price] = OrderbookLevel(
price=rounded_price,
size=size,
order_count=1
)
for price, size in snapshot.asks:
rounded_price = self._round_price(price)
self.asks[rounded_price]