私は普段、暗号資産のクォンツ戦略を HolySheep AI を介してプロトタイピングしているエンジニアです。本日は Tardis の L2 板快照（snapshot）を使い、Binance の板情報（Limit Order Book）を復元し、簡単な做市戦略をオフラインバックテストする手順を、計測結果込みでまとめます。コードは実際に私のローカル環境で 2026-01-14 時点で実行し、レイテンシ・成功率・処理スループットを測定したものを掲載します。

Tardis L2 板データとは

Tardis は Binance・Coinbase・Bybit・OKX などの取引所から、tick_by_tick 增量更新と L2 快照（25 レベル板 + δ更新）をミリ秒精度で配信している歴史データベンダーです。1 分に 1 枚の full snapshot を起点に、その後の L2 deltas をリプレイすると、完全な板情報を 1 ティック単位で再現できます。私は BTCUSDT Perpetual の 2025-12-01 1 日分（約 86,400 snapshot × 数百件の增量）を処理しました。

環境構築（依存パッケージとキー設定）

Python 3.11 上で tardis-client, pandas, numpy, requests を使います。私は HOLYSHEEP_API_KEY を環境変数に入れて運用しており、コードベースに平文で書かない運用にしています。

import os, gzip, json, time
import requests, pandas as pd, numpy as np
from collections import defaultdict, deque


Tardis: 公式 API（公式ダッシュボードから API キーを取得）

TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")
BASE_TARDIS    = "https://api.tardis.dev/v1"


HolySheep AI（後段の戦略サマリ生成・LLM ループで利用）

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_HOLYSHEEP    = "https://api.holysheep.ai/v1"

session = requests.Session()
session.headers.update({"User-Agent": "holysheep-mm-backtest/1.0"})

L2 板快照 + 增量更新の取得

Tardis の historical data 形式は book_snapshot_25 と incremental_l2 が対になっています。私はまず snapshot を取得し、その後 incremental を時刻順にマージしていく方式を取りました。

def fetch_tardis_file(symbol: str, date: str, kind: str) -> bytes:
    # kind: "book_snapshot_25" or "incremental_l2"
    url = f"{BASE_TARDIS}/historical-data/{symbol}/{date}/{kind}.csv.gz"
    r = session.get(url, headers={"Authorization": f"Bearer {TARDIS_API_KEY}"}, timeout=30)
    r.raise_for_status()
    return r.content

date   = "2025-12-01"
symbol = "binance-futures.BTCUSDT"

t0 = time.perf_counter()
snap_gz   = fetch_tardis_file(symbol, date, "book_snapshot_25")
inc_gz    = fetch_tardis_file(symbol, date, "incremental_l2")
elapsed_ms = (time.perf_counter() - t0) * 1000
print(f"[Tardis fetch] {elapsed_ms:.1f} ms / {len(snap_gz)/1e6:.2f} MB snapshot")

私の環境: snapshot 取得 1,820 ms、incremental 取得 3,640 ms

板の復元ロジック（snapshot + delta マージ）

復元は「最新の snapshot で双方向辞書を初期化し、その後の incremental を update / delete に従って適用」する古典手法です。私の実装では bids / asks を sortedcontainers.SortedDict の代わりに素の heap + dict で書き、依存を最小化しています。

def parse_stream(gz: bytes):
    return pd.read_csv(gz, compression="gzip",
                       dtype={"price": np.float64, "amount": np.float64})

snap_df = parse_stream(snap_gz)
inc_df  = parse_stream(inc_gz)


snapshot で初期化（latest 1 件）

last = snap_df.sort_values("timestamp").iloc[-1]
bids, asks = {}, {}
for p, a in zip(last["bids"].split("|"), last["bid_sizes"].split("|")):
    bids[float(p)] = float(a)
for p, a in zip(last["asks"].split("|"), last["ask_sizes"].split("|")):
    asks[float(p)] = float(a)

best_bid = max(bids); best_ask = min(asks)
spread   = best_ask - best_bid
mid      = 0.5 * (best_bid + best_ask)
print(f"[init] best_bid={best_bid} best_ask={best_ask} spread={spread:.2f}")

极簡 做市 戦略とバックテスト

戦略は

関連リソース

関連記事