天勤TqSdk从零基础安装部署到独立策略编写【保姆喂饭级教程】 第 1 天

手把手详细学习内容:从 TqApi 到第一个实时行情脚本

今天不是看计划,而是跟着文档一步一步完成:环境检查、账户模式判断、实时行情订阅、wait_update 更新循环、is_changing 过滤、常见错误排查。

能解释 TqApi、TqAuth、get_quote、wait_update、is_changing 分别解决什么问题。

能运行一个只读行情脚本,看到合约名称、最新价、买一价、卖一价等字段。

能说清 get_quote 返回的是会刷新的对象引用,不是一次性快照。

知道为什么不应该在循环里反复 get_quote,也不应该靠 sleep 等待行情。

遇到空值、NaN、没有更新、卡住时,能按检查表定位问题。

学习方式:你可以直接照着本文每一步做。看到"现在动手"就写代码或运行命令;看到"停下来想一想"就先回答,再看后面的参考答案。

0. 今天只学一个核心模型:TqSdk 是靠更新循环驱动的

先不要急着下单,也不要急着写策略。TqSdk 文档里反复出现的关键点是:创建 API 对象以后,行情、账户、持仓、委托等对象会随着 wait_update() 的推进而更新。今天只用行情对象练习这个模型。

1. 环境检查:先确认 Python 能导入 TqSdk

现在先把运行环境打通。建议你新建一个空目录,例如 tqsdk_day1,然后在这个目录里操作。

1.1 现在动手:安装或升级 TqSdk

在命令行运行:

python -m pip install -U tqsdk

如果你的系统里同时有 python 和 python3,可以改成:

python3 -m pip install -U tqsdk

1.2 现在动手:写一个最小导入检查文件

在 tqsdk_day1 目录下新建 check_tqsdk.py,输入下面代码。

check_tqsdk.py

from tqsdk import TqApi, TqAuth
print("TqSdk 导入成功")
print("下一步将创建 TqApi 会话")

运行:

python check_tqsdk.py

**你应该看到:**终端打印"TqSdk 导入成功"。这一步只检查包是否能导入,还没有连接行情服务器。

**如果报错:**ModuleNotFoundError: No module named tqsdk,说明当前 Python 环境没有安装成功。请确认安装命令和运行脚本用的是同一个 Python。

2. 先选对账户模式:今天是"只读行情",不是实盘交易

文档把账户模式区分得很清楚。今天只想获取行情,不需要连接期货公司实盘账户,也不需要股票账户。

今天你要形成的第一个习惯
写 TqSdk 代码前,先问自己:我是在读行情、做期货模拟、做股票模拟、还是连接实盘?
只读行情:TqApi(auth=TqAuth(...))。
期货下单或回测:以后再明确 TqSim、TqKq、TqAccount。
股票相关:以后使用股票账户与 Security 系列对象,不能照搬期货 offset 逻辑。

3. 第一个可运行脚本:订阅一个主连合约的实时行情

现在写第一个真正连接行情的脚本。这个脚本只订阅行情、打印一次字段,然后关闭 API。

3.1 现在动手:创建 day1_quote_once.py

day1_quote_once.py

from tqsdk import TqApi, TqAuth
# 1. 用主连合约作为第一天示例,避免具体交割月过期
SYMBOL = "KQ.m@SHFE.rb"

# 2. 运行前把下面两个字符串换成你的快期账户和密码
USER = "你的快期账户"
PASSWORD = "你的账户密码"

api = TqApi(auth=TqAuth(USER, PASSWORD))
quote = api.get_quote(SYMBOL)

try:
    print("正在等待第一笔行情数据...")
    api.wait_update()

    print("合约代码:", SYMBOL)
    print("合约名称:", quote.instrument_name)
    print("行情时间:", quote.datetime)
    print("最新价:", quote.last_price)
    print("买一价:", quote.bid_price1)
    print("卖一价:", quote.ask_price1)
    print("最小变动价位:", quote.price_tick)
    print("合约乘数:", quote.volume_multiple)
    print("是否过期:", quote.expired)
finally:
    api.close()

3.2 逐行带你理解

3.3 现在运行

python day1_quote_once.py

**你可能看到的正常情况:**行情时间、最新价、买一价、卖一价等字段被打印出来。若在非交易时段,部分字段可能不跳动,但合约名称、价格单位等静态字段通常可以观察。

**如果出现 NaN 或空字段:**先不要改成实盘账户。第一检查是否调用了 wait_update;第二检查合约是否过期;第三检查当前是否有该品种行情更新。

4. 停下来想一想:为什么 get_quote 后还要 wait_update?

很多初学者会写成"get_quote 以后马上读 price",然后发现字段为空或不动。文档中的运行模型不是这样的。get_quote 更像是"登记我要订阅这个合约",wait_update 才负责向前推进数据接收与对象刷新。

一句话记忆
get_quote:我要看这个合约。
wait_update:现在去收一次新数据,并把新数据合并到 quote 里。
is_changing:这一次更新里,我关心的字段有没有变化?

4.1 错误写法对比

不要这样写

