4、数据采集智能体部署：实时行情数据源接入
数据采集，说白了就是交易系统的「眼睛」。
眼睛要是瞎了，后面策略再牛也没用。我个人习惯把数据采集智能体比作「水管工」——你得把交易所的水（行情数据）干净、稳定地引到自家水池里。今天咱们就聊聊这个水管怎么铺。
4.1 实时行情数据源接入
接入行情，主流就两种方式：WebSocket 和 REST。我建议你优先用 WebSocket，为什么？
你想想看，REST 是「你问一句，它答一句」。高频交易里，每秒可能来几十笔成交，你轮询得越快，被交易所封 IP 的风险越大。WebSocket 是「长连接」，交易所主动推给你，延迟低、效率高。
核心原则：WebSocket 做实时推送，REST 做补数据和校验。
我在项目中遇到过一个问题：某次用 REST 轮询币安深度数据，每秒请求一次，结果跑了半小时就被限流了。后来改成 WebSocket，稳如老狗。
代码示例（Python + websockets 库）：
import asyncio
import websockets
import json
async def subscribe_depth(symbol="btcusdt"):
uri = "wss://stream.binance.com:9443/ws"
async with websockets.connect(uri) as ws:
# 订阅深度数据
sub_msg = {
"method": "SUBSCRIBE",
"params": [f"{symbol}@depth20@100ms"],
"id": 1
}
await ws.send(json.dumps(sub_msg))
while True:
data = await ws.recv()
# 这里交给清洗模块处理
yield data
# 使用示例
async def main():
async for msg in subscribe_depth():
print(msg[:100]) # 打印前100字符看看小技巧：订阅时尽量用「合并深度」而不是原始深度。原始深度每次推送几百条，合并后只推送变化的部分，带宽省一大半。
4.2 数据清洗与标准化
行情数据进了门，不能直接用。为什么？
不同交易所的字段名不一样。币安叫「bids」，OKX 叫「bids」但格式不同，火币可能叫「tick.bids」。你写策略时，总不能每个交易所写一套解析逻辑吧？
我习惯定义一个统一的数据模型：
from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime
@dataclass
class OrderBookSnapshot:
exchange: str
symbol: str
timestamp: datetime
bids: List[Tuple[float, float]] # (价格, 数量)
asks: List[Tuple[float, float]]
@classmethod
def from_binance(cls, raw: dict):
return cls(
exchange="binance",
symbol=raw["s"].lower(),
timestamp=datetime.fromtimestamp(raw["E"]/1000),
bids=[(float(p), float(q)) for p, q in raw["b"]],
asks=[(float(p), float(q)) for p, q in raw["a"]]
)清洗这一步，我踩过最大的坑是什么？
曾经有一次，币安推送的深度数据里，某个价格字段变成了字符串 "null"。我的解析器直接崩溃，整个交易系统停了 3 分钟。嗯，从那以后我加了一层「类型校验」：
def safe_float(value):
"""安全转换浮点数，异常数据直接丢弃"""
try:
return float(value)
except (TypeError, ValueError):
return None # 或者返回上一个有效值注意：清洗时不要「静默丢弃」异常数据。一定要打日志 + 告警。否则你永远不知道交易所那边出了什么幺蛾子。
4.3 数据存储策略
清洗好的数据存哪？我推荐两套方案：
| 数据库 | 适用场景 | 我的经验 |
|---|---|---|
| InfluxDB | 高频 Tick 数据、实时监控 | 写入快，查询快，但复杂聚合不行 |
| ClickHouse | 历史回测、批量分析 | 压缩比高，分析能力强，但写入延迟稍高 |
我个人习惯是「双写」：实时数据进 InfluxDB 给策略用，同时异步写入 ClickHouse 做归档。你想想看，策略跑的时候需要毫秒级查询，回测的时候需要 TB 级扫描，一个库搞不定。
InfluxDB 写入示例：
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
client = InfluxDBClient(url="http://localhost:8086", token="my-token")
write_api = client.write_api(write_options=SYNCHRONOUS)
def write_orderbook(snapshot: OrderBookSnapshot):
point = Point("orderbook") \
.tag("exchange", snapshot.exchange) \
.tag("symbol", snapshot.symbol) \
.field("bid_price_0", snapshot.bids[0][0]) \
.field("bid_qty_0", snapshot.bids[0][1]) \
.field("ask_price_0", snapshot.asks[0][0]) \
.field("ask_qty_0", snapshot.asks[0][1]) \
.time(snapshot.timestamp)
write_api.write(bucket="market_data", record=point)避坑指南：InfluxDB 的 tag 不要放太多唯一值。我曾经把「symbol」当 field 存，结果查询慢到怀疑人生。tag 是索引，field 是值，别搞反了。
4.4 异常处理与重连机制
网络断连是常态，不是异常。
我见过最离谱的事：某团队 WebSocket 断了之后，程序直接抛异常退出，第二天才发现数据停了 8 小时。嗯，这种事故在量化圈其实不少见。
重连机制的核心逻辑：
- 指数退避：第一次断连等 1 秒，第二次 2 秒，第三次 4 秒... 最大 60 秒
- 心跳检测：每 30 秒发一次 ping，收不到 pong 就主动重连
- 数据连续性校验：重连后检查是否有数据空洞，用 REST 补
代码实现：
import asyncio
import random
class WebSocketManager:
def __init__(self, url, max_retries=10):
self.url = url
self.max_retries = max_retries
self.retry_count = 0
async def connect_with_retry(self):
while self.retry_count < self.max_retries:
try:
async with websockets.connect(self.url) as ws:
self.retry_count = 0 # 连接成功，重置计数
await self.handle_messages(ws)
except Exception as e:
self.retry_count += 1
wait_time = min(2 ** self.retry_count, 60)
# 加一点随机抖动，防止多个客户端同时重连
wait_time += random.uniform(0, 1)
print(f"连接断开，{wait_time}秒后重试...")
await asyncio.sleep(wait_time)
async def handle_messages(self, ws):
async for msg in ws:
# 处理消息
pass重要：重连后一定要做「数据对齐」。比如深度数据，重连后先拉一次全量快照（REST），再继续接收增量推送。否则你的订单簿会越偏越远。
4.5 整体架构图
下面这张图，是我自己项目里用的数据采集层架构。你看一眼就明白了：
这张图里，最关键的其实是中间那层「WebSocket 连接管理器」。我见过太多人把精力花在存储优化上，结果连接层一断全白搭。记住：连接稳定性 > 存储性能。
总结一下我的经验：
- WebSocket 为主，REST 为辅
- 清洗时一定要做类型校验，别信交易所的数据
- InfluxDB + ClickHouse 双写，各取所长
- 重连用指数退避 + 随机抖动，别把交易所打崩了
数据采集这块，说白了就是「稳」字当头。你把这层做好了，后面的策略、风控、执行才有底气。下一层咱们聊聊策略智能体怎么部署，不过那是后话了。