沪深300ETF量化策略

首图为聚宽真实回测截图，经裁剪拼接。蓝线是本文策略，红线是沪深300指数；红线不代表下文“买入后一直持有”的ETF账户。

同样100万元投资沪深300ETF，从上市首日算起，买入后一直持有，期末约204.40万元；按本文规则交易，期末约306.97万元。

相差约102.57万元，最大回撤从42.15%降到25.22%。 这套策略只交易一只ETF，全程19笔成交订单。差别来自三个决定：什么时候持有，什么时候退出，退出后什么时候回来。

先说清比较条件：两组初始资金都是100万元，每次入场均以总资产的90%作为预算，含佣金，费用与成交限制相同。“买入后一直持有”从首个可成交日尝试买入，此后不卖出、不补仓，现金分红留在账户，不再投入。因此，本文对照的是这一具体持有方式，不等同于满仓且分红再投资的收益。

1、收益与回撤对比

标的是华泰柏瑞沪深300ETF（510300），聚宽代码510300.XSHG。

回测区间为2012年5月28日至2026年9月9日，共3474个交易日，覆盖上市以来全部可交易历史。 基金成立于2012年5月4日，5月28日上市；上市前没有场内成交数据。“至今”以本次回测截止日2026年9月9日为准。

年化收益采用聚宽展示口径；收益已计入设定的佣金和滑点。策略的均线预热期、空仓期和亏损年份全部计入。

收益之外，还要看持有过程。最大回撤是账户从历史高点到随后低点的最大跌幅：以某次高点100万元举例，回撤42.15%后只剩57.85万元。它不是相对初始本金的亏损。

以上两图由聚宽导出的每日账户数据重绘，均包含现金、持仓与设定费用，比较的是两个ETF账户。

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

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

沪深300ETF的长期收益并不是匀速积累的，上涨中也会夹着回撤。这套策略试图保留持续上涨中的持仓，在趋势转弱或急跌时退出，等条件恢复再参与。避开的下跌要足以弥补错过的上涨和交易费用，择时才有收益价值。

趋势向上时持有

MA120、MA250分别是最近120个、250个交易日的平均收盘价，约对应半年和一年。

MA120高于MA250，允许入场；低于MA250，退出持仓。 两线相等时不产生新的趋势信号，急跌检查仍然有效。程序每天比较均线位置，不要求当天刚好交叉；已有持仓时不会因为重复出现买入条件而补仓。

较长的均线窗口用来过滤短期波动，代价是入场和离场都会慢一些。上市初期不足250个有效日线时，账户保持现金。

急跌时提前退出

长期均线还向上，价格却已经快速下跌，这时等待年线转弱可能太迟。以下两项同时满足，触发急跌退出：

• 昨收盘价低于60日均价的92%，即向下偏离超过8%。
• 最近20个日对数收益的年化波动率超过40%。

前一项检查跌得有多深，后一项检查近期波动是否明显放大。波动率由21个收盘价生成20个对数收益，计算总体标准差后乘以√252年化。

8%不是从买入价计算的止损线，40%也不是账户亏损上限。 信号发出后仍要等到下单时刻，跳空与成交限制都会影响退出价格。

修复后重新入场

急跌卖出后，程序用g.defensive_exit记住等待状态。再次买入需要同时满足：MA120高于MA250、昨收盘价高于MA20的101%、急跌条件已经解除。

这一步让程序等价格修复后再参与，但无法保证反弹一定持续。等待期间若长期趋势转空，就结束这次修复等待，之后重新按长期趋势判断。空仓时出现急跌条件，同样不能买入。

代码按prepare_signal()读取昨日日线、classify()计算指标、decide()判断状态、execute_signal()提交委托的顺序运行。计算信号只用前一交易日及更早的数据，均线与波动率使用同一时点的前复权价格；具体下单时间与成交限制见第3节。

看两轮真实买卖

2014年10月17日，趋势确认后买入。 前一天的MA120约2.2854，高于MA250的2.2846，策略以2.494元成交353600份。此后保持持仓，参与了这一轮上涨。

