第四章 ETF套利实战：搭建实时监控系统
ETF套利，说白了就是吃价差。但价差转瞬即逝，你得有套趁手的工具。今天我就带你亲手搭一套实时监控系统，把IOPV和实时价差算得明明白白。
4.1 系统架构设计
先看整体结构。我习惯把系统拆成三层：数据层、计算层、展示层。
三层之间用消息队列串起来。我个人偏好用Redis的pub/sub，轻量又好用。你想想看，行情数据每秒几十笔，用HTTP轮询肯定扛不住。
4.2 IOPV计算原理
IOPV，全称是实时估算净值。交易所每5秒算一次，但咱们自己算可以更快。
核心公式：
IOPV = (成分股1价格 × 权重1 + 成分股2价格 × 权重2 + ...) / 基金份额
嗯，这里要注意。权重数据每天更新一次，你得在开盘前拉取最新的ETF申赎清单。
我曾经踩过一个坑：某天成分股停牌了，但权重没变。结果算出来的IOPV和实际净值差了0.3%。后来我加了个停牌处理逻辑——停牌股用上一交易日收盘价。
4.3 实时价差计算
价差 = ETF市价 - IOPV。正价差说明ETF贵了，负价差说明便宜了。
| 价差方向 | 套利方向 | 操作 |
|---|---|---|
| 正价差 > 阈值 | 溢价套利 | 买入成分股 → 申购ETF → 卖出ETF |
| 负价差 < -阈值 | 折价套利 | 买入ETF → 赎回成分股 → 卖出成分股 |
阈值怎么设？我一般用万三。低于这个数，扣除手续费基本不赚钱。
4.4 代码实现：实时监控核心
直接上代码。这是监控系统的核心模块：
import redis
import json
import time
from typing import Dict, List
class ETFMonitor:
def __init__(self, etf_code: str):
self.etf_code = etf_code
self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
self.components = self._load_components()
self.threshold = 0.0003 # 万三阈值
def _load_components(self) -> Dict:
"""加载成分股权重"""
# 实际项目中从数据库或文件读取
return {
'600519': {'weight': 0.15, 'name': '贵州茅台'},
'000858': {'weight': 0.12, 'name': '五粮液'},
# ... 更多成分股
}
def calculate_iopv(self, prices: Dict[str, float]) -> float:
"""计算实时IOPV"""
iopv = 0.0
for code, info in self.components.items():
price = prices.get(code, 0)
iopv += price * info['weight']
return iopv
def check_arbitrage(self, etf_price: float, iopv: float):
"""检查套利机会"""
spread = (etf_price - iopv) / iopv
if spread > self.threshold:
self._send_signal('溢价套利', spread)
elif spread < -self.threshold:
self._send_signal('折价套利', spread)
def _send_signal(self, signal_type: str, spread: float):
"""发送套利信号"""
message = {
'time': time.time(),
'etf': self.etf_code,
'type': signal_type,
'spread': round(spread * 10000, 2) # 转为BP
}
self.redis_client.publish('arbitrage_signals', json.dumps(message))
小技巧：实际生产环境中，我会用asyncio做异步处理。行情来了直接回调，不用轮询。性能能提升3-5倍。
4.5 模拟交易引擎
模拟交易，说白了就是纸上谈兵。但别小看它，我见过太多实盘亏钱的策略，模拟跑得飞起。
模拟引擎的核心逻辑：
class SimulatedTrader:
def __init__(self, initial_capital: float = 1000000):
self.capital = initial_capital
self.positions = {} # 持仓
self.trades = [] # 交易记录
def execute_arbitrage(self, signal: dict):
"""执行套利操作"""
if signal['type'] == '溢价套利':
# 买入成分股
cost = self._buy_components(signal['amount'])
# 申购ETF
etf_shares = self._subscribe_etf(signal['amount'])
# 卖出ETF
revenue = self._sell_etf(etf_shares, signal['etf_price'])
profit = revenue - cost
self._record_trade(signal, profit)
elif signal['type'] == '折价套利':
# 反向操作
pass
def _record_trade(self, signal: dict, profit: float):
"""记录交易"""
self.trades.append({
'time': signal['time'],
'type': signal['type'],
'profit': profit,
'capital_after': self.capital + profit
})
self.capital += profit
注意：模拟交易和实盘最大的区别是滑点和冲击成本。我建议模拟时主动加0.1%的滑点，这样结果更接近真实情况。
4.6 监控面板展示
数据算出来了，得让人看得懂。我习惯用表格展示实时数据：
| 时间 | ETF代码 | ETF市价 | IOPV | 价差(BP) | 信号 |
|---|---|---|---|---|---|
| 09:30:05 | 510050 | 3.456 | 3.452 | +1.16 | 无 |
| 09:30:10 | 510050 | 3.468 | 3.453 | +4.34 | 溢价套利 |
| 09:30:15 | 510050 | 3.445 | 3.454 | -2.61 | 折价套利 |
你看，09:30:10那笔溢价4.34个BP，超过万三阈值了。系统立刻发出信号。这时候你就要判断：是手动跟，还是让程序自动执行？
我的建议：刚开始做套利，先手动跟一个月。等摸清了市场的脾气，再考虑自动化。我当年就是太心急，一上来就全自动，结果某天行情剧烈波动，程序来回刷单，一天亏了2万多。
4.7 避坑指南
做ETF套利监控，有几个坑你一定要注意：
- 数据延迟：行情推送有延迟，我建议用Level-2行情。普通行情延迟3-5秒，套利机会早没了。
- 成分股变更：ETF每半年调一次成分股。我吃过这个亏，调仓后没更新权重，算出来的IOPV全是错的。
- 停牌处理：成分股停牌时，用收盘价还是用估值？我建议用估值模型算，但保守点就用收盘价。
- 手续费计算：ETF申购赎回有固定费用，别漏算。我见过有人模拟跑得欢，实盘一算手续费直接亏。
好了，这套监控系统搭起来后，你就能实时看到ETF的套利机会了。下一节我们聊聊怎么优化交易执行，把滑点降到最低。
公众号:蓝海资料掘金营，微信deep3321