基于 AD (Accumulation/Distribution) 指标的 QMT 策略

AD（Accumulation/Distribution Line，累积/派发线）是一种基于成交量和价格的动量指标，用于评估资金是流入（累积）还是流出（派发）证券。

策略逻辑

由于 AD 指标本身是一个累积值，其绝对数值意义不大，主要观察其趋势和背离。为了在量化策略中实现自动化交易，我们通常采用 AD 指标与其移动平均线（MA）的交叉 作为交易信号，这种方法比单纯的背离判断更具操作性。

策略规则如下：

1. 计算 AD 值：利用最高价、最低价、收盘价和成交量计算每日 AD 值。
2. 计算 AD 均线：计算 AD 值的 $N$ 周期简单移动平均（SMA），例如 20 日均线。
3. 买入信号：当 AD 值上穿 AD 均线时，视为资金流入增强，执行买入。
4. 卖出信号：当 AD 值下穿 AD 均线时，视为资金流出增强，执行卖出。

策略代码实现

以下是完整的 QMT Python 策略代码。代码使用了 talib 库来计算 AD 指标，并使用 ContextInfo.get_market_data_ex 获取数据。

# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import talib

def init(ContextInfo):
    """
    策略初始化函数
    """
    # 设置策略运行的股票列表 (示例：平安银行)
    ContextInfo.stock_code = '000001.SZ'
    ContextInfo.set_universe([ContextInfo.stock_code])
    
    # 设置策略参数
    ContextInfo.period = '1d'  # 周期：日线
    ContextInfo.ma_period = 20 # AD均线周期
    ContextInfo.account_id = '6000000000' # 请替换为您的实际资金账号
    
    # 设置交易账号
    ContextInfo.set_account(ContextInfo.account_id)
    
    print("AD策略初始化完成")

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前正在处理的K线索引
    index = ContextInfo.barpos
    
    # 获取当前图表的时间戳
    realtime = ContextInfo.get_bar_timetag(index)
    
    # 转换时间戳为可读格式 (用于日志)
    # date_str = timetag_to_datetime(realtime, '%Y-%m-%d %H:%M:%S')
    
    # 获取历史行情数据
    # 我们需要足够的数据来计算MA，所以取 ma_period + 额外缓冲
    count = ContextInfo.ma_period + 50 
    
    # 使用 get_market_data_ex 获取数据 (推荐接口)
    # 注意：subscribe=True 在回测和实盘中会自动处理数据订阅
    data_map = ContextInfo.get_market_data_ex(
        fields=['high', 'low', 'close', 'volume', 'open'],
        stock_code=[ContextInfo.stock_code],
        period=ContextInfo.period,
        count=count,
        dividend_type='front', # 前复权
        fill_data=True,
        subscribe=True
    )
    
    if ContextInfo.stock_code not in data_map:
        return

    df = data_map[ContextInfo.stock_code]
    
    # 确保数据量足够计算指标
    if len(df) < ContextInfo.ma_period + 2:
        return

    # 提取数据为 numpy 数组，供 talib 使用
    high = df['high'].values
    low = df['low'].values
    close = df['close'].values
    volume = df['volume'].values
    
    # --- 计算指标 ---
    
    # 1. 计算 AD (Chaikin A/D Line)
    # talib.AD(high, low, close, volume)
    try:
        ad_line = talib.AD(high, low, close, volume)
    except Exception as e:
        print(f"计算AD指标出错: {e}")
        return

    # 2. 计算 AD 的移动平均线 (Signal Line)
    try:
        ad_ma = talib.SMA(ad_line, timeperiod=ContextInfo.ma_period)
    except Exception as e:
        print(f"计算AD均线出错: {e}")
        return
        
    # 获取最新和上一个周期的指标值
    # -1 代表当前K线（最新），-2 代表上一根K线
    current_ad = ad_line[-1]
    prev_ad = ad_line[-2]
    
    current_ma = ad_ma[-1]
    prev_ma = ad_ma[-2]
    
    # 检查是否包含无效值 (NaN)
    if np.isnan(current_ad) or np.isnan(current_ma) or np.isnan(prev_ad) or np.isnan(prev_ma):
        return

    # --- 交易逻辑 ---
    
    # 获取当前持仓
    positions = ContextInfo.get_position(ContextInfo.stock_code, ContextInfo.account_id)
    current_holding = 0
    if positions:
        # 注意：get_position 返回的是对象列表，通常取第一个
        current_holding = positions[0].m_nVolume

    # 信号判断：金叉 (AD 上穿 MA)
    buy_signal = (prev_ad < prev_ma) and (current_ad > current_ma)
    
    # 信号判断：死叉 (AD 下穿 MA)
    sell_signal = (prev_ad > prev_ma) and (current_ad < current_ma)
    
    # --- 执行交易 ---
    
    # 只有在当前K线是最新K线时才下单（避免回测中重复下单逻辑，虽然handlebar机制保证了顺序）
    # 在实盘中，通常在K线结束或即时触发，这里演示标准逻辑
    
    if buy_signal:
        if current_holding == 0:
            print(f"产生买入信号: AD={current_ad:.2f}, MA={current_ma:.2f}")
            # 全仓买入 (示例：按最新价买入 1000 股)
            # opType: 23 (股票买入), orderType: 1101 (单股单账号普通下单)
            # price: -1 (使用策略设定的报价方式，如最新价)
            passorder(23, 1101, ContextInfo.account_id, ContextInfo.stock_code, 5, -1, 1000, ContextInfo)
            
    elif sell_signal:
        if current_holding > 0:
            print(f"产生卖出信号: AD={current_ad:.2f}, MA={current_ma:.2f}")
            # 清仓卖出
            # opType: 24 (股票卖出)
            passorder(24, 1101, ContextInfo.account_id, ContextInfo.stock_code, 5, -1, current_holding, ContextInfo)

    # 绘图 (可选，用于回测界面观察)
    # 注意：AD值通常很大，直接画在主图会压缩K线，建议画在副图
    # ContextInfo.paint('AD', current_ad, -1, 0)
    # ContextInfo.paint('AD_MA', current_ma, -1, 0)

