🚀 AI 一键生成 qmt 策略代码

如何在 QMT 副图中绘制当日的成交量分布图(筹码分布)?

qmt | 发布时间: 2025/12/19
以下内容由 EasyQuant 生成。

问题描述

如何统计一天内每个价格档位上的成交量,并把它画成一个横着的柱状图?

解决方案

在QMT中统计一天内每个价格档位的成交量(即筹码分布或成交量分布),最准确的方法是使用 Tick(分笔)数据

由于QMT的绘图函数 ContextInfo.paint 主要用于绘制随时间变化的序列指标(如K线、均线),它不支持直接绘制“横向”的直方图(即Y轴为价格,X轴为成交量的图表)。

为了实现您的需求,我将提供一段代码,它执行以下操作:

  1. 获取数据:获取当前主图品种当天的所有 Tick 数据。
  2. 数据处理:计算每一笔 Tick 的成交量差值(因为Tick的成交量是累加的),并按价格进行分组汇总。
  3. 可视化替代方案:由于无法画横向柱状图,我将使用 ContextInfo.draw_text 函数,在K线图的右侧对应价格位置上,直接标注该价格当天的总成交量。这能起到类似的效果,直观展示哪个价位成交最密集。

策略代码

# -*- coding: gbk -*-
import pandas as pd
import time

def init(ContextInfo):
    # 策略初始化,这里不需要特别设置,但必须存在
    pass

def handlebar(ContextInfo):
    # 只在最后一根K线(最新行情)进行计算和绘制,避免历史K线重复计算导致卡顿
    if not ContextInfo.is_last_bar():
        return

    # 1. 获取当前主图的品种代码
    stock_code = ContextInfo.stockcode + '.' + ContextInfo.market
    
    # 2. 获取当前K线的时间,确定"今天"的日期范围
    # 获取当前bar的时间戳(毫秒)
    timetag = ContextInfo.get_bar_timetag(ContextInfo.barpos)
    # 转换为结构化时间
    time_local = time.localtime(timetag / 1000)
    # 格式化为 YYYYMMDD
    current_date = time.strftime("%Y%m%d", time_local)
    
    # 设置Tick数据的起止时间 (当天全天)
    start_time = current_date + '000000'
    end_time = current_date + '235959'

    # 3. 获取Tick数据
    # 使用 get_market_data_ex 获取分笔数据
    # 注意:Tick数据量较大,如果品种非常活跃,计算可能会有轻微延迟
    tick_data_dict = ContextInfo.get_market_data_ex(
        fields=['lastPrice', 'volume'],  # 只需要最新价和成交量
        stock_code=[stock_code], 
        period='tick', 
        start_time=start_time, 
        end_time=end_time,
        count=-1,
        subscribe=False
    )

    if stock_code not in tick_data_dict:
        return

    df_tick = tick_data_dict[stock_code]

    if df_tick.empty:
        return

    # 4. 数据处理:计算每个价格的成交量
    # Tick数据中的 volume 是当日累计成交量,需要做差分计算单笔成交量
    # diff() 计算当前行与上一行的差值,第一行会是NaN,用第一行的原始值填充(fillna)
    df_tick['single_vol'] = df_tick['volume'].diff().fillna(df_tick['volume'])
    
    # 过滤掉成交量小于等于0的数据(有时候Tick数据会有噪点或非交易Tick)
    df_tick = df_tick[df_tick['single_vol'] > 0]

    # 按价格分组求和
    price_volume_dist = df_tick.groupby('lastPrice')['single_vol'].sum()

    # 5. 绘图(使用文字标注模拟)
    # 遍历统计结果,在图表上绘制
    # 注意:为了防止文字过于密集,可以考虑只绘制成交量较大的价格档位,或者全部绘制
    
    # 获取当前图表可视范围的最高价和最低价(简单估算,防止画到屏幕外)
    # 这里直接全部绘制,QMT会自动处理重叠或用户缩放查看
    
    for price, vol in price_volume_dist.items():
        # 格式化显示文本,例如 "1000手"
        # vol 是股数,除以100转换为手
        vol_hand = int(vol / 100)
        display_text = "  <-- Vol: %d" % vol_hand
        
        # 使用 draw_text 在对应价格位置绘制
        # 参数说明: draw_text(条件, 价格位置, 文本)
        # 条件设为 True 也就是 1>0,表示一直显示
        ContextInfo.draw_text(True, price, display_text)

    # 可以在日志输出一下,方便核对
    print(f"日期: {current_date}, 价格档位数量: {len(price_volume_dist)}")

代码逻辑详解

  1. 数据获取 (get_market_data_ex):

    • 我们使用 period='tick' 来获取最精细的逐笔成交数据。
    • fields 选择了 lastPrice (最新价) 和 volume (累计成交量)。
    • 时间范围设定为当前K线所在的自然日。
  2. 成交量还原 (diff):

    • 交易所推送的 Tick 数据中,volume 字段通常是开盘以来的累计总成交量
    • 为了知道在某个特定价格 lastPrice 发生了多少成交,我们需要计算当前 Tick 和上一个 Tick 的成交量差值:df['volume'].diff()
  3. 分组统计 (groupby):

    • 利用 Pandas 的 groupby('lastPrice') 功能,将所有相同价格的单笔成交量加总,得到该价格当天的总成交量。
  4. 可视化 (draw_text):

    • 局限性说明:QMT 的 paint 函数是基于时间轴(X轴是时间)绘制竖向柱状图的。要画“横向”柱状图(X轴是量,Y轴是价)需要底层的绘图接口支持,目前 Python API 暂未开放此类自定义图层接口。
    • 解决方案:使用 ContextInfo.draw_text(True, price, text)。这个函数允许我们在指定的 Y 轴坐标(即价格)上写字。我们在每个有成交的价格旁边写上成交量(手),这样您在看K线图时,右侧就会出现一排数字,数字越大代表该价位成交越多,起到了分布图的作用。

如何使用

  1. 打开 QMT 客户端,进入【模型研究】或【策略编辑器】。
  2. 新建一个 Python 策略。
  3. 将上述代码完整复制进去。
  4. 点击【运行】或将策略加载到具体的 K 线图(如日线或分钟线)上。
  5. 注意:由于 Tick 数据量较大,建议在盘中实时运行或盘后复盘时使用。如果在非常长周期的历史K线上回测,可能会因为请求大量 Tick 数据而变慢。代码中加了 if not ContextInfo.is_last_bar(): return 限制,确保只在最新的那根K线(即当天)进行计算和显示。

Q&A

Q: 为什么不能画成真正的横向柱子(色块)?
A: QMT 的 Python API 主要侧重于量化策略逻辑和基于时间序列的指标绘制(如 MACD, KDJ)。paint 函数只能在对应的时间点(Bar)上画图。虽然可以使用 draw_vertline 画竖线,但没有提供画横向矩形或横向线条长度随变量变化的接口。使用文字标注是目前最清晰的替代方案。

Q: 为什么显示的成交量和行情软件上的“分时成交”对不上?
A: 代码中计算的是Tick级的统计。部分行情软件的“分时成交”或“筹码分布”可能是基于分钟线估算的,或者对 Tick 数据进行了切片采样。此代码使用的是全量 Tick 数据,理论上是最精确的“价格-成交量”对应关系。

Q: 如果价格档位太多,文字重叠怎么办?
A: 对于波动剧烈或高价股,价格档位可能非常多。可以在代码中加入逻辑,例如:将价格按 0.050.1 进行取整(分箱/Binning)后再 groupby,这样可以减少显示的行数,使数据更聚合。例如:df_tick['lastPrice'] = df_tick['lastPrice'].round(2)