如何用 JoinQuant 编写行业轮动策略？（附完整 Python 代码）

这是一个基于 申万一级行业（Shenwan Level 1） 的动量轮动策略。

策略逻辑

1. 行业选择：使用申万一级行业分类（共31个行业，如银行、医药生物、电子等）。
2. 动量计算：
   • 每月月底进行一次轮动。
   • 计算每个行业过去 20个交易日 的涨跌幅作为动量指标。
   • 注意：为了计算效率和代表性，我们选取每个行业中市值最大的前10只股票的平均涨跌幅来代表该行业的涨跌幅。
3. 筛选行业：选取动量排名最高的 前3个 行业。
4. 选股：在选中的每个行业中，选取 市值最大 的3只股票进行买入。
5. 资金分配：资金等权重分配到选中的股票中。
6. 风险控制：过滤停牌、ST股票、退市股票。

策略代码

# -*- coding: utf-8 -*-
from jqdata import *
import pandas as pd

def initialize(context):
    """
    初始化函数
    """
    # 1. 设定基准（沪深300）
    set_benchmark('000300.XSHG')
    
    # 2. 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 3. 设置手续费（股票类：买入万三，卖出万三加千一印花税）
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 4. 策略参数设置
    g.momentum_days = 20      # 动量回顾周期（天）
    g.top_industry_num = 3    # 持有排名前几的行业
    g.stock_per_industry = 3  # 每个行业买入几只股票
    
    # 5. 设置定时运行：每月第一个交易日开盘运行
    run_monthly(rebalance, monthday=1, time='09:30')

def rebalance(context):
    """
    调仓主逻辑
    """
    log.info("开始进行行业轮动调仓...")
    
    # 1. 获取所有申万一级行业代码
    # sw_l1: 申万一级行业
    industries = get_industries(name='sw_l1', date=context.previous_date)
    industry_codes = list(industries.index)
    
    # 2. 计算每个行业的动量得分
    industry_scores = []
    
    for ind_code in industry_codes:
        # 获取该行业下的所有股票
        stocks = get_industry_stocks(ind_code, date=context.previous_date)
        
        if not stocks:
            continue
            
        # 过滤停牌和ST
        stocks = filter_stocks(stocks, context)
        
        if len(stocks) == 0:
            continue
            
        # 为了计算效率和代表性，取该行业市值最大的前10只股票来代表行业走势
        # 获取市值数据
        q = query(
            valuation.code,
            valuation.market_cap
        ).filter(
            valuation.code.in_(stocks)
        ).order_by(
            valuation.market_cap.desc()
        ).limit(10)
        
        df_cap = get_fundamentals(q, date=context.previous_date)
        
        if df_cap.empty:
            continue
            
        representative_stocks = list(df_cap['code'])
        
        # 计算这些代表性股票的平均动量（过去N天的收益率）
        # 获取收盘价：需要过去 momentum_days + 1 天的数据来计算收益率
        # 动量 = (当前价格 - N天前价格) / N天前价格
        prices = get_price(representative_stocks, end_date=context.previous_date, count=g.momentum_days + 1, fields=['close'], panel=False)
        
        if prices.empty:
            continue
            
        # 计算每只股票的收益率
        # 转换数据格式方便计算
        pivot_price = prices.pivot(index='time', columns='code', values='close')
        
        # 计算 (最新收盘价 - 期初收盘价) / 期初收盘价
        # iloc[-1] 是最近一天， iloc[0] 是N天前
        returns = (pivot_price.iloc[-1] - pivot_price.iloc[0]) / pivot_price.iloc[0]
        
        # 行业得分为成分股收益率的平均值
        avg_return = returns.mean()
        
        industry_scores.append((ind_code, avg_return))
    
    # 3. 对行业按动量得分从高到低排序
    industry_scores.sort(key=lambda x: x[1], reverse=True)
    
    # 选取排名前N的行业
    target_industries = [x[0] for x in industry_scores[:g.top_industry_num]]
    
    log.info("选中行业: %s" % target_industries)
    
    # 4. 构建目标股票池
    target_stocks = []
    for ind_code in target_industries:
        # 获取该行业股票
        stocks = get_industry_stocks(ind_code, date=context.previous_date)
        stocks = filter_stocks(stocks, context)
        
        # 选取该行业市值最大的M只股票
        q = query(
            valuation.code
        ).filter(
            valuation.code.in_(stocks)
        ).order_by(
            valuation.market_cap.desc()
        ).limit(g.stock_per_industry)
        
        df = get_fundamentals(q, date=context.previous_date)
        target_stocks.extend(list(df['code']))
    
    log.info("目标持仓股票: %s" % target_stocks)
    
    # 5. 执行交易
    adjust_position(context, target_stocks)

def filter_stocks(stock_list, context):
    """
    过滤停牌、ST、退市股票
    """
    current_data = get_current_data()
    filtered_list = []
    for stock in stock_list:
        # 过滤停牌
        if current_data[stock].paused:
            continue
        # 过滤ST
        if current_data[stock].is_st:
            continue
        # 过滤退市（虽然get_industry_stocks通常返回上市的，但双重保险）
        if 'ST' in current_data[stock].name or '*' in current_data[stock].name:
            continue
        filtered_list.append(stock)
    return filtered_list

def adjust_position(context, target_stocks):
    """
    调仓执行函数
    """
    # 获取当前持仓
    current_holdings = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在目标池中的股票
    for stock in current_holdings:
        if stock not in target_stocks:
            order_target_value(stock, 0)
            
    # 2. 买入目标池中的股票
    if len(target_stocks) > 0:
        # 等权重分配资金
        # 注意：这里简单处理，每次全仓轮动。实际中可能需要预留现金。
        position_count = len(target_stocks)
        value_per_stock = context.portfolio.total_value / position_count
        
        for stock in target_stocks:
            order_target_value(stock, value_per_stock)

策略详解

1. 行业动量计算方法：

   • 代码中没有直接使用行业指数（因为行业指数代码可能变动或不可交易），而是采用了**“成分股合成法”**。
   • 对于每个申万一级行业，我们取出该行业内市值最大的10只股票。
   • 计算这10只股票过去20天的平均收益率，以此作为该行业的动量得分。这种方法既能代表行业龙头的走势，又能保证计算速度。

2. 选股逻辑：

   • 在动量最强的3个行业中，我们并没有买入所有股票，而是选择了市值最大的3只股票。
   • 原因：大市值股票流动性好，且通常是该行业的龙头股，最能承载该行业的上涨趋势。

3. 交易执行：

   • 使用 order_target_value 进行调仓。如果股票已经在持仓中且属于目标池，它会根据资金总额自动调整仓位（再平衡）；如果不在目标池，则卖出。

4. 过滤逻辑：

   • filter_stocks 函数确保了我们不会买入停牌或ST的股票，避免踩雷。

常见问题 (Q&A)

Q: 为什么选择20天作为动量周期？
A: 20天大约对应一个月（4周）的交易日。这是一个经典的短期动量参数。你可以根据需要修改 g.momentum_days，例如改为60（季度动量）或120（半年动量）。

Q: 为什么只选市值最大的股票？
A: 行业轮动本质上是配置Beta。行业龙头股（大市值）通常与行业指数的相关性最高，且流动性最好，滑点低。如果你想做小盘股轮动，可以修改 order_by(valuation.market_cap.asc())。

Q: 如何查看回测结果？
A: 将上述代码复制到聚宽的“策略研究” -> “新建策略”中，点击“编译运行”或“运行回测”，设置好回测时间段（例如2019-01-01至今）即可查看收益曲线。