3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在使用 xtdata.get_full_tick 获取实时 tick 数据时,如果发现获取到的数据不是最新的(例如下午调用却返回上午的数据),这通常是因为没有订阅该股票的实时行情。
get_full_tick 默认从本地缓存中读取最新的 tick 数据。如果没有订阅实时行情,本地缓存的数据就不会更新,从而导致获取到的数据停留在过去的某个时间点。
要获取最新的实时 tick 数据,必须先使用 xtdata.subscribe_quote 订阅目标股票的实时行情。
from xtquant import xtdata
import time
# 目标股票列表
stock_list = ['000967.SZ', '600084.SH']
# 1. 订阅实时行情
# period='tick' 表示订阅 tick 级别的数据
# count=-1 表示订阅最新的数据
for stock in stock_list:
xtdata.subscribe_quote(stock, period='tick', count=-1)
# 等待一小段时间,确保数据已经推送到本地缓存
time.sleep(1)
# 2. 获取最新的 tick 数据
res = xtdata.get_full_tick(stock_list)
print(res)
sleep 一下,或者使用回调函数(callback 参数)来处理推送过来的数据,以确保拿到的是最新数据。xtdata.unsubscribe_quote 取消订阅,以节省系统资源和网络带宽。相比于主动调用 get_full_tick,使用回调函数处理实时推送的数据更加高效和及时:
from xtquant import xtdata
# 定义回调函数
def on_data(datas):
for stock_code, data in datas.items():
print(f"收到 {stock_code} 的最新数据: {data}")
# 订阅并绑定回调函数
for stock in ['000967.SZ', '600084.SH']:
xtdata.subscribe_quote(stock, period='tick', count=-1, callback=on_data)
# 保持程序运行,以便持续接收数据
xtdata.run()