企业官网建设流程全解析

100万元，策略和买入后一直持有都变成222.67万元。新增的分红再投试验，也没有增加收益。这次最有价值的结果，是看清哪些操作确实没有带来帮助。

首图由聚宽已完成回测的账户数据重绘；下文保留真实平台指标截图。

1、收益与回撤对比

标的为华泰柏瑞中证红利低波动ETF（512890），聚宽代码512890.XSHG。

回测从上市首日2019-01-18到2026-09-16，共1859个交易日，覆盖全部可交易历史。基金成立至上市之间没有场内成交，不能把这段时间虚构为可买入；本轮全部实验统一截至2026年9月16日。

两组起始资金均为100万元，最初买入预算均为90万元。“买入后一直持有”是从首个可成交日起按预算完成建仓，之后一直不动，不主动卖出、不补仓，现金分红留在账户。余款保留现金，不追加本金。

最大回撤是账户从历史高点向后的最大跌幅；收益已计入平台记录的分红、份额调整和费用，年化收益沿用聚宽口径。

本文使用聚宽开发与回测，环境准备可参照基础篇。

2、量化策略的思路与实现

红利低波的第一道问题，是有没有必要频繁操作。本文保留“先持有、明显破位才退出”的规则，先把干预条件写清楚。

高点回落衡量最近一段持有损失，MA60用于判断是否已经离开中期重心；只有两项同时成立才卖出。恢复线用较短的MA20，避免退出后等待完整长趋势才买回来。

完整历史没有发生风险退出，策略与持有账户相同，因此不能把122.67%的收益归功于择时。

这次额外检验“每月将已到账分红再买入同一ETF”，并锁定初次建仓后留下的全部现金储备。结果没有新增买入。现有聚宽账户记录没有现金分红入账，只记录到2021-10-25份额翻倍；份额拆分不会产生可再投现金。这只说明当前回测账本没有支持再投资的现金记录，不能据此断言基金现实中从未分红。

因此，再投规则只作为实验结果披露，正文仍使用已验证的保护规则。涉及分红再投资的进一步结论，应先用基金分红公告与平台账本逐笔核对。

prepare_signal()只读取已经完成的日线；strategy_rule()判断继续持有或退出；execute_signal()提交今天的限价单；close_report()按真实成交核算现金、份额和费用。信号出现不代表一定成交，买回也不保证比卖出价更低。

完整历史有成交订单3笔，累计佣金274.88元。订单数量包含建仓和退出的分批成交，不等于买卖轮数。

改变条件后，收益还能留下多少

后三行来自同一区间此前已完成的原保护规则检验。退出门槛从16%收紧为12%后，收益反而下降、回撤扩大；放宽到20%仍与持有相同。这进一步说明，当前历史没有支持主动干预的稳定优势。

滚动252交易日收益超过持有的窗口占0.00%；按不重叠的252交易日分段，胜出比例为0.00%。滚动窗口相互重叠，不是不相关的独立样本。

以上均沿用原账户，不在分段起点清仓或重新投入100万元；末段不足252交易日的部分不纳入分段胜出比例。

当前证据还不足以称为“稳定高收益”。本轮规则是在已经查看过的历史上修订的，此前也进行过多轮研究；全历史、分段、成本检验均属于回顾性验证。下一步应冻结规则，用尚未发生的行情做模拟跟踪，记录每次信号、未成交和实际成本，再判断能否投入实盘。

3、在聚宽复现回测

1. 在“我的策略”中新建Python 3空白策略，粘贴下一节全部代码。
2. 回测起点设为2019-01-18，终点设为2026-09-16，初始资金1000000元，运行频率选分钟。
3. 检查代码中的ETF代码512890.XSHG，保持主参数和费用不变，运行到显示“回测完成”。
4. 在“收益概述”核对收益、回撤，在“交易详情”核对成交，在“每日持仓&收益”核对资产。平台的“基准收益”是沪深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。 # mode="hold"为同ETF买入后一直持有；初始资金100万元，首次预算90%。 # initial_invested=True表示上市后先建仓，攒足历史再应用退出规则。 # 双倍成本及邻近参数是独立对照，不用于自动寻找最高收益。 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*(1+p['band']), fast < slow*(1-p['band']) elif family in ('price_band', 'monthly', 'invested_trend'): # 收盘越过均线上沿才买，跌破下沿才卖；中间区域不追涨杀跌。 mean = c[-p['slow']:].mean() buy, sell = c[-1] > mean*(1+p['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日涨跌：需要slow+1个收盘价；不是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_tax=0, close_tax=0, open_commission=g.cfg['commission'], close_commission=g.cfg['commission'], close_today_commission=0, min_commission=5), type='fund', ref=g.cfg['security']) # 此API参数表示买卖总价差，因此单边0.1%需要传0.2%。 set_slippage(PriceRelatedSlippage(2*g.cfg['slippage']),type='fund',ref=g.cfg['security']) run_daily(prepare_signal, time='09:05') run_daily(execute_signal, time='10:00') run_daily(cancel_remaining, time='10:05') run_daily(close_report, time='15:10') write_file(g.out+'_daily.csv', 'date,total_value,cash,shares,position_value,exposure,target,signal_date,reason\n',append=False) write_file(g.out+'_orders.csv', 'date,order_id,is_buy,requested,filled,price,commission,status\n',append=False) write_file(g.out+'_trades.csv','time,trade_id,order_id,amount,price\n',append=False) log.info('CONFIG '+json.dumps(g.cfg,ensure_ascii=True,sort_keys=True)) 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_paused=False,df=True,fq='pre') 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*(1+offset),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*(1+commission))/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(date=str(today),target=g.target, source=g.source_day,amount=amount,limit=lp,reason=g.reason),ensure_ascii=True)) order(security,amount,style=LimitOrderStyle(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.out+suffix,buffer.getvalue(),append=True) 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)] = (amount+t.amount, turnover+t.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_nav=p.total_value/p.starting_cash,exposure=value/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])

红利低波这一篇最值得检验的是干预的必要性。没有触发过的保护规则，需要未来新增行情验证，不能拿与持有相同的结果证明择时能力。