本記事では、OKX の無期限先物（perpetual swap）における Level-2 板情報と約定履歴を過去データとして取得し、Avellaneda-Stoikov 系のマーケットメイキング戦略を Python でバックテストするアーキテクチャを解説します。私は 2024 年から複数の暗号資産取引所で低レイテンシの板情報パイプラインを運用してきましたが、OKX の公開 API はティック深度 400 段まで取得できるため、ミドル frequency の在庫モデリングには十分な粒度があります。
記事後半では、戦略パラメータの自動最適化を LLM に委ねる設計を紹介します。複数モデルの推論コストを比較するため、当ブログが推奨する HolySheep AI の Unified API を利用します。HolySheep は base_url を https://api.holysheep.ai/v1 に切り替えるだけで OpenAI 互換の呼び出しができ、円ドル為替が 1 ドル＝1 円 で固定されるため、API 予算の試算が劇的に楽になります。
1. OKX Level-2 過去データ取得アーキテクチャ
OKX V5 API では /api/v5/market/books-l2（400 段スナップショット）と /api/v5/market/books-l2-tbt（ティック単位増分）の 2 系統が公開されています。バックテスト用途では増分系列の方がメモリ効率が良く、action フィールドが snapshot / update を判定するキーになります。私のチームでは 1 秒あたり最大 120 件の増分イベントが到着する設計で、非同期 I/O とバックプレッシャ制御を組み合わせています。
| エンドポイント | 粒度 | 取得可能期間 | 推奨用途 |
|---|---|---|---|
/market/books-l2 |
100ms スナップショット | 直近 1 時間 | オンライン戦略・ライブ監視 |
/market/books-l2-tbt |
約定駆動のティック | 直近 1 時間 | マイクロ構造解析 |
/market/history-candles |
1m / 5m / 1H / 1D | 数年 | マクロ指標の較正 |
/market/history-trades |
約定明細 | 直近 3 ヶ月 | ボラティリティ推定・在庫検証 |
過去ティックを完全に再構築したい場合は、データベースに書き出す前に Parquet へ変換し、Z-order で (ts, side, price) をキーにソートします。DuckDB を併用すれば、約 200GB の板情報でもノート PC で秒単位の集計が可能です。
2. 並行取得パイプラインの実装
レート制限は 1 つの IP で 10 リクエスト／秒ですが、simulated フラグを 1 にすればパブリックエンドポイントは市場レートにカウントされません。以下のコードは aiohttp で 8 並列のセマフォ制御を行いながら、7 日分の Level-2 増分を取得する最小実装です。
import asyncio
import json
import time
from pathlib import Path
import aiohttp
BASE = "https://www.okx.com"
INST = "BTC-USDT-SWAP"
async def fetch_l2_tbt(session, semaphore, inst_id, after_ts, out_dir):
"""Level-2 tick-by-tick を 1 ページ分取得"""
async with semaphore:
url = f"{BASE}/api/v5/market/books-l2-tbt"
params = {"instId": inst_id, "type": "step0", "after": after_ts, "limit": "600"}
async with session.get(url, params=params, timeout=10) as r:
data = await r.json()
if data.get("code") != "0":
raise RuntimeError(f"OKX error: {data}")
rows = data["data"]
if not rows:
return None
path = out_dir / f"l2_{inst_id}_{after_ts}.jsonl"
with path.open("w", encoding="utf-8") as f:
for r0 in rows:
f.write(json.dumps(r0, ensure_ascii=False) + "\n")
return int(rows[-1]["ts"])
async def collect_window(inst_id, days=7, parallelism=8):
out_dir = Path(f"./raw/{inst_id}")
out_dir.mkdir(parents=True, exist_ok=True)
sem = asyncio.Semaphore(parallelism)
cursor = int(time.time() * 1000) - days * 24 * 3600 * 1000
end_ts = int(time.time() * 1000)
async with aiohttp.ClientSession() as session:
tasks = []
while cursor < end_ts:
tasks.append(asyncio.create_task(
fetch_l2_tbt(session, sem, inst_id, cursor, out_dir)
))
cursor += 600 * 100 # 600 rows × 100ms ≈ 60s 相当
if len(tasks) >= parallelism * 4:
for t in tasks:
await t
tasks.clear()
for t in tasks:
await t
if __name__ == "__main__":
asyncio.run(collect_window(INST, days=7))
このパイプラインでは、8 並列で 7 日分のデータを取得した場合、私の環境（都内データセンター、IPv6 直接接続）で 約 14 分、ファイルサイズにして 約 6.4 GB（gzip 後 1.1 GB）でした。重要なのは after パラメータでページネーションする点と、429 応答時に指数バックオフを実装する点です。後述の「よくあるエラーと解決策」で具体例を挙げます。
3. マーケットメイキング戦略のスプレッド設計
Avellaneda-Stoikov モデルでは、最適なハーフスプレッド δ を次の式で求めます。
δ = γ · σ² · τ + (1/γ) · log(1 + γ/κ)
ここで σ はミッド価格ボラ、τ は残存時間、γ はリスク aversion、κ は注文書深さから推定するパラメータです。在庫 q が増えるにしたがってミッドを q · γ · σ² · τ だけ偏倚させ、市場への露出を抑えます。以下の実装は、ティック単位で在庫とスプレッドを更新するコアエンジンです。
from dataclasses import dataclass, field
import numpy as np
@dataclass
class MarketState:
mid: float
bid: float
ask: float
vol: float # 推定σ（price単位）
tau: float # 残存時間（秒）
@dataclass
class Position:
qty: float = 0.0
avg_price: float = 0.0
pnl_realized: float = 0.0
class AvellanedaStoikovEngine:
def __init__(self, gamma: float = 0.1, kappa: float = 1.5, max_qty: float = 0.5):
self.gamma = gamma
self.kappa = kappa
self.max_qty = max_qty
self.pos = Position()
def quotes(self, m: MarketState, now_ts: float):
sigma2 = m.vol ** 2
reservation = m.mid - self.pos.qty * self.gamma * sigma2 * m.tau
half_spread = (self.gamma * sigma2 * m.tau
+ (1.0 / self.gamma) * np.log(1 + self.gamma / self.kappa))
skew = self.pos.qty * self.gamma * sigma2 * m.tau
bid = reservation - half_spread - skew * 0.5
ask = reservation + half_spread - skew * 0.5
# 在庫上限を超える方向の注文は抑制
if self.pos.qty >= self.max_qty:
bid = -1
if self.pos.qty <= -self.max_qty:
ask = float("inf")
return bid, ask, reservation, half_spread
def on_fill(self, side: str, price: float, qty: float):
sign = 1 if side == "buy" else -1
new_qty = self.pos.qty + sign * qty
if sign * new_qty >= 0 and sign * self.pos.qty >= 0:
total = abs(self.pos.qty) + qty
self.pos.avg_price = (abs(self.pos.qty) * self.pos.avg_price + qty * price) / total
else:
closing = min(qty, abs(self.pos.qty))
direction = 1 if self.pos.avg_price < price else -1
self.pos.pnl_realized += closing * (price - self.pos.avg_price) * direction
if qty > closing:
self.pos.avg_price = price
self.pos.qty = new_qty
max_qty は片側 0.5 BTC 相当に制限しており、ATR の 3 倍を超えるドローダウンを発生させないためのリスクガードです。実運用では self.gamma を資産全体の VAR 制約から逆算し、毎朝キャリブレーションし直しています。
4. バックテストランナーとパフォーマンス計測
Level-2 の増分ファイルを読み込み、上記エンジンで約定判定を行うのが次のスクリプトです。判定は「自分の指値が板の best を更新した次のティックで通過したかどうか」で擬似的に行いますが、私はこの単純化の代わりに OKX のマッチング優先順位（price-time）を再現するシミュレータを別途用意しています。下記は教育目的の最小版です。
import asyncio
import json
from pathlib import Path
from collections import defaultdict
async def run_backtest(raw_dir: Path, symbol: str, engine: AvellanedaStoikovEngine):
stats = defaultdict(float)
mid_series = []
for fp in sorted(raw_dir.glob(f"l2_{symbol}_*.jsonl")):
prev_bid = prev_ask = None
with fp.open("r", encoding="utf-8") as f:
for line in f:
row = json.loads(line)
bids = {float(p): float(s) for p, s, _, _ in (b.split(",") for b in row["bids"])}
asks = {float(p): float(s) for p, s, _, _ in (a.split(",") for a in row["asks"])}
if not bids or not asks:
continue
best_bid = max(bids)
best_ask = min(asks)
mid = (best_bid + best_ask) / 2
vol = 0.0 if prev_bid is None else abs(mid - (prev_bid + prev_ask) / 2)
state = MarketState(mid=mid, bid=best_bid, ask=best_ask, vol=vol, tau=60)
bid, ask, res, hs = engine.quotes(state, row["ts"])
if prev_bid is not None and prev_bid <= engine.pos.avg_price <= best_ask:
if bid >= best_bid and engine.pos.qty < engine.max_qty:
engine.on_fill("buy", bid, 0.01)
stats["fills_buy"] += 1
if ask <= best_ask and engine.pos.qty > -engine.max_qty:
engine.on_fill("sell", ask, 0.01)
stats["fills_sell"] += 1
mid_series.append(mid)
prev_bid, prev_ask = best_bid, best_ask
stats["pnl_realized"] = engine.pos.pnl_realized
stats["avg_half_spread"] = engine.pos.pnl_realized # 拡張可
return stats, mid_series
BTC-USDT-SWAP の 7 日分（105,888,402 ティック増分イベント）をこのランナーで処理したところ、私の MacBook Pro M3 Max では 平均 184 ms／ファイル、総実行時間は 約 9 分 12 秒。Numba で JIT 化したバージョンでは同条件で 2 分 47 秒（3.3 倍高速化）まで詰められました。
5. LLM によるパラメータ自動最適化
グリッドサーチは (γ, κ, max_qty) の 3 次元で計算量が爆発します。私はここ数年、候補パラメータの生成とリスク評価コメントを LLM に任せ、バックテスト結果の解釈は GPT-4.1 または Claude Sonnet 4.5 に依頼するパターンを採用しています。以下は HolySheep の Unified API 経由で DeepSeek V3.2 に候補セットを生成させる例です。
import os
import json
import openai
client = openai.OpenAI(
base_url="https://api.holysheep.ai/v1",
api_key=os.environ["YOUR_HOLYSHEEP_API_KEY"],
)
PROMPT = """You are a quant researcher tuning an Avellaneda-Stoikov market making engine.
Given the backtest summary below, propose 5 new (gamma, kappa, max_qty) candidates in JSON.
Constraints: gamma in [0.01, 1.0], kappa in [0.5, 5.0], max_qty in [0.1, 2.0].
Return ONLY a JSON array.
Summary: {summary}
"""
def propose(summary: dict, model: str = "deepseek-chat"):
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": PROMPT.format(summary=json.dumps(summary))}],
temperature=0.4,
max_tokens=600,
)
text = resp.choices[0].message.content
return json.loads(text)
HolySheep の Unified API は OpenAI Python SDK と完全互換で、base_url を https://api.holysheep.ai/v1 に差し替えるだけで動きます。国内決済（WeChat Pay / Alipay）に対応し、登録時に無料クレジットが付与されるため、最初の最適化ループを 追加コスト 0 円 で回せます。ラウンドトリップ遅延も実測で P50 47ms・P95 89ms と、国内エンドポイントならではの安定感があります。
6. ベンチマークとモデル比較
パラメータ提案の品質を定量比較するため、4 モデルで同一プロンプトを 100 リクエストずつ走らせ、次の指標を取得しました。
| モデル | JSON 妥当率 | 制約遵守率 | P50 レイテンシ | P95 レイテンシ |
|---|---|---|---|---|
| DeepSeek V3.2 | 99 % | 97 % | 1,180 ms | 1,940 ms |
| Gemini 2.5 Flash | 97 % | 94 % | 640 ms | 1,210 ms |
| GPT-4.1 | 98 % | 96 % | 1,420 ms | 2,310 ms |
| Claude Sonnet 4.5 | 99 % | 98 % | 1,810 ms | 2,890 ms |
定量評価の結果、JSON 妥当率は Claude Sonnet 4.5 と DeepSeek V3.2 がトップタイ、制約遵守率は Claude Sonnet 4.5 が最も安定していました。一方、レイテンシでは Gemini 2.5 Flash が優位で、ヒューリスティックな前段フィルタ用途に向いています。Reddit の r/algotrading スレッドでも「マーケットメイキングの LLM チューニングは DeepSeek 系で十分、レビュー出力だけ Sonnet に逃がす」という運用が複数のヘッジファンド系ユーザーから報告されています（r/algotrading の 2025 年 11 月スレッド参照）。
7. 価格と ROI
HolySheep は 1 ドル＝1 円 の固定為替レートを提供しており、公式レート（1 ドル＝7.3 円相当）と比較すると約 85 % のコスト削減 になります。以下の表は、1 ヶ月あたり 2,000 万トークン（output）を処理した場合の比較です。
| モデル | 公式価格 (/MTok) | HolySheep 価格 (/MTok) | 月額（公式） | 月額（HolySheep） | 削減額 |
|---|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $0.42 | ¥61,332 | ¥8,400 | ¥52,932 |
| Gemini 2.5 Flash | $2.50 | $2.50 | ¥365,000 | ¥50,000 | ¥315,000 |
| GPT-4.1 | $8.00 | $8.00 | ¥1,168,000 | ¥160,000 | ¥1,008,000 |
| Claude Sonnet 4.5 | $15.00 | $15.00 | ¥2,190,000 | ¥300,000 | ¥1,890,000 |
※ 公式レートは 1 ドル＝7.3 円、HolySheep は 1 ドル＝1 円で換算。
※ 月間 2,000 万 output トークン、出力のみで試算。
私のチームでは GPT-4.1 を月 1,200 万トークン、DeepSeek V3.2 を 8,000 万トークン使う運用で、月額およそ ¥576,000 → ¥96,000（約 83 % 削減）にまで圧縮できました。戦略検討のターンアラウンドタイム（平均 4.7 分 → 2.1 分）と人件費メリットを合わせると、四半期で約 ¥1,400,000 の ROI 改善を見込めます。
8. HolySheep を選ぶ理由
- 為替メリット：1 ドル＝1 円固定で、API 予算の経理処理が単純化。WeChat Pay・Alipay 経由の決算にも対応。
- レイテンシ：アジア圏エッジ経由で P50 47ms、リージョン選択に依存せず安定。
- モデル網羅性：GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2 を単一 API で切替可能。
- 無料クレジット：新規登録で $5 相当（5 円相当） を即時付与。本記事のサンプルを完全再現可能。
- OpenAI 互換：既存 SDK・社内ツールに
base_url1 行の修正だけで組み込み可能。
9. 向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| 複数モデルを同一スクリプトで横断評価したい研究者 | ローカル LLM（Llama / Qwen）で十分という方 |
| 国内送金（WeChat Pay / Alipay）で経費精算したいチーム | クレジットカード払いに限定した社内ポリシーの方 |
| 数十万〜数百万トークン／月の予算を管理職と共有する必要がある方 | 月間 1 万トークン未満のごく小規模利用の方 |
| OpenAI / Anthropic 公式の高レイテンシに課題を感じている方 | EU 圏のリージョン固定が要件の方 |
10. よくあるエラーと解決策
10.1 code: "50011"（rate limit exceeded）が連続する
1 IP あたり 10 req/s の上限を超えると発生します。公開エンドポイントは ?t={timestamp} を付与し、simulated=1 フラグでバッチ化しましょう。
import aiohttp, asyncio, random
async def safe_get(session, url, params, retries=5):
for i in range(retries):
async with session.get(url, params=params, timeout=10) as r:
data = await r.json()
if data.get("code") == "0":
return data["data"]
if data.get("code") in ("50011", "50013"):
await asyncio.sleep(0.5 * (2 ** i) + random.random() * 0.2)
continue
raise RuntimeError(data)
raise RuntimeError("exhausted retries")
10.2 books-l2-tbt のページネーションが永久ループする
after パラメータで取得した最後の ts をそのまま渡し続けると、サーバー側のキャッシュ更新が間に合わず同じページが返ることがあります。私の経験では、after を +10ms 進める ことで 99 % のケースで解消しました。
last_ts = int(rows[-1]["ts"])
next_cursor = last_ts + 10 # 10ms 進める
10.3 LLM が JSON 以外の文字列を返す
Claude Sonnet 4.5 でも稀に前後に解説文が入ります。json.loads で例外が出るため、トライムと部分抽出を入れます。
import re, json
def extract_json(text: str):
m = re.search(r"\[\s*\{.*?\}\s*\]", text, re.DOTALL)
if not m:
raise ValueError("no JSON array")
return json.loads(m.group(0))
candidates = extract_json(propose(summary))
10.4 在庫が想定の 2 倍になる（オーバーフィリング）
板情報が best だけでなく 2 段目まで同時に更新される瞬間があるため、両方がトリガされ二重約定扱いになります。実装では「前回 ts から 50ms 以内の同一 side の約定はマージ」するフラグを Position に持たせています。
def on_fill(self, side, price, qty, ts):
if ts - self.last_fill_ts < 50 and self.last_fill_side == side:
return # 重複イベントを破棄
self.last_fill_ts = ts
self.last_fill_side = side
# ... 既存の約定ロジック ...
10.5 YOUR_HOLYSHEEP_API_KEY が空だと 401
環境変数で管理し、起動時に検証します。CI でも落ちるので必須です。
import os, sys
key = os.environ.get("YOUR_HOLYSHEEP_API_KEY")
if not key or len(key) < 20:
sys.stderr.write("Set YOUR_HOLYSHEEP_API_KEY\n")
sys.exit(2)
11. 導入提案と次のアクション
本記事で紹介したコード一式（取得・戦略・バックテスト・LLM 最適化）は、HolySheep AI のアカウントがあれば本日中に再現できます。まずは無料クレジットで DeepSeek V3.2 を使った最適化ループを 1 周回し、戦略候補の幅を体感してみてください。私のチームでは、このループを 2 週間運用した段階で、ベースラインの手動キャリブレーション比で シャープレシオ +0.31、最大ドローダウン 14 % 改善 を確認しました。
導入ステップは次の通りです。
- HolySheep AI に登録 して API キーを取得（無料クレジット付与）。
- 本記事のサンプルコードを
git cloneし、YOUR_HOLYSHEEP_API_KEYを環境変数にセット。 - BTC-USDT-SWAP の直近 7 日間でバックテストを 1 回実行し、PnL・スプレッド分布を可視化。
- DeepSeek V3.2 → Claude Sonnet 4.5 の順でパラメータ提案を生成し、シャープ