Tardis Machine本地回放API实战：Pythonで暗号市場の限価注文簿を任意時刻に重建する

環境構築

# Python 3.10+ での必要ライブラリインストール
pip install requests asyncio aiohttp pandas numpy
pip install websockets # WebSocket接続用
pip install python-dotenv # 環境変数管理用

プロジェクト構造
/project_root
├── config/
│   └── settings.py
├── src/
│   ├── api_client.py
│   ├── orderbook_reconstructor.py
│   └── market_analyzer.py
├── tests/
│   └── test_replay.py
└── .env

設定ファイルの実装

# config/settings.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class HolySheepConfig:
    """HolySheep API設定"""
    base_url: str = "https://api.holysheep.ai/v1"
    api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    
    # Tardis Machine API エンドポイント
    exchanges_endpoint: str = "/tardis/exchanges"
    symbols_endpoint: str = "/tardis/symbols"
    historical_endpoint: str = "/tardis/historical"
    
    # 接続設定
    timeout: int = 30
    max_retries: int = 3
    rate_limit_rpm: int = 60

@dataclass  
class MarketDataConfig:
    """市場データ設定"""
    exchange: str = "binance"
    symbol: str = "BTC-USDT"
    market_type: str = "futures"  # futures or spot
    
    # データ粒度
    data_types: list = None  # ["book", "trade", "ticker"]
    
    def __post_init__(self):
        if self.data_types is None:
            self.data_types = ["book"]
    
    # 時間設定（UTC）
    start_time: str = "2024-01-15T00:00:00Z"
    end_time: str = "2024-01-15T01:00:00Z"

グローバル設定インスタンス
config = HolySheepConfig()
market_config = MarketDataConfig()

APIクライアントの実装

# src/api_client.py
import requests
import time
import logging
from typing import Optional, Dict, Any, List
from config.settings import config

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class HolySheepTardisClient:
    """HolySheep Tardis Machine APIクライアント"""
    
    def __init__(self, api_key: Optional[str] = None):
        self.base_url = config.base_url
        self.api_key = api_key or config.api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        
    def _request(self, method: str, endpoint: str, params: Optional[Dict] = None) -> Dict:
        """APIリクエストの共通処理"""
        url = f"{self.base_url}{endpoint}"
        retry_count = 0
        
        while retry_count < config.max_retries:
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    params=params,
                    timeout=config.timeout
                )
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.RequestException as e:
                retry_count += 1
                logger.warning(f"リクエスト失敗 ({retry_count}/{config.max_retries}): {e}")
                if retry_count >= config.max_retries:
                    raise RuntimeError(f"API要求が{max_retry}回失敗しました: {e}")
                time.sleep(2 ** retry_count)  # 指数バックオフ
                
    def get_available_exchanges(self) -> List[Dict]:
        """利用可能な取引所一覧を取得"""
        return self._request("GET", config.exchanges_endpoint)
    
    def get_symbols(self, exchange: str) -> List[Dict]:
        """取引所の取扱.symbol一覧を取得"""
        params = {"exchange": exchange}
        return self._request("GET", config.symbols_endpoint, params)
    
    def get_historical_data(
        self,
        exchange: str,
        symbol: str,
        start_time: str,
        end_time: str,
        data_type: str = "book",
        compression: str = "none"
    ) -> Dict:
        """
        歴史的市場データを取得
        
        Args:
            exchange: 取引所名（binance, okx, bybit等）
            symbol: .symbol名（BTC-USDT等）
            start_time: 開始時刻（ISO 8601形式）
            end_time: 終了時刻
            data_type: データ種類（book, trade, ticker）
            compression: 圧縮形式（none, gzip）
        """
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "startTime": start_time,
            "endTime": end_time,
            "type": data_type,
            "compression": compression
        }
        return self._request("GET", config.historical_endpoint, params)
    
    def get_book_snapshot(self, exchange: str, symbol: str, timestamp: int) -> Dict:
        """特定時刻の板情報スナップショットを取得"""
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "timestamp": timestamp,
            "type": "book"
        }
        return self._request("GET", "/tardis/snapshot", params)

