算法研究--ETF套利交易算法研究

风险提示：本内容为量化技术及平台操作的纯知识分享，不构成任何投资建议，不推荐任何具体证券或交易时机。部分内容可能理解包含错误，请自行核实。市场有风险，投资需谨慎

最近在看ptrade的策略，发现可以利用ptrade计算出来etf的溢价率，etf独特的一级市场，二级市场机制，因为标的的价格和市场的价格不一样，存在套利机会，一般个人投资者可能套利lof，比较多，lof基金的套利门槛比较低，我先研究lof，看看可以实现自动套利不，etf套利门槛比较高，一般需要100万起，同时标的有份额限制，可能应该标的一条酒几个人可以抢到份额，

ETF套利的核心原理，是利用同一商品在不同市场之间的价格差异来获取利润。

ETF具有双重交易机制：一级市场（申赎市场）

：投资者可以用一篮子成分股向基金公司申购ETF份额，或者将ETF份额赎回成一篮子成分股。交易价格基于基金净值（IOPV或NAV）。二级市场（买卖市场）

：投资者可以在证券交易所像买卖股票一样买卖ETF份额。交易价格由市场供需关系决定

溢价率计算公式

溢价率(%) = (当前价格 - 参考净值) / 参考净值 × 100

第一步需要读取标的的实时盘中价格，整理利用读取快照数据接口

返回的参数

amount:持仓量(str:int)(期货字段,股票返回0)；
bid_grp:委买档位(第一档包含委托队列(仅L2支持))(str:dict[int:list[float,int,int,{int:int,...}],int:list[float,int,int]...])；
business_amount:总成交量(str:int)；
business_amount_in:内盘成交量(str:int)；
business_amount_out:外盘成交量(str:int)；
business_balance:总成交额(str:float)；
business_count:成交笔数(str:int)
circulation_amount:流通股本(str:int)；
current_amount:最近成交量(现手)(str:int)；
down_px:跌停价格(str:float)；
end_trade_date:最后交易日(str:str)
entrust_diff:委差(str:float)；
entrust_rate:委比(str:float)；
high_px:最高价(str:float)；
hsTimeStamp:时间戳(str:float)；
last_px:最新成交价(str:float)；
low_px:最低价(str:float)；
offer_grp:委卖档位(第一档包含委托队列(仅L2支持))(str:dict[int:list[float,int,int,{int:int,...}],int:list[float,int,int]...])；
open_px:今开盘价(str:float)；
pb_rate:市净率(str:float)；
pe_rate:动态市盈率(str:float)；
preclose_px:昨收价(str:float)；
prev_settlement:昨结算(str:float)(期货字段,股票返回0.0)；
px_change_rate:涨跌幅(str:float)；
settlement:结算价(str:float)(期货字段,股票返回0.0)
start_trade_date:首个交易日(str:float)
tick_size:最小报价单位(str:float)
total_bid_turnover:委买金额(str:int)；
total_bidqty:委买量(str:int)；
total_offer_turnover:委卖金额(str:int)
total_offerqty:委卖量(str:int)；
trade_mins:交易分钟数(str:int)
trade_status:交易状态(str:str)；
turnover_ratio:换手率(str:int)；
up_px:涨停价格(str:float)；
vol_ratio:量比(str:float)；
wavg_px:加权平均价(str:float)；
iopv:基金份额参考净值(str:float)；

还需要读取etf t-1,上一日的单位值，利用读取标的etf信息函数

返回的结果

etf_redemption_code -- 申赎代码(str:str)；
publish -- 是否需要发布IOPV(str:int)；
report_unit -- 最小申购、赎回单位(str:int)；
cash_balance -- 现金差额(str:float)；
max_cash_ratio -- 现金替代比例上限(str:float)；
pre_cash_component -- T-1日申购基准单位现金余额(str:float)；
nav_percu -- T-1日申购基准单位净值(str:float)；
nav_pre -- T-1日基金单位净值(str:float)；
allot_max -- 申购上限(str:float)；
redeem_max -- 赎回上限(str:float)

代码没有问题挂交易就可以

不懂的问我就可以

简单的参考

'''
ETF盘中实时溢价率监控策略 - 实盘专用版
作者: 小果量化
微信: xg_quant
时间: 20260730
功能: 计算ETF盘中实时溢价率，监控折溢价机会
说明: 本策略仅适用于实盘交易环境，不兼容回测
参考净值来源: get_etf_info获取的T-1日基金单位净值(nav_pre)
'''
import pandas as pd
import numpy as np
from datetime import datetime
def initialize(context):
    """
    策略初始化函数
    """
    # ETF观察列表
    g.etf_list = [
        "50.SS",  
        "0.SS",  
        
    ]
    
    # 设置股票池
    set_universe(g.etf_list)
    
    # 设置定时任务：每天15:08执行一次溢价率检查
    run_daily(context, func=check_premium_rate, time='15:20')
    
    log.info("ETF盘中实时溢价率监控策略初始化完成！")
    log.info("监控ETF数量: %d 只" % len(g.etf_list))
    log.info("执行时间: 15:08")
