4、数据源接入：券商API、量化平台（聚宽/米筐）、WebSocket实时行情
做ETF套利，说白了就是跟时间赛跑。
你策略再牛，模型再准，数据源跟不上，一切都是白搭。我见过太多人把精力花在优化参数上，结果行情接口一堵，几秒钟的延迟就把利润全吞了。嗯，这里咱们得认真聊聊数据源接入这件事。
4.1 数据源的三种主流姿势
我个人习惯把数据源分成三类，每一类都有它的脾气：
| 类型 | 延迟 | 成本 | 适用场景 |
|---|---|---|---|
| 券商API | 低（10-50ms） | 高（需开户+流量费） | 实盘交易、高频套利 |
| 量化平台（聚宽/米筐） | 中（100-500ms） | 低（免费/会员） | 策略回测、模拟交易 |
| WebSocket实时行情 | 极低（1-10ms） | 中（交易所授权） | 专业套利、做市商 |
你想想看，套利机会往往就存在那么几毫秒。我曾经在实盘中遇到过，就因为券商API的推送慢了20ms，眼睁睁看着价差从0.3%缩到0.05%。所以选对数据源，比选对策略还重要。
4.2 券商API接入实战
券商API是最直接的路径。国内主流券商基本都提供了Python接口，比如华泰的xtquant、中泰的xtrade、国信的qmt等。
我建议你优先选支持「极速交易」的券商。什么叫极速？就是行情推送频率至少是500ms一次，最好能到100ms。普通散户用的免费版，往往是3秒推送一次，做套利根本不够用。
核心要点：券商API的行情数据是「订阅制」的。你需要先订阅代码，然后等回调函数推送数据。千万别用轮询方式去拉数据，延迟会高到你怀疑人生。
下面是一个典型的券商API接入代码框架：
# 伪代码示例 - 券商API接入
from broker_api import BrokerClient
client = BrokerClient(account='your_account', password='your_pwd')
client.connect()
# 订阅ETF行情
def on_quote(quote):
# 这里拿到实时行情
etf_price = quote['price']
nav_price = quote['nav'] # 部分券商支持实时净值
spread = (etf_price - nav_price) / nav_price
if abs(spread) > 0.002: # 0.2%阈值
print(f"套利机会！价差：{spread:.4%}")
client.subscribe_quote(['510050.SH', '510300.SH'], callback=on_quote)
client.run()这里有个坑——回调函数里千万别做耗时操作。我曾经犯过这个错，在回调里写了个复杂的计算逻辑，结果行情堆积，内存直接爆了。正确的做法是：回调里只做数据入队，计算逻辑放到另一个线程去处理。
避坑指南：券商API的行情数据通常不包含「实时净值」。你需要自己算，或者从其他渠道获取。我一般用聚宽的数据来补净值，虽然慢一点，但胜在稳定。
4.3 量化平台接入（聚宽/米筐）
如果你还在策略开发阶段，或者做回测验证，量化平台是更好的选择。聚宽和米筐都提供了丰富的历史数据和模拟交易环境。
我个人习惯用聚宽做回测，原因很简单——它的数据清洗做得比较干净。ETF的复权、分红、拆分这些，聚宽都帮你处理好了。你直接拿数据跑策略就行，不用自己写一堆数据预处理代码。
米筐呢？它的优势在于数据维度更全。比如ETF的申赎清单、成分股权重这些，米筐都有现成的API。做套利策略时，这些数据能帮你更精准地计算瞬时净值。
# 聚宽数据接入示例
from jqdatasdk import *
auth('your_account', 'your_password')
# 获取ETF历史行情
df = get_price('510050.XSHG',
start_date='2024-01-01',
end_date='2024-12-31',
frequency='minute',
fields=['open', 'close', 'high', 'low', 'volume'])
# 获取实时行情（聚宽支持WebSocket）
from jqdatasdk import get_current_tick
tick = get_current_tick('510050.XSHG')
print(f"当前价格：{tick['current']}，实时净值：{tick['nav']}")小技巧：量化平台的实时行情延迟一般在200-500ms，做日频套利够用。但如果你做的是「瞬时价差捕捉」，建议还是用券商API或WebSocket。我自己的做法是：回测用聚宽，实盘用券商API，两边数据做交叉验证。
4.4 WebSocket实时行情
这是专业玩家的选择。WebSocket能提供毫秒级的行情推送，延迟通常在1-10ms。国内主流的行情提供商有：万得、东方财富、同花顺，以及一些专业的金融数据服务商。
WebSocket的好处是「全双工通信」。你订阅了行情后，服务端会主动推给你，不用你反复去问。这在套利场景下太重要了——你想想看，价差变化可能就在一瞬间，如果还用HTTP轮询，等你问到的时候，黄花菜都凉了。
# WebSocket行情接入示例
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 解析行情数据
symbol = data['symbol']
price = data['price']
timestamp = data['timestamp']
print(f"{symbol} 最新价：{price} @ {timestamp}")
def on_error(ws, error):
print(f"连接异常：{error}")
def on_close(ws, close_status_code, close_msg):
print("连接关闭，尝试重连...")
def on_open(ws):
# 订阅ETF行情
subscribe_msg = {
"action": "subscribe",
"symbols": ["510050.SH", "510300.SH", "510500.SH"]
}
ws.send(json.dumps(subscribe_msg))
# 启动连接
ws = websocket.WebSocketApp("wss://market.example.com/ws",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close)
ws.run_forever()注意：WebSocket连接需要处理断线重连。交易所的行情服务器偶尔会维护，或者网络波动导致断开。我建议你写一个心跳检测机制，每5秒发一次ping，如果连续3次没收到pong，就主动重连。
4.5 数据源架构设计
做套利系统，数据源架构比你想的重要。我见过有人把所有数据源混在一起，结果一个挂了，整个系统瘫痪。
我的建议是：主备分离。主数据源用WebSocket或券商API，备数据源用量化平台。主数据源负责实时行情，备数据源负责数据校验和补漏。
下面是我常用的架构图，用SVG画出来供你参考：
这个架构的好处是：任何一个数据源挂了，系统都能自动切换。我曾在实盘中遇到过券商API突然断连的情况，系统自动切到WebSocket，整个过程不到1秒，套利策略完全没受影响。
4.6 数据质量校验
数据源接进来了，不代表就万事大吉了。数据质量才是真正的隐形杀手。
我遇到过最离谱的一次：某数据源推送的ETF价格突然跳了10%，但实际市场根本没动。后来查出来是数据源内部的一个bug。如果当时我的策略直接用了那个价格，就会触发虚假的套利信号，后果不堪设想。
所以，我建议你做三件事：
- 价格合理性校验：当前价格与前一价格偏差超过5%，直接丢弃，等下一个tick
- 多源交叉验证：主数据源和备数据源的价格差异超过0.1%，发出告警
- 延迟监控：记录每个tick的时间戳，如果延迟超过500ms，自动切换数据源
我的经验：数据校验的代码一定要写在数据汇聚层，不要写在策略层。这样策略代码可以保持干净，只关注交易逻辑。数据脏活累活，交给底层去处理。
好了，数据源接入这块就聊到这儿。记住一句话：数据是套利策略的命根子。你花再多时间在数据源上都不为过。毕竟，没有靠谱的数据，再牛的策略也只是空中楼阁。