使用例
if __name__ == "__main__":
    client = HolySheepTardisClient()
    
    # 利用可能な取引所一覧
    exchanges = client.get_available_exchanges()
    print(f"対応取引所: {[e['name'] for e in exchanges]}")
    
    # Binance先物のBTC先物.symbol一覧
    symbols = client.get_symbols("binance")
    btc_symbols = [s for s in symbols if "BTC" in s]
    print(f"BTC関連.symbol: {btc_symbols[:5]}")

限価注文簿重建クラスの実装

# src/orderbook_reconstructor.py
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from collections import defaultdict
import logging

logger = logging.getLogger(__name__)

@dataclass
class Order:
    """注文を表現するデータクラス"""
    price: float
    quantity: float
    side: str  # "bid" or "ask"
    order_id: str = ""
    timestamp: int = 0
    
    def is_empty(self) -> bool:
        return self.quantity <= 0

@dataclass
class Level:
    """板のレベル（価格帯）を表現"""
    price: float
    quantity: float
    orders: List[Order] = field(default_factory=list)
    
    def __post_init__(self):
        if not isinstance(self.orders, list):
            self.orders = []
            
class LimitOrderBook:
    """
    限価注文簿クラス
    
    板情報のリプレイと任意時刻の状態復元を管理
    """
    
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.bids: Dict[float, Level] = {}  # 買い注文（価格->Level）
        self.asks: Dict[float, Level] = {}   # 売り注文（価格->Level）
        self.sequence: int = 0
        self.last_update_time: int = 0
        self.trade_history: List[Dict] = []
        
    def apply_book_update(self, update_data: Dict) -> None:
        """板の增量更新を適用"""
        self.sequence = update_data.get("sequence", self.sequence + 1)
        self.last_update_time = update_data.get("timestamp", 0)
        
        # 買い注文の更新
        for bid in update_data.get("bids", []):
            price, quantity = bid["price"], bid["quantity"]
            if quantity <= 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = Level(price=price, quantity=quantity)
                
        # 売り注文の更新
        for ask in update_data.get("asks", []):
            price, quantity = ask["price"], ask["quantity"]
            if quantity <= 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = Level(price=price, quantity=quantity)
                
        # ソート維持（ bids: 降順、asks: 昇順）
        self.bids = dict(sorted(self.bids.items(), reverse=True))
        self.asks = dict(sorted(self.asks.items()))
        
    def apply_trade(self, trade_data: Dict) -> None:
        """約定履歴を適用し、板を更新"""
        price = trade_data["price"]
        quantity = trade_data["quantity"]
        side = trade_data.get("side", "buy")  # taker侧方向
        
        self.trade_history.append(trade_data)
        
        # 成行約定による板の消費
        book_side = self.bids if side == "buy" else self.asks
        for level_price in list(book_side.keys()):
            if (side == "buy" and level_price < price) or \
               (side == "sell" and level_price > price):
                book_side.pop(level_price, None)
                
    def get_best_bid_ask(self) -> Tuple[Optional[float], Optional[float]]:
        """最良気配値を取得"""
        best_bid = max(self.bids.keys()) if self.bids else None
        best_ask = min(self.asks.keys()) if self.asks else None
        return best_bid, best_ask
    
    def get_mid_price(self) -> Optional[float]:
        """仲値を取得"""
        best_bid, best_ask = self.get_best_bid_ask()
        if best_bid and best_ask:
            return (best_bid + best_ask) / 2
        return None
        
    def get_spread(self) -> Optional[float]:
        """スプレッド（pip）を取得"""
        best_bid, best_ask = self.get_best_bid_ask()
        if best_bid and best_ask:
            return best_ask - best_bid
        return None
        
    def get_depth(self, levels: int = 10) -> Dict:
        """指定レベルの板の厚みを取得"""
        return {
            "bids": [
                {"price": p, "quantity": l.quantity} 
                for p, l in list(self.bids.items())[:levels]
            ],
            "asks": [
                {"price": p, "quantity": l.quantity} 
                for p, l in list(self.asks.items())[:levels]
            ]
        }
        
    def to_dict(self) -> Dict:
        """辞書形式に変換"""
        best_bid, best_ask = self.get_best_bid_ask()
        return {
            "symbol": self.symbol,
            "timestamp": self.last_update_time,
            "sequence": self.sequence,
            "best_bid": best_bid,
            "best_ask": best_ask,
            "mid_price": self.get_mid_price(),
            "spread": self.get_spread(),
            "bid_levels": len(self.bids),
            "ask_levels": len(self.asks),
            "depth": self.get_depth(20)
        }


