零基础10分钟上手天勤量化TqSdk：从期货实时行情到自动盯盘报警工具

【免费下载链接】tqsdk-python天勤量化开发包, 期货量化, 实时行情/历史数据/实盘交易项目地址: https://gitcode.com/gh_mirrors/tq/tqsdk-python

做期货的人，多少都被"盯盘"这件事折磨过——行情稍有风吹草动就得切窗口、刷报价，生怕错过买卖点。而天勤量化TqSdk，正是信易科技开源的一款Python量化开发包，它把期货实时行情、历史数据、策略回测、模拟盘与实盘交易打包成一套简洁的API，几行代码就能读到交易所的实时报价。这篇文章我们不聊抽象概念，直接动手搭一个"自动盯盘报警"小工具：行情异动、价差超限，程序替你看盘，你只管在价格到位时出手。

💡 全文所有代码都基于真实场景编写，复制即可运行。跟着步骤走，10分钟后你就拥有一台属于自己的"行情管家"。

一、开工前的三件准备：装库、开户、看懂合约代码

动手之前，把三样东西准备好，每一步都是必须的。

第 1 步：安装 TqSdk。一条命令搞定，前提是你有 Python 3.9 及以上的环境。

pip install tqsdk -U

国内网络下载慢的话，可以换清华镜像源，速度快很多：

pip install tqsdk -U -i https://pypi.tuna.tsinghua.edu.cn/simple --trusted-host=pypi.tuna.tsinghua.edu.cn

第 2 步：注册一个快期账户。行情服务器需要凭据才能登录，这是连接前的必要环节。去天勤量化官网免费注册，拿到账户名和密码，后面写代码时会用到。

第 3 步：看懂合约代码的格式。TqSdk 里所有品种统一使用"交易所代码.合约代码"的写法，常见的有：

• SHFE.rb2610—— 上期所螺纹钢 2610 合约
• DCE.i2609—— 大商所铁矿石 2609 合约
• CZCE.SR605—— 郑商所白糖 605 合约
• CFFEX.IF2506—— 中金所沪深300股指期货 2506 合约

记不住没关系，代码风格是固定的：SHFE、DCE、CZCE、CFFEX、INE、GFEX这些交易所前缀，配合品种代码和合约月份即可。

二、跑通第一条通路：三行代码把实时价格"接"进来

万事俱备，来写第一个脚本。目标是让行情服务器把某一合约的实时价格持续推到我们面前：

from tqsdk import TqApi, TqAuth api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) quote = api.get_quote("SHFE.rb2610") while True: api.wait_update() # 等待服务器推送新数据 print(quote.datetime, quote.last_price)

运行之后，终端会像下图的 demo 示例一样，不停滚动输出"时间 + 最新价"，这就说明你与行情服务器之间的通路已经打通了。

这里有两个核心点，新手容易卡壳，先讲透：

wait_update()是心跳，不是摆设。TqSdk 把数据放在本地内存里，网络只负责推送增量。wait_update()的作用就是"阻塞等待下一次数据更新"，更新一到就返回，我们趁机读取最新行情。整个程序就是"等待 → 读数据 → 处理 → 再等待"的循环，理解了这个节奏，后面所有代码都好写了。

get_quote()返回的不是快照，是"引用"。它返回一个行情对象，之后每次wait_update()返回，这个对象里的字段都会自动刷新为最新值。所以我们只管在循环里读它，不用重新订阅。

三、行情对象里藏着哪些宝贝？报警要用的字段就这几个

get_quote()返回的行情对象字段很丰富，但做盯盘报警，重点记住下面这些就够用了：

⚠️ 这里有个新手必踩的坑：在服务器还没推送第一笔数据前，行情字段是NaN，直接拿去做比较运算会得到错误结果。所以正式代码里，一定要先判断数值是否有效再处理，我们后面的报警器就会演示这个写法。

四、第一版报警器：单合约价格异动，第一时间喊你

环境跑通了，字段认识了，现在进入正题——写报警器。第一版目标很简单：盯着螺纹钢 2610 合约，价格相对上一次记录变动超过 0.5%，就打印一条报警。

from tqsdk import TqApi, TqAuth api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) quote = api.get_quote("SHFE.rb2610") last_price = None while True: api.wait_update() price = quote.last_price if price != price: # NaN 保护：第一笔数据到来前跳过 continue if last_price is None: last_price = price # 首次记录基准价 continue change = (price - last_price) / last_price * 100 if abs(change) >= 0.5: print(f"[报警] {quote.datetime} 螺纹2610 较上次变动 {change:+.2f}% -> {price}") last_price = price

到这里，你的第一个盯盘程序已经能干活了。跑起来之后可以挂机，价格一有风吹草动，终端就会蹦出报警信息。

五、第二版报警器：同时盯多个合约，跨期价差也一起算

单合约报警只是热身。实际交易里，我们经常要同时观察几个合约的关系。TqSdk 提供了get_quote_list()，一次订阅一整批合约，循环里逐个读取即可：

