QMT 平台 Python 策略：基于 DKX 多空线的均线交叉模型

这是一个基于 DKX（多空线）指标 的 QMT 量化交易策略。

策略原理

DKX（多空线）是一个趋势类指标，主要用于判断股价的中长期趋势。

1. MID 赋值：(3 * 收盘价 + 最低价 + 开盘价 + 最高价) / 6
2. DKX 线：MID 的 20 日加权移动平均（权重越近越大）。
3. MADKX 线：DKX 的 10 日简单移动平均。

交易信号：

• 买入（金叉）：当 DKX 线上穿 MADKX 线时，看多买入。
• 卖出（死叉）：当 DKX 线下穿 MADKX 线时，看空卖出。

QMT 策略代码

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

'''
策略名称：DKX多空线趋势策略
策略周期：日线 (1d)
'''

def init(ContextInfo):
    # ================= 策略参数设置 =================
    # 设置要操作的股票列表
    ContextInfo.stock_list = ['600000.SH', '000001.SZ'] 
    ContextInfo.set_universe(ContextInfo.stock_list)
    
    # 资金账号 (请替换为您自己的实盘或模拟账号)
    ContextInfo.account_id = 'YOUR_ACCOUNT_ID' 
    ContextInfo.account_type = 'STOCK'
    ContextInfo.set_account(ContextInfo.account_id)
    
    # DKX 参数
    ContextInfo.N = 20  # DKX计算周期
    ContextInfo.M = 10  # MADKX均线周期
    
    # 每次交易仓位比例 (0.5 代表 50%)
    ContextInfo.trade_ratio = 0.5 

def get_dkx_data(stock_code, context):
    """
    计算 DKX 指标数据
    """
    # 获取足够的历史数据，长度需要覆盖 N + M + 缓冲
    count = context.N + context.M + 20
    
    # 获取历史行情数据 (Open, High, Low, Close)
    # 注意：get_market_data_ex 返回的是 {code: dataframe}
    data_map = context.get_market_data_ex(
        ['open', 'high', 'low', 'close'], 
        [stock_code], 
        period='1d', 
        count=count, 
        dividend_type='front' # 前复权
    )
    
    if stock_code not in data_map:
        return None
        
    df = data_map[stock_code]
    
    if len(df) < count:
        return None

    # 1. 计算 MID
    # MID = (3*CLOSE + LOW + OPEN + HIGH) / 6
    df['mid'] = (3 * df['close'] + df['low'] + df['open'] + df['high']) / 6
    
    # 2. 计算 DKX (加权移动平均)
    # 权重为 1, 2, ..., 20，总和为 210
    weights = np.arange(1, context.N + 1)
    sum_weights = np.sum(weights)
    
    def weighted_avg(x):
        return np.dot(x, weights) / sum_weights

    # 使用 rolling apply 计算加权平均
    df['dkx'] = df['mid'].rolling(window=context.N).apply(weighted_avg, raw=True)
    
    # 3. 计算 MADKX (DKX 的 M 日简单移动平均)
    df['madkx'] = df['dkx'].rolling(window=context.M).mean()
    
    # 删除包含 NaN 的行
    df.dropna(inplace=True)
    
    return df

