【迅投qmt系列】3、获取实时数据

1、基本思想

在 xtquant 中，要获取指定资产的实时数据，需要先订阅它（subscribe_quote），然后再用 gmd 函数获取（gmd）使用。
这个逻辑与处理历史数据的一样。

2、常用函数

xtdata.get_full_tick()				# 主动获取全推数据

xtdata.subscribe_whole_quote()		# 订阅全推数据（被动）
xtdata.subscribe_quote()			# 订阅单个资产，K线周期可选，但按tick更新（被动）

xtdata.unsubscribe_quote()			# 取消订阅
1
2
3
4
5
6

以下分别介绍。

2.1、主动获取实时数据

get_full_tick 函数被调用一次，就主动实时获取指定资产的全推数据一次。使用方法是：

from xtquant import xtdata

code_list = ['000001.SH', '159649.SZ']
data = xtdata.get_full_tick(code_list)
1
2
3
4

返回值是一个 dict，key 是各个字段的名称，value 是各个字段获取到的最新值。如果是非交易时间，则返回上个最近交易日的最后一个全推数据。

2.2、被动获取实时数据

subscribe_whole_quote

subscribe_whole_quote 一次性订阅多个资产，并被动地获得这些资产的全推数据，同时可以定义回调函数，以及时处理数据：

from xtquant import xtdata

code_list = ['000001.SH', '159649.SZ']

def on_data(datas):
    print(list(datas.keys()))

handle = xtdata.subscribe_whole_quote(code_list, on_data)
xtdata.run()	# 使代码处于一直运行状态，订阅模式下，实时行情数据会一直返回
1
2
3
4
5
6
7
8
9

注意事项：
a、代码运行后，code_list 中的资产哪个有新的行情数据产生，就立即单独传回来这个新数据，并立即在回调函数 on_data 中处理
b、返回的数据只能是最新的 tick 数据，具体包含哪些字段，请查阅迅投的知识库
c、数据返回的最高频率是三秒一次，如果某个资产一直没有新数据产生（比如交易清淡）那就一直不会有它的数据返回

subscribe_quote

subscribe_quote 一次订阅一个资产，但是能指定频率，因此可以返回 1m、5m、1h、1d、tick 等数据，用法是：

from xtquant import xtdata

eqiuty_code = '000001.SH'
period = '1m'

def on_data(data):
    print(list(data.keys()))

handle = xtdata.subscribe_quote(eqiuty_code, period, on_data)
xtdata.run()	# 使代码处于一直运行状态，订阅模式下，实时行情数据会一直返回
1
2
3
4
5
6
7
8
9
10

注意事项：
a、订阅的周期 period 取值与 【迅投qmt系列】2、历史数据获取 中 period 的取值一样。
b、period 的值如果不是tick，而是比如1m，那么返回的数据格式就是1m的K线，但是返回频率仍然取决于tick数据的频率，即最高三秒一次（交易清淡的话可能多秒都不会有一次），只是每次返回的分钟线收盘价会持续变化，最高最低价有可能变化
c、非投研版的账号，用本方式订阅的资产有数量限制。同一个资产不同周期，各自单独占用订阅数，不合并计算，但是不同进程
d、subscribe_quote 的资源开销相对 subscribe_whole_quote 要大，如果订阅的资产数量超过50只，如果订阅的是tick数据时，建议改用 subscribe_whole_quote 订阅全推tick数据 ，如果订阅的是K线数据，建议改用 set_kline_mirror_enabled 订阅全推K线数据，这样得到的K线数据也是按tick更新的。

2.3、退订

订阅模式下，后台会有进程监听、返回订阅资产的数据，只要不退订，这个行为就一直会持续，如果订阅量大，系统资源会被大量占用。退订就由 unsubscribe_quote 来完成。

from time import sleep
from xtquant import xtdata

code_list = ['000001.SH', '159649.SZ']

def on_data (datas):
    print(datas)

handle = xtdata.subscribe_whole_quote(code_list, on_data)

time.sleep(10)

xtdata.unsubscribe_quote(handle)
print('退订成功')
1
2
3
4
5
6
7
8
9
10
11
12
13
14

注意事项：
a、因为只能按订阅号退订，如果在订阅了多只资产后，希望只退订其中部分资产，只能使用 subscribe_quote 实现，无法先全推订阅后再单独退订、新订。

3、小综合：同时获取到拼接完整的历史数据和实时行情数据

show me the code:

# 使用 miniqmt 的模式
from time import sleep
from xtquant import xtdata

start_date = '20240423090000'  # 格式"YYYYMMDD"，开始下载的日期，date = ""时全量下载
end_date = ""
period = "1m"
code_list = ["000001.SZ", "159649.SZ"]  # 股票列表

n = 1
num = len(stock_list)
for i in stock_list:
    print(f"当前正在下载{n}/{num}")
    xtdata.download_history_data(i, period, start_date, end_date)
    n += 1
print("下载任务结束")

for i in stock_list:
    xtdata.subscribe_quote(i, period=period)
sleep(1)  # 等待订阅完成

data = xtdata.get_market_data_ex(
	[], 
	code_list, 
	period=period, 
	start_time=start_date, 
	end_time=end_date, 
	count=-1
)  # count 设置为1，使返回值只包含最新行情

print(data[code_list[0]].tail())  # 行情数据查看
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31