通达信公式迁移聚宽：MACD、KDJ、CCI 指标如何转换为 Python 策略？

在量化交易中，将通达信等传统行情软件的指标公式迁移到聚宽（JoinQuant）平台是构建系统化交易策略的第一步。聚宽提供了强大的 Python API 和内置的技术指标库，可以非常方便地实现这一转换。

一、 迁移的核心思路与工具

通达信公式通常是基于时间序列的向量化计算，而在聚宽中，我们主要有以下两种方式进行迁移：

1. 使用聚宽内置的技术分析库 (jqlib.technical_analysis)：聚宽已经将常用的通达信、东财指标封装成了 Python 函数，支持直接调用。
2. 使用 Pandas/Numpy 自定义计算：对于复杂的自定义通达信公式，可以通过 attribute_history 或 get_bars 获取历史 K 线数据，再利用 Pandas 的滚动窗口函数（如 .rolling()、.ewm()）进行复现。

二、 核心指标转换与 Python 策略实现

1. MACD 指标转换

通达信公式：

DIF:EMA(CLOSE,12)-EMA(CLOSE,26);
DEA:EMA(DIF,9);
MACD:(DIF-DEA)*2,COLORSTICK;

聚宽 Python 策略实现：
聚宽因子库中已内置 MACDC 因子，或者你可以直接在策略中通过 Pandas 计算：

import pandas as pd
import numpy as np

def get_macd(security, count=100):
    # 获取收盘价历史数据
    close = attribute_history(security, count, '1d', ['close'])['close']
    # 计算EMA
    ema12 = close.ewm(span=12, adjust=False).mean()
    ema26 = close.ewm(span=26, adjust=False).mean()
    dif = ema12 - ema26
    dea = dif.ewm(span=9, adjust=False).mean()
    macd = (dif - dea) * 2
    return dif[-1], dea[-1], macd[-1]

2. KDJ 指标转换

通达信公式：

RSV:=(CLOSE-LLV(LOW,9))/(HHV(HIGH,9)-LLV(LOW,9))*100;
K:SMA(RSV,3,1);
D:SMA(K,3,1);
J:3*K-2*D;

聚宽 Python 策略实现：
通达信的 SMA(X,N,M) 算法等价于指数移动平均，其权重为 M/N。在 Python 中可以通过以下方式精确复现：

def get_kdj(security, N=9, M1=3, M2=3):
    # 获取高、低、收历史数据
    hist = attribute_history(security, N + 15, '1d', ['high', 'low', 'close'])
    close = hist['close']
    low_min = hist['low'].rolling(N).min()
    high_max = hist['high'].rolling(N).max()
    
    rsv = (close - low_min) / (high_max - low_min) * 100
    rsv = rsv.dropna()
    
    # 精确复现通达信 SMA(X,N,1) -> ewm(alpha=1/N)
    k = rsv.ewm(alpha=1.0/M1, adjust=False).mean()
    d = k.ewm(alpha=1.0/M2, adjust=False).mean()
    j = 3 * k - 2 * d
    return k.iloc[-1], d.iloc[-1], j.iloc[-1]

3. CCI 指标转换

通达信公式：

TYP:=(HIGH+LOW+CLOSE)/3;
CCI:(TYP-MA(TYP,N))/(0.015*AVEDEV(TYP,N));

聚宽 Python 策略实现：
聚宽因子库中直接支持 CCI10、CCI20 等因子。若需自定义周期 $N$，可使用如下代码：

def get_cci(security, N=14):
    hist = attribute_history(security, N + 10, '1d', ['high', 'low', 'close'])
    typ = (hist['high'] + hist['low'] + hist['close']) / 3
    
    ma_typ = typ.rolling(N).mean()
    # 计算平均绝对偏差 (AVEDEV)
    avedev = typ.rolling(N).apply(lambda x: np.fabs(x - x.mean()).mean(), raw=True)
    
    cci = (typ - ma_typ) / (0.015 * avedev)
    return cci.iloc[-1]

三、 完整交易策略模板（以 MACD 金叉买入死叉卖出为例）

你可以直接将以下代码复制到聚宽的策略编辑器中进行回测：

import jqdata

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 设定操作股票
    g.security = '000001.XSHE' 
    # 每天 09:30 运行
    run_daily(market_open, time='09:30')

def market_open(context):
    security = g.security
    # 计算当前 MACD 值
    dif, dea, macd = get_macd(security)
    
    # 获取当前持仓
    position = context.portfolio.positions[security]
    cash = context.portfolio.available_cash
    
    # 金叉买入：DIF 向上突破 DEA
    if dif > dea and position.total_amount == 0:
        order_value(security, cash)
        log.info("MACD金叉，买入: %s" % security)
        
    # 死叉卖出：DIF 向下突破 DEA
    elif dif < dea and position.closeable_amount > 0:
        order_target(security, 0)
        log.info("MACD死叉，卖出: %s" % security)

def get_macd(security):
    close = attribute_history(security, 100, '1d', ['close'])['close']
    ema12 = close.ewm(span=12, adjust=False).mean()
    ema26 = close.ewm(span=26, adjust=False).mean()
    dif = ema12 - ema26
    dea = dif.ewm(span=9, adjust=False).mean()
    macd = (dif - dea) * 2
    return dif[-1], dea[-1], macd[-1]

四、 如何验证计算结果与通达信一致？

1. 复权模式对齐：通达信默认展示的是前复权价格。在聚宽中，请确保在 initialize 中设置了 set_option('use_real_price', True)，并且在调用 attribute_history 或 get_price 时，复权参数 fq 保持默认的 'pre'（前复权）。
2. K线对齐：通达信的日线数据包含当天未收盘的实时数据。在聚宽回测中，attribute_history(..., '1d') 默认不包含当天未收盘的数据。若要在盘中获取包含当前周期的指标，请使用 get_bars(..., include_now=True)。
3. 算法细节对齐：
   • 通达信的 EMA 算法与 Pandas 的 ewm(adjust=False) 一致。
   • 通达信的 SMA(X, N, M) 必须转换为 ewm(alpha=M/N, adjust=False) 才能保证数值完全精确对齐。