2015年6月29日，急跌保护触发退出。 此时MA120仍高于MA250，长期趋势尚未转空，策略已按保护规则以4.356元卖出全部353600份。账户从入场前约98.11万元，变为退出后约165.12万元，这一轮净增加约67.01万元。

这67.01万元包含持有期间的现金分红，并已扣除买卖佣金；成交价已受设定滑点影响。它是这一轮账户增加的金额，不是整段回测收益，也不是卖在最高点的结果。

接下来的一次重新入场就失败了。 2015年7月24日，修复条件满足，策略以4.300元买入345100份；7月27日又触发急跌退出，以4.169元卖出。扣除两笔佣金后，这一轮净亏约4.61万元，账户从约165.12万元降到160.51万元。

两轮交易说明了同一件事：规则能让程序按条件退出，但修复信号也可能失败。收益要靠完整历史中的多轮交易累积，不能只看盈利案例。本次共完成9轮买卖，期末仍有持仓。

参数有依据，也要经得起检验

120/250日均线用来观察半年与一年趋势，60日和20日窗口分别观察季度偏离与月度修复。8%、40%、1%是本次实验设定的门槛，没有证据表明它们是最优值。

本次已经完成三组可核对的比较：

加入保护后，这段历史的收益提高、回撤下降，说明整套退出与重新入场机制有继续研究的价值；这张表不能分离出某个阈值的单独贡献。

双倍成本将买卖佣金各从0.03%提高到0.06%，单边滑点从0.10%提高到0.20%；限价偏移上限同步从0.2%扩大到0.4%，以容纳更高滑点。

将历史分为2012-05-28至2018-12-31、2019-01-01至2026-09-09，两段分别以100万元起投，结果如下：

后段使用此前已公开行情预热均线；两段独立起投，收益不能直接相加。

两段领先不代表每年都领先。全历史账户在2019年收益约13.30%，买入后一直持有约32.05%；策略在2018年、2021年和2023年也有年度亏损。

这些规则是在看过历史行情与前期回测后比较选择的，分段复核属于回顾性检查，不是独立样本外验证。相邻参数尚未完成逐项回测，例如将8%改为7%或9%、40%改为35%或45%、1%改为0.5%或1.5%，结果是否仍然接近，目前不能下结论。成本测试不能替代这项检验，新增行情也仍需持续 跟踪 。

3、在聚宽复现回测

复现本文需要分别运行两次回测：一次运行量化策略，一次运行“买入后一直持有”。两次使用相同的日期、资金和费用，才能比较择时规则带来的差异。

先运行本文策略

在聚宽进入策略回测 → 新建策略 → 空白模版，将编辑器中的 示例代码 替换为第4节的完整代码，保存后填写：

保留代码中CONFIG的默认设置，确认mode为'trend'、protection为True，点击“运行回测”，等待状态变为“回测完成”。

这里有两个容易混淆的设置：初始资金要在界面填写，代码里的initial_cash只做校验；回测频率选“分钟”，因为程序09:05根据昨日日线生成信号，10:00提交限价单，10:05撤销余单。使用分钟引擎，是为了模拟这些执行时刻。

仓位与费用已写入代码：每次入场预算为总资产的90%，含佣金；买卖佣金各0.03%、每笔最低5元，ETF印花税为0，单边滑点0.10%。限价偏移上限0.2%，撮合量限制1%，买入按100份整手计算，卖出遵守T+1可卖数量。复现时保持这些设置一致。

再运行买入后一直持有

另建一个策略，命名为“沪深300ETF买入后一直持有”，粘贴同一份完整代码。只在开头的CONFIG中修改以下两项：

'mode': 'hold',
'experiment': 'HS300_V2_FINAL_HOLD',

mode切换交易方式，experiment用于区分两次回测自动生成的数据记录，避免覆盖前一次结果。其余代码、日期、资金和回测频率均保持相同，再点击“运行回测”。

这一账户从首个可成交日尝试买入，此后不卖出、不补仓，现金分红留在账户。应将它的回测结果与本文策略比较，首图中的沪深300指数红线不能代替这个账户。

核对收益和成交

两次回测完成后，按下面顺序检查：