def handlebar(ContextInfo):
    # 获取当前 K 线索引
    index = ContextInfo.barpos
    # 获取当前时间
    realtime = ContextInfo.get_bar_timetag(index)
    current_date = timetag_to_datetime(realtime, '%Y-%m-%d')
    
    # 遍历股票池
    for stock in ContextInfo.stock_list:
        # 计算指标
        df = get_dkx_data(stock, ContextInfo)
        
        if df is None or len(df) < 2:
            continue
            
        # 获取最新一根K线（当前bar）和前一根K线的数据
        # iloc[-1] 是当前正在形成的 bar (如果是盘中) 或 最新收盘 bar
        # iloc[-2] 是前一根 bar
        
        curr_dkx = df['dkx'].iloc[-1]
        curr_madkx = df['madkx'].iloc[-1]
        
        prev_dkx = df['dkx'].iloc[-2]
        prev_madkx = df['madkx'].iloc[-2]
        
        # 获取当前持仓
        positions = get_trade_detail_data(ContextInfo.account_id, ContextInfo.account_type, 'POSITION')
        curr_pos = 0
        for pos in positions:
            if pos.m_strInstrumentID + '.' + pos.m_strExchangeID == stock:
                curr_pos = pos.m_nVolume
                break
        
        # ================= 交易逻辑 =================
        
        # 1. 金叉买入逻辑 (DKX 上穿 MADKX)
        # 前一天 DKX < MADKX 且 今天 DKX > MADKX
        if prev_dkx < prev_madkx and curr_dkx > curr_madkx:
            if curr_pos == 0:
                print(f"{current_date} [买入信号] {stock} 金叉: DKX={curr_dkx:.2f}, MADKX={curr_madkx:.2f}")
                # 目标市值下单：调整仓位到总资产的 trade_ratio 比例
                order_target_percent(stock, ContextInfo.trade_ratio, ContextInfo, ContextInfo.account_id)

        # 2. 死叉卖出逻辑 (DKX 下穿 MADKX)
        # 前一天 DKX > MADKX 且 今天 DKX < MADKX
        elif prev_dkx > prev_madkx and curr_dkx < curr_madkx:
            if curr_pos > 0:
                print(f"{current_date} [卖出信号] {stock} 死叉: DKX={curr_dkx:.2f}, MADKX={curr_madkx:.2f}")
                # 清仓
                order_target_percent(stock, 0, ContextInfo, ContextInfo.account_id)

代码使用说明

1. 设置账号：
   请务必将 init 函数中的 ContextInfo.account_id = 'YOUR_ACCOUNT_ID' 修改为您实际的 QMT 资金账号。

2. 设置标的：
   在 ContextInfo.stock_list 中修改您想要交易的股票代码列表。

3. 数据获取：
   代码使用了 get_market_data_ex 接口，这是 QMT 推荐的高效数据接口。为了计算 DKX，我们需要获取 Open, High, Low, Close 四个字段。

4. DKX 计算逻辑：
   DKX 的核心是加权移动平均。代码中使用了 numpy 的 rolling().apply() 方法来实现加权计算：

   • 权重数组 weights 生成为 [1, 2, ..., 20]。
   • rolling 窗口的最右侧（最新数据）对应权重的最大值（20），符合 DKX 对近期价格赋予更高权重的定义。

5. 交易执行：

   • 使用了 order_target_percent 函数。
   • 买入：当金叉发生且当前无持仓时，买入资金账户总资产的 50%（由 trade_ratio 控制）。
   • 卖出：当死叉发生且有持仓时，将仓位调整为 0（清仓）。

注意事项

• 回测与实盘：此代码可直接用于回测。若用于实盘，建议在 handlebar 中增加 ContextInfo.is_last_bar() 的判断，或者结合定时器，确保信号稳定（例如只在收盘前几分钟或收盘后确认信号）。
• 复权：代码中设置了 dividend_type='front'（前复权），这对于技术指标策略非常重要，可以避免分红除权导致的技术指标失真。

Q&A

Q: 如何调整 DKX 的灵敏度？
A: 修改 init 函数中的 ContextInfo.N。N 值越小，指标越灵敏，但虚假信号可能增多；N 值越大，趋势越稳定，但信号会有滞后。

Q: 为什么使用 get_market_data_ex 而不是 get_market_data？
A: get_market_data_ex 是 QMT 的新版接口，支持订阅和更灵活的数据提取，且返回格式更统一（字典包含 DataFrame），处理多只股票时效率更高。

Q: 可以在分钟线上运行吗？
A: 可以。只需将 get_market_data_ex 中的 period='1d' 修改为 period='5m' 或其他分钟周期，并在运行策略时选择相应的分钟周期即可。