# 错误示例:创建 quote 后马上读字段,且没有推动更新
from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("你的快期账户", "你的账户密码"))
quote = api.get_quote("KQ.m@SHFE.rb")
print(quote.last_price)
api.close()

这段代码的问题不是语法,而是没有等待第一笔行情数据进入内存。正确做法至少要在读取关键行情字段前调用一次 wait_update。

这样才符合更新模型

# 正确示例:先等待一次更新,再读取字段
from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("你的快期账户", "你的账户密码"))
quote = api.get_quote("KQ.m@SHFE.rb")
api.wait_update()
print(quote.last_price)
api.close()

5. 第二个脚本:持续监听行情变化,而不是只打印一次

第一个脚本只打印一次。真实策略通常要持续处理行情,但不能无脑循环打印。下面用 is_changing 只在关键字段变化时输出。

5.1 现在动手:创建 day1_quote_loop.py

day1_quote_loop.py

import time
from tqsdk import TqApi, TqAuth

SYMBOL = "KQ.m@SHFE.rb"
USER = "你的快期账户"
PASSWORD = "你的账户密码"

api = TqApi(auth=TqAuth(USER, PASSWORD))
quote = api.get_quote(SYMBOL)

printed = 0
max_print = 10

try:
    print("开始监听行情变化,最多打印 10 次。")
    while printed < max_print:
        # deadline 使用绝对时间戳:最多等待 5 秒
        updated = api.wait_update(deadline=time.time() + 5)

        if not updated:
            print("5 秒内没有新的业务数据,继续等待...")
            continue

        if api.is_changing(quote, ["datetime", "last_price", "bid_price1", "ask_price1"]):
            printed += 1
            print(
                printed,
                quote.datetime,
                "最新价=", quote.last_price,
                "买一=", quote.bid_price1,
                "卖一=", quote.ask_price1,
            )
finally:
    api.close()

5.2 逐行带你理解循环

你应该观察到什么
交易时段:价格、买一、卖一、行情时间可能持续变化。
非活跃时段:可能多次出现“5 秒内没有新的业务数据”。这不是代码一定错了。
如果长期没有任何字段,优先检查账号密码、网络、行情权限、合约符号。

6. 第三个练习:证明 quote 是"活对象",不是一次性快照

文档中的重要语义是:get_quote 返回的 Quote 对象会在 wait_update 推进时原地刷新。你不需要在循环里反复 get_quote。

day1_live_object.py

mport time
from tqsdk import TqApi, TqAuth

SYMBOL = "KQ.m@SHFE.rb"
api = TqApi(auth=TqAuth("你的快期账户", "你的账户密码"))
quote = api.get_quote(SYMBOL)

try:
    print("quote 对象 id:", id(quote))
    for i in range(5):
        api.wait_update(deadline=time.time() + 5)
        print(i + 1, "对象 id=", id(quote), "时间=", quote.datetime, "最新价=", quote.last_price)
finally:
    api.close()

运行后你会发现 quote 对象 id 不变,但字段值可能变化。这说明你手里的 quote 是同一个对象引用,TqSdk 在 wait_update 后把新数据合并到了它里面。

今天必须改掉的坏习惯
不要在循环里面反复 quote = api.get_quote(SYMBOL)。
不要用 time.sleep() 替代 wait_update()。sleep 只会让程序睡觉,不会驱动 TqSdk 收发业务数据。
不要硬编码很久以前的具体合约月份。想看当前行情,优先考虑主连或先查询当前合约。

7. 读懂 Quote 字段:今天只掌握最常用的一组

Quote 字段很多,第一天不需要全背。先掌握"看价格、看合约、看是否可用"的最小集合。

7.1 现在动手:把字段打印得更整齐

day1_quote_fields.py

from tqsdk import TqApi, TqAuth

SYMBOL = "KQ.m@SHFE.rb"
api = TqApi(auth=TqAuth("你的快期账户", "你的账户密码"))
quote = api.get_quote(SYMBOL)

try:
    api.wait_update()
    fields = {
        "行情时间": quote.datetime,
        "合约名称": quote.instrument_name,
        "最新价": quote.last_price,
        "买一价": quote.bid_price1,
        "卖一价": quote.ask_price1,
        "成交量": quote.volume,
        "持仓量": quote.open_interest,
        "最小变动价位": quote.price_tick,
        "合约乘数": quote.volume_multiple,
        "是否过期": quote.expired,
    }
    for name, value in fields.items():
        print(f"{name}: {value}")
finally:
    api.close()

8. 合约符号:第一天先用主连,再学会查询

文档建议当前行情示例不要硬编码过期合约。第一天最简单的方法是用主连符号,例如 KQ.m@SHFE.rb。等你要换品种时,再通过查询接口确认当前可用合约。

可选练习:查询螺纹钢连续合约

from tqsdk import TqApi, TqAuth

with TqApi(auth=TqAuth("你的快期账户", "你的账户密码")) as api:
    conts = api.query_cont_quotes(exchange_id="SHFE", product_id="rb")
    print("螺纹钢连续合约列表:")
    print(conts)

    info = api.query_symbol_info(list(conts))
    print(info[["instrument_id", "instrument_name", "ins_class", "price_tick", "volume_multiple"]])