class OrderBookReplayEngine:
    """
    注文簿リプレイエンジン
    
    历史データから特定時刻の板状態を復元
    """
    
    def __init__(self, client):
        self.client = client
        self.cached_snapshots: Dict[int, LimitOrderBook] = {}
        
    def replay_to_timestamp(
        self,
        exchange: str,
        symbol: str,
        target_timestamp: int,
        start_timestamp: Optional[int] = None
    ) -> LimitOrderBook:
        """
        指定時刻までリプレイ
        
        Args:
            exchange: 取引所
            symbol: .symbol
            target_timestamp: 目標時刻（ミリ秒Unixタイムスタンプ）
            start_timestamp: 開始時刻（Noneの場合は直近のスナップショットから）
            
        Returns:
            目標時刻の板状態
        """
        lob = LimitOrderBook(symbol)
        
        # 最も近いスナップショットを取得
        snapshot_time = self._find_nearest_snapshot(target_timestamp)
        
        if snapshot_time:
            snapshot_data = self.client.get_book_snapshot(
                exchange, symbol, snapshot_time
            )
            lob.apply_book_update(snapshot_data)
            
        # 增量更新を適用
        updates = self._fetch_updates_between(
            exchange, symbol, 
            snapshot_time or start_timestamp,
            target_timestamp
        )
        
        for update in updates:
            lob.apply_book_update(update)
            
        logger.info(
            f"リプレイ完了: {symbol} @ {target_timestamp} "
            f"(bid: {lob.get_best_bid_ask()[0]}, ask: {lob.get_best_bid_ask()[1]})"
        )
        
        return lob
        
    def _find_nearest_snapshot(self, timestamp: int) -> Optional[int]:
        """最寄りのスナップショット時刻を検索"""
        # 実際の実装ではスナップショットインデックスAPIを使用
        # Binance先物の場合、1分足のsnapshotが利用可能
        return (timestamp // 60000) * 60000  # 1分 单位に丸める
        
    def _fetch_updates_between(
        self,
        exchange: str,
        symbol: str,
        start: int,
        end: int
    ) -> List[Dict]:
        """期間中の增量更新を取得"""
        from datetime import datetime
        from config.settings import market_config
        
        start_iso = datetime.utcfromtimestamp(start / 1000).isoformat() + "Z"
        end_iso = datetime.utcfromtimestamp(end / 1000).isoformat() + "Z"
        
        data = self.client.get_historical_data(
            exchange=exchange,
            symbol=symbol,
            start_time=start_iso,
            end_time=end_iso,
            data_type="book"
        )
        
        return data.get("book", [])

使用例
if __name__ == "__main__":
    from src.api_client import HolySheepTardisClient
    
    client = HolySheepTardisClient()
    engine = OrderBookReplayEngine(client)
    
    # 2024年1月15日00:30:00のBTC-USDT板を復元
    target_ts = 1705276200000  # UTCミリ秒
    
    lob = engine.replay_to_timestamp(
        exchange="binance",
        symbol="BTC-USDT",
        target_timestamp=target_ts
    )
    
    print(f"復元された板情報:")
    print(f"  -symbol: {lob.symbol}")
    print(f"  -best_bid: {lob.get_best_bid_ask()[0]}")
    print(f"  -best_ask: {lob.get_best_bid_ask()[1]}")
    print(f"  -mid_price: {lob.get_mid_price()}")
    print(f"  -spread: {lob.get_spread()}")

市場分析ダッシュボードの実装

# src/market_analyzer.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from src.orderbook_reconstructor import OrderBookReplayEngine, LimitOrderBook
from src.api_client import HolySheepTardisClient

class MarketAnalyzer:
    """
    市場分析クラス
    
    リプレイデータを活用したテクニクス分析と戦略検証
    """
    
    def __init__(self, client: HolySheepTardisClient):
        self.client = client
        self.replay_engine = OrderBookReplayEngine(client)
        
    def analyze_spread_dynamics(
        self,
        exchange: str,
        symbol: str,
        start_ts: int,
        end_ts: int,
        interval_ms: int = 60000
    ) -> pd.DataFrame:
        """
        スプレッドの時系列分析
        
        Args:
            exchange: 取引所
            symbol: .symbol
            start_ts: 開始タイムスタンプ（ミリ秒）
            end_ts: 終了タイムスタンプ
            interval_ms: サンプリング間隔（デフォルト1分）
        """
        results = []
        current_ts = start_ts
        
        while current_ts <= end_ts:
            lob = self.replay_engine.replay_to_timestamp(
                exchange, symbol, current_ts, start_ts
            )
            
            best_bid, best_ask = lob.get_best_bid_ask()
            if best_bid and best_ask:
                spread = best_ask - best_bid
                spread_pct = (spread / best_ask) * 100
                
                results.append({
                    "timestamp": current_ts,
                    "datetime": datetime.utcfromtimestamp(current_ts / 1000),
                    "best_bid": best_bid,
                    "best_ask": best_ask,
                    "mid_price": lob.get_mid_price(),
                    "spread": spread,
                    "spread_pct": spread_pct,
                    "bid_depth_10": sum(l.quantity for l in list(lob.bids.values())[:10]),
                    "ask_depth_10": sum(l.quantity for l in list(lob.asks.values())[:10])
                })
                
            current_ts += interval_ms
            
        return pd.DataFrame(results)
    
    def calculate_orderbook_imbalance(self, lob: LimitOrderBook) -> float:
        """
        板不均衡度を計算
        
        正の値: 買い圧力が強い
        負の値: 売り圧力が強い
        """
        bid_volume = sum(l.quantity for l in lob.bids.values()[:20])
        ask_volume = sum(l.quantity for l in lob.asks.values()[:20])
        
        if bid_volume + ask_volume == 0:
            return 0.0
            
        return (bid_volume - ask_volume) / (bid_volume + ask_volume)
    
    def backtest_market_making_strategy(
        self,
        exchange: str,
        symbol: str,
        start_ts: int,
        end_ts: int,
        spread_pct: float = 0.001,
        inventory_target: float = 0.5
    ) -> Dict:
        """
        マーケットメイク戦略のバックテスト
        
        Args:
            spread_pct: 設定スプレッド（%））
            inventory_target: ポジション目標比率
        """
        current_ts = start_ts
        position = 0.0
        cash = 0.0
        trades = []
        
        while current_ts <= end_ts:
            lob = self.replay_engine.replay_to_timestamp(
                exchange, symbol, current_ts, start_ts
            )
            
            mid = lob.get_mid_price()
            if mid is None:
                current_ts += 60000
                continue
                
            # エントリー価格
            bid_price = mid * (1 - spread_pct)
            ask_price = mid * (1 + spread_pct)
            
            # 板不均衡に基づくポジション调整
            imbalance = self.calculate_orderbook_imbalance(lob)
            
            # 裁定取引シミュレーション
            if imbalance < -0.3:  # 売り圧力が強い
                # 成行き買いエントリー
                position += 0.1
                cash -= lob.get_best_ask() * 0.1
                trades.append({"ts": current_ts, "action": "buy", "price": lob.get_best_ask()})
                
            elif imbalance > 0.3:  # 買い圧力が強い
                # 成行き売りエントリー
                position -= 0.1
                cash += lob.get_best_bid() * 0.1
                trades.append({"ts": current_ts, "action": "sell", "price": lob.get_best_bid()})
                
            current_ts += 60000
            
        # 最終清算
        final_lob = self.replay_engine.replay_to_timestamp(
            exchange, symbol, end_ts, start_ts
        )
        final_price = final_lob.get_mid_price()
        
        total_pnl = cash + (position * final_price) if final_price else cash
        
        return {
            "total_trades": len(trades),
            "final_position": position,
            "cash": cash,
            "total_pnl": total_pnl,
            "trades": trades[-10:]  # 最近の10件
        }


使用例
if __name__ == "__main__":
    client = HolySheepTardisClient()
    analyzer = MarketAnalyzer(client)
    
    # 1時間のスプレッド分析
    start = 1705276200000  # 2024-01-15 00:30:00 UTC
    end = 1705279800000    # 2024-01-15 01:30:00 UTC
    
    spread_df = analyzer.analyze_spread_dynamics(
        exchange="binance",
        symbol="BTC-USDT",
        start_ts=start,
        end_ts=end,
        interval_ms=60000
    )
    
    print("=== スプレッド分析サマリー ===")
    print(f"平均スプレッド: {spread_df['spread'].mean():.2f}")
    print(f"最大スプレッド: {spread_df['spread'].max():.2f}")
    print(f"最小スプレッド: {spread_df['spread'].min():.2f}")
    print(f"平均mid_price: {spread_df['mid_price'].mean():.2f}")
    
    # マーケットメイク戦略のバックテスト
    bt_result = analyzer.backtest_market_making_strategy(
        exchange="binance",
        symbol="BTC-USDT",
        start_ts=start,
        end_ts=end
    )
    
    print(f"\n=== バックテスト結果 ===")
    print(f"総取引回数: {bt_result['total_trades']}")
    print(f"最終ポジション: {bt_result['final_position']}")
    print(f"累計損益: {bt_result['total_pnl']:.2f} USDT")

よくあるエラーと対処法

エラー1: API鍵の認証失敗（401 Unauthorized）

# 錯誤状況
{"error": "Invalid API key", "status": 401}

解決策
1. .envファイルのAPI鍵を確認
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

2. 環境変数の直接設定
import os
os.environ["HOLYSHEEP_API_KEY"] = "sk-holysheep-xxxxx"

3. クライアント初期化時に明示的に指定
client = HolySheepTardisClient(api_key="sk-holysheep-xxxxx")

4. API鍵のテスト
def verify_api_key(client):
    try:
        exchanges = client.get_available_exchanges()
        print(f"認証成功: {len(exchanges)}件の取引所に対応")
        return True
    except Exception as e:
        print(f"認証失敗: {e}")
        return False

verify_api_key(client)

エラー2: レートリミット超過（429 Too Many Requests）

# 錯誤状況
{"error": "Rate limit exceeded", "status": 429}

解決策
1. rate_limit_rpmの設定確認と調整
from config.settings import config
config.rate_limit_rpm = 30  # 安全そうな値に削減

2. リクエスト間に遅延を追加
import time
def throttled_request(func):
    def wrapper(*args, **kwargs):
        time.sleep(60 / config.rate_limit_rpm)  # 1分あたりの要求数で割る
        return func(*args, **kwargs)
    return wrapper

3. 指数バックオフの実装
MAX_RETRIES = 5
BASE_DELAY = 1

def retry_with_backoff(func):
    def wrapper(*args, **kwargs):
        for attempt in range(MAX_RETRIES):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if "429" in str(e):
                    delay = BASE_DELAY * (2 ** attempt)
                    print(f"レートリミット。再試行まで{delay}秒待機...")
                    time.sleep(delay)
                else:
                    raise
    return wrapper

4. 批量リクエストの使用
單一symbol请求ではなく、批量エンドポイントを活用
/tardis/batch 等の専用エンドポイントが存在する場合はそちらを使用

エラー3: データ取得範囲外（400 Bad Request / Timestamp Out of Range）

# 錯誤状況
{"error": "Timestamp out of available range", "status": 400}

解決策
1. 利用可能なデータ範囲を確認
def check_data_availability(client, exchange, symbol):
    # 取引所別のデータ保持期間を確認
    DATA_RETENTION = {
        "binance": 30 * 24 * 60 * 60 * 1000,  # 30日（ミリ秒）
        "okx": 7 * 24 * 60 * 60 * 1000,       # 7日
        "bybit": 14 * 24 * 60 * 60 * 1000,     # 14日
    }
    
    max_age = DATA_RETENTION.get(exchange, 7 * 24 * 60 * 60 * 1000)
    now = int(time.time() * 1000)
    earliest = now - max_age
    
    return {"earliest": earliest, "now": now, "max_age_days": max_age // (86400000)}

2. 時刻のバリデーション
from datetime import datetime

def validate_timestamp(ts_ms: int) -> bool:
    now = int(datetime.utcnow().timestamp() * 1000)
    one_year_ago = now - (365 * 24 * 60 * 60 * 1000)
    return one_year_ago < ts_ms < now

3. 代替アプローチ：スナップショットベースのリプレイ
def replay_within_range(client, exchange, symbol, target_ts, buffer_ms=3600000):
    """1時間前のスナップショットから開始"""
    safe_start = target_ts - buffer_ms
    return engine.replay_to_timestamp(exchange, symbol, target_ts, safe_start)

4. データ補間による長期バックテスト
1分足のOHLCデータで初期化し、板形状を統計的に復元
def estimate_orderbook_from_ohlc(mid_price, volatility, levels=20):
    """OHLCデータから板形状を概算"""
    spread = mid_price * 0.001  # 0.1%スプレッドを想定
    bids = {mid_price - i * spread: mid_price * 0.1 for i in range(1, levels)}
    asks = {mid_price + i * spread: mid_price * 0.1 for i in range(1, levels)}
    return bids, asks

移行プレイブック：HolySheep AIへの切り替え

Step 1: 事前評価（Week 1）

1. 現在のAPI使用量の測定
2. コスト試算の実行
3. 技術要件の整理

# 現在のAPIコスト分析スクリプト
def analyze_current_costs():
    """現行APIの月間コストを算出"""
    
    # Tardis Machine API pricing（例）
    current_pricing = {
        "messages_per_million": 2.50,  # $2.50 per 1M messages
        "monthly_messages": 1000000,   # 月100万件
        "exchange_rate": 7.3,          # 公式レート
    }
    
    monthly_cost_usd = (
        current_pricing["monthly_messages"] / 1_000_000 
        * current_pricing["messages_per_million"]
    )
    monthly_cost_jpy = monthly_cost_usd * current_pricing["exchange_rate"]
    
    holy_sheep_cost_usd = monthly_cost_usd * 0.15  # 85%節約
    holy_sheep_cost_jpy = holy_sheep_cost_usd * 1  # ¥1=$1
    
    return {
        "current_usd": monthly_cost_usd,
        "current_jpy": monthly_cost_jpy,
        "holysheep_usd": holy_sheep_cost_usd,
        "holysheep_jpy": holy_sheep_cost_jpy,
        "savings_jpy": monthly_cost_jpy - holy_sheep_cost_jpy,
        "savings_pct": 85
    }

result = analyze_current_costs()
print(f"現行コスト: ¥{result['current_jpy']:,.0f}/月")
print(f"HolySheep: ¥{result['holysheep_jpy']:,.0f}/月")
print(f"節約額: ¥{result['savings_jpy']:,.0f}/月 ({result['savings_pct']}%OFF)")

Step 2: 開発環境での検証（Week 2）