美股盘前盘后交易的流动性陷阱：如何过滤假突破信号
开篇
“你的策略回测年化收益 47%，实盘第一周亏损 12%。”
这不是策略失效，而是样本污染——你可能把盘前盘后的极端行情当成了正常交易信号。
美股盘前盘后占总交易时长的 40%，但成交量仅占全天的 5%-15%。在流动性真空的时段里，一笔 500 股的市价单就能推动股价跳动 3%，技术指标全面失真，止损被反复扫掉，趋势跟踪策略变成“高买低卖”反向指标。
本文拆解盘前盘后的微观结构特征，给出交易时段过滤 + 成交量阈值 + 订单簿深度校验的三层过滤框架，并提供生产级的 Python 实现。核心目标：让你的策略在盘前盘后只做“看得懂的行情”，不做“流动性陷阱的猎物”。
一、盘前盘后的微观结构：数据不会说谎
1.1 时间窗口与成交量分布
美股典型交易日的成交量分布并非均匀。以纳斯达克上市的大型科技股为例（非特定标的，仅作结构说明）：
| 时段 | 交易时间（美东时） | 时长占比 | 成交量占比 | 平均买卖价差 |
|---|---|---|---|---|
| 盘前 | 04:00 - 09:30 | 23% | 5-8% | 常规的 3-8 倍 |
| 正常交易 | 09:30 - 16:00 | 27% | 80-85% | 基准 |
| 盘后 | 16:00 - 20:00 | 17% | 5-10% | 常规的 2-5 倍 |
| 非交易时段 | 其余时间 | 33% | <1% | 无 |
数据来源基于公开市场研究及 TickDB 历史 K 线数据的统计特征分析。关键结论：
- 盘前盘后合计贡献了 10%-18% 的成交量，但买卖价差是正常时段的 2-8 倍
- 流动性深度（订单簿厚度）在盘后 30 分钟内通常不足正常交易时段的 5%
1.2 价差扩大：订单簿失衡的量化表现
为什么价差会扩大？因为流动性提供者在低成交量环境下倾向于：
- 挂单价差保护：在不确定方向时，拉开买卖报价以覆盖库存风险
- 减少挂单深度：在波动加剧时，不愿意用大单锁定价格
- 滑点预期调整：预期低流动性下的执行冲击更大
这种行为在订单簿上呈现为：买一和卖一之间的价差扩大，而每一档的挂单量急剧减少。
典型盘后订单簿快照（示意）：
- 买一：150.00，挂单量 200 股
- 卖一：150.15，挂单量 150 股
- 买卖价差：0.15 美元（正常时段可能仅为 0.01）
- 深度（5档合计）：买盘 800 股，卖盘 650 股
在此基础上，一次 1000 股的市价单扫过卖一、卖二、卖三，成交均价约 150.10，相比买一的滑点超过 0.6%。这不是“剧烈波动”，这是流动性枯竭下的正常冲击。
1.3 波动率失真：虚假突破的温床
低流动性时段的技术指标会失真，原因在于：
- RSI / MACD：基于价格变化率，低成交量下的价格跳动权重被放大，指标极易进入超买超卖区间
- 布林带：盘后布林带开口可能收窄至正常时段的 1/3，突破上下轨的阈值失去参考意义
- 支撑阻力：盘前形成的日内高低点，往往由少数几笔大单决定，不代表市场共识
核心问题：如果你的策略在 09:30 开盘时触发信号，而这个信号的“依据”是 04:00-09:25 盘前累积的数据，那么你的策略本质上是用 5% 的流动性数据做 100% 的交易决策。
二、三层过滤框架：从时段判断到信号生效
解决流动性陷阱不能靠“简单关闭盘前盘后交易”。有些事件驱动策略（如财报发布后的盘后行情）恰恰需要盘后参与。正确的做法是分层过滤，让信号在满足流动性门槛时才触发。
2.1 第一层：交易时段过滤（Time-Based Filter）
目标：识别当前是否处于流动性可交易时段。
from datetime import datetime, time
import pytz
def get_trading_session(utc_now: datetime) -> str:
"""
判断当前处于哪个交易时段。
基于美东时区的交易所日历。
Returns:
"pre_market" - 盘前交易时段 (04:00-09:30 ET)
"regular" - 正常交易时段 (09:30-16:00 ET)
"after_hours" - 盘后交易时段 (16:00-20:00 ET)
"closed" - 非交易时段
"""
et_zone = pytz.timezone("US/Eastern")
et_now = utc_now.astimezone(et_zone)
# 周末判断
if et_now.weekday() >= 5:
return "closed"
current_time = et_now.time()
if time(4, 0) <= current_time < time(9, 30):
return "pre_market"
elif time(9, 30) <= current_time < time(16, 0):
return "regular"
elif time(16, 0) <= current_time < time(20, 0):
return "after_hours"
else:
return "closed"
使用逻辑：
- 对于趋势跟踪策略：仅在
regular时段执行核心逻辑 - 对于事件驱动策略：可将
pre_market和after_hours纳入交易窗口，但需叠加第二、第三层过滤
2.2 第二层：成交量阈值过滤（Volume Threshold）
目标：确保当前 K 线（或时间窗口）的成交量足以支撑技术信号的可靠性。
关键问题：成交量阈值怎么设？
推荐方法：自适应阈值，以过去 N 个正常交易时段 K 线的成交量均值为基准，计算当前成交量的相对强度（Volume Ratio）。
def calculate_volume_ratio(
current_volume: int,
historical_volumes: list[int],
percentile: float = 0.25
) -> float:
"""
计算成交量相对强度。
Args:
current_volume: 当前 K 线成交量
historical_volumes: 历史正常时段 K 线成交量列表（建议取 20 日日均量）
percentile: 阈值百分位（低于此百分位认为流动性不足）
Returns:
volume_ratio: 当前成交量 / 历史中位数
is_acceptable: 是否满足最低流动性要求
"""
if not historical_volumes:
return 0.0, False
# 使用中位数而非均值，对异常值更鲁棒
median_volume = sorted(historical_volumes)[len(historical_volumes) // 2]
if median_volume == 0:
return 0.0, False
ratio = current_volume / median_volume
# 低于 25% 分位视为流动性不足
threshold = 0.25
is_acceptable = ratio >= threshold
return ratio, is_acceptable
实战建议：
| 交易场景 | 最低 Volume Ratio | 说明 |
|---|---|---|
| 趋势跟踪（止损触发） | 0.5 | 宁可错过，不可做错 |
| 事件驱动（财报后） | 0.3 | 事件本身就是催化剂 |
| 盘中高频scalping | 1.0+ | 需要高于均量才有足够深度 |
2.3 第三层：订单簿深度校验（Order Book Depth Validation）
目标：在盘前盘后时段，通过实时订单簿数据判断当前是否具备足够的流动性执行订单。
这是最精细的一层过滤，也是 TickDB depth 频道的核心价值所在。
def validate_depth_for_trade(
depth_snapshot: dict,
order_size: int,
max_slippage_bps: float = 50.0
) -> dict:
"""
校验订单簿深度是否足以承载指定交易量。
Args:
depth_snapshot: TickDB depth 频道快照
{
"bids": [[price, quantity], ...],
"asks": [[price, quantity], ...]
}
order_size: 计划执行的股数
max_slippage_bps: 最大可接受滑点（基点）
Returns:
{
"is_tradeable": bool,
"estimated_slippage_bps": float,
"required_depth": int,
"available_depth": int,
"rejection_reason": str | None
}
"""
bids = depth_snapshot.get("bids", [])
asks = depth_snapshot.get("asks", [])
if not bids or not asks:
return {
"is_tradeable": False,
"estimated_slippage_bps": 0.0,
"required_depth": order_size,
"available_depth": 0,
"rejection_reason": "订单簿为空，数据不可用"
}
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread_bps = (best_ask - best_bid) / best_bid * 10000
# 计算扫单到达的价格
# 对于买入，模拟扫光卖盘各档
remaining = order_size
cumulative_cost = 0.0
cumulative_quantity = 0
for ask_price, ask_qty in asks:
fill_qty = min(remaining, int(ask_qty))
cumulative_cost += fill_qty * float(ask_price)
cumulative_quantity += fill_qty
remaining -= fill_qty
if remaining <= 0:
break
if remaining > 0:
return {
"is_tradeable": False,
"estimated_slippage_bps": float('inf'),
"required_depth": order_size,
"available_depth": cumulative_quantity,
"rejection_reason": f"订单簿深度不足：需要 {order_size} 股，仅有 {cumulative_quantity} 股可扫"
}
# 计算执行均价和滑点
avg_fill_price = cumulative_cost / cumulative_quantity
slippage_bps = (avg_fill_price - best_ask) / best_ask * 10000
return {
"is_tradeable": slippage_bps <= max_slippage_bps,
"estimated_slippage_bps": round(slippage_bps, 2),
"required_depth": order_size,
"available_depth": cumulative_quantity,
"rejection_reason": None if slippage_bps <= max_slippage_bps else f"滑点超限：预计 {slippage_bps:.1f} bps > 阈值 {max_slippage_bps} bps"
}
深度校验的决策矩阵：
| 场景 | 订单簿深度 | 滑点估算 | 决策 |
|---|---|---|---|
| 正常时段，买入 1000 股 | 卖盘合计 50,000 股 | < 5 bps | ✅ 执行 |
| 盘后，买入 1000 股 | 卖盘合计 800 股 | > 80 bps | ❌ 拒绝 |
| 盘后，买入 200 股 | 卖盘合计 800 股 | < 10 bps | ⚠️ 减仓执行 |
三、生产级实现：盘前盘后监控系统
将三层过滤整合为一个可实时运行的监控模块。以下代码展示了从 TickDB 获取实时数据、执行过滤判断、到触发告警的完整链路。
3.1 架构概览
┌─────────────────────────────────────────────────────────────────┐
│ 盘前盘后流动性监控系统 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ 交易时段判断 │ -> │ 成交量校验 │ -> │ 订单簿深度校验 │ │
│ │ (第一层) │ │ (第二层) │ │ (第三层) │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │ │ │ │
│ v v v │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 决策引擎 │ │
│ │ All Pass -> 执行信号 | 任一 Fail -> 记录/告警 │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ v │
│ ┌────────────┐ ┌────────────┐ ┌────────────────────────┐ │
│ │ 执行交易 │ │ 发送告警 │ │ 记录拒绝原因 (日志) │ │
│ └────────────┘ └────────────┘ └────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
3.2 完整实现代码
import os
import json
import time
import logging
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import requests
import websockets
import asyncio
import random
# ============================================================
# 配置与常量
# ============================================================
@dataclass
class LiquidityConfig:
"""流动性过滤配置"""
# 滑点阈值（基点）
max_slippage_bps: float = 50.0
# 成交量阈值（相对历史中位数的比例）
min_volume_ratio: float = 0.25
# 历史基准 K 线数量
lookback_klines: int = 20
# 盘前盘后是否允许交易
allow_pre_market: bool = False
allow_after_hours: bool = False
# WebSocket 重连参数
ws_reconnect_base_delay: float = 1.0
ws_reconnect_max_delay: float = 60.0
ws_max_retries: int = 10
# HTTP 请求超时
http_timeout: tuple = (3.05, 10)
class TradingSession(Enum):
CLOSED = "closed"
PRE_MARKET = "pre_market"
REGULAR = "regular"
AFTER_HOURS = "after_hours"
# ============================================================
# 日志配置
# ============================================================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("liquidity_monitor")
# ============================================================
# 第一层：交易时段判断
# ============================================================
def get_trading_session(utc_now: datetime) -> TradingSession:
"""判断当前美东时区交易时段"""
try:
import pytz
et_zone = pytz.timezone("US/Eastern")
et_now = utc_now.astimezone(et_zone)
if et_now.weekday() >= 5:
return TradingSession.CLOSED
current_time = et_now.time()
if current_time.hour < 9 or (current_time.hour == 9 and current_time.minute < 30):
if current_time.hour >= 4:
return TradingSession.PRE_MARKET
return TradingSession.CLOSED
elif current_time.hour < 16:
return TradingSession.REGULAR
elif current_time.hour < 20:
return TradingSession.AFTER_HOURS
else:
return TradingSession.CLOSED
except ImportError:
# 如果没有 pytz，用简化逻辑（仅工作日 9:30-16:00 为正常时段）
logger.warning("pytz 未安装，使用简化时段判断")
return TradingSession.REGULAR
# ============================================================
# 第二层：成交量阈值计算
# ============================================================
def calculate_volume_ratio(current_volume: int, historical_volumes: list[int]) -> float:
"""计算成交量相对强度"""
if not historical_volumes:
return 0.0
sorted_vols = sorted(historical_volumes)
median_idx = len(sorted_vols) // 2
median_volume = sorted_vols[median_idx]
if median_volume == 0:
return 0.0
return current_volume / median_volume
# ============================================================
# 第三层：订单簿深度校验
# ============================================================
def validate_depth_for_trade(
bids: list[list],
asks: list[list],
order_size: int,
max_slippage_bps: float
) -> dict:
"""校验订单簿是否满足交易深度要求"""
if not bids or not asks:
return {
"is_tradeable": False,
"slippage_bps": 0.0,
"available_depth": 0,
"reason": "订单簿数据为空"
}
best_ask = float(asks[0][0])
best_bid = float(bids[0][0])
spread_bps = (best_ask - best_bid) / best_bid * 10000
# 模拟扫单（买入场景：扫卖盘）
remaining = order_size
cumulative_cost = 0.0
cumulative_qty = 0
for ask_price, ask_qty in asks:
fill_qty = min(remaining, int(float(ask_qty)))
cumulative_cost += fill_qty * float(ask_price)
cumulative_qty += fill_qty
remaining -= fill_qty
if remaining <= 0:
break
if remaining > 0:
return {
"is_tradeable": False,
"slippage_bps": float('inf'),
"available_depth": cumulative_qty,
"reason": f"深度不足：需要 {order_size} 股，仅 {cumulative_qty} 股"
}
avg_price = cumulative_cost / cumulative_qty
slippage_bps = (avg_price - best_ask) / best_ask * 10000
return {
"is_tradeable": slippage_bps <= max_slippage_bps,
"slippage_bps": round(slippage_bps, 2),
"available_depth": cumulative_qty,
"reason": None if slippage_bps <= max_slippage_bps else f"滑点超限 {slippage_bps:.1f} > {max_slippage_bps} bps"
}
# ============================================================
# TickDB API 封装
# ============================================================
class TickDBClient:
"""TickDB API 封装，含重连和限频处理"""
def __init__(self, api_key: str, config: LiquidityConfig):
self.api_key = api_key
self.base_url = "https://api.tickdb.ai/v1"
self.config = config
self.headers = {"X-API-Key": api_key}
self._last_request_time = 0
self._request_count = 0
def _rate_limit_wait(self):
"""简单限频：每秒不超过 10 次请求"""
current_time = time.time()
elapsed = current_time - self._last_request_time
if elapsed < 0.1 and self._request_count >= 10:
time.sleep(0.1 - elapsed + 0.01)
self._last_request_time = time.time()
self._request_count = min(self._request_count + 1, 10)
def get_historical_klines(self, symbol: str, interval: str = "1h", limit: int = 100) -> dict:
"""获取历史 K 线数据（用于计算成交量基准）"""
self._rate_limit_wait()
url = f"{self.base_url}/market/kline"
params = {"symbol": symbol, "interval": interval, "limit": limit}
try:
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=self.config.http_timeout
)
data = response.json()
if data.get("code") == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
logger.warning(f"限频触发，等待 {retry_after} 秒")
time.sleep(retry_after)
return self.get_historical_klines(symbol, interval, limit)
return data
except requests.exceptions.Timeout:
logger.error(f"K线请求超时: {symbol}")
raise
except Exception as e:
logger.error(f"K线请求异常: {e}")
raise
def get_latest_kline(self, symbol: str, interval: str = "1h") -> dict:
"""获取最新 K 线（用于实时成交量校验）"""
self._rate_limit_wait()
url = f"{self.base_url}/market/kline/latest"
params = {"symbol": symbol, "interval": interval}
try:
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=self.config.http_timeout
)
data = response.json()
return data
except Exception as e:
logger.error(f"最新K线请求异常: {e}")
raise
# ============================================================
# WebSocket 实时深度监控
# ============================================================
class DepthWebSocket:
"""订单簿深度 WebSocket 监控"""
def __init__(self, api_key: str, symbols: list[str], config: LiquidityConfig):
self.api_key = api_key
self.symbols = symbols
self.config = config
self.ws_url = "wss://stream.tickdb.ai/v1/ws"
self._running = False
self._depth_cache = {}
async def connect(self):
"""建立 WebSocket 连接"""
retry_count = 0
delay = self.config.ws_reconnect_base_delay
while retry_count < self.config.ws_max_retries:
try:
# WebSocket 鉴权通过 URL 参数传递
url = f"{self.ws_url}?api_key={self.api_key}"
async with websockets.connect(url) as ws:
logger.info("WebSocket 连接建立成功")
# 订阅 depth 频道
subscribe_msg = {
"cmd": "subscribe",
"params": {
"channels": [f"depth.{symbol}" for symbol in self.symbols]
}
}
await ws.send(json.dumps(subscribe_msg))
logger.info(f"已订阅频道: {subscribe_msg['params']['channels']}")
# 心跳保活
self._running = True
await self._heartbeat_loop(ws)
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"WebSocket 连接断开: {e.code} {e.reason}")
except Exception as e:
logger.error(f"WebSocket 异常: {e}")
# 指数退避 + 抖动重连
retry_count += 1
jitter = random.uniform(0, delay * 0.1)
wait_time = min(delay + jitter, self.config.ws_reconnect_max_delay)
logger.info(f"{wait_time:.1f} 秒后第 {retry_count} 次重连尝试")
await asyncio.sleep(wait_time)
delay = min(delay * 2, self.config.ws_reconnect_max_delay)
async def _heartbeat_loop(self, ws):
"""心跳保活循环"""
while self._running:
try:
await ws.send(json.dumps({"cmd": "ping"}))
await asyncio.sleep(30)
except Exception:
break
def get_cached_depth(self, symbol: str) -> Optional[dict]:
"""获取缓存的订单簿深度"""
return self._depth_cache.get(symbol)
async def process_message(self, msg: dict):
"""处理接收到的消息"""
channel = msg.get("channel", "")
if not channel.startswith("depth."):
return
symbol = channel.replace("depth.", "")
self._depth_cache[symbol] = msg.get("data", {})
# ============================================================
# 流动性信号过滤器
# ============================================================
class LiquiditySignalFilter:
"""三层流动性信号过滤器"""
def __init__(self, symbol: str, tickdb_client: TickDBClient, config: LiquidityConfig):
self.symbol = symbol
self.client = tickdb_client
self.config = config
self.historical_volumes = []
self._load_historical_data()
def _load_historical_data(self):
"""加载历史数据用于基准计算"""
try:
data = self.client.get_historical_klines(
self.symbol,
interval="1h",
limit=self.config.lookback_klines
)
if data.get("code") == 0:
klines = data.get("data", {}).get("klines", [])
# 过滤出正常交易时段的 K 线
self.historical_volumes = [
k.get("volume", 0)
for k in klines
if k.get("volume", 0) > 0
]
logger.info(f"已加载 {len(self.historical_volumes)} 条历史 K 线数据")
else:
logger.warning(f"历史数据加载失败: {data.get('message')}")
except Exception as e:
logger.error(f"历史数据加载异常: {e}")
def check_signal(self, depth: dict, current_kline_vol: int, order_size: int) -> dict:
"""
执行三层过滤检查
Returns:
{
"passed": bool,
"layers": {
"session": {"passed": bool, "session": str, "reason": str},
"volume": {"passed": bool, "ratio": float, "reason": str},
"depth": {"passed": bool, "slippage_bps": float, "reason": str}
},
"decision": str
}
"""
now = datetime.utcnow()
result = {
"passed": False,
"layers": {
"session": {"passed": False, "session": "", "reason": ""},
"volume": {"passed": False, "ratio": 0.0, "reason": ""},
"depth": {"passed": False, "slippage_bps": 0.0, "reason": ""}
},
"decision": ""
}
# ---- 第一层：交易时段判断 ----
session = get_trading_session(now)
result["layers"]["session"]["session"] = session.value
if session == TradingSession.REGULAR:
result["layers"]["session"]["passed"] = True
result["layers"]["session"]["reason"] = "正常交易时段"
elif session == TradingSession.PRE_MARKET:
if self.config.allow_pre_market:
result["layers"]["session"]["passed"] = True
result["layers"]["session"]["reason"] = "盘前时段（已配置允许）"
else:
result["layers"]["session"]["reason"] = "盘前时段（已配置拒绝）"
elif session == TradingSession.AFTER_HOURS:
if self.config.allow_after_hours:
result["layers"]["session"]["passed"] = True
result["layers"]["session"]["reason"] = "盘后时段（已配置允许）"
else:
result["layers"]["session"]["reason"] = "盘后时段（已配置拒绝）"
else:
result["layers"]["session"]["reason"] = "非交易时段"
# ---- 第二层：成交量阈值 ----
vol_ratio = calculate_volume_ratio(current_kline_vol, self.historical_volumes)
result["layers"]["volume"]["ratio"] = round(vol_ratio, 3)
if vol_ratio >= self.config.min_volume_ratio:
result["layers"]["volume"]["passed"] = True
result["layers"]["volume"]["reason"] = f"成交量比例 {vol_ratio:.2f} >= 阈值"
else:
result["layers"]["volume"]["reason"] = f"成交量比例 {vol_ratio:.2f} < 阈值 {self.config.min_volume_ratio}"
# ---- 第三层：订单簿深度 ----
bids = depth.get("bids", [])
asks = depth.get("asks", [])
depth_check = validate_depth_for_trade(
bids, asks, order_size, self.config.max_slippage_bps
)
result["layers"]["depth"]["slippage_bps"] = depth_check.get("slippage_bps", 0)
if depth_check["is_tradeable"]:
result["layers"]["depth"]["passed"] = True
result["layers"]["depth"]["reason"] = f"滑点 {depth_check['slippage_bps']:.1f} bps 可接受"
else:
result["layers"]["depth"]["reason"] = depth_check.get("reason", "深度校验失败")
# ---- 综合决策 ----
all_passed = all(
layer["passed"]
for layer in result["layers"].values()
)
result["passed"] = all_passed
if all_passed:
result["decision"] = "信号通过，执行交易"
else:
failed_layers = [
name for name, layer in result["layers"].items()
if not layer["passed"]
]
result["decision"] = f"信号拒绝：{', '.join(failed_layers)} 层未通过"
return result
# ============================================================
# 使用示例
# ============================================================
async def main():
"""演示：三只股票的流动性监控"""
# 配置
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
symbols = ["AAPL.US", "TSLA.US", "NVDA.US"]
config = LiquidityConfig(
allow_pre_market=False, # 趋势策略关闭盘前
allow_after_hours=False, # 趋势策略关闭盘后
min_volume_ratio=0.5, # 成交量需达到历史中位数的 50%
max_slippage_bps=30.0 # 滑点不超过 30 基点
)
# 初始化客户端
client = TickDBClient(api_key, config)
# 初始化过滤器
filters = {
symbol: LiquiditySignalFilter(symbol, client, config)
for symbol in symbols
}
# 初始化 WebSocket
ws = DepthWebSocket(api_key, symbols, config)
# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio
logger.info("启动流动性监控...")
# 示例：手动构造一个盘后订单簿场景进行测试
test_depth = {
"bids": [["150.00", "200"], ["149.95", "300"], ["149.90", "500"]],
"asks": [["150.15", "150"], ["150.20", "250"], ["150.30", "400"]]
}
test_volume = 5000 # 极低的成交量
for symbol, filter_obj in filters.items():
result = filter_obj.check_signal(test_depth, test_volume, order_size=1000)
print(f"\n{'='*50}")
print(f"标的: {symbol}")
print(f"{'='*50}")
print(f"综合决策: {result['decision']}")
print(f"\n过滤详情:")
for layer_name, layer_data in result["layers"].items():
status = "✅" if layer_data["passed"] else "❌"
print(f" {status} {layer_name}: {layer_data.get('reason') or layer_data.get('session', '')}")
if __name__ == "__main__":
asyncio.run(main())
四、策略适配：从理论到实战
4.1 策略类型与过滤强度矩阵
并非所有策略都需要同等的过滤强度。以下是不同策略类型的推荐配置：
| 策略类型 | 盘前交易 | 盘后交易 | 成交量阈值 | 滑点阈值 | 说明 |
|---|---|---|---|---|---|
| 趋势跟踪（止损/突破） | ❌ 关闭 | ❌ 关闭 | ≥ 0.5 | ≤ 30 bps | 宁错过不做错 |
| 日内均值回归 | ✅ 开启 | ✅ 开启 | ≥ 1.0 | ≤ 20 bps | 需要高流动性支撑 |
| 事件驱动（财报） | ✅ 开启 | ✅ 开启 | ≥ 0.3 | ≤ 50 bps | 事件催化可降低阈值 |
| 统计套利（配对） | ✅ 开启 | ⚠️ 审慎 | ≥ 0.8 | ≤ 25 bps | 配对关系在盘后可能失效 |
| 期权对冲 | ✅ 开启 | ✅ 开启 | ≥ 0.5 | ≤ 40 bps | 对冲需求优先于流动性 |
4.2 常见错误与避坑指南
错误一：只看时间窗口，不看成交量
# ❌ 错误做法
if 9 <= hour < 16:
execute_trade(signal)
正确做法：时间窗口是第一道门槛，但成交量和订单簿深度是必要条件。
错误二：固定滑点阈值，不区分波动率环境
在 VIX 超过 30 的高波动时段，同样的滑点阈值会导致大量拒绝。可以引入波动率调整因子：
def adjusted_slippage_threshold(
base_threshold: float,
current_vix: float,
normal_vix: float = 18.0
) -> float:
"""波动率调整后的滑点阈值"""
vix_ratio = current_vix / normal_vix
# 高波动环境下，放宽阈值（但不超过基础值的 2 倍）
return min(base_threshold * vix_ratio, base_threshold * 2.0)
错误三：回测中忽略盘前盘后数据
如果你的回测数据包含盘前盘后成交价，而实盘策略会过滤这些时段，会导致严重的过拟合。正确的回测做法：
- 仅使用正常交易时段的数据进行回测
- 或者在回测中实时模拟三层过滤逻辑
五、总结与行动建议
盘前盘后交易的流动性陷阱，本质上是**“数据密度不足导致的信号失真”。解决方案不是简单地“关掉盘前盘后”，而是建立一套分层过滤体系**，让每一笔交易决策都经过：
- 时间窗口过滤：判断当前是否处于可交易时段
- 成交量阈值过滤：确保信号有足够的市场参与者支撑
- 订单簿深度校验：确保订单能够以预期价格执行
三层过滤的权衡：过滤越严格，信号越少，但假突破率越低；过滤越宽松，信号越多，但被骗线的概率越高。最佳平衡点需要根据策略的盈亏结构和市场特征动态调整。
下一步行动
如果你想亲手实现本文的三层过滤框架：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 设置环境变量
TICKDB_API_KEY，复制本文代码即可运行
如果你需要验证盘前盘后的流动性特征：
使用 TickDB 的 /kline 接口拉取 1 小时 K 线数据，计算各时段的成交量中位数，代入 LiquiditySignalFilter 初始化你的基准数据。
如果你习惯用 AI 辅助开发：
在 AI 助手中搜索安装 tickdb-market-data SKILL，用自然语言描述需求，AI 可协助生成定制化的过滤配置。
风险提示：本文介绍的技术框架仅用于信息分享目的，不构成任何投资建议。历史流动性特征不代表未来表现，实际使用前请充分回测并评估风险。市场有风险，投资需谨慎。