QMT 在不同场景下使用的获取数据接口方式？可以直接上手模拟或实盘复现起来！

QUANT 2025-11-24 13:32:55 76 举报

通过前几篇对这几个接口的详细了解，那么我们如何运用到策略中去？先让我们再来回顾一下这几个接口：

downloda_history_data（下载历史数据）https://www.lianghuaba.net/thread-65.htm

Contextinfo._get_market_data_ex (获取数据）https://www.lianghuaba.net/thread-86.htm

get_full_tick （获取实时全推数据）https://www.lianghuaba.net/thread-108.htm

进行三个接口的对比，在实际使用过程中更好的用对接口。

目标：我们想开发一个经典的“双均线金叉/死叉”策略，并需要用过去3个月的日线数据来验证策略的有效性（即回测）

在“策略执行前”准备好充足的“粮草”（历史数据）。一次性将3个月的数据下载到本地，后续的回测和分析就可以直接从本地硬盘读取，速度极快，且不依赖网络连接。在策略的 init 初始化函数中，或者在一个专门用于数据准备的单独脚本中，在策略正式运行回测开始前，一次性调用。

#encoding:gbk
import pandas as pd
import numpy as np
import talib
 
def init(ContextInfo):
    print("策略初始化，开始准备数据...")
    # 下载 '600519.SH' 的日线数据
    stock_code = "600519.SH"
    period = "1d"
    start_time = "20250822"
    end_time = "20251122"
    
    try:
        result = download_history_data(stock_code, period, start_time, end_time)
 
        if result:
            print(f"{stock_code} 的历史数据已准备就绪！")
        else:
            print(f"数据下载失败，请检查网络或参数！")
            
    except Exception as e:
        print(f"数据下载出错: {e}")
        # 打印详细的错误信息
        import traceback
        print(traceback.format_exc())
 
def handlebar(ContextInfo):
    # 这里添加你的策略逻辑
    pass

将本段代码我们复制到qmt去，看看输出结果：

总结：凡是需要进行历史回测或基于长周期历史数据的统计分析，第一步永远是使用 download_history_data 把数据准备好

我们的双均线策略参数及股票代码已经准备完成，现在要投入实盘运行。

策略逻辑是：在每根分钟K线走完时，判断短期均线是否上穿长期均线，如果形成“金叉”，则买入。

在此场景下，我们应该使用：ContextInfo.get_market_data_ex

逻辑是在策略的 handlebar 循环函数中，每次 handlebar 被触发时（即每根分钟K线走完形成时）调用。

# coding:gbk
def handlebar(ContextInfo):
    # 策略在每分钟的k线结束时运行
    stock_code = "600519.SH"
    period = "1d"
    
    # 获取最新的100根1d的K线收盘价数据
    # 注意：这里subscribe=True（默认），QMT会自动将实时行情拼接到历史数据中
    close_prices_df = ContextInfo.get_market_data_ex(
        fields=["close"], 
        stock_code=[stock_code], 
        period=period, 
        count=100
    )
    
    if stock_code in close_prices_df:
        close_prices = close_prices_df[stock_code]["close"]
        
        # 计算短期和长期均线
        short_ma = close_prices.iloc[-10:].mean() # MA10
        long_ma = close_prices.iloc[-60:].mean()  # MA60
        
        print(f"短期均线: {short_ma:.2f}, 长期均线: {long_ma:.2f}")
        
        # 判断金叉逻辑 (此处仅为示例)
        if short_ma > long_ma:
            print("形成金叉信号，考虑买入！")
            # 此处添加下单代码   passorder

将本段代码我们复制到qmt去，看看输出结果：

总结：所有基于固定K线周期（分钟、小时、日线等）进行判断的常规技术指标策略，都应该在 handlebar 中使用 get_market_data_ex 来获取分析所需的K线序列.

但在实际策略交易中，我们可能会对价格相对的敏感，策略需要对市场做出“即时反应”。

比如我们持有一只股票，并为其设置了一个严格的止损线，比如 1470元。我希望一旦股价跌破1470元，就必须在第一时间卖出，而不是等到当前这根K线走完（不然跌的会更加的多），我们需要捕捉到市场的每一个价格变动（即分笔成交Tick），而不是等待周期性的K线一分钟走完。

在此场景下，我们应该使用：ContextInfo.get_full_tick

代码示例 (handlebar 函数中，高频运行):

# coding:gbk
def handlebar(ContextInfo):
    # 假设策略持仓 '600519.SH'，止损价为 1470
    stock_to_monitor = "600519.SH"
    stop_loss_price = 1470
    
    # 获取最新的实时分笔成交数据
    latest_ticks = ContextInfo.get_full_tick([stock_to_monitor])
    
    if stock_to_monitor in latest_ticks:
        # 获取最新成交价
        last_price = latest_ticks[stock_to_monitor].get("lastPrice", 0)
        
        if last_price > 0:
            print(f"{stock_to_monitor} 最新成交价: {last_price}")
        
            # 检查是否触发止损
            if last_price < stop_loss_price:
                print(f"紧急！价格跌破止损线 {stop_loss_price}，立即执行卖出！")
                # 此处添加立即市价卖出的下单代码

将本段代码我们复制到qmt去，看看输出结果：