我是Mr.看海，由于我开发的看海量化交易系统是基于miniQMT的行情和交易接口，所以在此我将miniQMT的使用教程采用案例集的形式进行讲解，该手册会不断完善和丰富，为做miniQMT的开发的朋友提供便利。
需要注意的是，本手册仅作为官方手册的辅助教程，最权威的接口请以官方说明文档为准：
官方手册请点击这里
1.实时数据获取的运行逻辑与演示案例
xtdata模块中，获取实时数据有三种常用方式：先下载再读取的方式、回调函数触发方式、获取市场快照方式，其中前两种方式需要先向服务器订阅(subscribe)数据，而第三种方式可以直接调用获取。
1.1 先下载再读取的方式
第一步：使用subscribe_quote()函数订阅单只股票的实时行情，或者使用subscribe_whole_quote()函数订阅全市场行情。经过订阅的数据实质上也是下载到了本地，这一点是和download_history_data函数一样的，只不过使用订阅的函数得到的数据会自动更新到本地。
第二步：使用get_market_data或get_market_data_ex函数，读取本地的数据。
需要注意的一点是，实时数据是从调用的那一刻起开始存储，比如从上午10点启动，那么9点30分到10点钟的数据本地是没有的，为了将数据补充完整，方便后续处理，可以先调用download_history_data补充数据，再启动订阅。当然如果你不需要用到之前的数据，那么直接订阅也可以。
先下载再读取的方式比较简单直观，不过实时性会稍微差一些。毕竟当调用get_market_data函数时，得到的数据可能已经是几秒钟前更新的了。
from xtquant import xtdata
import time
# 1.下载数据
xtdata.download_history_data(
stock_code="000004.SZ",
period="tick",
start_time="20241212", #这个时间要根据运行程序日期修改
end_time="20241212" #这个时间要根据运行程序日期修改，也可缺省
)
# 2.订阅数据
seq = xtdata.subscribe_quote(
stock_code="000004.SZ", # 平安银行
period="tick", # 1分钟K线
count=-1 # 获取当天所有数据
)
# 3.等待订阅完成
time.sleep(1)
# 4.定期获取数据
for i in range(5): # 演示获取5次
data = xtdata.get_market_data_ex(
field_list=["time", "open", "high", "low", "close", "volume"],
stock_list=["000004.SZ"],
period="tick"
)
print(f"\n第{i+1}次获取数据:")
print(data)
time.sleep(3) # 每3秒获取一次
# 5.取消订阅
xtdata.unsubscribe_quote(seq)
运行结果（因为程序是收盘后运行的，所以有全天的运行数据）：
第1次获取数据:
{'000004.SZ': time open high low close volume
20241213091500 1734052500000 0.00 0.00 0.00 NaN 0
20241213091509 1734052509000 0.00 0.00 0.00 NaN 0
20241213091518 1734052518000 0.00 0.00 0.00 NaN 0
20241213091527 1734052527000 0.00 0.00 0.00 NaN 0
20241213091536 1734052536000 0.00 0.00 0.00 NaN 0
... ... ... ... ... ... ...
20241213145924 1734073164000 20.03 20.84 19.65 NaN 348615
20241213145933 1734073173000 20.03 20.84 19.65 NaN 348615
20241213145942 1734073182000 20.03 20.84 19.65 NaN 348615
20241213145951 1734073191000 20.03 20.84 19.65 NaN 348615
20241213150000 1734073200000 20.03 20.84 19.65 NaN 351013
[4812 rows x 6 columns]}
1.2 回调函数触发方式
第一步：使用subscribe_quote()函数订阅单只股票的实时行情，或者使用subscribe_whole_quote()函数订阅全市场行情。
第二步：定义回调函数处理数据更新，订阅时设置callback参数，使用xtdata.run()保持程序运行，在回调函数中使用get_market_data或get_market_data_ex函数。
这种方式是通过回调函数来处理数据更新。需要注意的是，虽然我们可以设置period参数（如”1m”表示分钟K线），但回调函数的触发并不是按照这个时间间隔来的。实际上：
- 对于subscribe_quote：每当股票有新的成交时就会触发回调，period参数只是决定了返回数据的聚合周期
- 对于subscribe_whole_quote：每当任何股票有新的tick数据就会触发回调
这种方式能够及时捕获市场的每个变化，适合对实时性要求较高的场景。但需要注意回调函数的处理效率，避免阻塞后续数据的处理。经过实测，对于subscribe_quote订阅的股票，回调触发周期大概是3秒。这个应该是能给普通用户提供的服务限制所致，这也就制约了高频交易应用。不过话说回来，对于个人量化交易开发者来说，也基本不考虑高频交易这条路子了，即使接口能够提供支持，个人的网络和硬件条件也难以达标。
话说回来，下边的例子是使用回调函数获取实时数据的一个演示。案例中period=”1m”指的是聚合数据的周期，不是进入回调的周期，实测进入这个回调的周期也是3秒。对于实时性要求低的策略，使用先下载再读取的方式可能会更灵活。
from xtquant import xtdata
import time
# 1.定义回调函数
def on_data_update(data):
"""数据更新回调函数"""
# data是本次触发的数据(单条)
code_list = list(data.keys())
# 在回调中获取完整K线数据
klines = xtdata.get_market_data_ex(
field_list=["time", "open", "high", "low", "close", "volume"],
stock_list=code_list,
start_time="20241213",
period="1m"
)
print("\n数据更新:")
print(klines)
# 2.订阅数据并设置回调
xtdata.subscribe_quote(
stock_code="000001.SZ", # 平安银行
period="1m", # 1分钟K线
count=-1, # 获取当天所有数据
callback=on_data_update # 设置回调函数
)
# 3.保持程序运行
print("开始接收实时数据...")
try:
xtdata.run()
except KeyboardInterrupt:
print("程序结束")
运行结果（因为是收盘后运行的，只进入一次回调，其中有全天数据）：
开始接收实时数据...
数据更新:
{'000001.SZ': time open high low close volume
20241213132100 1734067260000 11.59 11.61 11.59 11.61 1735
20241213132200 1734067320000 11.60 11.61 11.60 11.60 942
20241213132300 1734067380000 11.60 11.60 11.60 11.60 0
20241213132400 1734067440000 11.60 11.60 11.60 11.60 0
20241213132500 1734067500000 11.60 11.60 11.60 11.60 0
... ... ... ... ... ... ...
20241213145600 1734072960000 11.56 11.57 11.56 11.57 9518
20241213145700 1734073020000 11.57 11.58 11.56 11.58 4688
20241213145800 1734073080000 11.58 11.58 11.58 11.58 154
20241213145900 1734073140000 11.58 11.58 11.58 11.58 0
20241213150000 1734073200000 11.56 11.56 11.56 11.56 21462
1.3 使用市场快照方式获取数据
除了订阅数据外，xtdata还提供了get_full_tick函数来获取市场的最新快照数据。这个函数可以一次性获取指定市场所有股票的最新tick数据，而不需要订阅，适合定期扫描市场。
from xtquant import xtdata
# 获取沪深两市的最新行情快照
snapshot = xtdata.get_full_tick(["SH", "SZ"])
# 打印数据
print(f"当前市场共{len(snapshot)}只股票")
if snapshot:
# 展示第一只股票的数据
first_stock = list(snapshot.keys())[0]
print(f"\n股票{first_stock}的tick数据:")
print(snapshot[first_stock])
运行结果：
当前市场共43070只股票
股票000001.SH的tick数据:
{'timetag': '20241213 15:00:09', 'lastPrice': 3391.878, 'open': 3442.926, 'high': 3442.926, 'low': 3390.754, 'lastClose': 3461.4998, 'amount': 850714863000, 'volume': 777464144, 'pvolume': 777464144, 'stockStatus': 0, 'openInt': 0, 'settlementPrice': 0, 'lastSettlementPrice': 0, 'askPrice': [0, 0, 0, 0, 0], 'bidPrice': [0, 0, 0, 0, 0], 'askVol': [0, 0, 0, 0, 0], 'bidVol': [0, 0, 0, 0, 0]}
2.相似函数的对比
上边提到的几个函数是比较相像的，这里集中讲一下他们的区别。
2.1 subscribe_quote与subscribe_whole_quote的对比
主要区别：
（1）订阅范围：
- subscribe_quote：订阅指定股票的行情
- subscribe_whole_quote：订阅全市场行情
（2）数据类型
- subscribe_quote：支持K线和tick数据
- subscribe_whole_quote：仅支持tick数据
（3）使用场景
- subscribe_quote：适合跟踪特定股票
- subscribe_whole_quote：适合做市场扫描、高频交易等
下面分别演示两种订阅方式：
（1）单只股票订阅示例
from xtquant import xtdata
import time
def on_quote_data(data):
"""单只股票数据回调"""
print("\n收到单只股票数据:")
print(data)
# 订阅平安银行的分钟K线
seq = xtdata.subscribe_quote(
stock_code="000001.SZ", # 平安银行
period="1m", # 1分钟K线
callback=on_quote_data # 回调函数
)
# 运行10秒后退出
time.sleep(10)
xtdata.unsubscribe_quote(seq)
print("程序结束")
运行结果：
收到单只股票数据:
{'000001.SZ': [{'time': 1734417420000, 'open': 11.52, 'high': 11.540000000000001, 'low': 11.52, 'close': 11.540000000000001, 'volume': 2627, 'amount': 3029681.0, 'settlementPrice': 0.0, 'openInterest': 13, 'dr': 1.0, 'totaldr': 98.9329302783749, 'preClose': 6.95196156517096e-310, 'suspendFlag': 1948377584}]}
收到单只股票数据:
{'000001.SZ': [{'time': 1734417420000, 'open': 11.52, 'high': 11.540000000000001, 'low': 11.52, 'close': 11.53, 'volume': 2647, 'amount': 3052754.0, 'settlementPrice': 0.0, 'openInterest': 13, 'dr': 1.0, 'totaldr': 98.9329302783749, 'preClose': 6.95196156517096e-310, 'suspendFlag': 1611694256}]}
（2）全市场订阅示例：
from xtquant import xtdata
import time
def on_market_data(data):
"""全市场数据回调"""
# 只打印股票数量和第一只股票的数据作为示例
print(f"\n收到{len(data)}只股票的数据")
if data: # 如果有数据
first_stock = list(data.keys())[0] # 获取第一只股票
print(f"第一只股票 {first_stock} 的数据:")
print(data[first_stock])
# 订阅沪深两市的tick数据
seq = xtdata.subscribe_whole_quote(
code_list=["SH", "SZ"], # 上海和深圳市场
callback=on_market_data
)
# 运行程序
print("开始接收数据...")
try:
xtdata.run()
except KeyboardInterrupt:
xtdata.unsubscribe_quote(seq)
print("程序结束")
运行结果：
开始接收数据...
收到23425只股票的数据
第一只股票 000001.SH 的数据:
{'time': 1734073209000, 'lastPrice': 3391.878, 'open': 3442.926, 'high': 3442.926, 'low': 3390.754, 'lastClose': 3461.4998, 'amount': 850714863000.0, 'volume': 777464144, 'pvolume': 777464144, 'stockStatus': 0, 'openInt': 0, 'transactionNum': 0, 'lastSettlementPrice': 0.0, 'settlementPrice': 0.0, 'pe': 0.0, 'askPrice': [0.0, 0.0, 0.0, 0.0, 0.0], 'bidPrice': [0.0, 0.0, 0.0, 0.0, 0.0], 'askVol': [0, 0, 0, 0, 0], 'bidVol': [0, 0, 0, 0, 0], 'volRatio': 0.0, 'speed1Min': 0.0, 'speed5Min': 0.0}
2.2 get_full_tick与subscribe_whole_quote的主要区别：
（1）调用方式
- get_full_tick：主动获取，单次调用返回当前市场快照
- subscribe_whole_quote：被动接收，需要设置回调函数
（2）使用场景
- get_full_tick：适合定期获取市场概况，如每分钟扫描一次市场
- subscribe_whole_quote：适合实时监控市场变化，如高频交易
（3）数据完整性
- get_full_tick：返回所有股票的最新状态
- subscribe_whole_quote：只推送有变化的股票数据
3.关于取消订阅
当我们不再需要接收实时数据时,应该及时取消订阅以释放资源。xtdata提供了unsubscribe_quote函数用于取消订阅:
from xtquant import xtdata
import time
def on_quote_data(data):
"""数据回调函数"""
print("\n收到数据:")
print(data)
# 订阅数据
seq = xtdata.subscribe_quote(
stock_code="000001.SZ",
period="1m",
callback=on_quote_data
)
try:
# 运行一段时间
print("开始接收数据...")
time.sleep(10) # 接收10秒数据
finally:
# 取消订阅
xtdata.unsubscribe_quote(seq)
print("已取消订阅")
需要注意的几点：
- 每次subscribe都会返回一个订阅号seq
- unsubscribe_quote需要传入对应的订阅号
- 建议在程序结束前主动取消订阅
- 可以在try-finally结构中确保订阅被正确取消
4.注意事项
- 使用前确保miniQMT已启动并正常运行
- 在使用或者测试上述接口时，要选在开盘时间内
- 注意控制订阅数量，避免数据处理压力过大
- 建议在回调函数中加入异常处理
- Level2行情需要额外申请开通
- 实时数据处理要注意效率，避免阻塞回调函数
以上就是XtQuant行情模块的主要功能和使用方法。通过这些案例,您可以快速上手使用xtdata获取各类行情数据。每个案例都是独立可行的完整代码,您可以直接复制使用并根据需要修改参数。
5. 附录2：函数接口说明
5.1 获取全推数据
get_full_tick(code_list)
- 释义
- 获取全推数据
- 参数
- 传入市场代码代表订阅全市场，示例：[‘SH’, ‘SZ’]
- 传入合约代码代表订阅指定的合约，示例：[‘600000.SH’, ‘000001.SZ’]
- code_list – 代码列表，支持传入市场代码或合约代码两种方式
- 返回
- dict 数据集 { stock1 : data1, stock2 : data2, … }
- 备注
- 无
5.2 订阅单股行情
subscribe_quote(stock_code, period='1d', start_time='', end_time='', count=0, callback=None)
- 释义
- 订阅单股的行情数据，返回订阅号
- 数据推送从callback返回，数据类型和period指定的周期对应
- 数据范围代表请求的历史部分的数据范围，数据返回后会进入缓存，用于保证数据连续，通常情况仅订阅数据时传count = 0即可
- 参数
- 回调定义形式为on_data(datas)，回调参数datas格式为 { stock_code : [data1, data2, …] }
- stock_code – string 合约代码
- period – string 周期
- start_time – string 起始时间
- end_time – string 结束时间
- count – int 数据个数
- callback – 数据推送回调
defon_data(datas): for stock_code in datas: print(stock_code, datas[stock_code]) - 返回
- 订阅号，订阅成功返回大于0，失败返回-1
- 备注
- 单股订阅数量不宜过多，详见 接口概述-请求限制
5.3 订阅全推行情
subscribe_whole_quote(code_list, callback=None)
- 释义
- 订阅全推行情数据，返回订阅号
- 数据推送从callback返回，数据类型为分笔数据
- 参数
- 回调定义形式为on_data(datas)，回调参数datas格式为 { stock1 : data1, stock2 : data2, … }
- 传入市场代码代表订阅全市场，示例：[‘SH’, ‘SZ’]
- 传入合约代码代表订阅指定的合约，示例：[‘600000.SH’, ‘000001.SZ’]
- code_list – 代码列表，支持传入市场代码或合约代码两种方式
- callback – 数据推送回调
defon_data(datas): for stock_code in datas: print(stock_code, datas[stock_code])
- 返回
- 订阅号，订阅成功返回大于0，失败返回-1
- 备注
- 订阅后会首先返回当前最新的全推数据
5.4 反订阅行情数据
unsubscribe_quote(seq)
- 释义
- 反订阅行情数据
- 参数
- seq – 订阅时返回的订阅号
- 返回
- 无
- 备注
- 无
开通miniQMT
如果你还没有开通miniQMT，可以点击此处开通，低费率、低门槛，欢迎咨询。
开发日志
我正在紧锣密鼓地开发系统，完善功能，点击此处了解最新进展。