深入分析MiniQMT实时订阅延迟测试代码

摘要

本文将深入分析两段MiniQMT平台的Python代码，这些代码用于测试实时数据订阅的延迟情况。我们将详细探讨代码的结构、关键功能以及它们在实时交易策略中的应用，并通过代码示例展示其工作原理。

背景介绍

MiniQMT是一个量化交易平台，支持股票、期货等多种金融产品的交易。它提供了丰富的API，允许开发者编写自定义的交易策略。实时数据订阅是量化交易中的关键环节，因为数据的延迟可能直接影响交易决策的准确性。

MiniQMT具体开通方法及要求，可以参看《QMT开通规则分享》

代码分析

订阅延迟测试代码一：`subscribe_quote_callback.py`

功能：该代码段通过设置订阅回调函数，实时接收股票的报价数据，并计算数据到达的延迟。

关键点：

订阅设置：
```
xtdata.subscribe_quote(stock_code=stock, period=period, count=1, callback=on_data)
```
这行代码订阅了指定股票的实时报价数据，period参数指定了数据的周期，count参数指定了订阅的数据条数，callback参数指定了接收数据时调用的回调函数。

回调函数：

def on_data(res, stock=stock):if not ready:returndff = calculate_seconds_difference(res[stock][0]['time'])if period != 'tick':close_value = res[stock][0]['close']else:close_value = res[stock][0]['lastPrice']logging.info(f'时间戳：{res[stock][0]["time"]}, 股票代码：{stock}, 当前价格：{close_value}, 延迟：{dff}')

回调函数on_data在接收到数据时被调用，计算当前时间与数据时间戳的差值（dff），从而得到延迟，并记录相关信息。

进度条显示：

def update_progress(progress):bar_length = 40  # 进度条长度block = int(round(bar_length * progress))text = f"\r[{'#' * block + '-' * (bar_length - block)}] {progress * 100:.2f}%"if progress < 1:print(text, end='', flush=True)else:print(text, flush=True)

update_progress函数用于显示订阅进度，通过控制台输出进度条。

延迟计算：

def calculate_seconds_difference(specified_time):current_time = datetime.now().timestamp()time_difference =  current_time - (specified_time / 1000)return time_difference

calculate_seconds_difference函数计算当前系统时间与数据时间戳的差值，并返回差值的秒数。

订阅延迟测试代码二：`subscribe_quote_get.py`

功能：与第一个代码段类似，但使用xtdata.get_market_data_ex函数主动获取数据，并通过调度器定期执行策略。

关键点：

订阅设置：
```
xtdata.subscribe_quote(stock_code=stock, period=period, count=1)
```
这行代码与第一个代码段类似，订阅了指定股票的实时报价数据。

数据获取：

def strategy(stock_list):logging.info('运行策略')scheduler = sched.scheduler(time.time, time.sleep)def task():scheduler.enter(interval, 1, task)res = xtdata.get_market_data_ex(stock_list=stock_list, period=period, count=1)for key in res.keys():if not res[key].empty:dff = calculate_seconds_difference(res[key].index[-1])if period != 'tick':close_value = res[key]['close'].iloc[-1]else:close_value = res[key]['lastPrice'].iloc[-1]if dff > 3:logging.info(f'时间戳：{res[key].index[-1]}, 股票代码：{key}, 当前价格：{close_value}, 延迟：{dff}')scheduler.enter(interval, 1, task)scheduler.run()

strategy函数通过调度器定期执行数据获取任务，获取最新的市场数据，并计算延迟。

进度条显示：

def update_progress(progress):bar_length = 40  # 进度条长度block = int(round(bar_length * progress))text = f"\r[{'#' * block + '-' * (bar_length - block)}] {progress * 100:.2f}%"if progress < 1:print(text, end='', flush=True)else:print(text, flush=True)

进度条显示功能与第一个代码段相同。

延迟计算：

def calculate_seconds_difference(time_str):specified_time = datetime.strptime(time_str, "%Y%m%d%H%M%S")current_time = datetime.now()time_difference =  current_time - specified_timeseconds_difference = time_difference.total_seconds()return seconds_difference