from tqsdk import TqApi, TqAuth api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) symbols = ["SHFE.rb2610", "SHFE.rb2701", "DCE.i2609"] quotes = api.get_quote_list(symbols) while True: api.wait_update() for q in quotes: if q.last_price != q.last_price: # 同样做 NaN 保护 continue print(f"{q.symbol} 最新价 {q.last_price} 持仓量 {q.open_interest}")

再进一步，我们做一个跨期价差报警——螺纹钢 2610 和 2701 两个合约，价差超过 250 个点就提示可能的高抛低吸机会：

from tqsdk import TqApi, TqAuth api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) rb_10 = api.get_quote("SHFE.rb2610") rb_01 = api.get_quote("SHFE.rb2701") while True: api.wait_update() if rb_10.last_price != rb_10.last_price or rb_01.last_price != rb_01.last_price: continue spread = rb_10.last_price - rb_01.last_price if spread > 250: print(f"[价差报警] {rb_10.datetime} RB2610-2701 价差 {spread:.0f}，偏大！") elif spread < 200: print(f"[价差报警] {rb_10.datetime} RB2610-2701 价差 {spread:.0f}，偏小！")

把第一版和第二版拼在一起，就是一个能同时监控"单合约异动 + 跨期价差"的多面手报警器了。批量订阅还有一个隐性好处：订阅的合约越多，网络与本地内存的复用效率越高，性能上明显优于一次只盯一个。

六、给报警器穿上盔甲：日志落盘与断线自愈

程序能跑只是及格，扛得住真实环境才算合格。两个问题必须解决：

问题一：报警信息刷屏即失，过后想复盘没数据。解决办法是加一个日志器，把行情快照攒批写入文件，既不影响性能又能留痕：

import json from tqsdk import TqApi, TqAuth class TickLogger: def __init__(self, path="tick.log"): self.path = path self.buffer = [] def write(self, quote): self.buffer.append({ "time": str(quote.datetime), "symbol": quote.symbol, "last_price": quote.last_price, "volume": quote.volume, }) if len(self.buffer) >= 50: # 攒够50条再落盘，减少IO with open(self.path, "a", encoding="utf-8") as f: for row in self.buffer: f.write(json.dumps(row, ensure_ascii=False) + "\n") self.buffer.clear() logger = TickLogger() api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) quote = api.get_quote("DCE.i2609") while True: try: api.wait_update() logger.write(quote) except Exception as e: print("出现异常，TqSdk 会自动重连并补发数据：", e)

问题二：担心半夜断网、程序掉线。这里可以放心：TqSdk 内置了完善的重连机制（源码里由ReconnectTimer和重连处理器负责管理 WebSocket 连接），断线后会自动重连，并在恢复后补发断线期间错过的数据。我们只需在wait_update()外层做好异常兜底即可，无需自己写复杂的断线重连逻辑。

七、还想再进一步？K线数据与进阶路线图

报警器能看实时行情了，但很多时候我们还想回看走势、验证信号。这时候轮到get_kline_serial()登场——它能拉取任意周期的K线序列，周期以秒为单位：60 是一分钟线，3600 是一小时线，86400 是日线：

from tqsdk import TqApi, TqAuth api = TqApi(auth=TqAuth("你的快期账户", "账户密码")) klines = api.get_kline_serial("SHFE.rb2610", duration_seconds=60, data_length=200) while True: api.wait_update() if api.is_changing(klines.iloc[-1]): # 最新一根K线有更新时 last = klines.iloc[-1] print(f"{last.datetime} 开{last.open} 高{last.high} 低{last.low} 收{last.close}")

至此，从"单合约报价"到"多合约价差"再到"K线序列"，你已经摸清了 TqSdk 数据侧的完整脉络。如果你想继续往下走，官方的进阶路线很清晰：

1. 策略回测：把报警逻辑升级为完整的买卖信号，用历史数据验证有效性，详见 doc/usage/backtest.rst 和 doc/advanced/backtest.rst；
2. 模拟盘交易：用TargetPosTask让策略真正下单，先在模拟环境跑熟；
3. 实盘接入：连接真实账户前，建议通读 doc/usage/trade.rst，把风控规则理清楚；
4. 参考源码：核心 API 定义都在 tqsdk/api.py，入门示例在 tqsdk/demo/，遇到疑问直接查源码是最快的学习路径。

写在最后

回过头看，我们其实只做了一件事：用 TqSdk 的get_quote、get_quote_list、get_kline_serial三个接口，加上一个wait_update()循环，就搭出了一个能自动盯盘、跨合约算价差、断线自愈、落盘留痕的行情管家。整个过程没有一行难懂的代码，也没有复杂的配置。

量化交易的第一步，永远是先把行情数据牢牢握在手里。现在你的工具箱里已经有了这件趁手的兵器，接下来无论做信号研究还是写交易策略，都只是在这个地基上继续盖楼的事。从模拟盘开始，一步步验证你的想法，天勤量化TqSdk 会让这条路比想象中平坦很多。去跑通你的第一份实时行情吧，剩下的交给时间。

【免费下载链接】tqsdk-python天勤量化开发包, 期货量化, 实时行情/历史数据/实盘交易项目地址: https://gitcode.com/gh_mirrors/tq/tqsdk-python