年化收益100%，11年1700倍，聚宽小市值策略回测及源码

admin

在任何一个圈子里，总有那么一两个“扫地僧”，平时不显山不露水，扔下两句话就足以让后辈琢磨好几年。

量化圈里，就有这么一位蒋老师。

5年前，他在聚宽论坛发了个帖子，算上标点符号，全文也就100来字。但就是这百十来个字，成了那几年最火的“镇店之宝”。9万多人拜读，5000多人大呼“卧槽”并直接抄了作业，评论区盖了几百层楼。

这“圣经”写了个啥？

核心逻辑，简单到令人发指：

去中小综指（代码399101）的成分股里，挑出流通市值最小的5只股票，然后买了它。

没了。

就这？对，就这。

当年贴出来的回测曲线，那是相当的性感，年化收益100%，夏普率2.49。简直就是一台印钞机。

神迹消亡还是回归人间？

可快进到今天，你再跑一遍这个最原始的版本， magic 消失了。年化收益变成了40%，夏普率也降到了1.08。

很多人一看，嗨，没意思了，说好的100%呢？这不成“渣男”了么。

但你换个角度想，一个“毛坯房”策略，没装修、没精洗，光靠一个骨架子，能在残酷的A股市场拿到年化40%，这已经不是“不错”，而是相当凶悍了。想想看，我们之前捣鼓的那些ETF轮动，改了又改，加了无数条件，最后也才勉强摸到这个收益。

这恰恰说明，这个粗糙的逻辑里，藏着巨大的可挖掘潜力。

全网最关心的问题：到底作弊没？

但真正圈内人关心的，是蒋老师在标题里拍的胸脯——“绝无未来函数”。

什么叫“未来函数”？通俗点讲，就是你用还没发生的信息做了交易决策，这属于考试翻书，属于作弊。在量化回测里，这是最致命的原罪。

蒋老师说他没有，怎么证明？他说的话就一定是金科玉律吗？

换个炉子，重新炼一遍“神丹”

空口无凭。证明一个东西是不是纯金，最好的办法不是听它自己说，而是把它扔到另一坨火里，重新炼一遍。

我就干了这事。我把这个策略的逻辑，从聚宽那个“炼丹炉”里原封不动地搬出来，放到了QMT这个可以直接上战场的“工业熔炉”里，并且给它上了最严苛的“纪律枷锁”，彻底杜绝任何作弊的可能：

• 1. 今天选股，市值数据只用昨天的收盘价。
• 2. 成分股列表，只用上个月官方公布的。
• 3. 所有涨跌停的、ST戴帽的，一律踢出皮球。
  

你看，规矩有多严。我就像个刻薄的包工头，不让它有半点“提前看答案”的机会。

然后，在另一套数据源上，我重新点火了。

那么，在这么严苛的家法下，重新炼出来的这颗丹，疗效如何？

Curve is still稳步向上，虽然没有当年100%那么疯，但年化收益稳稳地站在了42.7%。最大回撤也控制在了一个比较合理的范围内。夏普比率也守住了1的关口，说明这收益不是靠赌博赌出来的，而是踏踏实实赚的风险调整后回报。

这个结果，对我来说，比最初那个100%的野路子更值钱。因为它明确地告诉我一个最重要的结论：逻辑，是站得住脚的。

它没作弊，它只是褪去了神话的光环，变回了那个朴实无华但依然高效的样子。

这一整套操作下来，它就已经不再是一个简单的“小市值策略”了。它是我亲手锻造的一把“归零之剑”：一把剔除了所有嫌疑和杂质，可以直接拿来分析、拿来魔改、拿来上战场的武器。

这已经不是在复制一个策略了，而是在验证一种思想：一个最朴素、最简单的想法，只要它在现实中逻辑自洽、执行到位，就依然是一条值得走下去的宽客之路。

• 
  