这段查询代码是可选练习。今天的重点仍然是理解更新循环;查询接口只是帮你避免选错或选到过期合约。

9. 常见错误排查:照这张表一步步查

第一天不要用这些方式解决问题
不要因为行情字段为空就改成 TqAccount 实盘账户。只读行情不需要实盘交易账户。
不要因为 wait_update 卡住就换成 sleep。要用 deadline 学会控制等待。
不要把股票与期货代码混在一起。今天的例子是期货行情,不涉及股票下单语义。

10. 课堂练习:你现在来写,写完再看答案

练习 1:把打印次数从 10 次改成 3 次

要求:修改 day1_quote_loop.py,让它最多打印 3 次行情变化。

**参考答案:**把 max_print = 10 改成 max_print = 3。不要改 while 条件,也不要删除 api.close。

练习 2:新增一个字段 open_interest

要求:在持续监听脚本中,当 open_interest 变化时也触发打印,并把它打印出来。

参考答案片段

if api.is_changing(quote, ["datetime", "last_price", "bid_price1", "ask_price1", "open_interest"]):
    printed += 1
    print(
        printed,
        quote.datetime,
        "最新价=", quote.last_price,
        "买一=", quote.bid_price1,
        "卖一=", quote.ask_price1,
        "持仓量=", quote.open_interest,
    )

练习 3:解释下面这句话

题目:quote 是会自动更新的对象引用,不是一次性快照。用你自己的话解释。

**参考答案:**quote 不是一次性复制出来的行情字典,而是一个会被 TqSdk 持续刷新的对象。只要程序继续调用 wait_update,TqSdk 收到的新行情就会合并进这个对象,所以我们保留同一个 quote 引用即可。

练习 4:找出错误

问题代码

while True:
    quote = api.get_quote("KQ.m@SHFE.rb")
    time.sleep(1)
    print(quote.last_price)

**参考答案:**错误有两个:第一,循环里反复 get_quote;第二,用 sleep 替代 wait_update。正确做法是在循环外 get_quote 一次,循环里调用 api.wait_update(),再用 is_changing 判断是否打印。

11. 今日复盘:你要能独立说出来的 8 句话

TqApi 是 TqSdk 会话入口;今天只读行情,不连接实盘交易账户。

TqAuth 用于传入快期账户和密码,运行前要替换占位符。

get_quote 用来订阅并取得 Quote 对象。

Quote 对象会随着 wait_update 被刷新,不是一次性快照。

第一次读取关键行情字段前,至少要调用一次 wait_update。

持续监听时,不要在循环里反复 get_quote。

用 is_changing 过滤字段变化,避免重复执行逻辑。

wait_update 可以使用 deadline 控制最长等待时间。

12. 今天的完整最终版脚本

把下面脚本保存为 day1_final.py。它包含了今天应该掌握的全部关键点:主连合约、只读行情、wait_update、deadline、is_changing、退出循环和关闭 API。

day1_final.py

import time
from tqsdk import TqApi, TqAuth

SYMBOL = "KQ.m@SHFE.rb"
USER = "你的快期账户"
PASSWORD = "你的账户密码"

api = TqApi(auth=TqAuth(USER, PASSWORD))
quote = api.get_quote(SYMBOL)

try:
    print("TqSdk 第 1 天最终脚本:监听主连合约行情")
    print("合约:", SYMBOL)
    print("等待第一笔行情...")

    api.wait_update()
    print("合约名称:", quote.instrument_name)
    print("最小变动价位:", quote.price_tick)
    print("合约乘数:", quote.volume_multiple)
    print("是否过期:", quote.expired)
    print("开始监听价格变化。")

    printed = 0
    while printed < 10:
        updated = api.wait_update(deadline=time.time() + 5)
        if not updated:
            print("等待超时:5 秒内暂无新业务数据。")
            continue

        fields = ["datetime", "last_price", "bid_price1", "ask_price1", "open_interest"]
        if api.is_changing(quote, fields):
            printed += 1
            print(
                printed,
                quote.datetime,
                "last=", quote.last_price,
                "bid1=", quote.bid_price1,
                "ask1=", quote.ask_price1,
                "oi=", quote.open_interest,
            )
finally:
    api.close()
    print("API 已关闭。")

13. 明天开始前的准备

完成今天后,把 day1_final.py 保留好。明天会在这个基础上继续学习 K 线、Tick 或更复杂的数据序列时,你会反复用到同一个更新循环模型。

今日完成标准
你至少成功运行 check_tqsdk.py。
你能说明 TqApi(auth=TqAuth(...)) 今天为什么足够。
你成功运行 day1_quote_once.py 或 day1_final.py。
你能解释 wait_update、is_changing 的作用。
你能根据排错表处理 NaN、无更新、阻塞、过期合约等问题。

提示:本文档中的代码用于学习行情读取与更新循环,不构成交易建议,也不包含实盘下单逻辑。