代码关键点解析

1. 数据获取 (get_market_data_ex):

   • 我们使用了 get_market_data_ex 接口，这是 QMT 推荐的高效数据接口。
   • count 参数设置为 ma_period + 50，确保有足够的数据来计算移动平均线，避免计算结果为 NaN。
   • dividend_type='front' 设置为前复权，这对于技术指标计算非常重要，可以消除分红配股造成的价格跳空影响。

2. 指标计算 (talib):

   • QMT 内置了 talib 库。
   • talib.AD(high, low, close, volume)：直接计算累积/派发线。注意输入必须是 numpy 数组。
   • talib.SMA：对 AD 线进行平滑处理，生成交易信号线。

3. 交易信号:

   • 金叉：prev_ad < prev_ma 且 current_ad > current_ma。表示 AD 线从下方突破均线，资金流入加速。
   • 死叉：prev_ad > prev_ma 且 current_ad < current_ma。表示 AD 线从上方跌破均线，资金流出加速。

4. 下单函数 (passorder):

   • opType=23 代表股票买入，24 代表股票卖出。
   • orderType=1101 代表单股单账号普通交易。
   • prType=5 代表使用最新价下单（在回测中通常以收盘价成交）。

常见问题 (Q&A)

Q: 为什么 AD 指标有时候会一直上升，但股价却不涨？
A: AD 指标基于 CLV (Close Location Value)，如果收盘价持续接近当日最高价，即使成交量不大，AD 也会上升。这可能意味着有人在悄悄吸筹，但也可能只是市场流动性不足导致的虚假信号。这就是为什么要配合 MA (均线) 使用，或者结合 RSI 等其他指标进行过滤。

Q: 如何在 QMT 中查看 AD 指标的图形？
A: 在代码末尾的 ContextInfo.paint 函数可以用来绘图。但在 QMT 界面中，AD 的数值通常非常大（累积成交量），如果直接画在主图（和股价在一起）会导致 K 线被压缩成一条直线。建议在模型设置中将“位置”设置为“副图”。

Q: 这个策略可以直接用于实盘吗？
A: 提供的代码是一个基础的策略模板。在实盘前，您需要：

1. 修改 ContextInfo.account_id 为您的真实账号。
2. 添加风控逻辑（如止损、止盈）。
3. 考虑资金管理（如根据资金比例下单，而不是固定的 1000 股）。
4. 在 QMT 的回测模块中进行充分的历史回测。