作为一名在加密货币做市领域摸爬滚打 4 年的工程师，我见过太多团队因为缺乏精准的 PnL 分析能力而陷入库存风险失控的困境。2023 年第三季度，BTC 市场波动率飙升至 180%，我所在团队因为库存模型滞后，单日浮亏超过 12 万美元。这个惨痛教训让我下定决心打造一套基于 Tardis Order Book 数据的实时 PnL 分析系统。今天这篇文章，我将毫无保留地分享这套系统的架构设计、核心代码和性能优化经验。
一、做市商 PnL 核心原理与库存风险本质
在深入代码之前，我们必须先理清做市商 PnL 的数学本质。做市商的收益来源不是单边方向，而是买卖价差（Spread）和库存调整的复合收益。经典的 Bizer & Glosten 模型指出，做市商的期望收益由三部分组成：价差收益、库存成本、价格冲击损失。
假设我们在 Binance Futures 上做 BTC/USDT 做市商，当 BTC 价格从 45000 跌至 43000 时，如果我们持有 2 BTC 净多头，库存损失就是 4000 USDT。但真正的复杂性在于，库存是动态变化的——每一笔成交都会改变我们的持仓，而持仓变化又会影响我们报价的价差决策。
二、系统架构：实时流处理 + 异步分析双轨制
我做过的第一版 PnL 系统是同步架构，所有数据处理都在收到 Tick 的同一进程中完成。实测在 Bybit 500ms 切片数据下，延迟从 15ms 飙升到 120ms，CPU 使用率超过 85%。这个性能瓶颈让我被迫重构为双轨架构。
"""
HolySheep AI 驱动的做市商 PnL 分析系统架构
核心设计：Tardis 实时流 → Redis 缓存 → AI 异步分析
作者实战经验：异步分析层使用 Claude Sonnet 4.5 做自然语言 PnL 归因
"""
import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import redis.asyncio as redis
from tardis_client import TardisClient, TardisFilters
import httpx
HolySheep AI 配置 - 用于 PnL 归因分析
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 从 https://www.holysheep.ai/register 获取
@dataclass
class OrderBookSnapshot:
"""订单簿快照结构"""
exchange: str
symbol: str
timestamp: int
asks: List[tuple] # [(price, size), ...]
bids: List[tuple]
@property
def mid_price(self) -> float:
if not self.asks or not self.bids:
return 0.0
return (float(self.asks[0][0]) + float(self.bids[0][0])) / 2
@property
def spread_bps(self) -> float:
"""价差，单位：基点 (bps)"""
if not self.asks or not self.bids:
return 0.0
return (float(self.asks[0][0]) - float(self.bids[0][0])) / self.mid_price * 10000
@dataclass
class InventoryState:
"""库存状态机"""
symbol: str
base_position: float = 0.0 # 标的资产数量
quote_position: float = 0.0 # 计价资产数量
unrealized_pnl: float = 0.0
realized_pnl: float = 0.0
trades: List[dict] = field(default_factory=list)
def update_from_trade(self, side: str, price: float, size: float, fee_rate: float):
"""成交后更新库存状态"""
notional = price * size
fee = notional * fee_rate
if side == "buy":
self.base_position += size
self.quote_position -= (notional + fee)
else:
self.base_position -= size
self.quote_position += (notional - fee)
self.trades.append({
"side": side,
"price": price,
"size": size,
"fee": fee,
"timestamp": time.time()
})
def mark_to_market(self, current_price: float):
"""按市价计价库存盈亏"""
self.unrealized_pnl = self.base_position * current_price + self.quote_position
class MarketMakerPnLSystem:
"""
做市商 PnL 分析系统
架构说明：
1. Tardis WebSocket 实时接收 Order Book 数据
2. Redis 缓存最新订单簿快照（TTL 1秒）
3. 后台任务定期计算 VaR 和库存风险
4. HolySheep AI 异步进行自然语言 PnL 归因分析
"""
def __init__(self, symbols: List[str], exchanges: List[str]):
self.symbols = symbols
self.exchanges = exchanges
self.order_books: Dict[str, OrderBookSnapshot] = {}
self.inventory: Dict[str, InventoryState] = {
s: InventoryState(symbol=s) for s in symbols
}
# Redis 连接池 - 延迟初始化
self._redis: Optional[redis.Redis] = None
self._http_client: Optional[httpx.AsyncClient] = None
# 性能指标
self.metrics = {
"messages_processed": 0,
"latency_p50_ms": 0.0,
"latency_p99_ms": 0.0,
"last_update": None
}
async def initialize(self):
"""异步初始化连接"""
self._redis = await redis.from_url("redis://localhost:6379/0")
self._http_client = httpx.AsyncClient(timeout=30.0)
# 初始化 Tardis 连接
self._tardis_client = TardisClient()
async def subscribe_orderbook(self, exchange: str, symbol: str):
"""订阅 Tardis Order Book 实时数据"""
filters = TardisFilters().symbol(symbol).exchange(exchange)
# 实战经验：Tardis 的 WebSocket 重连机制需要自己实现
# 这里使用指数退避策略，最长等待 30 秒
retry_count = 0
max_retries = 10
while retry_count < max_retries:
try:
# 异步迭代 Order Book 数据
async for timestamp, action, data in self._tardis_client.replay(
filters=filters,
from_datetime=datetime.utcnow(),
to_datetime=None, # 持续接收直到手动停止
validate=True
):
if action == "book_snapshot": # 快照数据
await self._process_orderbook_snapshot(exchange, symbol, data)
elif action == "book_update": # 增量更新
await self._process_orderbook_update(exchange, symbol, data)
self.metrics["messages_processed"] += 1
self.metrics["last_update"] = time.time()
except Exception as e:
retry_count += 1
wait_time = min(2 ** retry_count, 30