def get_etf_premium_rate(context, etf_code):
    """
    计算单只ETF的盘中实时溢价率（实盘专用）
    
    溢价率计算公式：
    溢价率(%) = (当前价格 - 参考净值) / 参考净值 * 100
    
    参考净值来源：get_etf_info获取的T-1日基金单位净值(nav_pre)
    
    参数:
        context: Context对象
        etf_code: ETF代码，如 "00.SS"
    
    返回:
        dict: 包含溢价率及相关信息的字典
    """
    result = {
        'etf_code': etf_code,
        'current_price': 0.0,      # 当前价格
        'reference_nav': 0.0,      # 参考净值
        'premium_rate': 0.0,       # 溢价率(%)
        'nav_pre': 0.0,            # T-1日基金单位净值
        'status': 'unknown'        # 状态: 'normal', 'no_data', 'error'
    }
    
    try:
        # ============ 第一步：获取当前价格 ============
        # 实盘环境使用get_snapshot获取tick级别行情
        snapshot = get_snapshot(etf_code)
        
        if not snapshot or etf_code not in snapshot:
            log.warning("%s: 无法获取快照数据" % etf_code)
            result['status'] = 'no_data'
            return result
        
        # 获取最新价
        current_price = float(snapshot[etf_code].get('last_px', 0))
        result['current_price'] = current_price
        
        if current_price <= 0:
            log.warning("%s: 当前价格无效: %.4f" % (etf_code, current_price))
            result['status'] = 'no_data'
            return result
        
        # ============ 第二步：获取参考净值（使用get_etf_info） ============
        reference_nav = 0.0
        
        # 使用get_etf_info获取T-1日基金单位净值(nav_pre)
        try:
            # 获取ETF基本信息
            etf_info = get_etf_info(etf_code)
            
            if etf_info and etf_code in etf_info:
                # 获取T-1日基金单位净值
                nav_pre = float(etf_info[etf_code].get('nav_pre', 0))
                result['nav_pre'] = nav_pre
                
                if nav_pre > 0:
                    reference_nav = nav_pre
                    log.info("%s: 使用T-1日净值作为参考净值: %.4f" % (etf_code, reference_nav))
                else:
                    log.warning("%s: T-1日净值无效: %.4f" % (etf_code, nav_pre))
                    result['status'] = 'no_data'
                    return result
            else:
                log.warning("%s: 无法获取ETF信息" % etf_code)
                result['status'] = 'no_data'
                return result
        except Exception as e:
            log.error("%s: 获取ETF信息异常: %s" % (etf_code, str(e)))
            result['status'] = 'error'
            return result
        
        result['reference_nav'] = reference_nav
        
        # ============ 第三步：计算溢价率 ============
        # 溢价率(%) = (当前价格 - 参考净值) / 参考净值 * 100
        premium_rate = (current_price - reference_nav) / reference_nav * 100
        result['premium_rate'] = round(premium_rate, 4)
        result['status'] = 'normal'
        
        return result
    
    except Exception as e:
        log.error("%s: 计算溢价率异常: %s" % (etf_code, str(e)))
        result['status'] = 'error'
        return result
def check_premium_rate(context):
    """
    定时检查所有ETF的盘中实时溢价率
    """
    current_dt = context.blotter.current_dt
    current_time = current_dt.strftime('%H:%M:%S')
    
    log.info("========== %s ETF盘中实时溢价率检查 ==========" % current_time)
    
    # 存储所有ETF的溢价率数据
    premium_data = []
    
    for etf_code in g.etf_list:
        result = get_etf_premium_rate(context, etf_code)
        
        if result['status'] == 'normal':
            premium_data.append(result)
            
            # 判断溢价状态，只标记异常状态
            if result['premium_rate'] > 0.5:
                status_tag = "⚠️ 高溢价"
            elif result['premium_rate'] > 0.2:
                status_tag = "↑ 溢价"
            elif result['premium_rate'] < -0.5:
                status_tag = "⚠️ 高折价"
            elif result['premium_rate'] < -0.2:
                status_tag = "↓ 折价"
            else:
                # 正常状态不输出日志，减少信息量
                continue
            
            log.info("%s: 价格=%.4f, T-1净值=%.4f, 溢价率=%.4f%% %s" % 
                     (etf_code, result['current_price'], result['nav_pre'], 
                      result['premium_rate'], status_tag))
        else:
            log.warning("%s: 数据获取失败，状态: %s" % (etf_code, result['status']))
    
    # 输出溢价率排名（前5名）
    if premium_data:
        # 按溢价率从高到低排序
        sorted_data = sorted(premium_data, key=lambda x: x['premium_rate'], reverse=True)
        
        log.info("--- 溢价率排名 Top 5 ---")
        for i, data in enumerate(sorted_data[:5], 1):
            log.info("第%d名: %s, 价格=%.4f, 溢价率=%.4f%%" % 
                     (i, data['etf_code'], data['current_price'], data['premium_rate']))
        
        # 输出最高溢价和最高折价
        max_premium = sorted_data
        min_premium = sorted_data[-1]
        
        log.info("最高溢价: %s, 溢价率=%.4f%%" % 
                 (max_premium['etf_code'], max_premium['premium_rate']))
        log.info("最高折价: %s, 溢价率=%.4f%%" % 
                 (min_premium['etf_code'], min_premium['premium_rate']))
    
    log.info("========== ETF盘中实时溢价率检查结束 ==========")
def handle_data(context, data):
    """
    PTrade框架必需的handle_data函数
    """
    pass