JoinQuant 平台如何编写小市值轮动策略？（附 Python 源码）

这是一个基于聚宽（JoinQuant）API 实现的小市值轮动策略。

策略逻辑说明

1. 选股标准：
   • 剔除停牌股票。
   • 剔除 ST/*ST 股票（避免退市风险）。
   • 剔除上市不满 6 个月的次新股（避免上市初期的剧烈波动）。
   • 在剩余股票中，选取总市值最小的 10 只股票。
2. 调仓频率：
   • 每月调仓一次，设定在每月的第一个交易日开盘时运行。
3. 资金分配：
   • 等权重买入（即每只股票买入总资金的 1/10）。

策略代码

# -*- coding: utf-8 -*-
from jqdata import *

def initialize(context):
    """
    初始化函数，设定基准、手续费、滑点、定时运行等
    """
    # 设定基准为中证1000指数（小市值代表指数）
    set_benchmark('000852.XSHG')
    
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 设定日志级别
    log.set_level('order', 'error')
    
    # 设定手续费：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 设定持仓数量
    g.stock_num = 10
    
    # 设定每月第一个交易日 09:30 运行调仓函数
    run_monthly(rebalance, 1, time='09:30')

def rebalance(context):
    """
    调仓主函数
    """
    # 1. 获取选股列表
    target_list = get_stock_list(context)
    
    # 2. 执行交易
    trade(context, target_list)

def get_stock_list(context):
    """
    选股逻辑：剔除ST、停牌、次新股，选取市值最小的股票
    """
    # 获取当前所有股票代码
    all_stocks = list(get_all_securities(['stock']).index)
    
    # 获取当前时间数据（用于判断停牌和ST）
    current_data = get_current_data()
    
    # 过滤掉停牌的股票
    all_stocks = [stock for stock in all_stocks if not current_data[stock].paused]
    
    # 过滤掉ST的股票
    all_stocks = [stock for stock in all_stocks if not current_data[stock].is_st]
    
    # 过滤掉上市不满180天的次新股
    # 获取股票上市日期
    stock_infos = get_all_securities(types=['stock'])
    today = context.current_dt.date()
    all_stocks = [stock for stock in all_stocks if (today - stock_infos.loc[stock]['start_date']).days > 180]
    
    # 查询财务数据，获取市值最小的股票
    # query对象：查询股票代码
    # filter：在经过初步筛选的股票池中筛选
    # order_by：按总市值(market_cap)升序排列
    # limit：取前 g.stock_num 只
    q = query(
        valuation.code
    ).filter(
        valuation.code.in_(all_stocks)
    ).order_by(
        valuation.market_cap.asc()
    ).limit(
        g.stock_num
    )
    
    df = get_fundamentals(q)
    
    if len(df) > 0:
        return list(df['code'])
    else:
        return []

def trade(context, target_list):
    """
    交易逻辑
    """
    # 获取当前持仓
    current_holdings = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在目标列表中的股票
    for stock in current_holdings:
        if stock not in target_list:
            order_target_value(stock, 0)
            log.info("卖出 %s" % stock)
    
    # 如果没有选出股票，则直接返回
    if len(target_list) == 0:
        return

    # 2. 买入目标列表中的股票
    # 计算每只股票应该分配的资金（等权重）
    # 使用 total_value 保证满仓操作
    position_per_stock = context.portfolio.total_value / len(target_list)
    
    for stock in target_list:
        order_target_value(stock, position_per_stock)
        log.info("买入/调整 %s" % stock)

代码关键点解析

1. run_monthly(rebalance, 1, time='09:30'):

   • 这是聚宽的定时函数，指定策略在每月第 1 个交易日的 09:30 执行 rebalance 函数。

2. get_fundamentals(q):

   • 这是获取财务数据的核心函数。
   • valuation.market_cap.asc() 表示按总市值从小到大排序。
   • limit(g.stock_num) 限制只取前 10 只。

3. 过滤逻辑:

   • 如果不剔除停牌和 ST 股票，策略可能会买入无法交易的股票，或者风险极高的退市预警股，导致回测结果失真或实盘亏损。
   • 剔除次新股是因为新股上市初期市值通常较小且波动剧烈，不符合小市值策略“捡便宜”的初衷。

4. order_target_value:

   • 这是一个非常方便的下单函数。它会自动计算需要买入或卖出多少股，使该股票的持仓价值达到指定金额。如果需要卖出，它会自动卖出；如果需要加仓，它会自动买入。

Q&A

Q: 为什么小市值策略在回测中表现通常很好？
A: 小市值因子（Size Factor）是A股市场长期有效的风格因子之一。小盘股通常具有高波动性和高弹性的特征，在市场流动性充裕或风险偏好上升时，往往能获得超过大盘股的超额收益（壳价值、并购重组预期等）。但需要注意，近年来随着注册制推行，小市值策略的超额收益有所衰减，且面临流动性风险。

Q: 如何修改持仓数量，比如改成持有 20 只？
A: 在 initialize 函数中，修改 g.stock_num = 10 为 g.stock_num = 20 即可。

Q: 策略中的 set_benchmark 有什么用？
A: 这不影响策略的实际买卖，只影响回测结果图中“基准收益”的曲线。对于小市值策略，通常建议使用中证1000（000852.XSHG）或国证2000（399303.XSHE）作为基准，这样对比收益更客观。代码中已设置为中证1000。

Q: 为什么使用 order_target_value 而不是 order_value？
A: order_target_value 是“目标调仓”。例如你原本持有 10000 元 A 股票，现在因为股价上涨变成了 12000 元，而目标是持仓 11000 元。order_target_value 会自动卖出 1000 元。如果是 order_value(11000)，则会再买入 11000 元，导致仓位错误。轮动策略使用 Target 系列函数更准确。