聚宽回测策略实现指南

基于上交所与深交所股票的多条件策略实现示例

核心要点

股票池筛选：选择上交所与深交所所有股票，同时排除次新股、创业版、科创板、新股及ST股。
买入条件：连续三天涨停且首次出现最低点比昨日收盘下跌9%以上（可能先跌后涨形成涨停），在跌破9%的时点买入。
卖出逻辑：如果次日10点未涨停则在当日10点卖出；若次日涨停，则持股至第三天10点后卖出。

策略逻辑详细说明

该策略基于对上交所与深交所全量股票的严格筛选，加上对股票价格行为的判断，旨在捕捉连续三天涨停且伴随特定价格波动的股票机会。策略在筛选过程中排除了速新股、创业板、科创板、新股及ST股，以规避由于这些股票可能带来的数据不完整、流动性较差或特殊风险带来的干扰，并使策略适用性更广。

交易逻辑主要分为两部分：买入逻辑与卖出逻辑。首先在买入逻辑中，程序需检测股票是否连续三天涨停。随后，特别关注该阶段中第一次交易日的价格波动情况，若该交易日中某个时段最低点比前一日收盘价低9%以上，即可认为该股票存在短期调整后回弹涨停的信号，此时即触发买入操作。在卖出逻辑方面，策略要求若在买入后第二天的10点时发现股票并未涨停，则立即平仓卖出；若第二天达到涨停状态，则持有至第三天10点后出售，从而获取潜在盈利。

实现示例代码

以下代码示例基于聚宽平台的Python回测框架，详细展示了如何按照上述要求构建回测策略。代码中定义了初始化、交易前的股票筛选、以及买卖操作的具体逻辑，各环节都做了详细注释说明。请确保您在使用前已安装并配置好聚宽的开发环境。


# 导入聚宽数据接口及常用包
import jqdata
import pandas as pd

# 初始化函数：设置回测参数与股票池
def initialize(context):
    # 设置回测基准及手续费
    set_benchmark('000300.XSHG')  # 可修改为您所需的回测基准
    set_option('use_real_price', True)  # 使用真实价格进行回测
    set_order_cost(StockTransactionFee(0.001, 0.00003), type='stock')
    
    # 初始股票池：获取上交所与深交所所有股票（不包括被排除的板块和次新股）
    context.universe = get_all_securities(['stock']).index.tolist()
    
    # 记录买入股票持仓和订单信息
    context.buy_list = {}
    
    # 设置回测状态跟踪标记，用以执行卖出逻辑
    context.sell_schedule = {}
    
    # 每天盘前调仓前进行股票筛选，排除次新股、创业版、科创板、新股和ST股票
    run_daily(before_trading_start, time='before_open')

# 辅助函数：判断是否为次新股、新股、创业板、科创板或者ST
def is_excluded(stock_code):
    try:
        basics = get_extras('is_st', stock_code)
        # 如果股票存在ST标记或其他特殊标识，则排除
        if basics:
            return True
    except:
        pass
    # 若获取上市日期信息，排除上市不足一年的股票视为次新股与新股
    try:
        info = get_security_info(stock_code)
        if pd.Timestamp.now() - info.start_date < pd.Timedelta(days=365):
            return True
    except:
        pass
    # 加入对创业板与科创板的排除判断（可根据股票代码规则进行简单筛选）
    if stock_code.startswith('300') or stock_code.startswith('688'):
        return True
    return False

# 盘前函数：每日更新股票池，排除不符合条件的股票
def before_trading_start(context):
    qualified_stocks = []
    for stock in context.universe:
        if not is_excluded(stock):
            qualified_stocks.append(stock)
    context.stocks_pool = qualified_stocks

