【实战教程】Python量化交易入门:搞懂xtquant订阅机制,轻松获取实时行情数据!

EasyXT项目介绍

EasyXT是基于miniqmt中xtquant的二次开发封装库，旨在简化xtquant的使用，提供更友好的API接口。通过统一的接口设计、智能参数处理和完善的错误处理，让量化交易开发变得更加简单高效。

项目地址: https://github.com/quant-king299/EasyXT

🛠️ 环境准备

系统要求

• 操作系统：Windows 10/11（PowerShell 7）
• Python：3.9+（建议 3.10+），并将 Python 加入 PATH

ptrade/QMT账号获取指导

📱 还没有ptrade/QMT账号的朋友，可以扫码加我微信，全程指导搞定Ptrade/QMT账号！

📌 本文导读很多初学者在使用xtquant获取实时行情时，常常遇到"五档数据为空"、"回调函数不执行"等问题。本文将从订阅机制的核心原理出发，手把手教你正确获取实时行情数据，避坑又高效！

🔍 一、什么是订阅机制？

1.1 传统方式 vs 订阅方式

在讲解订阅机制之前，我们先来对比两种数据获取方式：

【传统方式】主动获取（拉取模式）

你的程序 → 主动请求数据 → 服务器 → 返回数据

• ✅ 优点：简单直接
• ❌ 缺点：可能有延迟，需要轮询

【订阅方式】推送模式

你的程序 → 订阅股票 → 服务器持续推送最新数据

• ✅ 优点：实时性好，服务器主动推送
• ❌ 缺点：需要理解回调函数

1.2 订阅机制的工作流程

┌─────────────────────────────────────────────┐│           订阅机制工作流程                    │├─────────────────────────────────────────────┤│ 1. subscribe_quote()  →  发送订阅请求        ││ 2. QMT服务器          →  接收订阅            ││ 3. 有新数据时          →  服务器主动推送      ││ 4. 客户端缓存          →  保存推送数据        ││ 5. get_full_tick()    →  从缓存读取最新数据  │└─────────────────────────────────────────────┘

1.3 三个关键点

⚠️ 订阅后数据不是立即返回的，而是异步推送⚠️ 需要等待一小段时间让数据推送完成⚠️ 使用回调函数可以实时接收每一笔推送

💡 类比理解：订阅模式就像订阅报纸——你订阅后，报社每天主动送报纸到家里，不需要每天去报社查询！

⚖️ 二、订阅 vs 主动获取：实战对比

2.1 方式1：主动获取（不推荐）

import xtquant.xtdata as xtcode = '000001.SZ'tick_data = xt.get_full_tick([code])if tick_data and code in tick_data:    tick = tick_data[code]print(f"最新价: {tick.get('lastPrice', 0):.2f}")# ❌ 五档数据可能为空（因为未订阅）

问题：五档数据可能获取不到！

2.2 方式2：先订阅再获取（推荐✅）

import xtquant.xtdata as xtimport timecode = '000001.SZ'# 步骤1: 订阅xt.subscribe_quote(code, period='tick')# 步骤2: 等待数据推送time.sleep(2.0)# 步骤3: 获取数据tick_data = xt.get_full_tick([code])if tick_data and code in tick_data:    tick = tick_data[code]print(f"最新价: {tick.get('lastPrice', 0):.2f}")# ✓ 五档数据完整    ask_price = tick.get('askPrice', [])if ask_price andlen(ask_price) > 0:print(f"卖一: {ask_price[0]:.2f}")

2.3 总结对比

📥 三、使用回调函数接收推送

3.1 什么是回调函数？

回调函数是你定义的一个函数，当有新数据时，系统会自动调用这个函数并把数据传给你。

3.2 回调函数示例

import xtquant.xtdata as xtimport timefrom datetime import datetimecode = '000001.SZ'counter = {'count': 0, 'max': 5}defon_tick_data(data):"""tick数据回调函数"""if code in data:        tick = data[code]        counter['count'] += 1print(f"\n[推送 #{counter['count']}] {datetime.now().strftime('%H:%M:%S')}")print(f"  最新价: {tick.get('lastPrice', 0):.2f}")# 显示五档        ask_price = tick.get('askPrice', [])        bid_price = tick.get('bidPrice', [])if ask_price andlen(ask_price) > 0:print(f"  卖一: {ask_price[0]:.2f}  买一: {bid_price[0]:.2f}")# 订阅并设置回调xt.subscribe_whole_quote(code_list=[code], callback=on_tick_data)# 等待推送while counter['count'] < counter['max']:    time.sleep(0.1)