• 收益概述：确认起止日期、100万元本金和完成状态，再对照第1节的累计收益、期末资产、年化收益与最大回撤。
• 交易详情：本文策略有19笔成交订单，对照账户有1笔买入订单。找到2014年10月17日的买入、2015年6月29日的卖出，与第2节的数量和价格核对。
• 每日持仓&收益：确认策略退出后持仓归零，对照账户持续持仓。需要追查某笔交易时，再查看日志中的SIGNAL（信号）、ORDER（委托）和FILL（成交）。

策略上市初期需要等待足够的均线数据，这段时间没有交易是正常的，仍计入完整回测。结果不一致时，先检查日期、分钟频率、mode和费用设置；尤其要保留本文的截止日，延长区间后的收益不能再与表中数字直接核对。

本次已用每日账户数据独立复算收益与回撤，并核对成交记录。代码已完成聚宽回测，尚未接入券商实盘；25.22%的历史最大回撤也不是未来亏损上限。

4、完整策略代码

将以下完整代码复制到聚宽策略编辑器，按第3节设置日期、资金和回测频率后即可运行。代码包含参数配置、信号判断、下单与记录的中文注释，默认运行本文量化策略；测试“买入后一直持有”时，按第3节修改CONFIG中的两项设置。

# -*- coding: utf-8 -*-
"""沪深300ETF量化策略：聚宽Python 3策略编辑器使用。

回测界面设置：初始资金100万元，分钟频率，开始日期2012-05-28；
本篇结束日期为2026-09-09，后续复现时按平台完整数据日填写。
代码中的initial_cash用于校验界面资金，不会替你修改回测设置。

trend模式：MA120 > MA250时允许入场；急跌退出后等待价格修复。
hold模式：从首个可成交日尝试买入，持有后不主动卖出或补仓。
两种模式分别运行，均使用90%入场预算及相同的费用、成交限制。

本策略用于教学研究，未接入券商实盘，历史收益不保证未来收益。
"""

from jqdata import *  # 聚宽提供的行情、回测、下单及文件接口。
# 显式使用Python内置函数，避免平台导入的NumPy同名函数误判生成器。
from builtins import all, any, max, min
import csv
import io
import json
import math
import numpy as np


CONFIG = {
    'security': '510300.XSHG',  # 华泰柏瑞沪深300ETF，XSHG表示上海市场。
    'initial_cash': 1000000,   # 回测界面的初始资金必须同为100万元。
    'mode': 'trend',           # 主实验用trend；买入持有对照改为hold。
    'fast': 120,               # 半年趋势：最近120个已完成交易日。
    'slow': 250,               # 年线趋势：最近250个已完成交易日。
    'protection': True,        # 正式版开启急跌保护；设False可复现消融对照。
    'crash_window': 60,        # 急跌参考均线。
    'crash_band': 0.08,        # 昨收低于MA60的92%，且波动放大才触发保护。
    'vol_window': 20,          # 用20个日对数收益计算波动率。
    'vol_threshold': 0.40,     # 年化波动率超过40%，不是预测下一年波动。
    'reentry_window': 20,      # 保护退出后，用短期价格修复确认再入场。
    'reentry_band': 0.01,      # 昨收重新高于MA20的101%。
    'entry_weight': 0.90,      # 空仓入场时的含佣金预算比例，不是每日目标仓位。
    'commission': 0.0003,      # 买、卖各收成交额的万分之三。
    'min_commission': 5.0,     # 每笔最低佣金5元，须按实际账户费率核对。
    'slippage_one_way': 0.001, # 每个方向千分之一滑点，交给回测引擎处理。
    'limit_offset': 0.002,     # 限价最多偏离参考价0.2%，不是额外扣收的费用。
    'experiment': 'HS300_V2_FINAL', # 不同区间或参数须更换编号，避免覆盖。
}