# 核心函数：实时处理数据，执行买卖逻辑
def handle_data(context, data):
    # 遍历股票池中的每只股票，检测是否满足买入条件
    for stock in context.stocks_pool:
        try:
            # 获取过去三天的日线数据
            history_data = attribute_history(stock, 3, '1d', ['close', 'high', 'low'], skip_paused=True)
            if len(history_data) < 3:
                continue
            # 判断连续三天是否涨停：比较当天收盘价与前一日收盘价是否都达到或超过10%的涨幅
            close_day_minus2 = history_data['close'][0]
            close_day_minus1 = history_data['close'][1]
            close_day0 = history_data['close'][2]
            # 对涨停条件进行检测（简单估计10%为涨停幅度，实际情况需结合股票具体涨停规则）
            if close_day_minus1 >= close_day_minus2 * 1.095 and close_day0 >= close_day_minus1 * 1.095:
                # 进一步检查第一次出现最低价是否比昨日收盘下跌9%以上
                # 对于跌幅检测，获取第一天的分钟级价格数据（假定可利用get_price获取部分实时数据）
                # 使用data接口获取当天的分钟数据
                current_day_price = get_price(stock, end_date=context.current_dt, frequency='1m',
                                              fields=['close', 'low'], skip_paused=True)
                if current_day_price is None or current_day_price.empty:
                    continue
                min_price = current_day_price['low'].min()
                # 昨日收盘价
                pre_close = history_data['close'][1]
                if min_price < pre_close * 0.91:
                    # 如果满足条件且当前未持仓，则执行买入逻辑
                    if stock not in context.portfolio.positions:
                        # 买入操作：以市值买入，单位资金可调整为固定金额或仓位比例
                        order_target_value(stock, 10000)  # 此处设置下单金额为10000元
                        # 记录买入日期，用于后续卖出逻辑判定
                        context.buy_list[stock] = context.current_dt
                        # 标记卖出调度，初步设定于第二天10点成交，条件异动后转至第三天10点
                        context.sell_schedule[stock] = 'T+1'
        except Exception as e:
            # 捕捉异常，避免因个股数据问题影响整体策略
            log.info("股票%s处理异常: %s" % (stock, e))
    
    # 卖出逻辑部分
    for stock in list(context.portfolio.positions.keys()):
        # 获取昨晚当前持仓的买入日期，确保策略在流动时间点进行判断
        buy_date = context.buy_list.get(stock, None)
        if buy_date is None:
            continue

        # 判断当前日期与买入日期是否达到 T+1 或 T+2
        # 通过context.current_dt.time()获取当前时刻，判断是否到10点
        current_time = context.current_dt.time()
        target_time = pd.to_datetime("10:00:00").time()
        
        # T+1：买入后第二天10点未涨停，则卖出
        if context.sell_schedule.get(stock) == 'T+1' and context.current_dt.date() != buy_date.date() and current_time >= target_time:
            # 检查当前是否涨停, 用前一日收盘价对比当前价格判断
            current_price = data[stock].close
            pre_close = attribute_history(stock, 1, '1d', ['close'], skip_paused=True)['close'][0]
            # 若价格没有达到涨停（假设涨停价为前收盘价的1.10倍），则进行卖出
            if current_price < pre_close * 1.10:
                order_target_value(stock, 0)
                # 移除记录，完成调仓
                context.buy_list.pop(stock, None)
                context.sell_schedule.pop(stock, None)
            else:
                # 若达到涨停，则更新调度为第三天T+2卖出
                context.sell_schedule[stock] = 'T+2'
        
        # T+2：在第三天10点卖出操作，无论价格如何变化
        if context.sell_schedule.get(stock) == 'T+2' and current_time >= target_time and context.current_dt.date() != buy_date.date():
            order_target_value(stock, 0)
            context.buy_list.pop(stock, None)
            context.sell_schedule.pop(stock, None)

# 设置回测参数与运行周期
def before_backtest(context):
    # 设定回测起始与结束日期
    set_params(
        start_date='2023-01-01',
        end_date='2025-03-21',
        initial_cash=1000000  # 设置初始资金1,000,000元
    )
    