# 导入函数库from jqdata import *# 初始化函数，设定基准等参数def initialize(context):    # 设定沪深300指数作为业绩基准    set_benchmark('000300.XSHG')    # 开启动态复权模式（使用真实交易价格）    set_option('use_real_price', True)    # 输出内容到日志系统 log.info()    log.info('初始化函数开始运行，全局仅执行一次')    # 过滤掉order系列API产生的低于error级别的日志    log.set_level('order', 'error')    # 股票池配置    g.security_universe_index = "399101.XSHE"  # 中小板指数代码    g.buy_stock_count = 5  # 持仓股票数量上限
    ### 股票交易成本设定 ###    # 股票类每笔交易手续费规则：    # 买入时佣金万分之三，卖出时佣金万分之三 + 千分之一印花税，每笔交易佣金最低5元    set_order_cost(        OrderCost(            close_tax=0.001,        # 卖出印花税：0.1%            open_commission=0.0003, # 买入佣金：0.03%            close_commission=0.0003,# 卖出佣金：0.03%            min_commission=5        # 最低佣金门槛：5元        ),        type='stock'  # 交易品种类型：股票    )
    ## 定时运行配置（reference_security仅用于区分运行周期类型，    ## 传入'000300.XSHG'或'510300.XSHG'效果一致）    # 每日14:40执行交易逻辑    run_daily(my_trade, time='14:40', reference_security='000300.XSHG')    # 收盘后执行复盘逻辑    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')## 开盘期间运行的交易函数def my_trade(context):    # 步骤1：获取中小板指数成分股列表    check_out_lists = get_index_stocks(g.security_universe_index)
    # 步骤2：查询成分股中流通市值最小的股票（取3倍目标持仓数作为候选）    q = query(valuation.code).filter(        valuation.code.in_(check_out_lists)  # 筛选指数成分股    ).order_by(        valuation.circulating_market_cap.asc()  # 按流通市值升序排列    ).limit(        g.buy_stock_count * 3  # 取前N*3只股票作为初选池    )    check_out_lists = list(get_fundamentals(q).code)  # 转换为股票代码列表
    # 步骤3：过滤风险股票    check_out_lists = filter_st_stock(check_out_lists)        # 过滤ST/*ST/退市股    check_out_lists = filter_limitup_stock(context, check_out_lists)  # 过滤涨停股    check_out_lists = filter_limitdown_stock(context, check_out_lists)# 过滤跌停股    check_out_lists = filter_paused_stock(check_out_lists)    # 过滤停牌股
    # 步骤4：最终筛选出目标数量的股票    check_out_lists = check_out_lists[:g.buy_stock_count]
    # 步骤5：执行调仓操作    adjust_position(context, check_out_lists)## 收盘后运行的复盘函数def after_market_close(context):    log.info(f'复盘函数运行时间(after_market_close)：{context.current_dt.time()}')    # 获取当日所有成交记录    trades = get_trades()    for _trade in trades.values():        log.info(f'成交记录详情：{str(_trade)}')    log.info('当日交易流程结束')    log.info('##############################################################')# 自定义下单函数（封装底层API）# 说明：根据聚宽官方文档，报单函数（如order_target_value）为阻塞执行，# 返回即表示报单流程完成（不代表成交）# 返回值：报单成功返回Order对象，失败返回Nonedef order_target_value_(security, value):    if value == 0:        log.debug(f'清仓操作：卖出{security}全部持仓')    else:        log.debug(f'目标持仓操作：调整{security}持仓至{value:.2f}元')
    # 注意事项：    # 1. 股票停牌时，报单直接失败，返回None    # 2. 股票涨跌停时，报单会创建成功（返回Order），但会被交易所撤销    # 3. 部分成交后撤销的报单，聚宽状态为"已撤"，可通过filled字段判断实际成交量    return order_target_value(security, value)# 开仓函数：买入指定金额的证券# 返回值规则：# - 报单成功且有成交（全部/部分成交，成交量>0）→ 返回True# - 报单失败/报单成功但被撤销（成交量=0）→ 返回Falsedef open_position(security, value):    order = order_target_value_(security, value)    if order is not None and order.filled > 0:        return True    return False# 平仓函数：卖出指定持仓# 返回值规则：# - 报单成功且全部成交 → 返回True# - 报单失败/报单成功但被撤销（成交量=0）/部分成交 → 返回Falsedef close_position(position):    security = position.security    order = order_target_value_(security, 0)  # 清仓操作（停牌时可能失败）    if order is not None:        # 检查报单状态：已挂单且成交量等于委托量（全部成交）        if order.status == OrderStatus.held and order.filled == order.amount:            return True    return False# 核心调仓函数def adjust_position(context, buy_stocks):    # 第一步：清理持仓中不在目标列表的股票    for stock in context.portfolio.positions:        if stock not in buy_stocks:            log.info(f'持仓股票[{stock}]不在目标买入列表，执行平仓')            position = context.portfolio.positions[stock]            close_position(position)        else:            log.info(f'持仓股票[{stock}]已在目标买入列表，保留持仓')
    # 第二步：计算可用资金并分仓买入新标的    # 注：仅按可用现金平均分配，不保证最终持仓金额完全均等    position_count = len(context.portfolio.positions)    if g.buy_stock_count > position_count:        # 可用资金均分至待买入的股票        value = context.portfolio.cash / (g.buy_stock_count - position_count)        for stock in buy_stocks:            # 仅对无持仓的股票执行买入            if context.portfolio.positions[stock].total_amount == 0:                if open_position(stock, value):                    # 达到持仓上限时停止买入                    if len(context.portfolio.positions) == g.buy_stock_count:                        break# 辅助函数：过滤停牌股票def filter_paused_stock(stock_list):    current_data = get_current_data()    # 保留未停牌的股票    return [stock for stock in stock_list if not current_data[stock].paused]# 辅助函数：过滤ST及退市风险股票def filter_st_stock(stock_list):    current_data = get_current_data()    # 过滤条件：非ST标识、名称不含ST/*、无退市标识    return [        stock for stock in stock_list        if not current_data[stock].is_st        and 'ST' not in current_data[stock].name        and '*' not in current_data[stock].name        and '退' not in current_data[stock].name    ]# 辅助函数：过滤涨停股票def filter_limitup_stock(context, stock_list):    # 获取最新一分钟收盘价    last_prices = history(1, unit='1m', field='close', security_list=stock_list)    current_data = get_current_data()    # 保留条件：已持仓（避免涨停被清仓） 或 最新价 < 涨停价    return [        stock for stock in stock_list        if stock in context.portfolio.positions.keys()        or last_prices[stock][-1] < current_data[stock].high_limit    ]    # 可选阈值版本（过滤接近涨停的股票）：    # return [stock for stock in stock_list if stock in context.portfolio.positions.keys()    #         or last_prices[stock][-1] < current_data[stock].high_limit * 0.995]# 辅助函数：过滤跌停股票def filter_limitdown_stock(context, stock_list):    # 获取最新一分钟收盘价    last_prices = history(1, unit='1m', field='close', security_list=stock_list)    current_data = get_current_data()    # 保留条件：已持仓（避免跌停被清仓） 或 最新价 > 跌停价    return [        stock for stock in stock_list        if stock in context.portfolio.positions.keys()        or last_prices[stock][-1] > current_data[stock].low_limit    ]