PTrade ETF 套利实战：如何利用 API 实现一二级市场价差套利？

这是一个基于 PTrade 平台的 ETF 一二级市场瞬时套利策略。

策略逻辑说明

该策略主要监控 ETF 的 交易价格（二级市场） 与 IOPV/净值（一级市场） 之间的价差。

1. 溢价套利（正向套利）：

   • 条件：ETF 价格 > IOPV + 成本阈值
   • 操作：买入一篮子股票 -> 申购 ETF -> 卖出 ETF。
   • API：使用 etf_basket_order 买入股票篮子，etf_purchase_redemption 申购，order 卖出 ETF。

2. 折价套利（反向套利）：

   • 条件：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("执行折价套利完成")

关键点解析

1. IOPV 计算：

   • PTrade 的 get_snapshot 返回的是行情快照，通常包含 last_px。虽然部分券商的 Level-2 行情可能直接推送交易所计算好的 IOPV，但为了通用性，代码中演示了如何通过 get_etf_stock_list 和 get_etf_stock_info 获取成分股权重，并结合实时价格手动计算 IOPV。
   • 公式：IOPV = (∑(成分股价格 × 篮子内数量) + 现金差额) / 最小申赎单位。

2. 篮子交易 (etf_basket_order)：

   • 这是 PTrade 专门为 ETF 交易设计的强力接口。
   • amount=1：代表买入 1 个标准篮子的所有成分股。
   • amount=-1：代表卖出 1 个标准篮子的所有成分股。
   • 这大大简化了折价套利中“卖出股票”的步骤，不需要自己写循环去卖几百只股票。

3. 申购与赎回 (etf_purchase_redemption)：

   • amount 参数是 份额数量（如 900,000），而不是篮子数量。
   • 正数代表申购，负数代表赎回。

4. T+0 假设：

   • ETF 瞬时套利通常依赖于券商提供的 T+0 交易权限（即买入股票后可立即申购，申购得到的 ETF 可立即卖出；或买入 ETF 后可立即赎回，赎回得到的股票可立即卖出）。
   • 如果您的账户是普通账户（非做市商或专业机构），可能面临 T+1 或 T+2 的交收限制，此时该策略需要配合底仓（持有 ETF 或持有股票底仓）来进行变相 T+0。

5. 价格类型：

   • 在套利交易中，速度优于价格。代码中使用了 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申赎费用（部分券商有）+ 冲击成本（滑点）。建议根据您账户的实际费率进行调整。