1. 订单簿快照解析与状态重建
任务目标
从交易所API获取订单簿快照（order book snapshot），解析其中的买盘（bids）与卖盘（asks）数据，重建当前市场深度状态，并输出可查询的最优买卖价与各档位累计深度。
环境与依赖
本案例使用Python 3.9+，仅依赖标准库json与urllib.request，无需安装第三方包。以Binance现货公开API的BTCUSDT订单簿快照接口为例（无需）。
关键完整代码
import json
import urllib.request
from typing import List, Tuple, Dict
# ---------- 1. 获取订单簿快照 ----------
def fetch_snapshot(symbol: str = "btcusdt", limit: int = 10) -> dict:
"""从Binance公开接口获取订单簿快照"""
url = f"https://api.binance.com/api/v3/depth?symbol={symbol}&limit={limit}"
with urllib.request.urlopen(url, timeout=5) as resp:
return json.loads(resp.read().decode("utf-8"))
# ---------- 2. 解析并重建订单簿状态 ----------
class OrderBook:
def __init__(self, raw: dict):
self.bids: List[Tuple[float, float]] = [] # (价格, 数量)
self.asks: List[Tuple[float, float]] = []
self._parse(raw)
def _parse(self, raw: dict) -> None:
"""将API返回的字符串价格/数量转为浮点数，按价格降序(买)/升序(卖)排列"""
for price_str, qty_str in raw.get("bids", []):
self.bids.append((float(price_str), float(qty_str)))
for price_str, qty_str in raw.get("asks", []):
self.asks.append((float(price_str), float(qty_str)))
# 买盘按价格降序（最优买价在前），卖盘按价格升序（最优卖价在前）
self.bids.sort(key=lambda x: x[0], reverse=True)
self.asks.sort(key=lambda x: x[0])
def best_bid(self) -> Tuple[float, float]:
"""最优买价及对应数量"""
return self.bids[0] if self.bids else (0.0, 0.0)
def best_ask(self) -> Tuple[float, float]:
"""最优卖价及对应数量"""
return self.asks[0] if self.asks else (0.0, 0.0)
def spread(self) -> float:
"""买卖价差"""
return self.best_ask()[0] - self.best_bid()[0]
def cumulative_depth(self, levels: int = 5) -> Dict[str, float]:
"""前N档累计深度（按价格从最优到次优累加数量）"""
bid_depth = sum(qty for _, qty in self.bids[:levels])
ask_depth = sum(qty for _, qty in self.asks[:levels])
return {"bid_cum_qty": bid_depth, "ask_cum_qty": ask_depth}
def __str__(self) -> str:
lines = ["买盘(降序):"]
for p, q in self.bids:
lines.append(f" {p:.2f} x {q:.5f}")
lines.append("卖盘(升序):")
for p, q in self.asks:
lines.append(f" {p:.2f} x {q:.5f}")
return "\n".join(lines)
# ---------- 3. 最小调用入口 ----------
if __name__ == "__main__":
raw_snapshot = fetch_snapshot("btcusdt", limit=5)
book = OrderBook(raw_snapshot)
print("=== 订单簿状态 ===")
print(book)
print(f"\n最优买价: {book.best_bid()[0]:.2f} @ {book.best_bid()[1]:.5f}")
print(f"最优卖价: {book.best_ask()[0]:.2f} @ {book.best_ask()[1]:.5f}")
print(f"买卖价差: {book.spread():.2f}")
depth = book.cumulative_depth(5)
print(f"前5档买盘累计数量: {depth['bid_cum_qty']:.5f}")
print(f"前5档卖盘累计数量: {depth['ask_cum_qty']:.5f}")
运行方式
python orderbook_snapshot.py
验证方法与判定标准
程序应输出以下内容，满足即判定通过：
- 数据获取成功：打印出5档买盘与卖盘，每档包含价格和数量，且买盘价格从高到低排列、卖盘价格从低到高排列；
- 最优价正确：
最优买价等于买盘第一档价格，最优卖价等于卖盘第一档价格； - 价差为正：
买卖价差= 最优卖价 − 最优买价 > 0（正常市场状态下）； - 累计深度正确：前5档累计数量等于对应方向前5档数量之和。
若网络不可用，可替换fetch_snapshot中的URL为本地测试数据文件（如将JSON保存为snapshot.json后改用json.load(open("snapshot.json"))），其余解析与重建逻辑不变。
常见问题
- 价格精度：Binance返回的价格为字符串，直接转
float可能丢失极小精度；生产环境建议使用Decimal。本案例为教学演示，使用float足够。 - 档位数量：
limit参数控制获取档位数，最大为5000；本案例取5档演示。 - 实时性：快照仅代表请求时刻的状态，若要持续跟踪需轮询或使用WebSocket增量更新，不在本案例范围内。