def initialize(context):
    """初始化参数、调度与输出文件；context是聚宽传入的当前账户和时间。"""
    # g由聚宽提供，用于在各回调之间保存状态。复制参数后统一通过g.cfg读取。
    g.cfg = dict(CONFIG)
    assert g.cfg['mode'] in ('trend', 'hold')
    assert 1 < g.cfg['fast'] < g.cfg['slow']
    assert 1 < g.cfg['crash_window'] <= g.cfg['slow']
    assert 1 < g.cfg['reentry_window'] <= g.cfg['slow']
    assert 1 < g.cfg['vol_window'] < g.cfg['slow']
    assert 0 < g.cfg['entry_weight'] < 1
    assert abs(context.portfolio.starting_cash - g.cfg['initial_cash']) < 0.01, \
        'Set initial backtest cash to 1000000 CNY.'
    g.signal = None          # BUY买入、EXIT退出、HOLD维持；None表示没有有效信号。
    g.signal_day = None      # 信号在哪一天生成，用于阻止过期信号下单。
    g.attempt_day = None     # 最近一次尝试提交订单的日期，限制同日重复委托。
    g.signal_values = {}     # 保存昨收盘价、两条均线、行情日期，供日志和导出使用。
    g.defensive_exit = False # 是否因急跌退出，决定是否需要恢复确认。
    g.exit_pending = False   # 卖出未完全完成时，后续交易日继续退出。
    g.output = g.cfg['experiment'] + '_' + g.cfg['mode']

    # 沪深300指数作为图中参考线，避免ETF上市首日缺少前收盘价的问题。
    # 指数不是可交易账户；正式对照使用另一次hold回测的扣费后账户资产。
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    set_option('avoid_future_data', True)
    # 对撮合成交量施加1%的限制，可能导致部分成交；不能理解为实盘容量保证。
    set_option('order_volume_ratio', 0.01)
    # 对指定ETF设置基金交易费用。场内ETF不按股票收取印花税。
    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=g.cfg['min_commission']),
        type='fund', ref=g.cfg['security'])
    # 聚宽参数表示买卖价差，单边0.001应传入0.002，不能少算一半滑点。
    set_slippage(PriceRelatedSlippage(
        2 * g.cfg['slippage_one_way']),
        type='fund', ref=g.cfg['security'])
    # 日线用于判断趋势，分钟引擎用于模拟10:00委托和10:05撤单。
    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')
    # 回测运行时，在聚宽研究区自动创建三个CSV文件；同编号、同模式重跑会覆盖。
    # 日账户用于收益统计；订单保留方向和佣金；成交保留每次实际撮合的数量与价格。
    write_file(g.output + '_daily.csv',
               'date,total_value,cash,shares,position_value,exposure,'
               'signal,signal_date,close,ma_fast,ma_slow\n', append=False)
    write_file(g.output + '_orders.csv',
               'date,order_id,is_buy,requested,filled,average_price,'
               'commission,status\n', append=False)
    write_file(g.output + '_trades.csv',
               'time,trade_id,order_id,amount,price\n', append=False)
    emit(context, 'CONFIG', config=g.cfg)


def emit(context, event, **fields):
    """输出带时间和实验编号的JSON日志，便于追溯信号、委托与成交。"""
    item = {'time': str(context.current_dt), 'event': event,
            'experiment': g.output, 'security': g.cfg['security']}
    item.update(fields)
    log.info(json.dumps(item, ensure_ascii=True, sort_keys=True))


def classify(closes, fast, slow, cfg=None):
    """把按日期升序的收盘价序列转为信号；无效数据返回None，不触发卖出。

    closes须恰好包含slow个已完成日线收盘价。
    返回字典同时保留均线值，以便对照图表检查信号原因。
    """
    values = np.asarray(closes, dtype=float)
    if len(values) != slow or not np.isfinite(values).all():
        return None
    if (values <= 0).any():
        return None
    # 切片[-fast:]只取最近fast条；整个窗口恰好是slow条。
    last = float(values[-1])
    ma_fast = float(values[-fast:].mean())
    ma_slow = float(values.mean())
    # 长均线决定市场方向。短期跌破均线不再自动卖出，减少震荡交易。
    if ma_fast > ma_slow:
        signal = 'BUY'
    elif ma_fast < ma_slow:
        signal = 'EXIT'
    else:
        # HOLD表示两个分支都未触发；不是“出现任意相等就一律不交易”。
        signal = 'HOLD'
    result = {'signal': signal, 'close': last,
              'ma_fast': ma_fast, 'ma_slow': ma_slow}
    if cfg is not None:
        crash_ma = float(values[-cfg['crash_window']:].mean())
        reentry_ma = float(values[-cfg['reentry_window']:].mean())
        returns = np.diff(np.log(values[-(cfg['vol_window'] + 1):]))
        volatility = float(np.std(returns, ddof=0) * np.sqrt(252))
        result.update(
            volatility=volatility,
            crash=bool(cfg['protection']
                       and last < crash_ma * (1 - cfg['crash_band'])
                       and volatility > cfg['vol_threshold']),
            recovered=bool(last > reentry_ma * (1 + cfg['reentry_band'])))
    return result


