🚀 AI 一键生成 joinquant 策略代码
立即体验
👉 AI 生成代码 👈

在JoinQuant聚宽中实现通达信DMA动态移动平均函数(附Python源码)

joinquant | 发布时间: 2026/6/5
以下内容由 EasyQuant 生成。

问题描述

有人实现过动态移动平均线吗?如通达信里DMA函数

知道如何实现通达信中的DMA函数吗?比如以下语句中的DMA

CC:=ABS((2*CLOSE+HIGH+LOW)/4-MA(CLOSE,20))/MA(CLOSE,20);
DD:=DMA(CLOSE,CC);
是talib里说的WMA么?

解决方案

首先回答您的疑问:通达信的 DMA 函数并不是 TA-Lib 中的 WMA

  • WMA (Weighted Moving Average) 是线性加权移动平均,它的权重是固定的、按时间线性递减的。
  • DMA (Dynamic Moving Average) 是动态移动平均,它的算法类似于指数移动平均(EMA),但是平滑系数(权重)是动态变化的。通达信中 DMA(X, A) 的算法逻辑为:Y = A * X + (1 - A) * Y',其中 Y' 是上一周期的 Y 值,且平滑系数 A 必须小于 1。

您给出的公式实际上是经典技术指标**济安线(GDX)**的核心算法。在 JoinQuant(聚宽)平台中,有两种方式可以实现您的需求:

方法一:直接使用聚宽内置的 GDX 指标

聚宽的 jqlib.technical_analysis 库中已经内置了 GDX 指标,其内部计算逻辑与您提供的公式完全一致(只是默认周期 N=30,您可以修改为 20)。

from jqlib.technical_analysis import GDX

# 获取 000001.XSHE 的 GDX 指标,设置 N=20
# 返回的 gdx_jax 即为您公式中的 DD (济安线)
gdx_jax, gdx_ylx, gdx_zcx = GDX('000001.XSHE', check_date='2023-10-10', N=20, M=9)

print(gdx_jax['000001.XSHE'])

方法二:使用 Python (Pandas/Numpy) 手动实现 DMA 函数

如果您想在自己的策略中灵活计算,可以利用 Pandas 编写一个通用的 DMA 函数。以下是完整的实现代码:

import pandas as pd
import numpy as np

def calc_DMA(series_x, series_a):
    """
    实现通达信的 DMA(X, A) 函数
    :param series_x: 价格序列 (如 Close)
    :param series_a: 动态平滑系数序列 (如 CC)
    :return: DMA 序列
    """
    # 将输入转换为 numpy 数组以提高计算速度
    x = series_x.values
    a = series_a.values
    dma = np.zeros_like(x)
    
    # 初始值通常取 X 的第一个有效值
    dma[0] = x[0]
    
    # 迭代计算:Y = A * X + (1 - A) * Y'
    for i in range(1, len(x)):
        # 确保平滑系数 A 在 0~1 之间(通达信底层逻辑)
        weight = max(min(a[i], 1.0), 0.0)
        dma[i] = weight * x[i] + (1 - weight) * dma[i-1]
        
    return pd.Series(dma, index=series_x.index)

# --- 测试与应用 --- 
# 假设在 handle_data 或本地研究环境中获取数据
df = attribute_history('000001.XSHE', 60, '1d', ['close', 'high', 'low'])

close = df['close']
high = df['high']
low = df['low']

# 1. 计算 MA(CLOSE, 20)
ma20 = close.rolling(window=20).mean()

# 2. 计算 CC:=ABS((2*CLOSE+HIGH+LOW)/4-MA(CLOSE,20))/MA(CLOSE,20)
cc = np.abs((2 * close + high + low) / 4 - ma20) / ma20

# 3. 计算 DD:=DMA(CLOSE,CC)
# 注意:由于 MA20 前 19 天是 NaN,我们需要去除 NaN 后再计算 DMA
valid_idx = cc.dropna().index
dd = calc_DMA(close.loc[valid_idx], cc.loc[valid_idx])

print(dd.tail())

总结

  • 不要用 TA-Lib 的 WMA 来替代 DMA,两者的数学逻辑完全不同。
  • 如果您只是想用济安线,直接调包 jqlib.technical_analysis.GDX 最简单。
  • 如果您想复刻通达信的自定义公式,使用上述的 calc_DMA 循环迭代函数即可完美还原通关。