3.3 回调函数要点

1. ✅ 回调函数由系统自动调用，有新数据时触发
2. ✅ 回调函数的参数就是推送的数据
3. ✅ 可以在回调函数中实时处理数据
4. ✅ 适合持续监控行情的场景

📊 四、五档行情数据结构详解

4.1 五档数据的字段结构

五档行情包含：买5档 + 卖5档

# 原始数据字段（根据官方文档）askPrice  # 委卖价格数组 [卖1, 卖2, 卖3, 卖4, 卖5]bidPrice  # 委买价格数组 [买1, 买2, 买3, 买4, 买5]askVol    # 委卖量数组 [卖量1, 卖量2, ...]bidVol    # 委买量数组 [买量1, 买量2, ...]

4.2 五档盘口展示

tick_data = xt.get_full_tick([code])tick = tick_data[code]ask_price = tick.get('askPrice', [])bid_price = tick.get('bidPrice', [])ask_vol = tick.get('askVol', [])bid_vol = tick.get('bidVol', [])# 显示五档print(f"\n{'档位':<8}{'卖盘':<25}{'买盘':<25}")print("-" * 50)for i inrange(5):    idx = 4 - i  # 倒序显示    ask_p = ask_price[idx] if idx < len(ask_price) else0    ask_v = ask_vol[idx] if idx < len(ask_vol) else0    bid_p = bid_price[idx] if idx < len(bid_price) else0    bid_v = bid_vol[idx] if idx < len(bid_vol) else0    ask_str = f"{ask_p:.2f} ({ask_v:.0f})"if ask_p > 0else"--"    bid_str = f"{bid_p:.2f} ({bid_v:.0f})"if bid_p > 0else"--"print(f"  {i+1}档    {ask_str:<25}{bid_str:<25}")

输出示例：

档位     卖盘                     买盘--------------------------------------------------  1档    10.88 (560)             10.87 (1200)  2档    10.89 (890)             10.86 (2300)  3档    10.90 (450)             10.85 (1560)  4档    10.91 (680)             10.84 (890)  5档    10.92 (320)             10.83 (1120)

4.3 盘口分析

# 计算买卖价差spread = ask_price[0] - bid_price[0]spread_pct = (spread / bid_price[0]) * 100print(f"买卖价差: {spread:.2f} 元 ({spread_pct:.3f}%)")# 计算买卖盘总量total_bid_vol = sum(bid_vol)total_ask_vol = sum(ask_vol)ratio = total_bid_vol / total_ask_volprint(f"买卖比: {ratio:.2f}")if ratio > 1.2:print("市场情绪: 买盘强势 📈")elif ratio < 0.8:print("市场情绪: 卖盘强势 📉")else:print("市场情绪: 买卖平衡 ➡️")

🚀 五、使用EasyXT简化接口

5.1 EasyXT的优势

EasyXT已经封装了订阅机制，让你无需关心底层细节：

import easy_xtapi = easy_xt.get_api()api.init_data()# 一行代码获取五档行情order_book = api.get_order_book('000001.SZ')

EasyXT的优势：

• ✅ 自动处理订阅逻辑
• ✅ 自动等待数据推送（最多3次重试）
• ✅ 统一的DataFrame返回格式
• ✅ 简化字段名（bid1-5, ask1-5）

5.2 EasyXT订阅接口（新功能）

# 定义回调函数defon_tick(data):for code, tick_list in data.items():print(f"{code}: {tick_list.get('lastPrice', 0):.2f}")# 方式1: 订阅单只股票seq = api.subscribe('000001.SZ', callback=on_tick)# 方式2: 订阅多只股票（推荐）seq = api.subscribe_whole(['000001.SZ', '600000.SH'], callback=on_tick)# 方式3: 持续接收推送api.subscribe('000001.SZ', callback=on_tick)api.run_forever()  # 阻塞运行，持续接收

