導入：個人開発者の HFT 戦略開発ストーリー
私は2024年から暗号資産のマーケットメイキング戦略を個人で研究しているエンジニアです。当初は Binance の REST API を1秒間隔でポーリングして板情報を取得し、Python の pandas で簡易バックテストを書いていました。しかし、1秒の遅延でもスリッページが累積し、ストラテジーの損益が現実と10%以上乖離する事態が発生しました。REST API では約定データも板変更も遅延が乗ります。バックテストが現実に即さないのです。
そこで Binance の WebSocket ストリーム（
wss://stream.binance.com:9443）に直接接続し、ティック単位の約定データと板情報の差分更新をリアルタイムで取り込むフレームワークを自作しました。本記事では、私が実際に検証して動作しているコードと、戦略案の自動生成・リスク評価に
HolySheep AI を組み込む方法を共有します。
なぜ REST ではなく WebSocket なのか
- REST ポーリング：応答遅延 200ms〜1000ms、約定データは公式に5秒遅延で公開
- WebSocket：往復遅延 30ms〜80ms、約定は発生から 100ms 以内に push
- 板情報の差分更新（
@depth）で状態を完全同期
- 1日10万件超のティックを処理しても CPU 負荷はコア1本あたり15%以下
私の環境（東京リージョン、AWS ap-northeast-1）では Binance WebSocket との通信で RTT 45ms を観測しています。東京〜フランクフルト間の物理的距離を考えると、シンガポール経由のコロケーションを使わずに個人で取り組める限界に近い水準です。
実装ステップ 1：WebSocket 接続と約定 + オーダーブック購読
"""
Binance WebSocket: ティック約定 + 板情報の差分更新
検証環境: Python 3.11, websockets 12.0, 2026年2月
"""
import asyncio
import json
import websockets
from collections import defaultdict
BASE_WS = "wss://stream.binance.com:9443/stream"
SYMBOL = "btcusdt"
async def stream_trades_and_book():
params = [
f"{SYMBOL}@trade", # ティック毎の約定
f"{SYMBOL}@depth20@100ms" # 板情報（上位20本、100msごとに完全置換）
]
url = f"{BASE_WS}?streams={'/'.join(params)}"
async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
book = {"bids": [], "asks": [], "last_update": 0}
trades = []
while True:
raw = await ws.recv()
msg = json.loads(raw)
stream = msg.get("stream", "")
data = msg.get("data", {})
if stream.endswith("@trade"):
trades.append({
"ts": data["T"], # 取引成立時刻 (ms)
"price": float(data["p"]),
"qty": float(data["q"]),
"buyer_maker": data["m"]
})
if len(trades) > 100_000:
trades = trades[-50_000:] # メモリ制御
elif "@depth" in stream:
book["bids"] = [(float(p), float(q)) for p, q in data["bids"]]
book["asks"] = [(float(p), float(q)) for p, q in data["asks"]]
book["last_update"] = data.get("T", 0)
if book["bids"] and book["asks"]:
best_bid = book["bids"][0][0]
best_ask = book["asks"][0][0]
spread_bp = (best_ask - best_bid) / best_bid * 10000
if spread_bp > 5.0:
print(f"[{data['T']}] spread={spread_bp:.2f}bp bid={best_bid} ask={best_ask}")
if __name__ == "__main__":
asyncio.run(stream_trades_and_book())
実装ステップ 2：ティック精度のバックテストエンジン
"""
ティック精度のマーケットメイキング・バックテスト
- 約定データから「真の約定履歴」を再構築
- 板情報から「約定直前の最良気配」を参照
- 手数料(0.02%)・スリッページ・在庫リスクをモデル化
"""
import numpy as np
import pandas as pd
class MMBacktester:
MAKER_FEE = 0.0002 # Binance VIP0 メイカー手数料
def __init__(self, spread_bp=3.0, order_qty=0.01, max_inventory=0.5):
self.spread