问题描述
请帮我编写一个 ETF 套利策略，监控 ETF 的一二级市场价差，当出现套利机会时自动申购/赎回并卖出/买入一篮子股票。
解决方案
这是一个基于 PTrade 平台的 ETF 一二级市场瞬时套利策略。
策略逻辑说明
该策略主要监控 ETF 的 交易价格（二级市场） 与 IOPV/净值（一级市场） 之间的价差。
-
溢价套利（正向套利）：
- 条件：ETF 价格 > IOPV + 成本阈值
- 操作：买入一篮子股票 -> 申购 ETF -> 卖出 ETF。
- API：使用
etf_basket_order买入股票篮子，etf_purchase_redemption申购，order卖出 ETF。
-
折价套利（反向套利）：
- 条件：ETF 价格 < IOPV - 成本阈值
- 操作：买入 ETF -> 赎回 ETF -> 卖出一篮子股票。
- API：使用
order买入 ETF，etf_purchase_redemption赎回，etf_basket_order卖出股票篮子。
核心代码
import numpy as np
def initialize(context):
"""
策略初始化函数
"""
# 设定要套利的ETF代码 (例如: 510050.SS 上证50ETF)
g.etf_code = '510050.SS'
# 设定套利阈值 (例如 0.002 代表 0.2% 的价差才触发交易，覆盖手续费和滑点)
g.spread_threshold = 0.002
# 设定单次套利的篮子数量 (Basket Quantity)
g.trade_basket_num = 1
# 初始化全局变量
g.report_unit = 0 # 最小申赎单位 (例如 900,000 份)
g.cash_component = 0 # 预估现金差额
g.stock_list = [] # 成分股列表
g.stock_qty = {} # 成分股对应的每篮子数量
# 设置股票池
set_universe([g.etf_code])
# 开启交易日志
log.info("ETF套利策略初始化完成，目标: %s" % g.etf_code)
def before_trading_start(context, data):
"""
盘前处理：每日更新ETF的基础信息和成分股清单
"""
# 1. 获取ETF基础信息 (最小申赎单位, 现金差额等)
# 注意：get_etf_info 返回的是字典 {code: info_dict}
etf_info_dict = get_etf_info(g.etf_code)
if etf_info_dict and g.etf_code in etf_info_dict:
info = etf_info_dict[g.etf_code]
g.report_unit = int(info.get('report_unit', 0))
# 注意：cash_balance 可能随每日变动，这里取盘前数据作为参考
g.cash_component = float(info.get('cash_balance', 0.0))
log.info("ETF信息更新: 最小申赎单位=%s, 现金差额=%s" % (g.report_unit, g.cash_component))
else:
log.error("获取ETF信息失败: %s" % g.etf_code)
return
# 2. 获取成分股列表
g.stock_list = get_etf_stock_list(g.etf_code)
if not g.stock_list:
log.error("获取成分股列表失败")
return
# 将成分股加入股票池，以便获取行情
set_universe(g.stock_list + [g.etf_code])
# 3. 获取成分股每篮子对应的数量
# get_etf_stock_info 支持传入列表
stock_infos = get_etf_stock_info(g.etf_code, g.stock_list)
g.stock_qty = {}
if stock_infos:
for stock in g.stock_list:
if stock in stock_infos:
# code_num 是成分股数量
g.stock_qty[stock] = stock_infos[stock].get('code_num', 0)
log.info("成分股数据加载完成，共 %d 只" % len(g.stock_list))
def handle_data(context, data):
"""
盘中运行：计算IOPV，监控价差，执行套利
"""
if g.report_unit == 0 or not g.stock_list:
return
# 1. 获取ETF实时行情
etf_snap = get_snapshot(g.etf_code)
if not etf_snap or g.etf_code not in etf_snap:
return
etf_price = etf_snap[g.etf_code].get('last_px', 0.0)
# 过滤无效价格（如停牌或未开盘）
if etf_price <= 0:
return
# 2. 计算实时 IOPV (Indicative Optimized Portfolio Value)
# IOPV ≈ (∑(成分股价格 * 成分股数量) + 现金差额) / 最小申赎单位
# 批量获取成分股快照
stock_snaps = get_snapshot(g.stock_list)
if not stock_snaps:
return
basket_stock_value = 0.0
valid_calc = True
for stock in g.stock_list:
# 获取成分股价格
stock_data = stock_snaps.get(stock)
if stock_data:
price = stock_data.get('last_px', 0.0)
# 如果成分股停牌或无价格，使用昨收价作为估算 (preclose_px)
if price <= 0:
price = stock_data.get('preclose_px', 0.0)
qty = g.stock_qty.get(stock, 0)
basket_stock_value += price * qty
else:
# 如果获取不到关键成分股数据，放弃本次计算
valid_calc = False
break
if not valid_calc:
return
# 计算每股 IOPV
total_basket_value = basket_stock_value + g.cash_component
calculated_iopv = total_basket_value / g.report_unit
if calculated_iopv <= 0:
return
# 3. 计算价差率
# spread > 0 表示溢价 (Price > IOPV)，spread < 0 表示折价 (Price < IOPV)
spread_rate = (etf_price - calculated_iopv) / calculated_iopv
# 4. 执行套利逻辑
# --- 溢价套利 (买入股票 -> 申购 ETF -> 卖出 ETF) ---
if spread_rate > g.spread_threshold:
log.info("发现溢价套利机会: ETF价格=%.3f, IOPV=%.3f, 溢价率=%.2f%%" % (etf_price, calculated_iopv, spread_rate * 100))
# 步骤A: 买入一篮子股票
# etf_basket_order: amount正数为买入篮子
# price_style='S5' 表示以卖五价买入，保证成交速度
basket_oid = etf_basket_order(g.etf_code, g.trade_basket_num, price_style='S5', position=False)
if basket_oid:
# 步骤B: 申购 ETF
# etf_purchase_redemption: amount为份额数量 (篮子数 * 单位份额)
share_amount = g.trade_basket_num * g.report_unit
pr_oid = etf_purchase_redemption(g.etf_code, share_amount)
if pr_oid:
# 步骤C: 卖出 ETF
# 假设账户支持T+0回转（ETF申购后实时可用），直接卖出
# 使用对手方最优价格或本方最优价格卖出
order(g.etf_code, -share_amount)
log.info("执行溢价套利完成")
# --- 折价套利 (买入 ETF -> 赎回 ETF -> 卖出股票) ---
elif spread_rate < -g.spread_threshold:
log.info("发现折价套利机会: ETF价格=%.3f, IOPV=%.3f, 折价率=%.2f%%" % (etf_price, calculated_iopv, spread_rate * 100))
share_amount = g.trade_basket_num * g.report_unit
# 步骤A: 买入 ETF
# 检查资金是否足够略过，直接下单，依靠系统风控
buy_oid = order(g.etf_code, share_amount)
if buy_oid:
# 步骤B: 赎回 ETF
# amount为负数表示赎回
pr_oid = etf_purchase_redemption(g.etf_code, -share_amount)
if pr_oid:
# 步骤C: 卖出一篮子股票
# etf_basket_order: amount负数为卖出篮子
# price_style='B5' 表示以买五价卖出，保证成交
etf_basket_order(g.etf_code, -g.trade_basket_num, price_style='B5')
log.info("执行折价套利完成")
关键点解析
-
IOPV 计算：
- PTrade 的
get_snapshot返回的是行情快照，通常包含last_px。虽然部分券商的 Level-2 行情可能直接推送交易所计算好的 IOPV，但为了通用性，代码中演示了如何通过get_etf_stock_list和get_etf_stock_info获取成分股权重，并结合实时价格手动计算 IOPV。 - 公式：
IOPV = (∑(成分股价格 × 篮子内数量) + 现金差额) / 最小申赎单位。
- PTrade 的
-
篮子交易 (
etf_basket_order)：- 这是 PTrade 专门为 ETF 交易设计的强力接口。
amount=1：代表买入 1 个标准篮子的所有成分股。amount=-1：代表卖出 1 个标准篮子的所有成分股。- 这大大简化了折价套利中“卖出股票”的步骤，不需要自己写循环去卖几百只股票。
-
申购与赎回 (
etf_purchase_redemption)：amount参数是 份额数量（如 900,000），而不是篮子数量。- 正数代表申购，负数代表赎回。
-
T+0 假设：
- ETF 瞬时套利通常依赖于券商提供的 T+0 交易权限（即买入股票后可立即申购，申购得到的 ETF 可立即卖出；或买入 ETF 后可立即赎回，赎回得到的股票可立即卖出）。
- 如果您的账户是普通账户（非做市商或专业机构），可能面临 T+1 或 T+2 的交收限制，此时该策略需要配合底仓（持有 ETF 或持有股票底仓）来进行变相 T+0。
-
价格类型：
- 在套利交易中，速度优于价格。代码中使用了
S5（卖五价）买入和B5（买五价）卖出，这是一种激进的吃单方式，目的是确保立即成交，防止滑点导致套利失败。
- 在套利交易中，速度优于价格。代码中使用了
Q&A
Q: 为什么不直接使用 tick_data 函数？
A: tick_data 是处理高频数据的最佳位置。上述代码使用了 handle_data（通常是分钟级）作为演示框架，方便理解逻辑。在实盘中，建议将 handle_data 中的逻辑迁移到 tick_data(context, data) 中，以实现 3 秒一次甚至更快的监控频率。
Q: 如何处理成分股停牌的情况？
A: 代码中包含了一个简单的处理：if price <= 0: price = stock_data.get('preclose_px', 0.0)。如果成分股停牌，使用昨收价估算。在实际严谨的套利中，如果停牌股票权重较大，通常会暂停套利，或者使用行业指数收益率来修正停牌股票的估值（必须现金替代的部分除外）。
Q: g.spread_threshold = 0.002 是如何确定的？
A: 这个阈值必须覆盖：双边交易佣金 + 印花税（卖出股票时）+ ETF申赎费用（部分券商有）+ 冲击成本（滑点）。建议根据您账户的实际费率进行调整。