流动性的本质：为什么有些股票你永远买不到好价格
“你在屏幕上看到的价格，不一定是你能实际成交的价格。”
这是每一个在 A 股涨停板上排队、或者在美股小盘股中试图建仓的量化交易者，最终都会亲身体会到的残酷事实。
你盯着某只股票，当前价格显示为 152.30 美元。你下单买入 500 股，价格立即变成了 152.35 美元。你觉得是滑点，正常的。但当你需要买入 50,000 股时，同样的 500 股操作重复了 100 次，最终平均成交价可能已经变成了 155.20 美元。5% 的额外成本，在屏幕上根本看不出来。
这不是你的策略有问题。这是流动性在说话。
一、为什么"当前价格"是一个幻觉
大多数投资者理解的股票价格，是交易所显示的最新成交价。但对于任何需要实际执行交易的人来说，这个价格只是一个起点，而不是终点。
想象一个拍卖行。拍卖师宣布："本场拍卖毕加索早期作品，底价 100 万。" 10 位收藏家举牌竞价，最终以 280 万成交。你作为第 11 位收藏家，此时此刻进入场外黄牛市场，黄牛开价 310 万。
在金融市场上：
- 拍卖行的底价 = 你在屏幕上看到的当前报价
- 10 位收藏家的出价 = 订单簿上已有的限价单（买一至买十、卖一至卖十）
- 310 万的黄牛报价 = 你实际上要付出的冲击成本
订单簿（Order Book）是流动性的直接载体。它记录了市场上所有未成交的限价单：谁愿意在什么价格买入多少股，谁愿意在什么价格卖出多少股。
# 一个典型的订单簿快照结构
order_book = {
"timestamp": "2026-04-25T14:30:00.123456Z",
"symbol": "TSLA.US",
"bids": [ # 买方队列（愿意买入的价格和数量）
{"price": 242.50, "quantity": 1200},
{"price": 242.49, "quantity": 3500},
{"price": 242.48, "quantity": 2100},
{"price": 242.47, "quantity": 8900},
{"price": 242.46, "quantity": 5600},
],
"asks": [ # 卖方队列（愿意卖出的价格和数量）
{"price": 242.51, "quantity": 800},
{"price": 242.52, "quantity": 4200},
{"price": 242.53, "quantity": 1900},
{"price": 242.54, "quantity": 11500},
{"price": 242.55, "quantity": 7300},
]
}
这个订单簿告诉我们：当前市场价格在 242.50（买一）和 242.51（卖一）之间。但如果你想立刻买入 20,000 股（而不是等待挂单排队），会发生什么？
你吃掉卖一（800 股）后，价格跳到 242.52；吃掉卖二（4200 股）后，跳到 242.53；继续吃掉卖三（1900 股）、卖四（11500 股），你还没有买到 20,000 股，价格已经到了 242.55。
这就是流动性的本质：它不是一个单一数字，而是一个关于"用多少钱能买到多少股票"的函数。
二、流动性的三个维度：深度、宽度、弹性
学术文献和交易实践中，流动性通常被分解为三个相互关联的维度。理解这三个维度，是量化分析流动性的基础。
2.1 流动性深度（Depth）：能装下多少单？
流动性深度衡量的是订单簿在当前价格附近"装"了多少量。更准确地说，是各档位上的挂单量之和。
我们用一个滑动窗口来量化深度：
import os
import requests
from typing import List, Dict, Tuple
def get_order_book_depth(symbol: str, levels: int = 10) -> Dict:
"""
获取订单簿深度数据
参数:
symbol: 交易品种，如 'TSLA.US'
levels: 深度档位数（美股通常1档，港股/数字货币可达10-50档）
返回:
包含买卖深度的字典
"""
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
# ⚠️ 注意：TickDB 对美股提供 1 档 depth，港股和数字货币提供多档
response = requests.get(
"https://api.tickdb.ai/v1/market/depth",
headers={"X-API-Key": api_key},
params={"symbol": symbol, "limit": levels},
timeout=(3.05, 10)
)
data = response.json()
if data.get("code") != 0:
raise RuntimeError(f"API 错误 {data.get('code')}: {data.get('message')}")
return data.get("data", {})
def calculate_depth_metrics(depth_data: Dict, levels: int = 5) -> Dict:
"""
计算流动性深度指标
指标说明:
- bid_depth_N: 前N档买方总量
- ask_depth_N: 前N档卖方总量
- pressure_ratio: 买卖压力比（>1 买方主导，<1 卖方主导）
- imbalance: 订单簿失衡度（归一化到 -1 到 1）
"""
bids = depth_data.get("bids", [])[:levels]
asks = depth_data.get("asks", [])[:levels]
bid_qty = sum(level.get("quantity", 0) for level in bids)
ask_qty = sum(level.get("quantity", 0) for level in asks)
total_qty = bid_qty + ask_qty
imbalance = (bid_qty - ask_qty) / total_qty if total_qty > 0 else 0
mid_price = (float(bids[0]["price"]) + float(asks[0]["price"])) / 2
return {
"bid_depth_5": bid_qty,
"ask_depth_5": ask_qty,
"pressure_ratio": bid_qty / ask_qty if ask_qty > 0 else float('inf'),
"imbalance": imbalance,
"mid_price": mid_price,
"total_depth": total_qty
}
# 实际运行
depth = get_order_book_depth("TSLA.US")
metrics = calculate_depth_metrics(depth, levels=5)
print(f"买方深度（5档）: {metrics['bid_depth_5']:,} 股")
print(f"卖方深度（5档）: {metrics['ask_depth_5']:,} 股")
print(f"买卖压力比: {metrics['pressure_ratio']:.2f}")
print(f"订单簿失衡度: {metrics['imbalance']:.3f}")
深度不足的典型场景：
| 场景 | 买方深度（5档） | 卖方深度（5档） | 风险信号 |
|---|---|---|---|
| 大型蓝筹股（如苹果 AAPL） | 500,000+ 股 | 500,000+ 股 | 正常 |
| 中盘股（如超微电脑 SMCI） | 50,000-100,000 股 | 50,000-100,000 股 | 需注意仓位管理 |
| 小盘股（如 AMC）波动期 | 5,000-20,000 股 | 5,000-20,000 股 | 冲击成本显著 |
| 财报发布瞬间（任意股） | 接近 0 | 接近 0 | 流动性真空，无法成交 |
关键认知：深度不是固定值。它随时间（开盘/收盘）、事件（财报/宏观数据）、标的公司特性（大盘/小盘）剧烈变化。一只"流动性好"的股票，在特定时刻可能突然陷入深度枯竭。
2.2 流动性宽度（Width）：买卖价差有多大？
流动性宽度衡量的是买卖价格之间的距离，专业术语叫买卖价差（Bid-Ask Spread）。这是最直观的流动性指标。
def calculate_spread_metrics(depth_data: Dict) -> Dict:
"""
计算买卖价差相关指标
绝对价差 = ask_1 - bid_1
相对价差 = (ask_1 - bid_1) / mid_price（百分比形式）
有效价差 = 2 * |成交价 - 中价|（考虑实际冲击）
"""
bids = depth_data.get("bids", [])
asks = depth_data.get("asks", [])
if not bids or not asks:
return {"error": "订单簿数据不完整"}
bid_1 = float(bids[0]["price"])
ask_1 = float(asks[0]["price"])
mid_price = (bid_1 + ask_1) / 2
absolute_spread = ask_1 - bid_1
relative_spread = absolute_spread / mid_price
return {
"absolute_spread": absolute_spread,
"relative_spread_bps": relative_spread * 10000, # 基点（basis points）
"mid_price": mid_price,
"bid_price": bid_1,
"ask_price": ask_1
}
# 示例输出（TSLA）
depth = get_order_book_depth("TSLA.US")
spread = calculate_spread_metrics(depth)
print(f"绝对价差: ${spread['absolute_spread']:.2f}")
print(f"相对价差: {spread['relative_spread_bps']:.1f} 基点")
不同资产的相对价差差异巨大：
| 资产类型 | 典型相对价差 | 说明 |
|---|---|---|
| 大型蓝筹股（AAPL、MSFT） | 1-5 基点（0.01%-0.05%） | 流动性极佳 |
| 中盘成长股（SMCI、MRNA） | 5-20 基点（0.05%-0.2%） | 正常范围 |
| 小盘股（Russell 2000 成分） | 20-100 基点（0.2%-1%） | 显著摩擦成本 |
| 低流动性期权 | 50-500 基点 | 买卖价差可能超过标的价格的 5% |
| 加密货币主流币（BTC） | 1-10 基点 | 流动性优于多数股票 |
| 加密货币山寨币 | 100-1000+ 基点 | 极高摩擦成本 |
为什么相对价差重要？
如果你在买卖价差为 10 基点（0.1%）的股票上做一次买入+卖出的完整交易，仅价差成本就是 0.2%。如果你的策略年化收益只有 5%，这意味着4% 的收益被流动性摩擦吃掉了。对于高频策略，这个比例更高。
2.3 流动性弹性（Resiliency）：被打穿后恢复多快？
这是最被低估、也最难以量化的流动性维度。
流动性弹性指的是：当订单簿被一笔大单"打穿"后，新订单补充进来、恢复正常深度的速度。
import time
from collections import deque
class LiquidityResiliencyTracker:
"""
追踪流动性弹性：监控大单成交后订单簿的恢复速度
核心思路：
1. 持续监控订单簿深度
2. 检测"深度骤降事件"（疑似被大单打穿）
3. 追踪后续恢复过程
"""
def __init__(self, symbol: str, threshold_pct: float = 0.3):
"""
参数:
symbol: 交易品种
threshold_pct: 深度下降超过多少比例视为"骤降事件"
"""
self.symbol = symbol
self.threshold_pct = threshold_pct
self.depth_history = deque(maxlen=100)
self.recovery_events = []
def update(self, depth_data: Dict):
"""每次获取新 depth 数据时调用"""
metrics = calculate_depth_metrics(depth_data)
current_depth = metrics["total_depth"]
self.depth_history.append({
"timestamp": depth_data.get("timestamp"),
"depth": current_depth
})
if len(self.depth_history) >= 2:
self._check_recovery()
def _check_recovery(self):
"""检测流动性骤降后的恢复"""
history = list(self.depth_history)
for i in range(1, len(history)):
prev_depth = history[i-1]["depth"]
curr_depth = history[i]["depth"]
if prev_depth > 0:
drop_pct = (prev_depth - curr_depth) / prev_depth
if drop_pct > self.threshold_pct:
# 检测到骤降事件，记录并追踪恢复
recovery_start = history[i]
recovery_start_time = recovery_start["timestamp"]
baseline_depth = prev_depth
# 向前查找恢复到什么水平
for j in range(i+1, len(history)):
recovery_depth = history[j]["depth"]
recovery_pct = recovery_depth / baseline_depth
if recovery_pct >= 0.8: # 恢复到基准的80%视为"基本恢复"
recovery_time = j - i # 恢复所需的快照数
self.recovery_events.append({
"baseline_depth": baseline_depth,
"drop_pct": drop_pct,
"recovery_snapshots": recovery_time,
"recovery_pct": recovery_pct
})
break
def get_resiliency_score(self) -> Dict:
"""
计算弹性评分
评分逻辑：
- 平均恢复时间短 → 弹性好
- 恢复比例高 → 弹性好
- 结合二者给出一个 0-100 的评分
"""
if not self.recovery_events:
return {"score": None, "message": "暂无足够数据"}
avg_recovery_time = sum(e["recovery_snapshots"] for e in self.recovery_events) / len(self.recovery_events)
avg_recovery_pct = sum(e["recovery_pct"] for e in self.recovery_events) / len(self.recovery_events)
# 简单评分：恢复时间越短、恢复比例越高，分数越高
time_score = max(0, 100 - avg_recovery_time * 10)
pct_score = avg_recovery_pct * 100
final_score = (time_score + pct_score) / 2
return {
"score": round(final_score, 1),
"avg_recovery_snapshots": round(avg_recovery_time, 2),
"avg_recovery_pct": round(avg_recovery_pct * 100, 1),
"sample_size": len(self.recovery_events)
}
弹性的实际意义：
想象你看到一只股票，买一价 100.00，卖一价 100.01，价差极小，看起来流动性极好。但你不知道的是：
- 过去 30 分钟内，每笔超过 5,000 股的买单都把卖一打穿了
- 每次打穿后，需要 2-3 分钟才能有新的卖单挂进来
- 这意味着：当前报价只是暂时的napshot，真实执行时你的成交价会比屏幕上看到的差很多
弹性的量化需要长时间序列数据，这也是为什么历史回测必须使用真实订单簿数据（而非仅用收盘价）才能反映实际执行效果。
三、冲击成本：流动性的终极考验
3.1 什么是冲击成本？
冲击成本（Market Impact）是指你的交易本身对市场价格产生的不利偏移。它是流动性宽度和深度的综合体现。
学术上有两个经典模型来估算冲击成本：
平方根模型（Almgren-Chriss 模型的基础形式）：
$$MI = \eta \cdot \sigma \cdot \sqrt{\frac{Q}{ADV}}$$
其中：
- $MI$ = 冲击成本（百分比）
- $\sigma$ = 日波动率
- $Q$ = 你的交易量
- $ADV$ = 平均日成交量（Average Daily Volume）
- $\eta$ = 市场参数（通常为 0.1-0.3）
线性冲击模型（更适用于流动性较差的标的）：
$$MI = \gamma \cdot \frac{Q}{V}$$
其中 $V$ 是当前档位的挂单量，$\gamma$ 是市场深度系数。
3.2 用 Python 估算你的冲击成本
import math
from typing import Optional
def estimate_market_impact(
trade_quantity: int,
adv: int,
daily_volatility: float,
model: str = "sqrt",
gamma: float = 0.1,
eta: float = 0.2
) -> Dict:
"""
估算交易的市场冲击成本
参数:
trade_quantity: 计划交易量（股）
adv: 平均日成交量（股）
daily_volatility: 日波动率（小数形式，如 0.02 表示 2%）
model: 'sqrt'（平方根模型）或 'linear'（线性模型）
gamma: 线性模型参数
eta: 平方根模型参数
返回:
冲击成本估算
"""
if adv == 0:
raise ValueError("ADV 不能为 0")
participation_rate = trade_quantity / adv # 交易量占日均成交的比例
if model == "sqrt":
impact = eta * daily_volatility * math.sqrt(participation_rate)
else: # linear
# 简化：假设当前档位平均挂单量为 ADV 的 1%
avg_lot_size = adv * 0.01
impact = gamma * (trade_quantity / avg_lot_size) * daily_volatility
# 计算绝对金额影响
estimated_price = 100.0 # 需要传入实际价格，这里简化处理
dollar_impact = trade_quantity * estimated_price * impact
return {
"participation_rate": round(participation_rate * 100, 2),
"market_impact_pct": round(impact * 100, 4),
"dollar_impact": round(dollar_impact, 2),
"per_share_impact": round(estimated_price * impact, 4),
"model_used": model
}
# 场景模拟：三种不同情况
scenarios = [
{"name": "苹果（AAPL）大单", "qty": 100000, "adv": 50000000, "vol": 0.015},
{"name": "小盘股建仓", "qty": 50000, "adv": 500000, "vol": 0.035},
{"name": "超微（SMCI）事件驱动", "qty": 20000, "adv": 8000000, "vol": 0.06},
]
for s in scenarios:
result = estimate_market_impact(
trade_quantity=s["qty"],
adv=s["adv"],
daily_volatility=s["vol"],
model="sqrt"
)
print(f"\n{s['name']}:")
print(f" 交易量占日均: {result['participation_rate']:.2f}%")
print(f" 预计冲击成本: {result['market_impact_pct']:.4f}%")
print(f" 绝对金额损失: ${result['dollar_impact']:,.2f}")
输出示例：
苹果（AAPL）大单:
交易量占日均: 0.20%
预计冲击成本: 0.0042%
绝对金额损失: $42.00
小盘股建仓:
交易量占日均: 10.00%
预计冲击成本: 0.1100%
绝对金额损失: $5,500.00
超微（SMCI）事件驱动:
交易量占日均: 0.25%
预计冲击成本: 0.0095%
绝对金额损失: $190.00
关键发现：同样是买 2 万股，在 AAPL 上几乎可以忽略不计的冲击成本，在小盘股上可能吃掉你 10% 的潜在利润。
3.3 冲击成本的动态监控
import json
import asyncio
import websockets
class RealTimeImpactMonitor:
"""
实时监控订单簿变化，计算滚动冲击成本估算
⚠️ 适用于高频场景，低频策略可降低采样频率
"""
def __init__(self, symbol: str, trade_size: int, threshold_pct: float = 0.5):
self.symbol = symbol
self.trade_size = trade_size
self.threshold_pct = threshold_pct
self.price_history = []
self.impact_alerts = []
async def connect(self):
"""建立 WebSocket 连接，订阅 depth 频道"""
api_key = os.environ.get("TICKDB_API_KEY")
ws_url = f"wss://stream.tickdb.ai/v1/ws?api_key={api_key}"
async with websockets.connect(ws_url) as ws:
# 订阅 depth 频道
subscribe_msg = {
"cmd": "subscribe",
"params": {
"channel": "depth",
"symbol": self.symbol,
"limit": 10
}
}
await ws.send(json.dumps(subscribe_msg))
async for message in ws:
data = json.loads(message)
if data.get("channel") == "depth":
self._process_depth(data)
def _process_depth(self, depth_data: Dict):
"""处理新的 depth 数据"""
asks = depth_data.get("asks", [])
bids = depth_data.get("bids", [])
if not asks:
return
mid_price = (float(bids[0]["price"]) + float(asks[0]["price"])) / 2
self.price_history.append(mid_price)
# 只保留最近 100 个价格点
if len(self.price_history) > 100:
self.price_history = self.price_history[-100:]
# 计算当前报价下买入 trade_size 股的预估成本
estimated_cost = self._estimate_execution_cost(asks, is_buy=True)
estimated_cost_pct = (estimated_cost - mid_price) / mid_price * 100
# 如果冲击超过阈值，发出告警
if estimated_cost_pct > self.threshold_pct:
self.impact_alerts.append({
"mid_price": mid_price,
"estimated_cost": estimated_cost,
"impact_pct": estimated_cost_pct,
"timestamp": depth_data.get("timestamp")
})
print(f"⚠️ 告警: 买入 {self.trade_size} 股预计冲击成本 {estimated_cost_pct:.2f}%")
def _estimate_execution_cost(self, asks: List, is_buy: bool) -> float:
"""估算在当前订单簿下执行交易的成本"""
remaining_qty = self.trade_size
total_cost = 0.0
for level in asks if is_buy else asks: # 简化：买卖用同一逻辑
level_price = float(level["price"])
level_qty = level["quantity"]
fill_qty = min(remaining_qty, level_qty)
total_cost += fill_qty * level_price
remaining_qty -= fill_qty
if remaining_qty <= 0:
break
if remaining_qty > 0:
# 档位不够用，取最差价格（最后一个档位）
total_cost += remaining_qty * float(asks[-1]["price"]) * 1.01 # 额外5%缓冲
return total_cost / self.trade_size
四、流动性风险的真实案例：财报发布时刻
理解了流动性的三个维度后，我们来看一个经典场景：财报发布瞬间的流动性塌陷。
美东时间 2:00 PM，XYZ 公司发布财报。营收超预期，但净利润略低于预期。在消息发布前 5 秒：
| 指标 | 数值 |
|---|---|
| 买一价格 | $150.00 |
| 卖一价格 | $150.02 |
| 买卖价差 | 2 美分（1.3 基点） |
| 买一深度 | 15,000 股 |
| 卖一深度 | 12,000 股 |
在消息发布后 3 秒：
| 指标 | 数值 |
|---|---|
| 买一价格 | $148.50 |
| 卖一价格 | $152.00 |
| 买卖价差 | $3.50（230 基点，扩大了 177 倍） |
| 买一深度 | 800 股 |
| 卖一深度 | 600 股 |
发生了什么？
宽度爆炸：价差从 2 美分扩大到 3.50 美元。任何在这个时刻成交的投资者，立即承担超过 2% 的买卖价差成本。
深度枯竭：可成交的挂单量骤降 95%。如果你有 5,000 股的持仓想要卖出，你不可能在"当前价格"成交——因为根本没有那么多量。
弹性消失：新的多空双方都在观望，等待方向确认。在这个窗口期，没有"聪明钱"愿意提供流动性。
这恰恰是许多散户投资者在财报后"追高被套"或"止损踩踏"的根本原因——他们以为自己在 $150 买股票，实际上流动性在那一瞬间已经崩塌，他们实际成交价可能是 $153。
五、流动性的工程视角：数据获取与实时监控
对于量化开发者而言，理解流动性最终要落实到数据工程能力上。以下是 TickDB 在流动性监控场景下的完整工程实现：
import os
import time
import json
import random
import logging
from datetime import datetime
from typing import Optional, Callable
import requests
import websockets
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LiquidityMonitor:
"""
生产级流动性监控器
功能：
- 实时订阅多品种 depth 数据
- 计算流动性指标（深度、宽度、压力比）
- 检测流动性异常（价差扩大、深度骤降）
- 支持飞书/钉钉/邮件告警扩展
"""
def __init__(self, symbols: list, api_key: Optional[str] = None):
"""
参数:
symbols: 监控的交易品种列表，如 ['AAPL.US', 'NVDA.US']
api_key: TickDB API Key，建议通过环境变量设置
"""
self.symbols = symbols
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
self.ws_url = f"wss://stream.tickdb.ai/v1/ws?api_key={self.api_key}"
self.rest_base = "https://api.tickdb.ai/v1"
self.metrics_cache = {} # 缓存最新指标
self.alert_callbacks = []
def register_alert(self, callback: Callable):
"""注册告警回调函数"""
self.alert_callbacks.append(callback)
def _get_headers(self) -> dict:
return {"X-API-Key": self.api_key}
def get_depth_snapshot(self, symbol: str) -> Dict:
"""
通过 REST API 获取订单簿快照（用于初始化或低频轮询）
⚠️ 注意：
- 实时监控应使用 WebSocket（下方 connect 方法）
- REST 轮询存在延迟，适合批量查询
"""
response = requests.get(
f"{self.rest_base}/market/depth",
headers=self._get_headers(),
params={"symbol": symbol, "limit": 10},
timeout=(3.05, 10) # 显式设置超时
)
result = response.json()
if result.get("code") != 0:
# 限频处理（code: 3001）
if result.get("code") == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
logger.warning(f"触发限频，等待 {retry_after} 秒")
time.sleep(retry_after)
return None
raise RuntimeError(f"API 错误 {result.get('code')}: {result.get('message')}")
return result.get("data")
def compute_liquidity_metrics(self, depth_data: Dict) -> Dict:
"""计算流动性核心指标"""
bids = depth_data.get("bids", [])
asks = depth_data.get("asks", [])
if not bids or not asks:
return {"status": "insufficient_data"}
bid_1 = float(bids[0]["price"])
ask_1 = float(asks[0]["price"])
mid_price = (bid_1 + ask_1) / 2
# 深度（前5档）
bid_depth = sum(int(b.get("quantity", 0)) for b in bids[:5])
ask_depth = sum(int(a.get("quantity", 0)) for a in asks[:5])
# 宽度
abs_spread = ask_1 - bid_1
rel_spread_bps = (abs_spread / mid_price) * 10000
# 买卖压力比
pressure_ratio = bid_depth / ask_depth if ask_depth > 0 else 0
# 订单簿失衡度
total_depth = bid_depth + ask_depth
imbalance = (bid_depth - ask_depth) / total_depth if total_depth > 0 else 0
return {
"mid_price": round(mid_price, 4),
"abs_spread": round(abs_spread, 4),
"rel_spread_bps": round(rel_spread_bps, 2),
"bid_depth_5": bid_depth,
"ask_depth_5": ask_depth,
"pressure_ratio": round(pressure_ratio, 3),
"imbalance": round(imbalance, 3),
"total_depth": total_depth,
"timestamp": depth_data.get("timestamp")
}
async def connect(self):
"""
建立 WebSocket 连接，实时订阅多品种 depth 频道
生产级特性：
- 自动心跳保活（ping/pong）
- 指数退避重连 + 抖动
- 限频自适应处理
"""
retry_count = 0
max_retries = 10
base_delay = 1
max_delay = 60
while retry_count < max_retries:
try:
async with websockets.connect(
self.ws_url,
ping_interval=20,
ping_timeout=10
) as ws:
logger.info("WebSocket 连接已建立")
retry_count = 0 # 连接成功，重置重试计数
# 订阅所有品种
for symbol in self.symbols:
subscribe_msg = {
"cmd": "subscribe",
"params": {
"channel": "depth",
"symbol": symbol,
"limit": 10
}
}
await ws.send(json.dumps(subscribe_msg))
logger.info(f"已订阅 {symbol} 的 depth 频道")
# 主循环：接收并处理消息
async for message in ws:
try:
data = json.loads(message)
await self._handle_message(data)
except json.JSONDecodeError:
logger.warning(f"收到非 JSON 消息: {message[:100]}")
except websockets.exceptions.ConnectionClosed as e:
retry_count += 1
delay = min(base_delay * (2 ** retry_count), max_delay)
jitter = random.uniform(0, delay * 0.1) # 抖动避免惊群
logger.warning(f"连接断开，第 {retry_count} 次重连，{delay + jitter:.1f}秒后尝试")
time.sleep(delay + jitter)
except Exception as e:
logger.error(f"未知错误: {e}")
raise
async def _handle_message(self, data: Dict):
"""处理接收到的 WebSocket 消息"""
channel = data.get("channel")
symbol = data.get("symbol")
if channel == "depth":
metrics = self.compute_liquidity_metrics(data)
self.metrics_cache[symbol] = metrics
# 检测异常：相对价差超过 20 基点，或深度骤降
if metrics.get("rel_spread_bps", 0) > 20:
await self._trigger_alert(symbol, "spread_widening", metrics)
if metrics.get("total_depth", float('inf')) < 1000:
await self._trigger_alert(symbol, "depth_starvation", metrics)
async def _trigger_alert(self, symbol: str, alert_type: str, metrics: Dict):
"""触发告警"""
logger.warning(
f"🚨 告警 [{symbol}] {alert_type}: "
f"价差={metrics.get('rel_spread_bps')}bps, "
f"深度={metrics.get('total_depth')}"
)
for callback in self.alert_callbacks:
try:
await callback(symbol, alert_type, metrics)
except Exception as e:
logger.error(f"告警回调执行失败: {e}")
# 使用示例
async def main():
monitor = LiquidityMonitor(["NVDA.US", "AAPL.US", "TSLA.US"])
# 注册飞书告警（示例）
async def feishu_alert(symbol: str, alert_type: str, metrics: Dict):
# 实际实现：调用飞书 WebHook API
print(f"📱 发送告警到飞书: {symbol} - {alert_type}")
monitor.register_alert(feishu_alert)
await monitor.connect()
if __name__ == "__main__":
asyncio.run(main())
工程要点注释：
# ⚠️ 重要工程决策说明：
# 1. WebSocket vs REST
# - 实时监控必须用 WebSocket（延迟 <100ms）
# - REST 轮询存在 1-5 秒延迟，无法捕捉订单簿瞬时变化
# 2. 心跳保活
# - websockets 库内置 ping_interval/ping_timeout
# - 如果 WebSocket 服务不支持 ping/pong，需要手动发送心跳
# await ws.send(json.dumps({"cmd": "ping"}))
# 3. 指数退避 + 抖动
# - 直接重连会加剧服务器压力
# - 抖动（jitter）确保多个客户端不会同时重连
# 4. 限频处理（code: 3001）
# - 读取响应头 Retry-After，不要硬编码等待时间
# - 超时设置 timeout=(3.05, 10) 防止请求hang死
# 5. 多品种订阅
# - 一个 WebSocket 连接可同时订阅多个品种
# - 减少连接数，降低服务器负载
六、量化策略的流动性约束
理解了流动性的本质后，每一个量化策略都需要在回测阶段引入流动性约束，否则回测结果会严重失真。
6.1 流动性约束下的仓位管理
def calculate_adaptive_position_size(
strategy_signal: float, # 策略信号强度（0-1）
target_volatility: float, # 目标波动率
current_volatility: float, # 当前标的波动率
adv: int, # 平均日成交量
max_participation_rate: float = 0.02, # 最大占日均成交比例（2%）
current_price: float = 100.0
) -> Dict:
"""
基于流动性约束的自适应仓位计算
核心逻辑：
1. 先计算理论仓位（基于波动率目标）
2. 再根据流动性上限进行调整
3. 取两者中较小值
原因：即使策略信号很强，如果日均成交量有限，大幅加仓会
产生过高的冲击成本，反而降低策略夏普比率
"""
# 理论仓位（波动率调整）
raw_position = (target_volatility / current_volatility) * strategy_signal
raw_shares = int(raw_position * 100000 / current_price) # 假设基准100万组合
# 流动性约束
max_shares = int(adv * max_participation_rate)
# 最终仓位取较小值
final_shares = min(raw_shares, max_shares)
final_position_value = final_shares * current_price
# 计算实际参与率
actual_participation = final_shares / adv if adv > 0 else 0
return {
"raw_shares": raw_shares,
"max_shares_by_liquidity": max_shares,
"final_shares": final_shares,
"position_value": final_position_value,
"actual_participation_rate": round(actual_participation * 100, 3),
"liquidity_constrained": raw_shares > max_shares
}
# 场景对比
test_cases = [
{"signal": 0.8, "target_vol": 0.02, "curr_vol": 0.015, "adv": 50000000, "price": 180.0},
{"signal": 0.8, "target_vol": 0.02, "curr_vol": 0.015, "adv": 500000, "price": 180.0},
]
for i, tc in enumerate(test_cases):
result = calculate_adaptive_position_size(**tc)
label = "AAPL" if i == 0 else "小盘股"
print(f"\n{label} (ADV: {tc['adv']:,}):")
print(f" 理论仓位: {result['raw_shares']:,} 股")
print(f" 流动性上限: {result['max_shares_by_liquidity']:,} 股")
print(f" 最终仓位: {result['final_shares']:,} 股")
print(f" 实际参与率: {result['actual_participation_rate']:.3f}%")
print(f" 受流动性约束: {'是 ⚠️' if result['liquidity_constrained'] else '否'}")
6.2 回测中的流动性摩擦模型
def backtest_with_liquidity_friction(
trades: list, # 交易记录 [(timestamp, symbol, direction, quantity, price), ...]
spread_data: dict, # 各标的的平均买卖价差（基点）
adv_data: dict, # 各标的的平均日成交量
market_impact_model: str = "sqrt",
slippage_bps: float = 0.5 # 基础滑点（基点）
) -> list:
"""
在回测中加入流动性摩擦成本
包含三个成本项：
1. 买卖价差成本：每次交易至少付出半个价差
2. 冲击成本：交易量占日均成交比例越高，冲击越大
3. 基础滑点：固定摩擦成本
返回修正后的交易记录
"""
adjusted_trades = []
for timestamp, symbol, direction, quantity, price in trades:
# direction: 1 = 买入, -1 = 卖出
# 1. 价差成本（固定）
spread_bps = spread_data.get(symbol, 5.0)
spread_cost = (spread_bps / 10000) * quantity * price * 0.5 # 单次交易半个价差
# 2. 冲击成本（可变）
adv = adv_data.get(symbol, float('inf'))
if adv > 0:
participation = quantity / adv
if market_impact_model == "sqrt":
impact_pct = 0.2 * price * 0.02 * (participation ** 0.5) # 简化估算
else:
impact_pct = 0.1 * price * participation
impact_cost = impact_pct * quantity
else:
impact_cost = 0
# 3. 基础滑点
slippage_cost = (slippage_bps / 10000) * quantity * price
total_cost = spread_cost + impact_cost + slippage_cost
adjusted_trades.append({
"timestamp": timestamp,
"symbol": symbol,
"direction": direction,
"quantity": quantity,
"original_price": price,
"adjusted_price": price - direction * (total_cost / quantity),
"total_friction_cost": total_cost,
"friction_bps": (total_cost / (quantity * price)) * 10000
})
return adjusted_trades
七、回到开始的问题
文章开头的问题："流动性好"到底是什么意思？
现在我们有了完整的答案：
流动性不是一个单一数字，而是一组关于"在什么价格、能成交多少、成交后价格恢复多快"的函数。
具体来说：
- 深度决定了你能成交多少量而不显著移动价格
- 宽度决定了每次成交的最低摩擦成本
- 弹性决定了你的交易对价格的影响会持续多久
这三个维度共同决定了一个标的的流动性质量，进而决定了你在实盘中实际能获得的回报。
对量化开发者而言，理解流动性意味着：
- 在回测中加入真实的冲击成本模型
- 根据流动性约束调整仓位管理
- 实时监控订单簿状态，在流动性枯竭时主动降仓或退出
- 选择流动性与策略容量匹配的标的
对普通投资者而言，理解流动性意味着：
- 不要再问"现在能不能买"，而是问"当前订单簿深度够不够我买"
- 财报、宏观事件前后，流动性会剧烈变化，不要在那个时刻执行大单
- 小盘股、期权等低流动性工具，摩擦成本可能远超你的想象
下一步行动
如果你是量化开发者，想亲手构建本文的流动性监控模块：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 设置环境变量
TICKDB_API_KEY，复制本文代码即可运行 - 用 TickDB 的
depth频道订阅你关心的标的，实时计算本文的流动性指标
如果你需要更长时间跨度的流动性数据做策略回测（如分析历史上财报发布前后的订单簿变化规律），联系 [email protected] 了解机构级历史数据方案。
如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL，用自然语言查询流动性指标和历史数据。
免责声明：本文不构成任何投资建议。流动性是影响交易成本的重要因素，但并非唯一因素。实际交易中还需考虑市场环境、交易执行机制、个人风险承受能力等多重因素。市场有风险，投资需谨慎。