JoinQuant 平台如何构建 CCI 顺势指标交易策略？（附源码）

这是一个基于 CCI (Commodity Channel Index) 顺势指标 的量化交易策略。

策略逻辑说明

CCI 指标主要用于衡量股价是否超出常态分布范围。

• CCI > 100：通常被视为进入超买区域，但在顺势策略中，这往往意味着多头力量强劲，是买入信号（趋势确立）。
• CCI < -100：通常被视为进入超卖区域，意味着空头力量强劲，是卖出信号（趋势走坏）。
• 介于 -100 到 +100 之间：视为震荡区间，不进行开仓操作。

策略实现细节

1. 股票池：选取沪深300成分股。
2. 参数设置：CCI计算周期为 14 日。
3. 买入条件：当股票的 CCI 指标大于 100，且当前持仓未满时，买入。
4. 卖出条件：当持仓股票的 CCI 指标小于 -100 时，清仓卖出。
5. 仓位管理：最多持有 5 只股票，每只股票等权重分配资金。
6. 风控：过滤停牌、ST股票及退市股票。

以下是完整的策略代码：

# -*- coding: utf-8 -*-
# 导入函数库
from jqdata import *
import talib
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')
    
    # --- 策略参数设置 ---
    # CCI计算周期
    g.cci_period = 14
    # 最大持仓数量
    g.max_hold_count = 5
    # 股票池代码 (沪深300)
    g.index_code = '000300.XSHG'
    
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 每天开盘时运行
    run_daily(market_open, time='09:30')

def market_open(context):
    """
    每日交易逻辑
    """
    # 1. 获取当前持仓列表
    current_holdings = list(context.portfolio.positions.keys())
    
    # 2. 获取股票池（沪深300成分股）
    stock_pool = get_index_stocks(g.index_code)
    
    # 3. 过滤掉停牌、ST、退市的股票
    current_data = get_current_data()
    stock_pool = [stock for stock in stock_pool 
                  if not current_data[stock].paused 
                  and not current_data[stock].is_st
                  and 'ST' not in current_data[stock].name 
                  and '*' not in current_data[stock].name]
    
    # --- 卖出逻辑 ---
    for stock in current_holdings:
        # 获取计算CCI所需的数据
        # 我们多取一些数据以确保talib计算稳定，这里取 cci_period * 2 天
        h_data = attribute_history(stock, g.cci_period * 2, '1d', ['high', 'low', 'close'], skip_paused=True)
        
        # 如果数据长度不足，跳过
        if len(h_data) < g.cci_period:
            continue
            
        # 计算CCI
        cci_value = calculate_cci(h_data, g.cci_period)
        
        # 卖出条件：CCI < -100 (趋势转弱/进入超卖区，顺势策略离场)
        if cci_value < -100:
            order_target_value(stock, 0)
            log.info("卖出 %s，CCI数值: %.2f" % (stock, cci_value))

    # --- 买入逻辑 ---
    # 更新持仓列表（因为刚才可能卖出了一些）
    current_holdings = list(context.portfolio.positions.keys())
    
    # 如果持仓已满，则不再选股买入
    if len(current_holdings) >= g.max_hold_count:
        return

    # 计算剩余可买入的股票数量
    buy_count = g.max_hold_count - len(current_holdings)
    
    # 待买入列表
    to_buy_list = []
    
    for stock in stock_pool:
        # 如果已经在持仓中，跳过
        if stock in current_holdings:
            continue
            
        # 获取历史数据
        h_data = attribute_history(stock, g.cci_period * 2, '1d', ['high', 'low', 'close'], skip_paused=True)
        
        if len(h_data) < g.cci_period:
            continue
            
        # 计算CCI
        cci_value = calculate_cci(h_data, g.cci_period)
        
        # 买入条件：CCI > 100 (进入强势区间)
        if cci_value > 100:
            to_buy_list.append((stock, cci_value))
    
    # 对待买入股票按CCI值从大到小排序（优先买入趋势最强的）
    # 注意：这里也可以改为随机选择或其他因子排序
    to_buy_list.sort(key=lambda x: x[1], reverse=True)
    
    # 执行买入
    for i in range(min(len(to_buy_list), buy_count)):
        stock = to_buy_list[i][0]
        cci_val = to_buy_list[i][1]
        
        # 资金分配：将剩余可用资金平均分配给需要买入的股票
        # 注意：这里简单处理，每次重新计算单只股票应分配资金
        # 实际操作中，可以用 总资产 / max_hold_count 来计算目标仓位
        
        # 获取当前账户总资产
        total_value = context.portfolio.total_value
        # 计算每只股票的目标持仓金额
        target_value = total_value / g.max_hold_count
        
        # 如果当前可用资金足够
        if context.portfolio.available_cash > 0:
            order_target_value(stock, target_value)
            log.info("买入 %s，CCI数值: %.2f" % (stock, cci_val))

def calculate_cci(df, n):
    """
    辅助函数：计算CCI指标
    :param df: 包含 high, low, close 的 DataFrame
    :param n: 周期
    :return: 最新的 CCI 值
    """
    # 将数据转换为 numpy 数组，talib 需要这种格式
    high = df['high'].values
    low = df['low'].values
    close = df['close'].values
    
    # 使用 talib 计算 CCI
    # talib.CCI 返回的是一个数组，我们取最后一个值（最新值）
    try:
        cci = talib.CCI(high, low, close, timeperiod=n)
        return cci[-1]
    except Exception as e:
        # 如果计算出错（例如数据中有NaN），返回0
        return 0

常见问题 Q&A

Q1: 为什么使用 talib 库而不是手动编写公式？
A: talib 是专业的金融技术指标库，计算效率极高且经过广泛验证，能有效处理边缘情况。JoinQuant 平台原生支持该库。

Q2: 为什么买入条件是 CCI > 100 而不是 CCI 上穿 -100？
A: CCI > 100 代表股价进入了“强势区”，这是一种典型的动量（Momentum）策略。上穿 -100 通常被视为均值回归策略的买点，但在 A 股市场，捕捉强势股的动量效应通常比抄底更稳健。

Q3: 如何调整策略的风险？
A:

• 降低风险：可以增加 g.max_hold_count（例如改为 10 或 20），分散持仓。
• 调整灵敏度：修改 g.cci_period。周期越短（如 7），信号越频繁但假信号多；周期越长（如 28），信号越滞后但更稳定。

Q4: 代码中的 attribute_history 为什么要取 g.cci_period * 2 的数据？
A: 虽然 CCI 计算只需要 N 天的数据，但技术指标计算通常需要一定的“预热”数据来消除初始值的偏差，或者确保在剔除停牌日期后仍有足够的数据进行计算。