JoinQuant 实战：如何利用 get_concept_stocks 交易热门概念？

这是一个基于聚宽（JoinQuant）API 实现的锂电池概念股动量轮动策略。

策略思路

1. 获取股票池：在初始化时，通过 get_concepts 查找“锂电池”对应的概念代码，并使用 get_concept_stocks 获取成分股。
2. 过滤筛选：剔除停牌、ST、*ST 以及上市时间不足的股票。
3. 轮动因子：使用20日涨跌幅（动量因子）作为排序依据。
4. 交易逻辑：
   • 选取涨幅最高的 N 只股票（例如前 5 只）。
   • 采用等权重买入。
   • 定期（例如每 5 个交易日）进行调仓：卖出不在目标列表中的股票，买入新选中的股票。

策略代码

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

def initialize(context):
    """
    初始化函数，设定基准、全局变量、手续费等
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    # 设定持仓数量
    g.stock_num = 5
    # 设定调仓周期（天）
    g.period = 5
    # 记录运行天数
    g.days = 0
    # 目标概念名称
    g.concept_name = '锂电池'
    
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 获取锂电池概念的股票池（初始化时获取一次，也可以在rebalance中动态获取）
    g.target_concept_code = get_lithium_concept_code(g.concept_name)
    if g.target_concept_code:
        log.info("成功获取[%s]概念代码: %s" % (g.concept_name, g.target_concept_code))
    else:
        log.error("未找到[%s]概念代码，请检查名称" % g.concept_name)

    # 每天开盘时运行
    run_daily(rebalance, time='09:30')

def get_lithium_concept_code(name):
    """
    辅助函数：根据中文名称获取概念板块代码
    """
    all_concepts = get_concepts()
    # 筛选名称匹配的行
    target = all_concepts[all_concepts['name'] == name]
    if not target.empty:
        return target.index[0]
    return None

def filter_stocks(context, stock_list):
    """
    股票过滤函数：去停牌、去ST、去上市时间过短
    """
    curr_data = get_current_data()
    filtered_list = []
    
    for stock in stock_list:
        # 1. 过滤停牌
        if curr_data[stock].paused:
            continue
        # 2. 过滤ST, *ST
        if curr_data[stock].is_st:
            continue
        # 3. 过滤涨跌停（可选，防止无法买入卖出，此处暂不严格过滤以便计算排名）
        # if curr_data[stock].high_limit == curr_data[stock].last_price or curr_data[stock].low_limit == curr_data[stock].last_price:
        #     continue
            
        # 4. 过滤上市时间不足60天的次新股
        security_info = get_security_info(stock)
        if not security_info or (context.current_dt.date() - security_info.start_date).days < 60:
            continue
            
        filtered_list.append(stock)
    
    return filtered_list

def get_momentum_factor(stock_list, days=20):
    """
    计算动量因子：过去N天的涨跌幅
    """
    # 获取历史收盘价，多取1天用于计算收益率
    # history返回的是DataFrame，列是股票代码
    h = history(days + 1, '1d', 'close', stock_list)
    
    momentum_dict = {}
    for stock in stock_list:
        # 确保数据足够
        if stock in h.columns and len(h[stock]) == days + 1:
            # (当前价格 - N天前价格) / N天前价格
            # 注意：history不包含当天实时价格，这里使用的是昨天收盘价相对于(N+1)天前收盘价的涨幅
            # 如果需要包含今天的实时价格，需要用 get_current_data
            prev_close = h[stock].iloc[0]
            curr_close = h[stock].iloc[-1]
            
            if prev_close > 0:
                ret = (curr_close - prev_close) / prev_close
                momentum_dict[stock] = ret
    
    return momentum_dict

def rebalance(context):
    """
    调仓主逻辑
    """
    g.days += 1
    # 判断是否达到调仓周期
    if g.days % g.period != 1:
        return

    if not g.target_concept_code:
        return

    log.info("--- 开始调仓 ---")
    
    # 1. 获取概念成分股
    concept_stocks = get_concept_stocks(g.target_concept_code, date=context.current_dt)
    
    # 2. 基础过滤
    valid_stocks = filter_stocks(context, concept_stocks)
    
    # 3. 计算动量因子
    momentum_scores = get_momentum_factor(valid_stocks, days=20)
    
    # 4. 排序：按涨幅从大到小排序
    # sorted返回的是列表，元素是(stock, score)的元组
    sorted_stocks = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    
    # 5. 选取前N只
    target_list = [x[0] for x in sorted_stocks[:g.stock_num]]
    log.info("目标持仓: %s" % target_list)
    
    # 6. 交易操作
    # 6.1 卖出不在目标列表中的股票
    for stock in context.portfolio.positions:
        if stock not in target_list:
            order_target_value(stock, 0)
    
    # 6.2 买入目标股票
    if len(target_list) > 0:
        # 等权重分配资金
        position_count = len(target_list)
        # 获取当前总资产（包含现金和持仓）
        total_value = context.portfolio.total_value
        # 每只股票的目标市值
        target_value = total_value / position_count
        
        for stock in target_list:
            order_target_value(stock, target_value)

代码关键点解析

1. 动态获取概念代码 (get_lithium_concept_code)：

   • 聚宽的概念代码（如 GNxxx）可能会随时间调整。代码中通过 get_concepts() 获取所有概念板块的 DataFrame，然后筛选 name 为“锂电池”的行，动态获取其 index（即概念代码）。这样比硬编码代码更稳健。

2. 数据获取 (get_concept_stocks)：

   • 使用 get_concept_stocks(code, date=context.current_dt) 获取指定日期该板块的成分股。这避免了“幸存者偏差”（即只选到了现在属于该板块的股票，而忽略了历史上曾属于该板块但后来被剔除的股票）。

3. 动量计算 (get_momentum_factor)：

   • 使用 history 函数获取过去 21 天的收盘价。
   • 计算公式为：(昨天收盘价 - 20天前收盘价) / 20天前收盘价。
   • 这是经典的动量轮动逻辑：强者恒强。

4. 资金管理 (order_target_value)：

   • 使用 order_target_value 配合 context.portfolio.total_value / position_count 实现等权重轮动。
   • 该函数会自动计算需要买入或卖出的股数，非常适合轮动策略。

常见问题 Q&A

Q: 如何查看“锂电池”概念的具体代码是什么？
A: 在聚宽的研究环境中，运行 get_concepts() 可以得到一个包含所有概念的表格。你可以使用 df[df['name']=='锂电池'] 来查看具体代码。通常是 GN038 或类似代码，但建议使用代码中的动态查找方式。

Q: 为什么回测速度比较慢？
A: 概念股数量较多（通常有几十甚至上百只），每天计算所有成分股的 20 日涨跌幅并排序需要一定时间。可以通过增加调仓周期（如 g.period = 10）或在 filter_stocks 中先剔除市值过小的股票来减少计算量。

Q: 如何加入止损逻辑？
A: 可以在 rebalance 函数之外，增加一个 handle_data 或 run_daily 监控函数。遍历当前持仓 context.portfolio.positions，如果 (current_price - avg_cost) / avg_cost < -0.1 (亏损10%)，则执行 order_target(stock, 0) 进行止损。