问题描述
请帮我编写一个 ETF 一二级市场套利策略
解决方案
这是一个基于 PTrade 平台的 ETF 一二级市场瞬时套利策略。
策略原理
ETF 套利的核心在于利用 二级市场交易价格 与 一级市场净值（IOPV/NAV） 之间的价差进行无风险（或低风险）获利。
- 溢价套利（正向套利）：
- 条件：ETF 二级市场价格 > 一级市场参考净值 (IOPV) + 交易成本。
- 操作：买入一篮子股票（一级市场） -> 申购 ETF 份额 -> 在二级市场卖出 ETF 份额。
- 折价套利（反向套利）：
- 条件：ETF 二级市场价格 < 一级市场参考净值 (IOPV) - 交易成本。
- 操作：在二级市场买入 ETF 份额 -> 赎回 ETF 份额（得到一篮子股票） -> 卖出一篮子股票。
策略代码实现
本策略以 华夏上证50ETF (510050.SS) 为例。代码包含了 IOPV 的估算逻辑、套利信号触发及 PTrade 专用的 ETF 交易接口调用。
import numpy as np
def initialize(context):
"""
策略初始化函数
"""
# 1. 设定套利标的：510050.SS (上证50ETF)
g.etf_code = '510050.SS'
# 2. 设定套利阈值 (例如 0.2%，即 0.002)
# 需覆盖印花税、佣金、冲击成本等
g.spread_threshold = 0.002
# 3. 获取 ETF 的基本信息（最小申赎单位等）
# 注意：get_etf_info 返回的是字典，key 是 ETF 代码
etf_info = get_etf_info(g.etf_code)
if etf_info:
# report_unit: 最小申赎单位（例如 510050 通常是 900,000 份）
g.unit = int(etf_info[g.etf_code]['report_unit'])
log.info(f"ETF: {g.etf_code}, 最小申赎单位: {g.unit}")
else:
g.unit = 900000 # 默认兜底值
log.warning("未获取到ETF信息，使用默认申赎单位 900000")
# 4. 设定股票池为 ETF 本身及其成分股（用于行情获取）
# 获取成分股列表
g.components = get_etf_stock_list(g.etf_code)
# 将 ETF 和成分股加入 Universe
set_universe([g.etf_code] + g.components)
# 5. 预先获取成分股的数量结构（用于计算 IOPV）
# 格式: {stock_code: amount, ...}
g.basket_weights = {}
for stock in g.components:
# 获取成分券信息
info = get_etf_stock_info(g.etf_code, stock)
if info and stock in info:
# code_num 是成分股在申赎单位中的数量
g.basket_weights[stock] = info[stock]['code_num']
def handle_data(context, data):
"""
盘中运行函数 (建议在分钟或 Tick 级别运行)
"""
# 1. 获取 ETF 当前二级市场行情
etf_snap = get_snapshot(g.etf_code)
if not etf_snap or g.etf_code not in etf_snap:
return
etf_data = etf_snap[g.etf_code]
# 获取 ETF 买一价和卖一价
# bid_grp: 买盘 [价格, 量, 笔数]
# offer_grp: 卖盘
try:
etf_bid_1 = etf_data['bid_grp'][1][0] # 买一价 (用于我们卖出)
etf_ask_1 = etf_data['offer_grp'][1][0] # 卖一价 (用于我们买入)
except:
return # 行情数据不全，跳过
if etf_bid_1 == 0 or etf_ask_1 == 0:
return
# 2. 计算实时 IOPV (估算一篮子股票的买入成本和卖出所得)
# 注意：这里简化计算，实际生产环境可能需要考虑现金替代标志和现金差额
basket_buy_cost = 0.0 # 买入一篮子股票需要的钱
basket_sell_get = 0.0 # 卖出一篮子股票得到的钱
# 批量获取成分股快照
stock_snaps = get_snapshot(g.components)
valid_calc = True
for stock, amount in g.basket_weights.items():
if stock not in stock_snaps:
valid_calc = False
break
s_data = stock_snaps[stock]
# 获取成分股的卖一价（我们要买入股票）和买一价（我们要卖出股票）
try:
s_ask_1 = s_data['offer_grp'][1][0]
s_bid_1 = s_data['bid_grp'][1][0]
except:
valid_calc = False
break
if s_ask_1 == 0 or s_bid_1 == 0:
valid_calc = False
break
basket_buy_cost += s_ask_1 * amount
basket_sell_get += s_bid_1 * amount
if not valid_calc:
return
# 将一篮子总金额转换为单份 ETF 的净值价格
iopv_buy = basket_buy_cost / g.unit # 按照成分股卖一价买入计算出的 IOPV
iopv_sell = basket_sell_get / g.unit # 按照成分股买一价卖出计算出的 IOPV
# 3. 判断套利机会
# --- 溢价套利 (Premium Arbitrage) ---
# 逻辑：ETF 价格太贵。
# 操作：买入股票篮子 -> 申购 ETF -> 卖出 ETF
# 判定：ETF买一价 > 股票篮子买入成本 * (1 + 阈值)
if etf_bid_1 > iopv_buy * (1 + g.spread_threshold):
log.info(f"触发溢价套利: ETF Bid: {etf_bid_1}, IOPV(Buy): {iopv_buy:.4f}")
do_premium_arbitrage(context)
# --- 折价套利 (Discount Arbitrage) ---
# 逻辑：ETF 价格太便宜。
# 操作：买入 ETF -> 赎回 ETF -> 卖出股票篮子
# 判定：ETF卖一价 < 股票篮子卖出所得 * (1 - 阈值)
elif etf_ask_1 < iopv_sell * (1 - g.spread_threshold):
log.info(f"触发折价套利: ETF Ask: {etf_ask_1}, IOPV(Sell): {iopv_sell:.4f}")
do_discount_arbitrage(context)
def do_premium_arbitrage(context):
"""
执行溢价套利：买股票 -> 申购 -> 卖ETF
"""
# 检查资金是否足够 (粗略估算)
cash = context.portfolio.cash
estimated_cost = g.unit * 3.0 # 假设净值约 3.0
if cash < estimated_cost:
log.info("资金不足，无法执行溢价套利")
return
# 1. 买入一篮子股票
# etf_basket_order 是 PTrade 特有接口，方便一键买入
# amount: 篮子份数 (1 表示 1 个申赎单位)
# price_style: 'S1' 表示用卖一价买入 (对手方最优)
# position: False 表示不使用持仓替代，全部新买
log.info("Step 1: 买入一篮子股票")
etf_basket_order(g.etf_code, amount=1, price_style='S1', position=False)
# 2. 申购 ETF
# amount: 必须是 report_unit 的整数倍，正数表示申购
log.info("Step 2: 申购 ETF")
etf_purchase_redemption(g.etf_code, amount=g.unit)
# 3. 卖出 ETF
# 假设 T+0 可用（大部分货币、债券、跨境及部分股票ETF支持实时T+0，
# 但普通单市场股票ETF通常是 T+0 申购份额可用，需确认券商柜台配置）
log.info("Step 3: 卖出 ETF")
order(g.etf_code, amount=-g.unit)
def do_discount_arbitrage(context):
"""
执行折价套利：买ETF -> 赎回 -> 卖股票
"""
# 检查资金
cash = context.portfolio.cash
estimated_cost = g.unit * 3.0
if cash < estimated_cost:
log.info("资金不足，无法执行折价套利")
return
# 1. 买入 ETF
log.info("Step 1: 买入 ETF")
order(g.etf_code, amount=g.unit)
# 2. 赎回 ETF
# amount: 负数表示赎回
log.info("Step 2: 赎回 ETF")
etf_purchase_redemption(g.etf_code, amount=-g.unit)
# 3. 卖出一篮子股票
# amount: 负数表示卖出篮子
# price_style: 'B1' 表示用买一价卖出
# position: True 表示使用持仓卖出
log.info("Step 3: 卖出一篮子股票")
etf_basket_order(g.etf_code, amount=-1, price_style='B1', position=True)
代码关键点解析
get_etf_info与g.unit:- ETF 套利必须严格遵守最小申赎单位（Report Unit）。例如 510050 通常是 90 万股一篮子。如果下单数量不对，柜台会直接废单。
- IOPV 计算:
- 虽然 PTrade 的
get_snapshot有时包含 IOPV 字段，但为了策略的稳健性，代码中演示了通过成分股（g.components）的实时盘口价格（Bid/Ask）来手动计算“合成净值”。 - 买入成本 IOPV：用成分股的卖一价计算（因为我们要买股票）。
- 卖出所得 IOPV：用成分股的买一价计算（因为我们要卖股票）。
- 虽然 PTrade 的
etf_basket_order:- 这是 PTrade 专门为 ETF 交易设计的强力接口。
amount=1代表买入一个标准申赎单位对应的所有股票。price_style='S1'代表以卖一价下单（对手价），保证成交速度。
etf_purchase_redemption:- 这是向柜台发送申购（正数）或赎回（负数）指令的接口。
- 注意：申购赎回通常是 T+0 可用的，这意味着申购回来的份额当天可以卖出（具体视 ETF 品种和券商柜台设置而定，单市场 ETF 通常支持）。
实盘注意事项
- T+0 权限：请务必与券商确认该 ETF 是否支持 实时申赎 T+0（即申购后份额实时到账可卖，赎回后股票实时到账可卖）。如果不支持，该策略无法在日内闭环。
- 现金替代（Cash Substitution）：代码中简化了计算，未包含“必须现金替代”部分的溢价率计算。在实盘中，部分成分股停牌时，需要用现金替代，这会影响套利成本。
- Level-2 行情：套利对速度要求极高，建议使用
tick_data配合 Level-2 行情进行高频触发，而不是handle_data（分钟级）。 - 费率设置：
g.spread_threshold必须设置得足够覆盖双边交易佣金、印花税（卖出股票时收取）以及申赎费用（部分 ETF 有固定申赎费）。
Q&A
Q: 为什么使用 etf_basket_order 而不是自己循环下单买股票？
A: etf_basket_order 是 PTrade 封装好的原子指令，执行速度比在 Python 层面写 for 循环逐个下单要快得多，且能减少滑点风险。
Q: 如何在回测中查看效果？
A: PTrade 的回测系统支持 ETF 申赎逻辑，但回测通常是分钟级别的。由于套利机会稍纵即逝，分钟级回测可能无法捕捉到盘口的瞬时错价，或者成交价格与信号价格偏差较大。此策略更适合在模拟盘或实盘中观察。
Q: 什么是 IOPV？
A: IOPV (Indicative Optimized Portfolio Value) 是交易所发布的基金份额参考净值，每 15 秒更新一次。策略中我们手动计算是为了获得比交易所推送更实时的估值（基于实时 Tick）。