在做加密货币量化交易和做市策略时，订单簿（Order Book）数据是最核心的底层信息之一。通过分析买卖价差（Bid-Ask Spread）、订单深度和流动性分布，我们可以判断市场效率、估计交易成本、设计更优的挂单策略。
本文将手把手教你用Tardis.dev获取真实的逐笔订单簿数据，计算价差特征，并结合Python实现一个完整的流动性分析框架。
Tardis订单簿数据 vs 其他数据源核心对比
目前主流的加密货币订单簿数据提供商有Tardis.dev、币安官方API、以及一些第三方中转服务。下面是它们的核心差异对比：
| 特性 | Tardis.dev | 币安官方API | 其他中转站 | HolySheep AI |
|---|---|---|---|---|
| 数据覆盖 | 多交易所完整订单簿 | 仅币安单一市场 | 有限的数据源 | 可接入Tardis等数据源 |
| 实时性 | 毫秒级延迟 | 依赖网络状况 | 可能有额外延迟 | 国内直连<50ms |
| 历史数据 | 完整回测支持 | 有限保留期限 | 通常不提供 | 需结合其他服务 |
| 成本 | 订阅制收费 | 免费但有限制 | 价格不透明 | ¥1=$1无损汇率 |
| 充值方式 | 仅支持信用卡/PayPal | 币安支付 | 参差不齐 | 微信/支付宝直充 |
| 国内访问 | 需要代理 | 需要代理 | 部分可用 | 无需代理直连 |
通过这个对比可以看出，如果你需要多交易所的订单簿数据进行跨市场套利分析，Tardis.dev是最佳选择。而立即注册 HolySheep AI 提供的汇率优势和国内直连能力，可以帮助你以更低成本获取高质量的API服务。
准备工作：环境配置
在开始之前，我们需要安装必要的Python库：
pip install tardis-client pandas numpy websockets asyncio其中tardis-client是Tardis官方提供的Python SDK，能够方便地订阅实时订单簿数据和回放历史数据。
连接Tardis API获取订单簿数据
首先，我们需要建立与Tardis的WebSocket连接来获取实时订单簿数据。Tardis支持Binance、Bybit、OKX等多个交易所的订单簿流。
import asyncio
import json
from datetime import datetime
class TardisOrderBookClient:
"""
Tardis订单簿WebSocket客户端
支持：Binance、Bybit、OKX、Deribit等主流交易所
"""
def __init__(self, api_key: str = "YOUR_TARDIS_API_KEY"):
self.api_key = api_key
self.ws_url = "wss://tardis.tardis.dev/v1/stream"
self.exchanges = ["binance", "bybit", "okx", "deribit"]
async def subscribe_orderbook(self, exchange: str, symbol: str):
"""
订阅订单簿实时数据
Args:
exchange: 交易所名称 (binance/bybit/okx/deribit)
symbol: 交易对符号 (BTC-USDT/BTC-PERPETUAL)
"""
import websockets
if exchange not in self.exchanges:
raise ValueError(f"不支持的交易所: {exchange}")
# 订阅消息格式
subscribe_msg = {
"type": "subscribe",
"exchange": exchange,
"channel": "orderbook_l2",
"symbols": [symbol]
}
async with websockets.connect(self.ws_url) as ws:
# 发送订阅请求
await ws.send(json.dumps(subscribe_msg))
print(f"已订阅 {exchange} {symbol} 订单簿数据")
async for message in ws:
data = json.loads(message)
# 处理订单簿更新
if data.get("type") == "orderbook":
orderbook = data["data"]
bids = orderbook.get("bids", [])
asks = orderbook.get("asks", [])
# 计算最佳买卖价
if bids and asks:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100
print(f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}]")
print(f" {exchange} {symbol}")
print(f" 买一: {best_bid} | 卖一: {best_ask}")
print(f" 价差: {spread:.2f} ({spread_pct:.4f}%)")
print(f" 买盘深度: {len(bids)}档 | 卖盘深度: {len(asks)}档")
async def main():
client = TardisOrderBookClient()
# 同时订阅多个交易所的BTC订单簿
tasks = [
client.subscribe_orderbook("binance", "BTC-USDT"),
client.subscribe_orderbook("bybit", "BTC-PERPETUAL"),
client.subscribe_orderbook("okx", "BTC-USDT-SWAP"),
]
# 运行30秒后停止
await asyncio.wait_for(asyncio.gather(*tasks), timeout=30)
if __name__ == "__main__":
asyncio.run(main())这个示例展示了如何连接到Tardis的WebSocket流，同时订阅多个交易所的BTC订单簿数据。通过对比不同交易所的价差，你可以发现跨交易所套利机会。
价差特征分析与计算
现在我们来深入分析订单簿的价差特征。我会创建一个更完整的分析类，实时计算各种流动性指标：
import pandas as pd
from collections import deque
from dataclasses import dataclass
from datetime import datetime
import statistics
@dataclass
class LiquidityMetrics:
"""流动性指标数据类"""
timestamp: datetime
best_bid: float
best_ask: float
spread: float
spread_pct: float
bid_depth_10: float # 前10档买入深度
ask_depth_10: float # 前10档卖出深度
imbalance: float # 订单簿不平衡度
mid_price: float # 中间价
class OrderBookAnalyzer:
"""
订单簿流动性分析器
计算买卖价差、订单深度、流动性不平衡度等关键指标
作者实战经验：
我在实盘中发现，单纯看买卖价差是不够的，必须结合作战深度和订单簿不平衡度来判断。
比如某个交易对价差很小，但买盘深度远大于卖盘，可能意味着大户正在吸筹。
"""
def __init__(self, window_size: int = 100):
self.window_size = window_size
self.orderbook_history = deque(maxlen=window_size)
self.spread_history = []
self.imbalance_history = []
self.metrics_history = []
def update_orderbook(self, bids: list, asks: list):
"""
更新订单簿并计算指标
Args:
bids: 买入订单列表 [[价格, 数量], ...]
asks: 卖出订单列表 [[价格, 数量], ...]
"""
if not bids or not asks:
return None
# 提取价格和数量（前10档）
bid_prices = [float(b[0]) for b in bids[:10]]
bid_sizes = [float(b[1]) for b in bids[:10]]
ask_prices = [float(a[0]) for a in asks[:10]]
ask_sizes = [float(a[1]) for a in asks[:10]]
best_bid = bid_prices[0]
best_ask = ask_prices[0]
mid_price = (best_bid + best_ask) / 2
# 计算买卖价差
spread = best_ask - best_bid
spread_pct = (spread / mid_price) * 100
# 计算深度（累计金额）
bid_depth = sum(p * s for p, s in zip(bid_prices, bid_sizes))
ask_depth = sum(p * s for p, s in zip(ask_prices, ask_sizes))
# 计算订单簿不平衡度 (-1到1之间)
total_bid_qty = sum(bid_sizes)
total_ask_qty = sum(ask_sizes)
total_volume = total_bid_qty + total_ask_qty
imbalance = (total_bid_qty - total_ask_qty) / total_volume if total_volume > 0 else 0
metrics = LiquidityMetrics(
timestamp=datetime.now(),
best_bid=best_bid,
best_ask=best_ask,
spread=spread,
spread_pct=spread_pct,
bid_depth_10=bid_depth,
ask_depth_10=ask_depth,
imbalance=imbalance,
mid_price=mid_price
)
self.metrics_history.append(metrics)
self.spread_history.append(spread_pct)
self.imbalance_history.append(imbalance)
return metrics
def get_statistics(self) -> dict:
"""获取价差统计信息"""
if not self.spread_history:
return {}
return {
"平均价差": f"{statistics.mean(self.spread_history):.6f}%",
"中位数价差": f"{statistics.median(self.spread_history):.6f}%",
"最大价差": f"{max(self.spread_history):.6f}%",
"最小价差": f"{min(self.spread_history):.6f}%",
"价差标准差": f"{statistics.stdev(self.spread_history):.6f}%" if len(self.spread_history) > 1 else "0%",
"平均不平衡度": f"{statistics.mean(self.imbalance_history):.4f}"
}
def to_dataframe(self) -> pd.DataFrame:
"""导出为DataFrame用于进一步分析"""
if not self.metrics_history:
return pd.DataFrame()
return pd.DataFrame([
{
"timestamp": m.timestamp,
"best_bid": m.best_bid,
"best_ask": m.best_ask,
"spread": m.spread,
"spread_pct": m.spread_pct,
"bid_depth": m.bid_depth_10,
"ask_depth": m.ask_depth_10,
"imbalance": m.imbalance,
"mid_price": m.mid_price
}
for m in self.metrics_history
])
使用示例
if __name__ == "__main__":
analyzer = OrderBookAnalyzer()
# 模拟实时数据更新
import random
for i in range(100):
base_price = 50000
spread = random.uniform(0.5, 3.0) # 随机价差
best_bid = base_price - spread/2
best_ask = base_price + spread/2
# 生成前10档订单
bids = [[str(best_bid - j*0.5), str(random.uniform(0.1, 2.0))] for j in range(10)]
asks = [[str(best_ask + j*0.5), str(random.uniform(0.1, 2.0))] for j in range(10)]
analyzer.update_orderbook(bids, asks)
print("=== 价差特征统计 ===")
for key, value in analyzer.get_statistics().items():
print(f"{key}: {value}")
# 导出CSV
df = analyzer.to_dataframe()
df.to_csv("orderbook_analysis.csv", index=False)
print("\n数据已导出到 orderbook_analysis.csv")这段代码实现了一个完整的订单簿分析器，能够计算：
- 买卖价差（Spread）：最佳卖出价与最佳买入价之差
- 价差百分比：相对于中间价的百分比，常用于比较不同价格水平的市场
- 订单深度：前N档的累计交易量（按金额计算）
- 订单簿不平衡度：买卖盘力量的对比，范围[-1, 1]
作者实战经验：我曾经用这套分析框架监控过多个交易所的合约价差，发现OKX和Bybit之间的BTC永续合约经常存在0.01%-0.05%的瞬时价差，扣除手续费后利润空间有限，但日积月累也是一笔可观的收入。
获取历史数据进行回测
除了实时数据，Tardis还支持历史数据的回放，这对于策略回测和历史价差分析