def decide(values, held, defensive, pending):
    """纯状态转换：返回今日信号、是否等待修复、是否仍需完成卖出。"""
    trend = values['signal']
    if held:
        # 待完成卖出优先，避免部分成交后行情反弹导致残余仓位被遗忘。
        if pending or trend == 'EXIT' or values.get('crash', False):
            defensive = bool(trend != 'EXIT'
                             and (defensive or values.get('crash', False)))
            return 'EXIT', defensive, True
        return 'HOLD', defensive, False
    # 确认实际空仓后，才解除待卖出状态。趋势已转空也重置急跌状态。
    if trend == 'EXIT':
        return 'HOLD', False, False
    if values.get('crash', False):
        return 'HOLD', True, False
    if trend == 'BUY' and (not defensive or values.get('recovered', False)):
        return 'BUY', False, False
    return 'HOLD', defensive, False


def prepare_signal(context):
    """每天09:05生成信号，只使用截至前一交易日的数据。"""
    # 先清空昨天的信号。读取失败时保留None，避免把旧信号误当成今日指令。
    g.signal = None
    g.signal_day = None
    g.signal_values = {}
    security = g.cfg['security']
    if g.cfg['mode'] == 'hold':
        # 对照组不需要均线预热，从上市首日即可尝试买入。
        # 每天给出BUY不等于每天买：执行函数会拦截已有持仓的买入。
        g.signal = 'BUY'
        g.signal_day = context.current_dt.date()
        emit(context, 'SIGNAL', signal='BUY', source='buy_and_hold_control')
        return
    try:
        # attribute_history的日线不包含当天。前复权配合use_real_price，
        # 使均线使用相同时点的动态复权价格，委托仍依据当时的可交易价格。
        # skip_paused=False保留停牌日，随后显式检查paused字段。
        history = attribute_history(
            security, g.cfg['slow'], '1d', ['close', 'paused'],
            skip_paused=False, df=True, fq='pre')
        # 上市不足250个已完成交易日时保持现金，这段时间仍计入全历史收益。
        if history is None or len(history) != g.cfg['slow']:
            emit(context, 'SKIP', reason='insufficient_history')
            return
        # 校验最新数据属于上一交易日，同时拒绝重复或乱序的索引。
        latest = history.index[-1].date()
        if latest != context.previous_date:
            emit(context, 'SKIP', reason='stale_history')
            return
        if history.index.has_duplicates or not history.index.is_monotonic_increasing:
            emit(context, 'SKIP', reason='invalid_history_index')
            return
        # 本版本对停牌采取保守规则：窗口内有停牌就暂停生成新指令。
        # 暂停指令不会自动清仓，已有持仓仍承受市场风险。
        if history['paused'].isnull().any() or (history['paused'] != 0).any():
            emit(context, 'SKIP', reason='paused_in_window')
            return
        values = classify(history['close'].values,
                          g.cfg['fast'], g.cfg['slow'], g.cfg)
        if values is None:
            emit(context, 'SKIP', reason='invalid_prices')
            return
        # signal_date是行情所属日，signal_day是实际生成并执行信号的今天。
        values['signal_date'] = str(latest)
        position = (context.portfolio.positions[security]
                    if security in context.portfolio.positions else None)
        held = int(position.total_amount) if position else 0
        signal, g.defensive_exit, g.exit_pending = decide(
            values, held, g.defensive_exit, g.exit_pending)
        values['signal'] = signal
        g.signal_values = values
        g.signal = values['signal']
        g.signal_day = context.current_dt.date()
        emit(context, 'SIGNAL', **values)
    except Exception as error:
        # 数据异常只记录原因；不会把异常转换成退出指令。
        emit(context, 'DATA_ERROR', error=str(error))


