こんにちは、HolySheep AI の技術チームです。本日は金融市場の最前線にいる定量取引开发者の皆さんに向け、
私は以前、暗号資産取引所のグレードリングシステム構築に携わっており、実際の
Order Book の基本概念
Order Book とは、特定の金融商品に対する未約定の買い注文(Bid)と売り注文(Ask)を価格順に並べたものです。取引の約定可能性分析、板の厚みの評価、市場インパクトの算出などQuantitative Trading の根幹をなす数据结构입니다。
Order Book の3層構造
- Level 1(Best Bid/Ask):最良買い気配と最良売り気配のみ
- Level 2(Market Depth):複数の気配値を階層的に表示
- Level 3(Full Order Book):板の全体情報加上、個別の指値注文詳細
Level2 と Level3 の技術的違い
データ構造の比較
| 項目 | Level 2 | Level 3 |
|---|---|---|
| データ粒度 | 価格レベル単位 | 注文ID単位 |
| 更新頻度 | 比較的低め | 超高頻度(ミリ秒単位) |
| データ量 | 中程度 | 大量(数千〜数万件/秒) |
| Visibility | 板の厚みは見えるが注文者は不明 | 全注文の詳細が開示 |
| 主な用途 | 一般的なチャート分析、戦略立案 | 高頻度取引、グレードリング |
| 遅延要件 | 100-500ms 程度 | 1-10ms 级别 |
| コスト | 比較的安い | 非常に高い |
データ構造の違い(コード例)
# Level 2 データ構造の例
価格レベルごとの合計数量を示す
class Level2Data:
def __init__(self):
self.bids = {} # {price: total_quantity}
self.asks = {} # {price: total_quantity}
self.timestamp = None
self.sequence = None
def update_bid(self, price: float, quantity: float):
"""買い注文の更新"""
if quantity == 0:
self.bids.pop(price, None)
else:
self.bids[price] = quantity
def get_best_bid(self) -> tuple:
"""最良買い気配を取得"""
if not self.bids:
return None, None
best_price = max(self.bids.keys())
return best_price, self.bids[best_price]
def get_spread(self) -> float:
"""スプレッドを計算"""
best_bid, _ = self.get_best_bid()
best_ask, _ = self.get_best_ask()
if best_bid and best_ask:
return best_ask - best_bid
return None
def get_best_ask(self) -> tuple:
"""最良売り気配を取得"""
if not self.asks:
return None, None
best_price = min(self.asks.keys())
return best_price, self.asks[best_price]
# Level 3 データ構造の例
個別の注文をID単位で管理
from dataclasses import dataclass, field
from typing import Dict, List
from datetime import datetime
import heapq
@dataclass
class Order:
order_id: str
price: float
quantity: float
side: str # 'bid' or 'ask'
timestamp: datetime
participant_id: str = None # 参加者識別子(OTC市場等)
class Level3Data:
def __init__(self):
self.orders: Dict[str, Order] = {} # {order_id: Order}
self.bid_heap: List[tuple] = [] # max-heap(価格降順)
self.ask_heap: List[tuple] = [] # min-heap(価格昇順)
self.sequence = 0
def add_order(self, order: Order):
"""新規注文の追加"""
self.orders[order.order_id] = order
self.sequence += 1
# ヒープ構造の維持
if order.side == 'bid':
heapq.heappush(self.bid_heap, (-order.price, order.order_id))
else:
heapq.heappush(self.ask_heap, (order.price, order.order_id))
def modify_order(self, order_id: str, new_quantity: float):
"""注文数量の変更"""
if order_id in self.orders:
self.orders[order_id].quantity = new_quantity
def cancel_order(self, order_id: str):
"""注文のキャンセル"""
if order_id in self.orders:
del self.orders[order_id]
def get_market_impact(self, side: str, quantity: float) -> float:
"""指定数量を消化雰囲での市場インパクトを試算"""
remaining = quantity
total_cost = 0.0
if side == 'bid':
heap = self.ask_heap.copy()
while remaining > 0 and heap:
price, oid = heapq.heappop(heap)
if oid not in self.orders:
continue
order = self.orders[oid]
filled = min(remaining, order.quantity)
total_cost += filled * price
remaining -= filled
else:
heap = self.bid_heap.copy()
while remaining > 0 and heap:
price, oid = heapq.heappop(heap)
if oid not in self.orders:
continue
order = self.orders[oid]
filled = min(remaining, order.quantity)
total_cost += filled * price
remaining -= filled
return total_cost / (quantity - remaining) if quantity > remaining else 0