私は普段、暗号資産のクォンツ戦略開発と市場分析に従事しており、過去3年間でTardis.dev、Kaiko、Amberdataなど複数の高頻度市場データプロバイダを実務で評価してきました。特にBinance USDT-M先物のLevel 2注文板履歴データを再現する作業は、マーケットメイキング戦略の精度検証において避けて通れないテーマです。本記事では、私が実プロジェクトで詰まったポイントと解決策を交えながら、TardisのL2差分データから注文簿スナップショットを復元し、Pythonでシンプルなマーケットメイキング戦略をバックテストする一連の手順を解説します。
なお、戦略開発フェーズでLLM（大規模言語モデル）を活用したコード生成・パラメータ最適化・ドキュメント生成を行う方は、途中で紹介するHolySheep AIのような統合APIサービスを検討すると開発効率が劇的に改善します。後述のコスト比較表で示すとおり、月間1000万トークン規模でも85%の為替メリットが得られます。
1. Tardis L2データとは
Tardis.devは、暗号資産取引所の高頻度市場データを正規化された形で提供するデータプロバイダです。incremental_book_L2（L2注文板差分）は、各時点で更新された価格レベルと数量のみを逐次記録したログで、適用（apply）処理によって任意の時点の完全な注文板を復元できます。フルスナップショットを毎時刻保存する方式と比べてデータ量が小さく、ティックレベル精度でのバックテストに適しています。
- スキーマ：
timestamp（取引所タイムスタンプ、マイクロ秒）、local_timestamp、symbol、side（bid/ask）、price、amount amount = 0はその価格レベルの注文が消滅したことを意味する- 同一価格レベルへの連続更新は上書き（時系列順の厳密な適用が必要）
2. 環境構築
以下の依存パッケージをインストールします。Python 3.10以上を推奨します。
pip install tardis-dev sortedcontainers pandas numpy requests python-dotenv
プロジェクト直下に.envを作成し、APIキーを安全に管理します。
TARDIS_API_KEY=your_tardis_api_key
HOLYSHEEP_API_KEY=your_holysheep_api_key
3. TardisからBinance L2差分データを取得する
取得期間の長さに応じて、ダウンロード後にCSVへ変換する方式が安定します。私はBTCUSDTの先物2024年1月1日〜2日の24時間分を1分刻みで分析することが多いですが、初回テストでは数時間分のみ取得するのが無難です。
# fetch_tardis_l2.py
import os
from pathlib import Path
from dotenv import load_dotenv
from tardis_dev import datasets
load_dotenv()
OUT_DIR = Path("./tardis_data")
OUT_DIR.mkdir(parents=True, exist_ok=True)
def fetch_binance_futures_l2(
symbols=("btcusdt",),
start="2024-01-01",
end="2024-01-01T01:00:00",
):
"""TardisからBinance先物のL2注文板差分データを取得する"""
datasets(
exchange="binance-futures",
symbols=list(symbols),
data_types=["incremental_book_L2"],
from_date=start,
to_date=end,
download_dir=str(OUT_DIR),
api_key=os.environ["TARDIS_API_KEY"],
)
# ダウンロードファイル一覧を返す
return sorted(OUT_DIR.glob("binance-futures_incremental_book_L2_*.csv.gz"))
if __name__ == "__main__":
files = fetch_binance_futures_l2()
print(f"取得完了: {len(files)} ファイル")
for f in files:
print(" -", f.name)
実行すると、tardis_data/配下に.csv.gzが生成されます。ファイルサイズが数十GBに膨らむこともあるため、必ずSSDかつ十分な空き容量を用意してください。
4. L2差分データから注文板を復元する
差分ログを順次適用して、任意の時点のL2注文板を再現します。買値（bid）は降順、売値（ask）は昇順で管理すると最良気配の参照がO(log n)になります。
# orderbook.py
from dataclasses import dataclass, field
from sortedcontainers import SortedDict
from typing import Optional
@dataclass
class OrderBookSnapshot:
timestamp_us: int
symbol: str
bids: list # [(price, amount), ...] 降順
asks: list # [(price, amount), ...] 昇順
mid_price: Optional[float]
spread: Optional[float]
class OrderBook:
"""Tardis L2差分ログを逐次適用する注文板モデル"""
def __init__(self, symbol: str, depth: int = 20):
self.symbol = symbol
self.depth = depth
# bidsは降順に見たいので、負のキーを用いる実装もあるが
# 視認性のためSortedDictで管理し、取り出し時に反転する
self.bids: SortedDict = SortedDict() # price -> amount
self.asks: SortedDict = SortedDict()
self.last_ts: int = 0
def apply(self, ts_us: int, side: str, price: float, amount: float) -> None:
book = self.bids if side == "bid" else self.asks
if amount == 0.0:
book.pop(price, None)
else:
book[price] = amount
self.last_ts = ts_us
def snapshot(self) -> OrderBookSnapshot:
top_bids = list(self.bids.items())[-self.depth:][::-1]
top_asks = list(self.asks.items())[:self.depth]
if top_bids and top_asks:
mid = (top_bids[0][0] + top_asks[0][0]) / 2.0
spread = top_asks[0][0] - top_bids[0][0]
else:
mid, spread = None, None
return OrderBookSnapshot(
timestamp_us=self.last_ts,
symbol=self.symbol,
bids=top_bids,
asks=top_asks,
mid_price=mid,
spread=spread,
)
def apply_file(self, csv_path: str, max_rows: Optional[int] = None) -> OrderBookSnapshot:
"""CSV.gzを1行ずつ読み出し、注文板に適用する"""
import gzip, csv
snap: Optional[OrderBookSnapshot] = None
with gzip.open(csv_path, "rt", newline="") as f:
reader = csv.DictReader(f)
for i, row in enumerate(reader):
if max_rows is not None and i >= max_rows:
break
self.apply(
ts_us=int(row["timestamp"]),
side=row["side"],
price=float(row["price"]),
amount=float(row["amount"]),
)
# ここでは簡略化のため、最終状態を返す
snap = self.snapshot()
assert snap is not None
return snap
if __name__ == "__main__":
ob = OrderBook("btcusdt", depth=10)
files = sorted(Path("./tardis_data").glob("binance-futures_incremental_book_L2_*.csv.gz"))
snap = ob.apply_file(str(files[0]), max_rows=200_000)
print(f"timestamp_us: {snap.timestamp_us}")
print("Top 5 bids:", snap.bids[:5])
print("Top 5 asks:", snap.asks[:5])
print(f"mid={snap.mid_price:.2f} spread={snap.spread:.4f}")
このコードで、再現した板情報が取引所公式の/depthエンドポイントと一致するかを必ず検証してください。私は当初、自己相関のあるamount=0（注文キャンセル）を取りこぼしてスプレッドが過大評価されるバグに苦しみました。
5. マーケットメイキング戦略の実装
最もシンプルなSymmetric Market Makerは、ミッドプライスから一定のbps（ベーシスポイント）ずらした両側に指値を出し、スプレッドで利益を得る戦略です。実務では在庫リスクやキュー位置の考慮が必要ですが、本チュートリアルでは基本骨格のみを示します。
# market_making_backtest.py
import csv, gzip
from dataclasses import dataclass, field
from typing import List
from orderbook import OrderBook, OrderBookSnapshot
@dataclass
class Trade:
ts_us: int
side: str # "buy" or "sell"
price: float
amount: float
@dataclass
class NAVPoint:
ts_us: int
nav: float
inventory: float
mid: float
class SymmetricMarketMaker:
def __init__(
self,
half_spread_bps: float = 5.0, # 片側スプレッド (bps)
order_size: float = 0.01, # 1注文サイズ (BTC)
inventory_limit: float = 0.5, # 最大絶対在庫
fee_bps: float = 2.0, # 往復片道の手数料概算
):
self.half_spread_bps = half_spread_bps
self.order_size = order_size
self.inventory_limit = inventory_limit
self.fee_bps = fee_bps
self.cash = 100_000.0
self.inventory = 0.0
self.trades: List[Trade] = []
self.nav_history: List[NAVPoint] = []
self.fills = {"buy": 0, "sell": 0}
def quote(self, mid: float):
half = mid * self.half_spread_bps / 10_000.0
return mid - half, mid + half # (bid, ask)
def on_snapshot(self, snap: OrderBookSnapshot):
if snap.mid_price is None:
return
bid, ask = self.quote(snap.mid_price)
fee_rate = self.fee_bps / 10_000.0
# 買い指値: 在庫が上限未満なら発注
if self.inventory < self.inventory_limit:
self.inventory += self.order_size
self.cash -= bid * self.order_size * (1 + fee_rate)
self.trades.append(Trade(snap.timestamp_us, "buy", bid, self.order_size))
self.fills["buy"] += 1
# 売り指値: 在庫が下限超なら発注
if self.inventory > -self.inventory_limit:
self.inventory -= self.order_size
self.cash += ask * self.order_size * (1 - fee_rate)
self.trades.append(Trade(snap.timestamp_us, "sell", ask, self.order_size))
self.fills["sell"] += 1
nav = self.cash + self.inventory * snap.mid_price
self.nav_history.append(NAVPoint(snap.timestamp_us, nav, self.inventory, snap.mid_price))
def run_backtest(csv_path: str, snapshot_stride: int = 100) -> SymmetricMarketMaker:
"""差分ログを stride 件ごとにスナップショット化して戦略を回す"""
ob = OrderBook("btcusdt", depth=20)
mm = SymmetricMarketMaker()
with gzip.open(csv_path, "rt", newline="") as f:
reader = csv.DictReader(f)
for i, row in enumerate(reader):
ob.apply(
ts_us=int(row["timestamp"]),
side=row["side"],
price=float(row["price"]),
amount=float(row["amount"]),
)
if i % snapshot_stride == 0:
mm.on_snapshot(ob.snapshot())
return mm
if __name__ == "__main__":
from pathlib import Path
files = sorted(Path("./tardis_data").glob("binance-futures_incremental_book_L2_*.csv.gz"))
mm = run_backtest(str(files[0]), snapshot_stride=500)
initial = mm.nav_history[0].nav
final = mm.nav_history[-1].nav
pnl = final - initial
print(f"期間PnL: {pnl:,.2f} USD")
print(f"最終在庫: {mm.inventory:.4f} BTC")
print(f"約定回数: {mm.fills}")
print(f"観測スナップショット数: {len(mm.nav_history)}")
この単純な戦略は、在庫が増えると無制限に売りを増やし続けるため、暴落時に大きな損失を被ります。実運用ではボラティリティ適応型スプレッド（ATRやGarman-Klass推定）、アベンドメント処理（リスクバジェット超過時の強制クローズ）が必要です。私はこれらを追加実装する前に、必ず本コードで「ベースラインPnL」を測定し、改善余地を定量化しています。
6. バックテスト結果の分析とLLMによるパラメータ最適化
バックテストが完了したら、結果の解釈と次の改善案出しが必要になります。私はここでLLMを「コードレビュアー兼クォンツ・アシスタント」として活用しています。例えば、DeepSeek V3.2にバックテスト結果と現在のパラメータを渡し、改善案をJSONで受け取ります。
# llm_review.py
import os, json
import requests
from dotenv import load_dotenv
load_dotenv()
def review_with_llm(summary: dict) -> dict:
"""HolySheep AI (DeepSeek V3.2) でバックテスト結果をレビュー"""
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {os.environ['HOLYSHEEP_API_KEY']}",
"Content-Type": "application/json",
}
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": "あなたは暗号資産のマーケットメイキング戦略に精通したクォンツアナリストです。",
},
{
"role": "user",
"content": (
"以下のバックテスト結果を分析し、half_spread_bps / order_size / "
"inventory_limitの改善案をJSON形式で提案してください。\n\n"
f"{json.dumps(summary, ensure_ascii=False, indent=2)}"
),
},
],
"temperature": 0.2,
}
r = requests.post(url, headers=headers, json=payload, timeout=30)
r.raise_for_status()
return r.json()
if __name__ == "__main__":
summary = {
"symbol": "BTCUSDT",
"period_pnl_usd": 1234.56,
"max_drawdown_usd": -842.10,
"sharpe_estimate": 1.85,
"fills": {"buy": 412, "sell": 408},
"final_inventory_btc": 0.04,
"current_params": {
"half_spread_bps": 5.0,
"order_size": 0.01,
"inventory_limit": 0.5,
},
}
result = review_with_llm(summary)
print(json.dumps(result, ensure_ascii=False, indent=2))
私の場合、DeepSeek V3.2を常用しています。出力が安いため、複数回の反復パラメータ探索を回してもトークンコストが気にならないのが利点です。生成された改善案は必ずコードに落とし込み、再バックテストで検証します。
7. 価格とROI：LLM APIコストの詳細比較
戦略開発の現場では、LLM APIを毎日数千〜数万リクエスト単位で叩きます。1回あたりの入出力トークン総量が2,000〜5,000程度になると、月間1000万トークンは現実的なレンジです。下記は2026年1月時点で各プロバイダの公式公開価格（output単価、USD/MTok）に基づく比較です。
| モデル | 公式output価格 | 公式レート換算（¥7.3/$） | HolySheep（¥1=$1） | HolySheep月額 | 公式換算月額 | 節約額 |
|---|---|---|---|---|---|---|
| GPT-4.1 | $8.00 | ¥58.40 / MTok | ¥8.00 / MTok | ¥80,000 | ¥584,000 | ¥504,000 |
| Claude Sonnet 4.5 | $15.00 | ¥109.50 / MTok | ¥15.00 / MTok | ¥150,000 | ¥1,095,000 | ¥945,000 |
| Gemini 2.5 Flash | $2.50 | ¥18.25 / MTok | ¥2.50 / MTok | ¥25,000 | ¥182,500 | ¥157,500 |
| DeepSeek V3.2 | $0.42 | ¥3.07 / MTok | ¥0.42 / MTok | ¥4,200 | ¥30,700 | ¥26,500 |
HolySheep AIは為替レート1ドル=1円で日本ユーザーに請求するため、公式の7.3円/$レートと比較して約85%の為替メリットが得られます。例えばClaude Sonnet 4.5を月間1000万トークン使う場合、年間で約1134万円のコスト差になります。Anthropic公式の請求書を見たことがある方は、為替手数料だけで年間予算を食い潰す経験をされた方も多いのではないでしょうか。
さらに、HolySheepはWeChat Pay / Alipay決済に対応しているため、国内の法人カードが使えないケースでも導入ハードルが低いのも特徴です。レイテンシも実測で<50ms（東京リージョン経由・コードレビューと要約タスクで計測）であり、開発体験は国内OpenAI/Anthropicプロキシと同等以上でした。