推荐使用场景：

• 场景1：实时监控单只股票 → 使用 api.subscribe()
• 场景2：监控多只股票 → 使用 api.subscribe_whole()
• 场景3：持续监控行情 → 配合 api.run_forever()

⚠️ 六、常见问题及解决方案

❓ 问题1：订阅后获取不到五档数据

原因：

• 等待时间不够（数据还没推送完成）
• 字段名错误（应该是askPrice/bidPrice）
• 数据是数组格式，需要索引访问

解决：

# ✓ 至少等待2秒time.sleep(2.0)# ✓ 使用正确的字段名和索引ask_price = tick.get('askPrice', [])if ask_price andlen(ask_price) > 0:    first_ask = ask_price[0]  # 获取卖一价

❓ 问题2：非交易时间数据为空

原因：

• 非交易时间（周末、夜间）五档数据为0是正常的
• QMT模拟账户可能限制数据

解决：

• ✅ 在交易时间（9:30-15:00）内运行
• ✅ 检查QMT客户端是否显示五档数据
• ✅ 使用实盘账户测试

❓ 问题3：回调函数不执行

原因：

• 订阅后没有保持程序运行
• 回调函数定义不正确
• 使用了错误的订阅接口

解决：

# ✓ 使用 subscribe_whole_quote() 订阅xt.subscribe_whole_quote(code_list=[code], callback=on_tick)# ✓ 程序需要保持运行whileTrue:    time.sleep(0.1)  # 持续接收推送

❓ 问题4：获取的数据总是旧的

原因：

• 只获取了一次，没有持续订阅
• 没有使用回调函数

解决：

# ✓ 使用回调函数持续接收推送defon_tick(data):# 每次有新数据都会自动调用print("新数据:", data)xt.subscribe_whole_quote(code_list=[code], callback=on_tick)

🎯 七、实战练习

练习1：监控单只股票的实时行情

import xtquant.xtdata as xtfrom datetime import datetimeimport timecode = '000001.SZ'xt.subscribe_quote(code, period='tick')for i inrange(5):    time.sleep(1)    tick_data = xt.get_full_tick([code])if tick_data and code in tick_data:        tick = tick_data[code]        ask_price = tick.get('askPrice', [])        bid_price = tick.get('bidPrice', [])iflen(ask_price) > 0andlen(bid_price) > 0:            spread = ask_price[0] - bid_price[0]print(f"[{datetime.now().strftime('%H:%M:%S')}] "f"最新价: {tick.get('lastPrice', 0):.2f}  "f"价差: {spread:.3f}")

练习2：批量获取多只股票的五档行情

import easy_xtapi = easy_xt.get_api()api.init_data()codes = ['000001.SZ', '000002.SZ', '600000.SH']order_books = api.get_order_book(codes)ifnot order_books.empty:print(f"{'股票代码':<12}{'最新价':<10}{'买一':<10}{'卖一':<10}")print("-" * 50)for _, row in order_books.iterrows():print(f"{row['code']:<12}{row['lastPrice']:<10.2f} "f"{row['bid1']:<10.2f}{row['ask1']:<10.2f}")

📝 总结

通过本文的学习，你应该掌握了：

💡 核心要点

1. 先订阅后获取
   是获取五档数据的正确姿势
2. 等待2秒
   让数据推送完成
3. askPrice/bidPrice
    是数组，需要索引访问
4. 使用 EasyXT封装接口可以大幅简化代码

🎓 学习路径建议

01_基础入门.py → 08_订阅机制详解.py → 02_交易基础.py → 03_高级交易.py

📱 关注我们

欢迎扫码持续关注公众号，会持续分享

🔍 公众号名称: 王者quant📚 分享内容: 量化交易、Python编程、投资策略🎯 更新频率: 持续更新，干货满满

通过公众号您可以获得：

• 📈 最新的量化交易策略分享
• 💻 Python量化编程技巧
• 📊 市场分析和投资心得
• 🚀 EasyXT功能更新和使用技巧
• 💡 量化交易实战案例

本教程仅供学习参考，实际交易请谨慎操作！