红利低波ETF量化策略

100万元策略和买入后一直持有都变成222.67万元。新增的分红再投试验也没有增加收益。这次最有价值的结果是看清哪些操作确实没有带来帮助。首图由聚宽已完成回测的账户数据重绘下文保留真实平台指标截图。1、收益与回撤对比标的为华泰柏瑞中证红利低波动ETF512890聚宽代码512890.XSHG。回测从上市首日2019-01-18到2026-09-16共1859个交易日覆盖全部可交易历史。基金成立至上市之间没有场内成交不能把这段时间虚构为可买入本轮全部实验统一截至2026年9月16日。指标本文策略买入后一直持有期末资产 / 万元222.67222.67累计收益122.67%122.67%年化收益11.37%11.37%最大回撤15.54%15.54%平均实际仓位93.54%93.54%两组起始资金均为100万元最初买入预算均为90万元。“买入后一直持有”是从首个可成交日起按预算完成建仓之后一直不动不主动卖出、不补仓现金分红留在账户。余款保留现金不追加本金。最大回撤是账户从历史高点向后的最大跌幅收益已计入平台记录的分红、份额调整和费用年化收益沿用聚宽口径。本文使用聚宽开发与回测环境准备可参照基础篇。2、量化策略的思路与实现红利低波的第一道问题是有没有必要频繁操作。本文保留“先持有、明显破位才退出”的规则先把干预条件写清楚。环节执行规则初始持仓上市起分批建仓累计120个日线后启用保护退出昨收比最近120日最高收盘低16%以上且低于MA60重新入场风险条件消失且昨收高于MA20的101%优先级若恢复与退出同时成立退出优先高点回落衡量最近一段持有损失MA60用于判断是否已经离开中期重心只有两项同时成立才卖出。恢复线用较短的MA20避免退出后等待完整长趋势才买回来。完整历史没有发生风险退出策略与持有账户相同因此不能把122.67%的收益归功于择时。这次额外检验“每月将已到账分红再买入同一ETF”并锁定初次建仓后留下的全部现金储备。结果没有新增买入。现有聚宽账户记录没有现金分红入账只记录到2021-10-25份额翻倍份额拆分不会产生可再投现金。这只说明当前回测账本没有支持再投资的现金记录不能据此断言基金现实中从未分红。因此再投规则只作为实验结果披露正文仍使用已验证的保护规则。涉及分红再投资的进一步结论应先用基金分红公告与平台账本逐笔核对。prepare_signal()只读取已经完成的日线strategy_rule()判断继续持有或退出execute_signal()提交今天的限价单close_report()按真实成交核算现金、份额和费用。信号出现不代表一定成交买回也不保证比卖出价更低。完整历史有成交订单3笔累计佣金274.88元。订单数量包含建仓和退出的分批成交不等于买卖轮数。改变条件后收益还能留下多少检验累计收益最大回撤本文保护规则122.67%15.54%月度再投试验122.67%15.54%买入后一直持有122.67%15.54%原规则双倍成本122.38%15.54%退出回落改为12%62.83%22.80%退出回落改为20%122.67%15.54%后三行来自同一区间此前已完成的原保护规则检验。退出门槛从16%收紧为12%后收益反而下降、回撤扩大放宽到20%仍与持有相同。这进一步说明当前历史没有支持主动干预的稳定优势。滚动252交易日收益超过持有的窗口占0.00%按不重叠的252交易日分段胜出比例为0.00%。滚动窗口相互重叠不是不相关的独立样本。连续原账户分段本文策略一直持有2019-01-18 至 2020-02-068.83%8.83%2020-02-07 至 2021-02-1819.46%19.46%2021-02-19 至 2022-03-0323.57%23.57%2022-03-04 至 2023-03-164.78%4.78%2023-03-17 至 2024-03-2913.69%13.69%2024-04-01 至 2025-04-169.43%9.43%2025-04-17 至 2026-04-306.14%6.14%2026-05-06 至 2026-09-160.16%0.16%以上均沿用原账户不在分段起点清仓或重新投入100万元末段不足252交易日的部分不纳入分段胜出比例。当前证据还不足以称为“稳定高收益”。本轮规则是在已经查看过的历史上修订的此前也进行过多轮研究全历史、分段、成本检验均属于回顾性验证。下一步应冻结规则用尚未发生的行情做模拟跟踪记录每次信号、未成交和实际成本再判断能否投入实盘。3、在聚宽复现回测在“我的策略”中新建Python 3空白策略粘贴下一节全部代码。回测起点设为2019-01-18终点设为2026-09-16初始资金1000000元运行频率选分钟。检查代码中的ETF代码512890.XSHG保持主参数和费用不变运行到显示“回测完成”。在“收益概述”核对收益、回撤在“交易详情”核对成交在“每日持仓收益”核对资产。平台的“基准收益”是沪深300参考线不是本文独立运行的“买入后一直持有”账户。上图为对应主策略的聚宽原始回测截图包含回测区间、初始资金、完成状态、收益指标及完整收益曲线未重写平台数字。图中基准线为沪深300正文“买入后一直持有”对照为另行回测的同一ETF账户两者口径不同。程序09:05读取昨日及更早行情10:00挂限价单10:05撤余单15:10记账。上市起先按预算建仓积累120个完整交易日后才启用择时预热期的涨跌仍计入全历史。费用为买卖各0.03%、每笔最低5元、基金印花税为0单边滑点0.10%。买单按100份整手提交部分成交可能不是整手未完成买入按剩余原预算续单卖出持续到可卖份额处理完毕。复现持有对照时复制一份相同代码在CONFIG中只修改以下两项回测资金、区间、频率保持一致mode: hold, experiment: ETF17D_08_hold_compare,代码运行后会在研究文件中生成每日账户、订单和逐笔成交记录。复核总资产要用账面现金加持仓市值可用现金可能不包含暂时冻结的资金。以上是历史撮合模拟尚未接入券商实盘。4、完整策略代码复制以下全部代码即可运行。中文注释覆盖参数、信号、下单预算、成交限制和账户核对。# -*- coding: utf-8 -*- ETF单品种策略。聚宽Python3分钟频率初始资金100万元。 本代码完整独立回测参数与正文对应。 所有交易信号仅使用前一交易日及更早的数据真实成交由聚宽撮合。 from jqdata import * from builtins import all, any, max, min import csv import io import json import math import numpy as np # 红利低波ETF量化策略华泰柏瑞中证红利低波动ETF。 # modehold为同ETF买入后一直持有初始资金100万元首次预算90%。 # initial_investedTrue表示上市后先建仓攒足历史再应用退出规则。 # 双倍成本及邻近参数是独立对照不用于自动寻找最高收益。 CONFIG {security: 512890.XSHG, listing: 2019-01-18, mode: strategy, params: {family: fast_guard, slow: 60, recovery: 20, high_window: 120, drop: 0.16, frequency: daily}, warmup: 120, frequency: daily, sizing: entry, initial_invested: True, max_weight: 0.9, commission: 0.0003, slippage: 0.001, limit_offset: 0.002, rebalance_band: 0.05, ablation: False, experiment: ETF17D_08_main} def strategy_rule(h, held, cfg, state): 读取已完成日线。风险退出优先缓冲区内维持已有仓位。 p, c cfg[params], h.close.values family p[family] buy, sell False, False if family in (dual_ma, invested_dual): # fast/slow是日线窗口band避免均线贴近时反复买卖。 fast, slow c[-p[fast]:].mean(), c[-p[slow]:].mean() buy, sell fast slow*(1p[band]), fast slow*(1-p[band]) elif family in (price_band, monthly, invested_trend): # 收盘越过均线上沿才买跌破下沿才卖中间区域不追涨杀跌。 mean c[-p[slow]:].mean() buy, sell c[-1] mean*(1p[band]), c[-1] mean*(1-p[band]) elif family channel: # 历史上/下沿排除昨日自身避免使用包含待判断价格的通道。 buy c[-1] h.high.values[-p[entry]-1:-1].max() sell c[-1] h.low.values[-p[exit]-1:-1].min() elif family momentum: # slow日涨跌需要slow1个收盘价不是slow个价格。 buy, sell c[-1] c[-p[slow]-1], c[-1] c[-p[slow]-1] elif family crash_guard: # 先参与长期持有只有高点回落、价格转弱、均线转弱同时成立才退出。 slow, fast c[-p[slow]:].mean(), c[-p[fast]:].mean() sell (c[-1] c[-p[high_window]:].max()*(1-p[drop]) and c[-1] slow and fast slow) buy c[-1] c[-p[recovery]:].mean() and c[-1] c[-p[recovery]-1] elif family fast_guard: # 高点回落与跌破均线同时出现才退出短均线恢复后才考虑重入。 sell (c[-1] c[-p[high_window]:].max()*(1-p[drop]) and c[-1] c[-p[slow]:].mean()) buy c[-1] c[-p[recovery]:].mean()*1.01 elif family shock_guard: # 用短均线偏离幅度识别急跌恢复线比均线高1%减少立即追入。 sell c[-1] c[-p[slow]:].mean()*(1-p[band]) buy c[-1] c[-p[recovery]:].mean()*1.01 elif family trend_ensemble: votes sum(c[-1] c[-k*p[slow]:].mean() for k in (1,2,3)) buy, sell votes 2, votes 1 else: raise ValueError(未知策略规则: family) # 即使空仓也先判断sell。避免风险与恢复同时成立时隔日反复进出。 if sell: return 0., 风险或趋势退出条件成立 if held or buy: return cfg[max_weight], 继续持有 if held else 入场条件成立 return 0., 等待有效入场信号 def initialize(context): 只初始化一次资金检查、费用、调度和审计文件。 g.cfg dict(CONFIG) assert abs(context.portfolio.starting_cash - 1000000) .01, 初始资金须为1000000 assert g.cfg[mode] in (strategy, hold) g.state {armed: False, held_days: 0} g.observed_days 0 # 已经过的上市后交易日不使用未来价格。 g.target, g.signal_day, g.attempt_day None, None, None g.pending_exit False # 清仓未完成时次日继续卖出剩余可卖份额。 g.entry_remaining None # 首次建仓的剩余预算扣实际成交与佣金不随涨价扩大。 g.reason, g.source_day , g.out g.cfg[experiment] # 该指数仅供平台画参考线正式比较另跑同ETF的hold账户。 set_benchmark(000300.XSHG) set_option(use_real_price, True) set_option(avoid_future_data, True) set_option(order_volume_ratio, .01) # 1%分钟成交量上限不是容量保证。 set_order_cost(OrderCost(open_tax0, close_tax0, open_commissiong.cfg[commission], close_commissiong.cfg[commission], close_today_commission0, min_commission5), typefund, refg.cfg[security]) # 此API参数表示买卖总价差因此单边0.1%需要传0.2%。 set_slippage(PriceRelatedSlippage(2*g.cfg[slippage]),typefund,refg.cfg[security]) run_daily(prepare_signal, time09:05) run_daily(execute_signal, time10:00) run_daily(cancel_remaining, time10:05) run_daily(close_report, time15:10) write_file(g.out_daily.csv, date,total_value,cash,shares,position_value,exposure,target,signal_date,reason\n,appendFalse) write_file(g.out_orders.csv, date,order_id,is_buy,requested,filled,price,commission,status\n,appendFalse) write_file(g.out_trades.csv,time,trade_id,order_id,amount,price\n,appendFalse) log.info(CONFIG json.dumps(g.cfg,ensure_asciiTrue,sort_keysTrue)) def position_of(context): 空仓时返回None不依赖不存在持仓对象的默认值。 security g.cfg[security] return context.portfolio.positions[security] if security in context.portfolio.positions else None def is_new_week(today, previous): 按自然周比较前后两个交易日周一休市时周二仍可执行周频规则。 return today.isocalendar()[:2] ! previous.isocalendar()[:2] def prepare_signal(context): 先清理昨日信号再读取已完成日线失败时不沿用旧的买入指令。 g.target, g.signal_day None, None g.reason, g.source_day , today context.current_dt.date() position position_of(context) held int(position.total_amount) if position else 0 if g.pending_exit and held: # 已触发的清仓不依赖新的行情窗口但实际下单仍检查停牌和可卖份额。 g.target, g.signal_day, g.reason 0., today, 完成尚未成交的清仓 return g.pending_exit False if g.cfg[mode] hold: g.target, g.signal_day, g.reason g.cfg[max_weight], today, 买入后一直持有 return if g.cfg[initial_invested] and g.observed_days g.cfg[warmup]: # 保留上市初期的参与机会。历史不足时只完成最初建仓不凭缺失价格择时。 g.target, g.signal_day, g.reason g.cfg[max_weight], today, 上市初期先参与持有 return nondecision (g.cfg[frequency] weekly and not is_new_week(today, context.previous_date)) nondecision nondecision or (g.cfg[frequency] monthly and (today.year,today.month) (context.previous_date.year,context.previous_date.month)) if nondecision: # 这不是新的周频决策只是继续执行上次已经批准且未完成的初始建仓。 if g.entry_remaining is not None and g.cfg[sizing] entry: g.target, g.signal_day, g.reason g.cfg[max_weight], today, 继续最初建仓 return g.reason 本周期非决策日 return try: h attribute_history(g.cfg[security],g.cfg[warmup],1d, [close,high,low,paused],skip_pausedFalse,dfTrue,fqpre) if h is None or len(h) ! g.cfg[warmup]: g.reason 上市后历史不足保留原仓位 return # 聚宽可能以缺失值补足上市前的日线显式归为预热避免误记为数据故障。 if str(h.index[0].date()) g.cfg[listing]: g.reason 上市后有效历史尚不足预热窗口 return if h.index.has_duplicates or not h.index.is_monotonic_increasing: raise ValueError(日线索引重复或无序) if h.index[-1].date() ! context.previous_date: raise ValueError(日线不是上一交易日) prices h[[close,high,low]].values if not np.isfinite(prices).all() or (prices 0).any(): raise ValueError(价格缺失或非正数) # 窗口中停牌日保留其历史价格仅最近一日停牌时暂停新信号避免长期冻结。 if h.paused.isnull().any() or h.paused.iloc[-1] ! 0: raise ValueError(停牌字段异常或上日停牌) target, reason strategy_rule(h, held 0, g.cfg, g.state) if not math.isfinite(target) or not 0 target g.cfg[max_weight]: raise ValueError(目标仓位越界) g.target, g.signal_day float(target), today g.reason, g.source_day reason, str(h.index[-1].date()) if held and target 0: g.pending_exit True if target 0: g.entry_remaining None # 新退出或空仓条件撤销尚未完成的建仓预算。 except Exception as error: # 数据故障不是卖出信号。异常日志可用于检查是否被静默跳过了大量交易日。 g.reason DATA_ERROR str(error) log.warn(g.reason) def limit_price(price, high, low, buying, offset): 按0.001元报价精度计算限价涨停不追买、跌停不假定能卖出。 if not all(math.isfinite(x) and x 0 for x in (price,high,low)): return None if high low or not low price high: return None if buying: if price high-.001: return None return math.floor((min(price*(1offset),high-.001)1e-9)*1000)/1000 if price low.001: return None return math.ceil((max(price*(1-offset),low.001)-1e-9)*1000)/1000 def affordable_amount(budget, price, commission): 预算包含佣金买入必须是100份的整数倍。 if budget 5 or price 0: return 0 return max(0,int((budget-5)/(price*(1commission))/100)*100) def execute_signal(context): 10:00下限价单一次信号最多尝试一次不把委托成功当成全部成交。 today context.current_dt.date() if g.signal_day ! today or g.target is None or g.attempt_day today: return security g.cfg[security] if any(o.security security for o in get_open_orders().values()): return position position_of(context) held int(position.total_amount) if position else 0 entry_only g.cfg[mode] hold or g.cfg[sizing] entry # 完成最初建仓后不因价格波动或现金分红补仓未完成则按原预算继续。 if entry_only and g.target 0 and held and g.entry_remaining is None: return if g.target 0 and held 0: return quote get_current_data()[security] price float(quote.last_price) if quote.paused or not math.isfinite(price) or price 0: return total float(context.portfolio.total_value) exposure held*price/total # 可变仓位模型仅在仓位偏差至少5个百分点时调整降低小额反复交易。 if not entry_only and g.target 0 and abs(exposure-g.target) g.cfg[rebalance_band]: return buying (entry_only and g.target 0) or g.target exposure lp limit_price(price,float(quote.high_limit),float(quote.low_limit),buying,g.cfg[limit_offset]) if lp is None: return if buying: if entry_only: if g.entry_remaining is None: g.entry_remaining total*g.target # 此轮首次建仓预算只确定一次。 budget min(g.entry_remaining,context.portfolio.available_cash) else: budget min(max(0,total*g.target-held*price),context.portfolio.available_cash) amount affordable_amount(budget,lp,g.cfg[commission]) if entry_only and amount 0: g.entry_remaining None # 剩余预算不足100份及费用确认本轮建仓完成。 else: sellable int(position.closeable_amount) if position else 0 # 全清仓允许卖出最后不足一手的剩余份额减仓则按100份整手计算。 desired held if g.target 0 else int(max(0,held-total*g.target/price)/100)*100 amount -min(desired,sellable) if amount 0: return g.attempt_day today log.info(INTENT json.dumps(dict(datestr(today),targetg.target, sourceg.source_day,amountamount,limitlp,reasong.reason),ensure_asciiTrue)) order(security,amount,styleLimitOrderStyle(lp)) def cancel_remaining(context): 撤销本策略未成交委托次日按新信号或未完成清仓状态重新判断。 for item in get_open_orders().values(): if item.security g.cfg[security]: cancel_order(item) def append_csv(suffix, values): 逐行保存账户、订单和成交供独立复算避免仅凭图中曲线下结论。 buffer io.StringIO() csv.writer(buffer).writerow(values) write_file(g.outsuffix,buffer.getvalue(),appendTrue) def close_report(context): 记录完整资金曲线包括预热期和现金期分红和拆分交由引擎入账。 cancel_remaining(context) # Order.price可能只反映首笔成交分批成交须按Trade逐笔累加金额。 # 同时据此计算成交均价避免用首笔价格扣预算或展示为加权均价。 fills {} for t in get_trades().values(): if t.security g.cfg[security]: amount, turnover fills.get(str(t.order_id), (0, 0.)) fills[str(t.order_id)] (amountt.amount, turnovert.amount*t.price) append_csv(_trades.csv,[str(t.time),str(t.trade_id),str(t.order_id),t.amount,t.price]) for o in get_orders().values(): if o.security g.cfg[security]: amount, turnover fills.get(str(o.order_id), (0, 0.)) assert amount o.filled, 订单成交量与逐笔成交不一致 average_price turnover/amount if amount else 0. append_csv(_orders.csv,[str(context.current_dt.date()),str(o.order_id), bool(o.is_buy),o.amount,o.filled,average_price,o.commission,str(o.status)]) if o.is_buy and o.filled 0 and g.entry_remaining is not None: g.entry_remaining max(0.,g.entry_remaining-turnover-o.commission) g.observed_days 1 p context.portfolio position position_of(context) held int(position.total_amount) if position else 0 value float(position.value) if position else 0. if held: g.state[held_days] 1 g.state[armed] False else: g.state[held_days] 0 record(account_navp.total_value/p.starting_cash,exposurevalue/p.total_value) append_csv(_daily.csv,[str(context.current_dt.date()),p.total_value,p.available_cash, held,value,value/p.total_value,g.target,g.source_day,g.reason])红利低波这一篇最值得检验的是干预的必要性。没有触发过的保护规则需要未来新增行情验证不能拿与持有相同的结果证明择时能力。