# 运行策略
if __name__ == '__main__':
    # 注：在聚宽平台中通常通过运行策略文件的方式调用run()函数
    run_func(initialize=initialize, handle_data=handle_data, before_trading_start=before_trading_start, 
             before_backtest=before_backtest)

参数说明与注意事项

1. 股票池与数据筛选

此策略首先通过调用get_all_securities获取上交所与深交所所有股票数据，然后使用自定义的is_excluded函数排除那些不符合条件的股票（次新股、创业版、科创板、新股、ST股）。同时，也可以根据需求进一步扩展或细化排除条件，比如根据行业、流动性等因素进行筛选。

2. 买入逻辑

买入条件的关键在于连续三天涨停，此处以每日的收盘价的10%涨幅作为判断标准（实际交易中可能需要针对不同市场或不同股票做调整），并结合当日的分钟数据检测是否出现“最低点低于昨日收盘价9%”的情况。策略利用get_price函数来获取分钟级数据，以保证能够捕捉到当天出现的低价信号。注意，如果策略在执行买入操作时已有持仓，则需要根据策略要求避免重复下单。

3. 卖出逻辑

卖出部分分为两种情景：T+1的卖出与T+2的卖出。当股票在次日上午10点之前没有再次涨停，程序会按照市价单将股票卖出；若次日达到涨停，则持有到第三天的10点后再进行卖出。用以确保在不同的市场状态下，能够有效控制风险及锁定利润。实际中可根据交易成本、滑点及其他风险管理需求对卖出时机进行动态调整。

4. 交易成本与滑点考虑

该策略在初始化中设定了交易费用以及使用真实价格回测。建议在实际运用中，综合考虑滑点、手续费以及资金管理策略，进一步优化回测结果与实盘操作的一致性。同时，确保对数据质量和实时数据完整性的监控，加强风险控制。

5. 回测参数

策略代码中设定了起始日期、结束日期和初始资金。根据策略回测的时间段与市场环境，可对这些参数进行调整，以保证策略能够涵盖足够的市场周期并且具有较高的参考价值。

策略概览与效果指标

为便于更直观地把握策略逻辑，以下是该策略主要逻辑与操作步骤的概览表。该表同时概要介绍了每个操作的条件与买入/卖出时点，供日后策略调试与优化时参考。

操作	条件	买/卖操作	执行时点
股票筛选	上交所和深交所全部股票，不含次新股、创业版、科创板、新股和ST股	——	每日盘前
涨停检测	连续三天收盘价涨幅达到或超过10%	——	每日收盘后
首次跌幅监测	当天首次出现最低点低于昨日收盘价的91%	触发买入	当日内
卖出逻辑T+1	次日10点未达到涨停价	全部卖出	次日10点
卖出逻辑T+2	次日涨停信号成立	持有至第三天10点后全部卖出	第三天10点

扩展思路与策略优化

数据监控

对于如此细致的交易策略，数据的实时性和准确性显得尤为重要。建议策略开发者在回测阶段额外监控交易信号的生成过程及成交价格，确保所调用的分钟级数据与实际市场状态高度吻合。可以加入日志记录与调试窗口输出，帮助日后排查回测过程中的潜在问题。

风险管理

除了确定买卖时点，此策略同样需要考虑仓位控制与止损策略。建议投资者在实盘前结合其他风险控制技术，如每日最大亏损、单笔交易风险控制等，避免由于极端行情波动导致资金出现较大亏损。合理分散仓位与设定止损线可以有效提升策略的稳健性。

回测完善

在回测优化过程中，开发者可以考虑多种情形下的不同表现，如不同市场波动期间、不同金额配置以及滑点设置的敏感性测试。此外，可以通过聚宽社区与其它量化研究者分享策略，收集反馈意见，继续对算法进行性能调优与参数优化，以期达到更优的实盘表现。