def executable_limit(price, high_limit, low_limit, buying, offset):
    """计算符合0.001元报价单位的限价；报价异常或接近单边涨跌停时返回None。

    price为下单时参考价；high_limit/low_limit取当时行情中的涨跌停价。
    buying=True计算最高买价，False计算最低卖价；offset为允许偏离比例。
    """
    if not all(math.isfinite(x) and x > 0
               for x in (price, high_limit, low_limit)):
        return None
    if high_limit <= low_limit or not low_limit <= price <= high_limit:
        return None
    if buying:
        # 接近涨停时不追买；最高买价向下取整，避免舍入后突破预算价格上界。
        if price >= high_limit - 0.001:
            return None
        bound = min(price * (1 + offset), high_limit - 0.001)
        # 极小修正项只处理浮点表示误差，乘除1000对应0.001元价格单位。
        return math.floor((bound + 1e-9) * 1000) / 1000
    # 接近跌停时不提交本次卖单；最低卖价向上取整，避免低于保护价格。
    if price <= low_limit + 0.001:
        return None
    bound = max(price * (1 - offset), low_limit + 0.001)
    return math.ceil((bound - 1e-9) * 1000) / 1000


def buy_amount(budget, limit_price, commission, min_commission):
    """返回含佣金预算内的100份整数倍数量，资金不足返回0。

    先预留最低佣金，再按比例佣金保守计算，最后向下取整手。
    该方法可能少买一手，但不会为了用满资金而突破设定预算。
    """
    if budget <= min_commission or limit_price <= 0:
        return 0
    amount = int((budget - min_commission)
                 / (limit_price * (1 + commission)) / 100) * 100
    return max(amount, 0)


def execute_signal(context):
    """每天10:00尝试执行今日信号；信号、下单成功与实际成交是三个状态。"""
    day = context.current_dt.date()
    # 只有今天生成的有效信号才可执行，且同日最多尝试提交一次。
    if g.signal_day != day or g.signal is None:
        emit(context, 'SKIP', reason='no_valid_signal_today')
        return
    if g.attempt_day == day:
        emit(context, 'SKIP', reason='already_attempted_today')
        return
    security = g.cfg['security']
    # 未完成委托可能继续成交，在确认处理完之前不叠加新订单。
    if any(o.security == security for o in get_open_orders().values()):
        emit(context, 'SKIP', reason='outstanding_order')
        return
    position = (context.portfolio.positions[security]
                if security in context.portfolio.positions else None)
    held = int(position.total_amount) if position else 0
    if g.signal == 'BUY' and held > 0:
        # 即使此前只成交100份，也接受实际仓位；不自动追补，不每日再平衡。
        return
    if g.signal == 'HOLD' or (g.signal == 'EXIT' and held == 0):
        return
    # 用下单时行情检查停牌、价格和涨跌停。昨收只参与趋势判断，不作成交价。
    quote = get_current_data()[security]
    if quote.paused:
        emit(context, 'SKIP', reason='paused_now')
        return
    buying = g.signal == 'BUY'
    price = float(quote.last_price)
    limit_price = executable_limit(
        price, float(quote.high_limit), float(quote.low_limit),
        buying, g.cfg['limit_offset'])
    if limit_price is None:
        emit(context, 'SKIP', reason='invalid_or_limit_quote')
        return
    if buying:
        # 90%按当前总资产计算，同时不能超过可用现金；限价参与数量计算。
        budget = min(context.portfolio.total_value * g.cfg['entry_weight'],
                     context.portfolio.available_cash)
        amount = buy_amount(budget, limit_price, g.cfg['commission'],
                            g.cfg['min_commission'])
    else:
        # order约定正数买入、负数卖出。只卖可卖份额，遵守该ETF的T+1限制。
        amount = -min(held, int(position.closeable_amount))
    if amount == 0:
        emit(context, 'SKIP', reason='no_cash_or_sellable_shares')
        return
    # 提交前就标记尝试日期：即使接口超时或返回None，当天也不重复下单。
    # 仍需在订单和成交记录中核对真实状态，不能由返回None推断一定没有成交。
    g.attempt_day = day
    emit(context, 'INTENT', signal=g.signal, amount=amount,
         reference_price=price, limit_price=limit_price)
    result = order(security, amount, style=LimitOrderStyle(limit_price))
    if result is None:
        emit(context, 'REJECTED', requested_amount=amount)
        return
    # 这里是提交后即时状态；最终成交数量由收盘后的订单与成交明细确认。
    emit(context, 'ORDER', order_id=str(result.order_id),
         status=str(result.status), requested_amount=amount,
         filled=float(result.filled), fill_price=float(result.price))


