凌晨三点，你的策略为何还在"假死"？
凌晨三点，你被一条告警推送惊醒——策略显示触发止损信号，但你打开交易终端时发现，订单根本没有被执行。
这不是策略逻辑的问题，而是信号从产生到抵达执行层之间，那丢失的 200 毫秒。
真实交易环境中，行情数据的获取往往不是最慢的一环。真正的瓶颈发生在数据抵达之后：它经过哪些链路才能变成可执行的信号？策略引擎如何确保不漏掉任何一个有效信号？当行情剧烈波动时，10 个信号在 50 毫秒内同时涌入，你的系统如何不崩溃？
本文从架构层面拆解从 TickDB 实时数据到策略信号输出的完整链路，重点解决三个核心工程问题：
- 数据来了之后如何分发，才能让多个策略同时处理？
- 策略回调如何设计，才能兼顾实时性和健壮性？
- 信号如何防抖，避免在市场剧烈波动时被"反复收割"？
一、为什么你拿到的数据"用不上"
在讨论架构之前，先澄清一个常见误区。
大多数开发者以为接入了实时行情 API 就万事大吉。但实际的数据链路是这样的：
TickDB WebSocket → 你的网络接收层 → 数据解析层 → 分发层 → 策略引擎 → 信号输出层 → 执行层
问题往往出现在分发层到策略引擎这一段。典型症状包括：
- 单线程阻塞：行情解析和策略计算跑在同一个线程，策略卡顿时行情数据在缓冲区堆积，最终收到的是 3 秒前的"历史数据"
- 策略耦合：新增一个策略需要修改行情接入代码，每次改动都是风险
- 信号风暴：非农数据发布时，200ms 内涌入 500 条行情更新，策略被反复触发，信号在队列里排队等待，最终执行时行情早已变样
- 内存泄漏：长周期运行时回调函数未正确释放，内存持续增长，最终 OOM
要解决这些问题，你需要一套异步消息分发 + 策略回调注册的架构。下面我们直接给出可运行的工程方案。
二、架构总览：数据到信号的完整链路
在展开代码之前，先建立全局视图。整个系统的核心模块如下：
┌─────────────────────────────────────────────────────────────────┐
│ TickDB WebSocket │
│ tickdb.ai (wss://api.tickdb.ai/v1) │
└────────────────────────────┬────────────────────────────────────┘
│ 订阅 depth / trades 等频道
▼
┌─────────────────────────────────────────────────────────────────┐
│ DataReceiver（接收层） │
│ · 心跳保活 + 指数退避重连 │
│ · JSON 解析 + 合法性校验 │
│ · 消息分发到策略引擎（asyncio.Queue） │
└────────────────────────────┬────────────────────────────────────┘
│ asyncio.Queue.put()
▼
┌─────────────────────────────────────────────────────────────────┐
│ StrategyEngine（策略引擎） │
│ · 策略注册表（dict） │
│ · 策略回调调度（asyncio.TaskGroup / ThreadPoolExecutor） │
│ · 信号聚合与防抖 │
└────────────────────────────┬────────────────────────────────────┘
│ 策略信号
▼
┌─────────────────────────────────────────────────────────────────┐
│ SignalRouter（信号路由） │
│ · 信号验证（仓位、权限、频率） │
│ · 防抖逻辑（去重 + 时间窗口合并） │
│ · 下发到执行层 │
└─────────────────────────────────────────────────────────────────┘
下面逐层实现。
三、接收层：TickDB WebSocket 的健壮接入
这是整个链路的起点。很多人写的 WebSocket 接入代码只能跑在 Demo 里，一上生产就出问题——断线不知道、频率超了不知道、重连后数据对不上不知道。
3.1 标准接入代码（含完整工程要素）
import os
import asyncio
import json
import time
import random
from dataclasses import dataclass, field
from typing import Callable, Awaitable, Optional
import aiohttp
@dataclass
class TickDBConfig:
"""TickDB 连接配置"""
api_key: str
symbols: list[str]
channels: list[str] = field(default_factory=lambda: ["depth", "trades"])
base_url: str = "wss://api.tickdb.ai/v1/ws"
ping_interval: int = 20
max_reconnect_delay: float = 60.0
initial_reconnect_delay: float = 1.0
class TickDBReceiver:
"""
TickDB WebSocket 数据接收器。
职责：
1. 维护 WebSocket 连接（心跳保活）
2. 自动重连（指数退避 + 抖动）
3. 解析消息并分发到策略引擎
⚠️ 注意：生产环境中建议将数据接收与策略计算分离在不同线程/进程，
避免解析逻辑影响策略执行延迟。
"""
def __init__(self, config: TickDBConfig):
self.config = config
self._queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
self._running = False
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._reconnect_attempts = 0
async def start(self):
"""启动接收器，进入数据接收循环"""
self._running = True
while self._running:
try:
await self._connect_and_subscribe()
except (aiohttp.ClientError, asyncio.CancelledError) as e:
if not self._running:
break
delay = self._calculate_reconnect_delay()
print(f"[TickDBReceiver] 连接断开: {e}，{delay:.1f}秒后重连")
await asyncio.sleep(delay)
async def _connect_and_subscribe(self):
"""建立连接并订阅指定品种和频道"""
self._session = aiohttp.ClientSession()
params = {
"api_key": self.config.api_key,
"symbol": ",".join(self.config.symbols),
"channel": ",".join(self.config.channels),
}
url = f"{self.config.base_url}?{aiohttp.BasicAuth('', '')}"
# ⚠️ 实际鉴权通过 URL 参数传递 api_key
async with self._session.ws_connect(
self.config.base_url,
params=params,
receive_timeout=self.config.ping_interval + 5,
) as ws:
self._ws = ws
self._reconnect_attempts = 0
print(f"[TickDBReceiver] 已连接，订阅: {self.config.symbols} @ {self.config.channels}")
# 启动心跳任务
heartbeat_task = asyncio.create_task(self._heartbeat_loop())
receive_task = asyncio.create_task(self._receive_loop())
# 任意一个任务结束都意味着连接异常
done, pending = await asyncio.wait(
[heartbeat_task, receive_task],
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
for task in done:
if task.exception() and not isinstance(task.exception(), asyncio.CancelledError):
raise task.exception()
async def _heartbeat_loop(self):
"""心跳保活循环，防止 WebSocket 中间件断连"""
while True:
await asyncio.sleep(self.config.ping_interval)
if self._ws and not self._ws.closed:
try:
await self._ws.ping()
except Exception as e:
print(f"[TickDBReceiver] 心跳失败: {e}")
break
async def _receive_loop(self):
"""消息接收循环，处理限频和分发"""
while True:
msg = await self._ws.receive()
if msg.type == aiohttp.WSMsgType.PONG:
continue
if msg.type == aiohttp.WSMsgType.ERROR:
raise ConnectionError(f"WebSocket 错误: {msg.data}")
if msg.type == aiohttp.WSMsgType.CLOSE:
raise ConnectionError("服务器主动关闭连接")
if msg.type == aiohttp.WSMsgType.TEXT:
try:
data = json.loads(msg.data)
# 处理 TickDB 限频响应 (code: 3001)
await self._handle_message(data)
except json.JSONDecodeError as e:
print(f"[TickDBReceiver] JSON 解析失败: {e}")
except asyncio.QueueFull:
# ⚠️ 队列满说明策略引擎处理速度跟不上，
# 生产环境应触发告警而非无限阻塞
print("[TickDBReceiver] ⚠️ 队列已满，数据丢弃，触发告警！")
async def _handle_message(self, data: dict):
"""
统一消息处理入口。
⚠️ 重要：TickDB 在超限时返回 code:3001，
并在响应头中携带 Retry-After。处理不当会导致永久断连。
"""
if data.get("code") == 3001:
retry_after = int(data.get("retry_after", 5))
print(f"[TickDBReceiver] 请求超限，暂停 {retry_after}s")
await asyncio.sleep(retry_after)
return
if data.get("code", 0) != 0 and data.get("code") != 0:
print(f"[TickDBReceiver] API 错误: code={data.get('code')} msg={data.get('message')}")
return
# 将有效数据放入队列，供策略引擎消费
await self._queue.put(data)
def _calculate_reconnect_delay(self) -> float:
"""指数退避 + 抖动，避免惊群效应"""
base = self.config.initial_reconnect_delay
max_delay = self.config.max_reconnect_delay
exp_delay = min(base * (2 ** self._reconnect_attempts), max_delay)
jitter = random.uniform(0, exp_delay * 0.1)
self._reconnect_attempts += 1
return exp_delay + jitter
async def stop(self):
"""优雅停止接收器"""
self._running = False
if self._session:
await self._session.close()
async def get_queue(self) -> asyncio.Queue:
"""供策略引擎获取数据队列的引用"""
return self._queue
3.2 关键工程细节解析
重连策略：self._reconnect_attempts 每次断连递增，重连延迟从 1 秒指数增长到最大 60 秒，配合 10% 的随机抖动——这样即使 100 个客户端同时断连，也不会在同一秒集体重连造成 TickDB 服务器过载。
心跳机制：WebSocket 是长连接，中间件（如 Nginx）默认 60 秒无活动可能断连。TickDB 要求客户端每 20 秒发送一次 ping，本代码通过 ws.ping() 主动保活。
限频处理：TickDB 的 3001 错误意味着当前请求频率超出配额。错误处理中读取 retry_after 并等待，而不是立即重试——这是很多人踩过的坑：超限后立刻重试会被拉黑更长时间。
队列积压告警：Queue 设置 maxsize=10000，但当策略引擎处理速度跟不上行情速度时，队列会满。代码中没有用 put_nowait() 无限丢弃，而是打印告警——这是生产级代码的标志，因为无声丢弃数据是最危险的错误。
四、策略引擎：回调注册与并发调度
接收层把数据送进队列，接下来由策略引擎负责消费。
策略引擎的核心设计原则是：数据接入与策略逻辑完全解耦。新增策略不需要改一行行情接入代码，只需要注册一个回调函数。
4.1 引擎核心实现
import asyncio
from typing import Callable, Awaitable, Any
from dataclasses import dataclass
from datetime import datetime
import threading
@dataclass
class MarketData:
"""标准化的市场数据对象"""
symbol: str
channel: str # "depth" | "trades" | "ticker"
timestamp: float # Unix 时间戳（毫秒）
data: dict # 原始数据payload
received_at: float # 本地接收时间戳
@dataclass
class StrategySignal:
"""策略信号结构"""
strategy_name: str
symbol: str
action: str # "BUY" | "SELL" | "CLOSE" | "NO_ACTION"
quantity: float = 0.0
price: float = 0.0
confidence: float = 0.0 # 0.0~1.0，置信度
reason: str = "" # 信号来源描述
timestamp: float = 0.0
def __post_init__(self):
if self.timestamp == 0.0:
self.timestamp = time.time()
# 策略回调类型定义
StrategyCallback = Callable[[MarketData], Awaitable[Optional[StrategySignal]]]
class StrategyEngine:
"""
策略引擎：负责从接收队列消费数据，并分发给已注册的策略回调。
设计原则：
1. 策略与数据源解耦——新增/修改策略不影响行情接入
2. 每个策略独立运行——一个策略崩溃不影响其他策略
3. 策略执行与数据接收并行——不因策略计算慢而丢失行情
⚠️ 生产建议：
- CPU 密集型策略（复杂数学计算）建议放到 ThreadPoolExecutor
- I/O 密集型策略（网络请求、外部 API）建议用 asyncio
- 极高频策略建议用单独的进程，与主数据流隔离
"""
def __init__(self, queue: asyncio.Queue, max_workers: int = 4):
self._queue = queue
self._strategies: dict[str, StrategyCallback] = {}
self._running = False
self._executor = ThreadPoolExecutor(max_workers=max_workers)
# 信号输出回调
self._signal_handlers: list[Callable[[StrategySignal], Any]] = []
# 运行时统计
self._stats = {
"messages_processed": 0,
"signals_emitted": 0,
"last_processed_at": None,
}
def register(self, name: str, callback: StrategyCallback):
"""注册策略回调，name 唯一标识策略"""
if name in self._strategies:
raise ValueError(f"策略 '{name}' 已存在，请使用唯一名称")
self._strategies[name] = callback
print(f"[StrategyEngine] 注册策略: {name}")
def on_signal(self, handler: Callable[[StrategySignal], Any]):
"""注册信号处理器（如告警、下单）"""
self._signal_handlers.append(handler)
async def start(self):
"""启动引擎，进入消费循环"""
self._running = True
print(f"[StrategyEngine] 启动，注册策略数: {len(self._strategies)}")
while self._running:
try:
# 从队列取数据，设置超时以便定期打印统计
data = await asyncio.wait_for(self._queue.get(), timeout=5.0)
market_data = self._parse_market_data(data)
# ⚠️ 并发执行所有策略，但策略间相互隔离
# asyncio.gather 保证所有策略都能处理同一条数据
signals = await asyncio.gather(
*[self._run_strategy(name, cb, market_data)
for name, cb in self._strategies.items()],
return_exceptions=True,
)
# 收集有效信号并下发
for sig in signals:
if isinstance(sig, StrategySignal):
await self._emit_signal(sig)
self._stats["messages_processed"] += 1
self._stats["last_processed_at"] = datetime.now().isoformat()
except asyncio.TimeoutError:
# 定期打印状态（生产环境改为上报监控系统）
print(f"[StrategyEngine] 心跳，队列剩余: {self._queue.qsize()}")
except Exception as e:
print(f"[StrategyEngine] 消费循环异常: {e}")
def _parse_market_data(self, raw: dict) -> MarketData:
"""将 TickDB 原始消息解析为标准 MarketData 对象"""
return MarketData(
symbol=raw.get("symbol", ""),
channel=raw.get("channel", ""),
timestamp=raw.get("ts", time.time() * 1000),
data=raw.get("data", {}),
received_at=time.time() * 1000,
)
async def _run_strategy(
self,
name: str,
callback: StrategyCallback,
market_data: MarketData,
) -> Optional[StrategySignal]:
"""执行单个策略，捕获异常防止策略间相互影响"""
try:
signal = await callback(market_data)
if signal:
signal.strategy_name = name
return signal
except Exception as e:
# ⚠️ 策略异常不应影响引擎和其他策略
print(f"[StrategyEngine] ⚠️ 策略 {name} 执行异常: {e}")
return None
async def _emit_signal(self, signal: StrategySignal):
"""将信号分发给所有处理器"""
for handler in self._signal_handlers:
try:
result = handler(signal)
if asyncio.iscoroutine(result):
await result
except Exception as e:
print(f"[StrategyEngine] ⚠️ 信号处理器异常: {e}")
self._stats["signals_emitted"] += 1
async def stop(self):
"""停止引擎"""
self._running = False
self._executor.shutdown(wait=False)
def get_stats(self) -> dict:
return self._stats.copy()
4.2 策略注册示例：基于订单簿失衡的买卖信号
import time
from typing import Optional
# ─── 示例策略：订单簿失衡检测 ───────────────────────────────────────────
class OrderBookImbalanceStrategy:
"""
基于买卖盘量失衡的均值回归策略。
核心逻辑：
- 当 买盘压力比 > 2.5 且价格低于均线时 → 买入信号
- 当 卖盘压力比 > 2.5 且价格高于均线时 → 卖出信号
- 买卖压力比 = Σ(前3档买盘量) / Σ(前3档卖盘量)
⚠️ 此策略仅供演示用，实盘使用前请务必充分回测并设置风控阈值。
"""
def __init__(self, symbol: str, threshold: float = 2.5):
self.symbol = symbol
self.threshold = threshold
self.price_history: list[float] = []
async def on_market_data(self, market_data: MarketData) -> Optional[StrategySignal]:
"""
策略回调函数——TickDB 每推送一条数据，这里就会被调用一次。
⚠️ 关键性能点：
此函数必须高效执行（< 5ms 为佳）。
任何耗时操作（文件 I/O、数据库查询、外部 HTTP）都应放到
独立的 async 任务中，不阻塞策略执行。
"""
if market_data.symbol != self.symbol:
return None
if market_data.channel != "depth":
return None
depth_data = market_data.data
bids = depth_data.get("b", []) # 买盘 [[price, qty], ...]
asks = depth_data.get("a", []) # 卖盘 [[price, qty], ...]
if not bids or not asks:
return None
# 计算前3档买卖压力比
bid_volume = sum(float(b[1]) for b in bids[:3])
ask_volume = sum(float(a[1]) for a in asks[:3])
if ask_volume == 0:
return None
pressure_ratio = bid_volume / ask_volume
# 计算当前中间价
mid_price = (float(bids[0][0]) + float(asks[0][0])) / 2
self.price_history.append(mid_price)
# 维持最近30个价格用于均线计算
if len(self.price_history) > 30:
self.price_history.pop(0)
if len(self.price_history) < 10:
return None
ma = sum(self.price_history) / len(self.price_history)
# 信号生成逻辑
if pressure_ratio > self.threshold and mid_price < ma:
return StrategySignal(
strategy_name="OrderBookImbalance",
symbol=self.symbol,
action="BUY",
quantity=100,
price=mid_price,
confidence=min(pressure_ratio / 4.0, 1.0),
reason=f"压力比={pressure_ratio:.2f}, 中间价={mid_price:.2f}, MA={ma:.2f}",
)
elif pressure_ratio < (1.0 / self.threshold) and mid_price > ma:
return StrategySignal(
strategy_name="OrderBookImbalance",
symbol=self.symbol,
action="SELL",
quantity=100,
price=mid_price,
confidence=min((1 / pressure_ratio) / 4.0, 1.0),
reason=f"压力比={pressure_ratio:.2f}, 中间价={mid_price:.2f}, MA={ma:.2f}",
)
return None
# ─── 注册到引擎 ─────────────────────────────────────────────────────────
async def main():
# 初始化接收器
config = TickDBConfig(
api_key=os.environ.get("TICKDB_API_KEY", ""),
symbols=["BTC.USDT"],
channels=["depth"],
)
receiver = TickDBReceiver(config)
# 初始化引擎，从接收器获取数据队列
queue = await receiver.get_queue()
engine = StrategyEngine(queue, max_workers=4)
# 注册策略
strategy = OrderBookImbalanceStrategy(symbol="BTC.USDT", threshold=2.5)
engine.register("OrderBookImbalance", strategy.on_market_data)
# 注册信号处理器（示例：打印信号）
def print_signal(sig: StrategySignal):
emoji = "📈" if sig.action == "BUY" else "📉"
print(f"{emoji} [{sig.strategy_name}] {sig.action} {sig.quantity} @ {sig.price:.2f} "
f"(置信度: {sig.confidence:.0%}, {sig.reason})")
engine.on_signal(print_signal)
# 启动
await asyncio.gather(
receiver.start(),
engine.start(),
)
if __name__ == "__main__":
asyncio.run(main())
五、信号防抖：让策略在噪音中保持冷静
到这里，系统已经能跑了。但还有一个致命问题没解决：信号风暴。
5.1 信号风暴的成因
非农数据发布前后 500ms 内，行情数据量可能是平时的 50 倍。如果策略在每个 tick 都计算一次信号，你会在 1 秒内收到 30 个 "BUY" 信号——这对于日内短线策略是灾难性的。
核心问题：同一个策略在短时间内对同一标的产生多个相同方向的信号，只应执行第一个（或最后一个），其余全部丢弃。
5.2 防抖路由实现
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
import threading
@dataclass
class DebounceConfig:
"""防抖配置"""
window_ms: int = 2000 # 时间窗口（毫秒）
min_interval_ms: int = 500 # 同一方向信号最小间隔
cooldown_ms: int = 3000 # 反向信号冷却期（防止反复开平）
@dataclass
class SignalState:
"""追踪每个标的-策略组合的信号状态"""
last_signal_time: float = 0.0
last_signal_action: str = "NO_ACTION"
last_signal_price: float = 0.0
cooldown_until: float = 0.0
class SignalRouter:
"""
信号路由器：在下发执行前对信号进行防抖和验证。
防抖策略：
1. 时间窗口去重——窗口期内同方向信号只保留第一个
2. 冷却期——反向信号需等待冷却期结束
3. 最小间隔——防止高频刷信号
⚠️ 重要：此模块直接控制是否下单，是风控的第一道闸门。
生产环境务必配合人工确认或资金上限。
"""
def __init__(self, config: DebounceConfig = None):
self.config = config or DebounceConfig()
# 状态存储：{(strategy_name, symbol): SignalState}
self._states: dict[tuple[str, str], SignalState] = defaultdict(SignalState)
self._lock = threading.Lock()
def filter(self, signal: StrategySignal) -> Optional[StrategySignal]:
"""
过滤信号。返回 None 表示信号被丢弃，返回原信号表示放行。
⚠️ 线程安全：策略引擎中多个协程可能同时产生信号，
需要加锁保护共享状态。
"""
if signal.action == "NO_ACTION":
return None
key = (signal.strategy_name, signal.symbol)
now = signal.timestamp * 1000 # 毫秒
with self._lock:
state = self._states[key]
now_sec = now / 1000
# 检查冷却期
if state.cooldown_until > now_sec:
print(f"[SignalRouter] ⚠️ 冷却期中，丢弃 {signal.action} "
f"(剩余 {state.cooldown_until - now_sec:.1f}s)")
return None
# 检查最小间隔
interval = now - (state.last_signal_time * 1000)
if interval < self.config.min_interval_ms:
print(f"[SignalRouter] 间隔过短 ({interval:.0f}ms < {self.config.min_interval_ms}ms)，丢弃")
return None
# 检查时间窗口去重（同方向信号，窗口期内只取第一个）
if (state.last_signal_action == signal.action and
(now - state.last_signal_time * 1000) < self.config.window_ms and
state.last_signal_time > 0):
print(f"[SignalRouter] 窗口期内已有相同方向信号，丢弃")
return None
# 通过所有检查，更新状态
state.last_signal_time = now / 1000
state.last_signal_action = signal.action
state.last_signal_price = signal.price
# 设置反向信号冷却期
opposite_actions = {"BUY": "SELL", "SELL": "BUY"}
if signal.action in opposite_actions:
state.cooldown_until = now_sec + self.config.cooldown_ms / 1000
print(f"[SignalRouter] ✅ 信号放行: {signal.action} {signal.symbol} "
f"@ {signal.price:.2f} (置信度: {signal.confidence:.0%})")
return signal
def get_state(self, strategy_name: str, symbol: str) -> Optional[SignalState]:
key = (strategy_name, symbol)
with self._lock:
return self._states.get(key)
def reset_state(self, strategy_name: str, symbol: str):
"""手动重置某个策略-标的的信号状态（如手动平仓后）"""
key = (strategy_name, symbol)
with self._lock:
if key in self._states:
del self._states[key]
5.3 在引擎中集成防抖
只需将 SignalRouter 作为第一个信号处理器：
# 在 main() 中
router = SignalRouter(DebounceConfig(window_ms=2000, min_interval_ms=500, cooldown_ms=3000))
# 旧的直接打印改为通过路由过滤
engine.on_signal(lambda sig: router.filter(sig))
# 再接一个处理器处理通过路由的信号
def execute_trade(sig: StrategySignal):
if sig is not None:
print(f"🔔 执行下单: {sig.action} {sig.quantity} 手 @ {sig.price}")
engine.on_signal(execute_trade)
六、端到端整合：完整运行示例
将上述所有模块组合起来，加上配置和启动入口：
async def main():
"""完整系统启动流程"""
# 1. 配置
api_key = os.environ.get("TICKDB_API_KEY", "")
if not api_key:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
symbols = ["BTC.USDT", "ETH.USDT"]
channels = ["depth"]
# 2. 初始化组件
config = TickDBConfig(
api_key=api_key,
symbols=symbols,
channels=channels,
)
receiver = TickDBReceiver(config)
queue = await receiver.get_queue()
engine = StrategyEngine(queue, max_workers=4)
router = SignalRouter(DebounceConfig(window_ms=2000, min_interval_ms=500))
# 3. 注册策略
for symbol in symbols:
strategy = OrderBookImbalanceStrategy(symbol=symbol, threshold=2.5)
engine.register(f"OBImbalance-{symbol}", strategy.on_market_data)
# 4. 设置信号链路：策略 → 防抖路由 → 执行
def log_signal(sig: StrategySignal):
filtered = router.filter(sig)
if filtered:
print(f"🟢 执行: {filtered.action} {filtered.quantity} {filtered.symbol}")
engine.on_signal(log_signal)
# 5. 启动
print("[系统] TickDB 实时监控启动")
try:
await asyncio.gather(
receiver.start(),
engine.start(),
)
except KeyboardInterrupt:
print("[系统] 收到中断信号，优雅退出")
finally:
await receiver.stop()
await engine.stop()
if __name__ == "__main__":
asyncio.run(main())
七、性能调优与常见问题
7.1 分层调优指南
| 性能瓶颈 | 表现 | 解决方案 |
|---|---|---|
| 队列持续积压 | queue.qsize() 持续增长 |
增加策略引擎 worker 数量；或分离数据接收进程 |
| 策略回调耗时过长 | 信号延迟 > 100ms | 将计算密集逻辑移到 ThreadPoolExecutor；或异步化外部调用 |
| 信号风暴穿透防抖 | 队列满后数据丢弃 | 增加 maxsize；或开启背压机制通知 TickDB 降频 |
| 内存持续增长 | _stats 或 price_history 无上限 |
设置滑动窗口上限；定期写入外部存储 |
| 高可用要求 | 单进程宕机全链路中断 | 主备切换；或多实例 + 消息队列（Kafka/RabbitMQ） |
7.2 毫秒级响应的工程要点
要真正实现毫秒级响应，以下几点必须做到：
数据链路零拷贝：TickDB 的 JSON 数据进入队列后，策略回调中的 market_data.data 字段应尽量避免深拷贝。直接引用共享内存中的数据对象，减少 GC 压力。
避免同步阻塞：策略回调中任何 time.sleep()、同步文件 I/O、HTTP 请求都会阻塞协程。全部用 await asyncio.sleep() 和异步 HTTP 客户端（aiohttp）。
分离数据平面与控制平面：心跳检测、重连调度、统计上报等控制类任务，不要和行情处理抢同一个事件循环。建议用 asyncio.create_task(..., name="heartbeat") 并设置独立任务组。
预热热路径：系统启动时提前创建 aiohttp.ClientSession，不要在接收循环中按需创建（TCP 连接建立本身就有几十毫秒延迟）。
八、部署方案对比
| 维度 | 个人开发 | 团队协作 | 机构级 |
|---|---|---|---|
| 进程模型 | 单进程 asyncio | 多进程 + IPC | 分布式 + 消息队列 |
| 数据接收 | 1 个 Receiver | 1 主 + 1 热备 | N 主 + N 热备 + Kafka |
| 策略隔离 | asyncio 协程 | 多进程独立进程 | K8s Pod 独立部署 |
| 状态存储 | 内存 | Redis | 时序数据库 (InfluxDB) |
| 监控 | print 日志 | 结构化日志 + 飞书告警 | Prometheus + Grafana + 告警 |
九、结语：信号质量比信号数量更重要
回到开头的问题：凌晨三点止损信号没有执行，是因为信号死在链路里。
可能是队列积压导致信号排队等待，最终执行时价格早已穿破止损位；也可能是 WebSocket 断连，接收层在重连的 3 秒内完全"失聪"，错过了关键的波动窗口。
本文构建的链路，其核心设计原则就两个字：隔离。
- 数据接收与策略计算隔离——策略卡顿不丢行情
- 策略之间相互隔离——一个崩溃不影响全局
- 信号产生与信号执行隔离——防抖路由把住风控闸门
做到了这三点，你的策略引擎才能真正从"能收到数据"进化到"能稳定地产生高质量信号"。
下一步行动
如果你想亲手跑通这套系统：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key，设置环境变量
TICKDB_API_KEY - 将本文代码复制到本地 Python 环境（Python ≥ 3.10），安装依赖：
pip install aiohttp - 运行
main()函数，观察订单簿失衡信号的实时输出
如果你关注不同资产市场的深度数据差异：TickDB 的 depth 频道在不同市场上支持的档位数不同（美股 1 档、港股和数字货币最高 10 档），策略设计时需要根据市场特性调整压力比计算窗口。
如果你在机构环境中，需要高可用部署：联系 [email protected] 获取 TickDB 的专属数据通道和 SLA 保障方案。
本文不构成任何投资建议。策略回测结果不代表未来收益。实盘交易存在滑点、流动性枯竭等风险，请在充分测试后谨慎操作。