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