JoinQuant 行业轮动策略：如何基于申万一级行业动量进行资产配置？

这是一个基于聚宽（JoinQuant）平台的申万一级行业动量轮动策略。

策略逻辑说明

1. 核心思想：动量效应（Momentum），即“强者恒强”。我们假设过去一段时间表现最好的行业，在未来一段时间仍能保持强势。
2. 行业池：申万一级行业（共31个，如银行、医药生物、电子等）。
3. 动量计算：
   • 由于直接获取行业指数行情可能存在数据权限或代码后缀匹配问题，本策略采用**“成分股合成法”**。
   • 对于每个行业，选取市值最大的前10只股票作为代表。
   • 计算这10只股票过去20个交易日（约1个月）的平均涨跌幅，作为该行业的动量得分。
4. 选股逻辑：
   • 选取动量得分最高的 Top 3 行业。
   • 在每个入选行业中，选取市值最大的 3只 股票进行买入。
   • 总持仓约为 9 只股票。
5. 调仓频率：按月调仓（每月第一个交易日）。
6. 风控与过滤：过滤停牌、ST、上市不满半年的股票。

策略代码

# -*- coding: utf-8 -*-
# 导入函数库
from jqdata import *
import pandas as pd
import numpy as np

def initialize(context):
    # 设定基准为沪深300
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    # 动量回看天数 (20天约等于1个月)
    g.lookback_days = 20
    # 持仓行业数量
    g.top_industry_count = 3
    # 每个行业买入股票数量
    g.stocks_per_industry = 3
    
    # 设置交易费率：买入万三，卖出万三加千一印花税
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 按月运行，每月第一个交易日开盘后运行
    run_monthly(market_trade, monthday=1, time='09:30')

def market_trade(context):
    """
    月度调仓主函数
    """
    log.info("开始进行行业动量轮动调仓...")
    
    # 1. 获取所有申万一级行业代码
    # sw_l1: 申万一级行业
    industries = get_industries(name='sw_l1', date=context.previous_date)
    industry_codes = industries.index.tolist()
    
    # 2. 计算每个行业的动量得分
    industry_momentum = []
    
    for ind_code in industry_codes:
        # 获取行业名称，用于日志
        ind_name = industries.loc[ind_code, 'name']
        
        # 计算该行业的动量 (使用行业内龙头股的平均涨幅作为代理)
        mom_score = calculate_industry_momentum(context, ind_code)
        
        if mom_score is not None:
            industry_momentum.append({
                'code': ind_code,
                'name': ind_name,
                'score': mom_score
            })
    
    # 将结果转换为DataFrame并排序
    df_mom = pd.DataFrame(industry_momentum)
    if df_mom.empty:
        log.warn("未能计算出行业动量，跳过本次调仓")
        return
        
    # 按分数降序排列
    df_mom = df_mom.sort_values(by='score', ascending=False)
    
    # 3. 选取Top N行业
    target_industries = df_mom.head(g.top_industry_count)
    log.info("\n选中的行业及动量:\n" + str(target_industries))
    
    # 4. 在选中的行业中选股
    target_stocks = []
    for _, row in target_industries.iterrows():
        ind_code = row['code']
        # 获取该行业内市值最大的几只股票
        stocks = get_top_market_cap_stocks(context, ind_code, g.stocks_per_industry)
        target_stocks.extend(stocks)
    
    log.info("目标持仓股票: " + str(target_stocks))
    
    # 5. 执行交易
    adjust_position(context, target_stocks)

def calculate_industry_momentum(context, industry_code):
    """
    计算行业动量
    方法：选取该行业市值最大的10只股票，计算它们过去N天的平均涨跌幅
    """
    # 获取行业内市值前10的股票作为代表
    representative_stocks = get_top_market_cap_stocks(context, industry_code, 10)
    
    if not representative_stocks:
        return None
        
    # 获取历史价格数据
    # 我们需要 lookback_days 之前的价格和昨天的价格
    # count = g.lookback_days + 1
    h_data = history(g.lookback_days + 1, unit='1d', field='close', security_list=representative_stocks, df=True, skip_paused=True)
    
    if h_data.empty:
        return None
        
    # 计算个股收益率: (最新收盘价 - N天前收盘价) / N天前收盘价
    # iloc[-1] 是昨天收盘价， iloc[0] 是N天前收盘价
    returns = (h_data.iloc[-1] - h_data.iloc[0]) / h_data.iloc[0]
    
    # 计算平均收益率作为行业动量
    avg_return = returns.mean()
    
    # 如果数据全是NaN，返回None
    if np.isnan(avg_return):
        return None
        
    return avg_return

def get_top_market_cap_stocks(context, industry_code, n):
    """
    获取指定行业内，经过基本过滤后，市值最大的N只股票
    """
    # 获取行业成分股
    stocks = get_industry_stocks(industry_code, date=context.previous_date)
    
    # 过滤停牌、ST、上市时间不足的股票
    stocks = filter_stocks(context, stocks)
    
    if not stocks:
        return []
    
    # 查询市值
    q = query(
        valuation.code,
        valuation.market_cap
    ).filter(
        valuation.code.in_(stocks)
    ).order_by(
        valuation.market_cap.desc() # 按市值降序
    ).limit(n)
    
    df = get_fundamentals(q, date=context.previous_date)
    
    if df.empty:
        return []
        
    return df['code'].tolist()

def filter_stocks(context, stock_list):
    """
    股票过滤函数：去停牌、去ST、去上市不足180天
    """
    curr_data = get_current_data()
    valid_stocks = []
    
    for stock in stock_list:
        # 1. 过滤停牌
        if curr_data[stock].paused:
            continue
            
        # 2. 过滤ST
        if curr_data[stock].is_st:
            continue
            
        # 3. 过滤上市时间不足180天
        # 获取股票信息
        info = get_security_info(stock)
        if not info:
            continue
        # 计算上市天数
        days_listed = (context.current_dt.date() - info.start_date).days
        if days_listed < 180:
            continue
            
        # 4. 过滤科创板/创业板 (可选，这里暂时保留，如果不想买可以取消注释)
        # if stock.startswith('688') or stock.startswith('300'):
        #     continue
            
        valid_stocks.append(stock)
        
    return valid_stocks

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)

策略Q&A

Q: 为什么不直接交易行业指数？
A: A股市场中，申万一级行业指数本身不可直接交易。虽然有对应的ETF，但并非所有31个行业都有流动性好的ETF覆盖。因此，策略采用“选出强势行业 -> 买入该行业龙头股”的方式来模拟行业轮动。

Q: 动量计算为什么用“市值前10的平均涨幅”而不是所有成分股？
A: 申万一级行业中某些行业包含几百只股票，如果每次计算都获取所有股票的历史行情，回测速度会非常慢且容易超时。市值最大的前10只股票通常能代表该行业的整体走势，计算效率更高。

Q: 如何调整策略的敏感度？
A: 可以修改 g.lookback_days。

• 设置为 20（约1个月）：捕捉短期热点，轮动快，但可能受噪音干扰。
• 设置为 60（约3个月）或 120（约半年）：捕捉中期趋势，相对稳健。

Q: 资金是如何分配的？
A: 策略采用等权重模型。假设选出3个行业，每个行业买3只股，共9只股，每只股票分配总资金的 1/9。