创业板ETF量化策略
2026-10-242
100万元，策略期末606.73万元；买入后一直持有，期末379.55万元。
策略比买入后一直持有多约227.18万元，最大回撤分别为55.50%、67.96%。收益优势是否值得采用，还要看下文成本与邻近参数的结果。
首图由聚宽完成回测后的真实账户数据重绘，采用无损高清图；不是平台界面截图。
1、收益与回撤对比
本篇标的是易方达创业板ETF（159915），聚宽代码159915.XSHE。
回测覆盖2011-12-09至2026-09-16，共3588个交易日，为该ETF上市以来的全部可交易历史。 基金成立早于上市，成立后尚未上市的日期没有场内成交，不能虚构买入。本篇“至今”统一截至2026年9月16日。
| 指标 | 量化策略 | 买入后一直持有 |
|---|---|---|
| 期末资产 / 万元 | 606.73 | 379.55 |
| 累计收益 | 506.73% | 279.55% |
| 年化收益 | 13.39% | 9.74% |
| 最大回撤 | 55.50% | 67.96% |
| 平均实际仓位 | 46.78% | 94.93% |
| 有成交的订单 | 10 | 2 |
两组均投入100万元，初次入场预算为总资产的90%，使用相同费用和撮合限制。“买入后一直持有”就是从首个可成交日起完成建仓，此后一直不动，不主动卖出，也不补仓；现金分红留在账户。 它不等于100%满仓或分红即时再投资。
收益包含现金、持仓、分红及已扣费用。最大回撤是账户从历史高点向后的最大跌幅，年化收益沿用聚宽口径；订单可分批成交，不等于完整买卖轮数。图表根据真实账户数据重绘。
本文使用聚宽开发与回测，环境准备可参照基础篇。
2、量化策略的思路与实现
强势行情里，“涨多了就卖”可能恰好卖掉最赚钱的一段。创业板策略首先需要解决这个问题。
旧版既等回撤又等恢复，买入后还可能因RSI过热卖出，平均仓位约14%。新版取消短期超买、超卖条件，用120日与250日均线跟随更长的行情。
| 同区间方案 | 累计收益 | 最大回撤 |
|---|---|---|
| 趋势内RSI回撤恢复 | 30.33% | 25.19% |
| 本文修订规则 | 506.73% | 55.50% |
| 买入后一直持有 | 279.55% | 67.96% |
规则如何进入代码：
| 环节 | 执行规则 |
|---|---|
| 预热 | 先积累250个已完成交易日；此前保留现金 |
| 入场 | MA120 > MA250 × 1.01 |
| 退出 | MA120 < MA250 × 0.99 |
| 执行节奏 | 日频检查，缓冲区内维持现有份额 |
短均线代表较近的平均成本，长均线代表更长的价格重心。两者拉开后才确认方向，可以避免因为几天上涨就判断过热。代价也很明确：慢信号不能迅速躲开急跌，长期持仓仍可能承受较大回撤。
prepare_signal()先读取昨日及更早的已完成行情，strategy_rule()输出目标仓位，execute_signal()在今天提交限价单，close_report()记录实际成交。昨天的信号价，不是今天保证能成交的价格。
实际交易记录
完整历史中，有成交的订单共10笔，完成清仓的持仓阶段2段，累计佣金3785.58元。期末仍持有ETF。
| 完整持仓阶段 | 日期 | 账户变化 |
|---|---|---|
| 最好的一段 | 2019-06-11 至 2022-04-01 | 76.24% / +174.59万元 |
| 最差的一段 | 2013-03-29 至 2016-01-08 | 129.02% / +129.02万元 |
以上从入场前一日算到完全退出，包含期间费用、分红和调仓；不是单笔价差。
均价来自聚宽逐笔成交的加权汇总，佣金来自订单记录。展示对应阶段首尾交易；没有完成清仓时，展示最早几笔实际成交。
成本与参数检验
| 检验 | 累计收益 | 最大回撤 |
|---|---|---|
| 本文策略 | 506.73% | 55.50% |
| 佣金与滑点翻倍 | 502.76% | 55.50% |
| 长窗口改为200日 | 435.00% | 55.53% |
| 长窗口改为300日 | 428.15% | 55.50% |
双倍成本后，累计收益为502.76%；两组邻近参数为435.00%、428.15%。三项检验的终值都高于原持有对照。
双倍成本将买卖佣金费率各从0.03%提高到0.06%，最低佣金仍为5元；单边滑点从0.10%提高到0.20%，限价偏移同步从0.2%扩大到0.4%。成交路径可能改变，这属于执行压力检验。邻近检验只改一个参数；没有因为邻近参数赚得更多，就把它换成正文主策略。
这次属于看过历史后的策略修订。 先后研究了34组规则与参数，再选定本篇方案到聚宽复核。成本和邻近参数检验能发现脆弱性，却不能消除历史选择偏差。下面的后30%也已被研究过程查看，不能称为独立样本外。
| 连续账户区间 | 本文策略 | 一直持有 |
|---|---|---|
| 2011-12-09 至 2022-04-12 | 303.61% | 180.78% |
| 2022-04-13 至 2026-09-16 | 50.33% | 35.18% |
按交易日顺序分前70%与后30%；后段继承原持仓和资金，未重新起投。两段收益要连乘，不能相加。
上市初期的250日现金期保留在总收益中。它与“先持有再择时”的策略不同，复现时不能自行把预热期删掉或换成买入。
3、在聚宽复现回测
进入策略回测 → 新建策略 → 空白模版，粘贴第4节完整代码，保存后设置：
| 界面设置 | 填写内容 |
|---|---|
| 起止日期 | 2011-12-09 至 2026-09-16 |
| 初始资金 | 1000000元 |
| 回测频率 | 分钟 |
| 运行环境 | Python3 |
| 策略模式 | 'mode': 'strategy' |
上图为对应主策略的聚宽原始回测截图，包含回测区间、初始资金、完成状态、收益指标及完整收益曲线，未重写平台数字。图中基准线为沪深300；正文“买入后一直持有”对照为另行回测的同一ETF账户，两者口径不同。
程序09:05读昨日日线，10:00提交限价单，10:05撤销余单，15:10记录账户。每天检查一次策略条件。
本策略需要250个完整日线预热，历史不足时保留现金；预热期仍计入全历史收益。
代码已设定佣金各0.03%、每笔最低5元、基金印花税为0、单边滑点0.10%，按当时分钟成交量1%限制撮合。买入按100份整手提交，实际部分成交可能不是整手；未完成建仓使用原预算续单，按真实成交金额和佣金扣减，未完成卖出继续处理可卖份额。不要把委托成功当成全部成交。
要复现对照，再复制一份代码，只修改配置中的两项，其余资金、区间、费用不变：
'mode': 'hold',
'experiment': 'ETF17D_06_hold_compare',
运行完成后，核对“收益概述”的累计收益与回撤，再查看“交易详情”“每日持仓&收益”。代码会同时保存每日账户、订单和逐笔成交三 类 记录，文件名前缀由experiment决定；改参数重跑时换一个前缀，便于核对每次结果。
每日记录的cash是可用现金，暂时冻结的资金可能不在其中；核对总资产时，应使用平台账面现金加持仓市值，不能只加可用现金。
这是包含费用与成交约束的历史模拟，尚未接入券商实盘；历史回撤不是未来亏损上限。
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': '159915.XSHE',
'listing': '2011-12-09',
'mode': 'strategy',
'params': {'family': 'dual_ma', 'fast': 120, 'slow': 250, 'band': 0.01, 'frequency': 'daily'},
'warmup': 250,
'frequency': 'daily',
'sizing': 'entry',
'initial_invested': False,
'max_weight': 0.9,
'commission': 0.0003,
'slippage': 0.001,
'limit_offset': 0.002,
'rebalance_band': 0.05,
'ablation': False,
'experiment': 'ETF17D_06_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])
创业板这一篇不再用短期超买条件截断趋势。更高参与度是否带来值得承担的收益，需要连同最大回撤和成本检验一起判断。