def cancel_remaining(context):
    """仅撤销本策略ETF的未完成委托；撤单请求发出后再次检查。"""
    for pending in get_open_orders().values():
        if pending.security == g.cfg['security']:
            cancel_order(pending)
            emit(context, 'CANCEL_REQUEST', order_id=str(pending.order_id))
    if any(o.security == g.cfg['security']
           for o in get_open_orders().values()):
        # 该日志只提示撤单尚未确认，没有实现实盘推送告警或无限重试。
        emit(context, 'CANCEL_UNCONFIRMED')


def close_report(context):
    """每天15:10保存当日订单、逐笔成交及收盘账户，供回测结果复核。"""
    # 再次检查剩余委托，避免仅凭10:05的一次撤单请求就认定处理完成。
    cancel_remaining(context)
    # get_orders/get_trades均返回当天记录，一个订单可能对应多次成交。
    # 订单的amount/filled均为正数，买卖方向另存is_buy，佣金取订单最终值。
    for item in get_orders().values():
        if item.security == g.cfg['security']:
            append_csv('_orders.csv', [str(context.current_dt.date()),
                       str(item.order_id), bool(item.is_buy), item.amount,
                       item.filled, item.price, item.commission,
                       str(item.status)])
    for trade in get_trades().values():
        if trade.security == g.cfg['security']:
            append_csv('_trades.csv', [str(trade.time), str(trade.trade_id),
                       str(trade.order_id), trade.amount, trade.price])
            emit(context, 'FILL', order_id=str(trade.order_id),
                 amount=float(trade.amount), price=float(trade.price),
                 trade_time=str(trade.time))
    # 总资产包括现金和持仓市值，收益统计使用总资产，而不是仅用ETF价格涨幅。
    portfolio = context.portfolio
    position = (portfolio.positions[g.cfg['security']]
                if g.cfg['security'] in portfolio.positions else None)
    held = int(position.total_amount) if position else 0
    position_value = float(position.value) if position else 0.0
    exposure = position_value / portfolio.total_value if portfolio.total_value else 0.0
    # exposure为实际仓位比例，account_nav以初始100万元归一化为1。
    record(exposure=exposure, account_nav=portfolio.total_value / portfolio.starting_cash)
    values = g.signal_values
    # 对照组无需均线数据；趋势组记录的价格和均线属于昨日信号日期。
    if 'close' in values:
        record(signal_close=values['close'], ma_fast=values['ma_fast'],
               ma_slow=values['ma_slow'])
    # 无信号日也导出一行，保留预热期、空仓期以及数据异常日的账户轨迹。
    row = [str(context.current_dt.date()), portfolio.total_value,
           portfolio.available_cash, held, position_value, exposure,
           g.signal or 'INVALID', values.get('signal_date', ''),
           values.get('close', ''), values.get('ma_fast', ''),
           values.get('ma_slow', '')]
    append_csv('_daily.csv', row)
    emit(context, 'ACCOUNT', total_value=float(portfolio.total_value),
         cash=float(portfolio.available_cash), shares=held,
         exposure=exposure)


def append_csv(suffix, row):
    """用csv模块处理字段转义，再通过聚宽write_file追加一行。"""
    output = io.StringIO()
    csv.writer(output).writerow(row)
    write_file(g.output + suffix, output.getvalue(), append=True)

这套沪深300ETF策略，用长期均线确定持仓方向，遇到急跌先退出，价格修复后再入场。在2012年5月28日至2026年9月9日的回测中，100万元最终约为306.97万元，比买入后一直持有多约102.57万元，最大回撤从42.15%降至25.22%。这些优势来自已完成的历史回测，能否延续到未来，还需要在新增行情中持续检验。