军工ETF量化策略

100万元策略期末156.60万元买入后一直持有期末112.00万元。策略比买入后一直持有多约44.60万元最大回撤分别为17.38%、47.33%。收益优势是否值得采用还要看下文成本与邻近参数的结果。首图由聚宽完成回测后的真实账户数据重绘采用无损高清图不是平台界面截图。1、收益与回撤对比本篇标的是国泰中证军工ETF512660聚宽代码512660.XSHG。回测覆盖2016-08-08至2026-09-16共2456个交易日为该ETF上市以来的全部可交易历史。基金成立早于上市成立后尚未上市的日期没有场内成交不能虚构买入。本篇“至今”统一截至2026年9月16日。指标量化策略买入后一直持有期末资产 / 万元156.60112.00累计收益56.60%12.00%年化收益4.67%1.16%最大回撤17.38%47.33%平均实际仓位18.16%89.54%有成交的订单1083两组均投入100万元初次入场预算为总资产的90%使用相同费用和撮合限制。“买入后一直持有”就是从首个可成交日起完成建仓此后一直不动不主动卖出也不补仓现金分红留在账户。它不等于100%满仓或分红即时再投资。收益包含现金、持仓、分红及已扣费用。最大回撤是账户从历史高点向后的最大跌幅年化收益沿用聚宽口径订单可分批成交不等于完整买卖轮数。图表根据真实账户数据重绘。本文使用聚宽开发与回测环境准备可参照基础篇。2、量化策略的思路与实现军工ETF安静一段时间后突然突破能不能形成一套规则先要把“安静”变成可计算的指标。布林带宽等于4倍20日价格标准差除以20日均价。先用更早的120个带宽值计算30%分位阈值再检查突破前最近10个交易日有没有出现低于这个阈值的带宽。只有先收敛、随后昨收突破此前20日最高价才允许入场。环节执行规则历史尺度用更早120个带宽值的30%分位数定义相对收敛最近收敛突破日之前10个交易日至少一天带宽低于阈值向上突破昨收盘高于此前20个交易日最高价退出持有后昨收盘跌破MA20计算分位数时排除最近10个待检查的带宽值也排除昨日突破日。这样顺序是“先有历史尺度再观察收敛最后检查突破”。阈值来自历史分布不需要用完整回测区间的分位数倒填到过去。prepare_signal()先读取昨日及更早的已完成行情strategy_rule()输出目标仓位execute_signal()在今天提交限价单close_report()记录实际成交。昨天的信号价不是今天保证能成交的价格。实际交易记录完整历史中有成交的订单共108笔完成清仓的持仓阶段29段累计佣金18440.37元。期末仍持有ETF。完整持仓阶段日期账户变化最好的一段2020-07-03 至 2020-08-2431.57% / 27.17万元最差的一段2020-01-06 至 2020-02-04-8.67% / -8.17万元以上从入场前一日算到完全退出包含期间费用、分红和调仓不是单笔价差。均价来自聚宽逐笔成交的加权汇总佣金来自订单记录。展示对应阶段首尾交易没有完成清仓时展示最早几笔实际成交。成本与参数检验检验累计收益最大回撤本文策略56.60%17.38%佣金与滑点翻倍47.27%18.11%取消收敛前提只按20日突破入场18.93%28.20%带宽分位改为20%68.19%16.21%带宽分位改为40%21.57%25.03%双倍成本后累计收益为47.27%两组邻近参数为68.19%、21.57%。三项检验的终值都高于原持有对照。双倍成本将买卖佣金费率各从0.03%提高到0.06%最低佣金仍为5元单边滑点从0.10%提高到0.20%限价偏移同步从0.2%扩大到0.4%。成交路径可能改变这属于执行压力检验。邻近检验只改一个参数没有因为邻近参数赚得更多就把它换成正文主策略。下面按时间顺序拆分同一个连续账户检查收益是否集中在某一阶段属于回顾性归因。连续账户区间本文策略一直持有2016-08-08 至 2023-08-3118.51%1.80%2023-09-01 至 2026-09-1632.14%10.02%按交易日顺序分前70%与后30%后段继承原持仓和资金未重新起投。两段收益要连乘不能相加。军工篇最值得学习的是滚动分位数的时间边界。一次性对全历史带宽求分位数后回测会让早期决策知道未来分布得到的结果不应作为实战证据。3、在聚宽复现回测进入策略回测 → 新建策略 → 空白模版粘贴第4节完整代码保存后设置界面设置填写内容起止日期2016-08-08 至 2026-09-16初始资金1000000元回测频率分钟运行环境Python3策略模式mode: strategy上图为对应主策略的聚宽原始回测截图包含回测区间、初始资金、完成状态、收益指标及完整收益曲线未重写平台数字。图中基准线为沪深300正文“买入后一直持有”对照为另行回测的同一ETF账户两者口径不同。程序09:05读昨日日线10:00提交限价单10:05撤销余单15:10记录账户。每天检查一次策略条件。本策略需要161个完整日线预热历史不足时保留现金预热期仍计入全历史收益。代码已设定佣金各0.03%、每笔最低5元、基金印花税为0、单边滑点0.10%按当时分钟成交量1%限制撮合。买入按100份整手提交实际部分成交可能不是整手未完成建仓使用原预算续单按真实成交金额和佣金扣减未完成卖出继续处理可卖份额。不要把委托成功当成全部成交。要复现对照再复制一份代码只修改配置中的两项其余资金、区间、费用不变mode: hold, experiment: ETF17C_14_hold_compare,运行完成后核对“收益概述”的累计收益与回撤再查看“交易详情”“每日持仓收益”。代码会同时保存每日账户、订单和逐笔成交三类记录文件名前缀由experiment决定改参数重跑时换一个前缀便于核对每次结果。每日记录的cash是可用现金暂时冻结的资金可能不在其中核对总资产时应使用平台账面现金加持仓市值不能只加可用现金。这是包含费用与成交约束的历史模拟尚未接入券商实盘历史回撤不是未来亏损上限。4、完整策略代码复制到聚宽空白策略即可运行。中文注释覆盖参数、信号、预算、限价和真实成交核对。# -*- coding: utf-8 -*- ETF单品种策略。聚宽Python3分钟频率初始资金100万元。 军工ETF量化策略波动收敛后等待向上突破。 所有交易信号仅使用前一交易日及更早的数据真实成交由聚宽撮合。 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可运行买入后一直持有对照其他参数保持一致。 # frequency是决策频率回测界面仍选择分钟。params为本文策略指标参数。 CONFIG {security: 512660.XSHG, listing: 2016-08-08, mode: strategy, params: {channel: 20, band_window: 20, rank_window: 120, quantile: 0.3, recent: 10, exit: 20}, warmup: 161, frequency: daily, sizing: entry, max_weight: 0.9, commission: 0.0003, slippage: 0.001, limit_offset: 0.002, rebalance_band: 0.05, ablation: False, experiment: ETF17C_14_main} def strategy_rule(h, held, cfg, state): 先收敛再突破历史带宽阈值排除最近待判断的收敛窗口。 p, c cfg[params], h.close.values if held: return (0. if c[-1] c[-p[exit]:].mean() else cfg[max_weight]), 跌回短期均线 # 布林带宽(上轨-下轨)/中轨上下轨分别是均值加减2倍总体标准差。 widths np.array([4*c[i-p[band_window]:i].std(ddof0) /c[i-p[band_window]:i].mean() for i in range(p[band_window],len(c))]) past widths[-p[rank_window]-p[recent]:-p[recent]] # 聚宽旧版NumPy没有quantile百分位写法与同一线性插值的分位数等价。 threshold float(np.percentile(past,100*p[quantile])) was_quiet widths[-p[recent]:].min() threshold breakout c[-1] h.high.values[-p[channel]-1:-1].max() return (cfg[max_weight] if breakout and (cfg[ablation] or was_quiet) else 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.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[frequency] weekly and not is_new_week(today, context.previous_date): # 这不是新的周频决策只是继续执行上次已经批准且未完成的初始建仓。 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) 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])军工这一篇将“先收敛、后突破”拆成三个时间明确的步骤。它能否改善交易需要和普通突破规则比较形态越复杂越应该检查每个新增条件有没有实际贡献。