前提:已安装 SDK,在个人中心 → SDK Token 创建并保存令牌,认证域名已接入有效账户服务。将下方 YOUR_TOKEN 替换为自己的完整令牌再运行;它是占位文本,不是可用令牌。以下代码自动查询主力期货,避免把固定月份的过期合约直接用于程序。
SDK 通过 ExApi(token=...) 接收令牌,不会自动创建或读取 EXSDK_TOKEN。若希望使用环境变量保存令牌,参见 安装与环境 的可选用法。
第一个行情程序
from exsdk import ExApi
with ExApi(token="YOUR_TOKEN") as api:
symbols = api.query_symbols("FUTURE", exchange_id="SHFE", is_main=True)
if not symbols:
raise RuntimeError("没有找到合约,请检查交易所或查询条件")
symbol = symbols[0]
print("订阅", symbol)
quote = api.watch_quote(symbol)
while api.wait_update():
if quote.is_changing("last_price"):
snapshot = quote.data()
print(snapshot.timestamp, snapshot.last_price)watch_quote() 创建订阅句柄;wait_update() 推进并提交数据;is_changing() 检查最近一批变化;data() 获取不会随之后更新改变的独立快照。非交易时段可能长时间没有变化,不能据此推断连接失败。
同时观察 K 线
在同一个 with 内创建所有需要的句柄,再使用一层统一的更新循环:
quote = api.watch_quote(symbol)
klines = api.watch_kline(symbol, "1m", data_length=200)
while api.wait_update():
if quote.is_changing("last_price"):
print("最新价", quote.data().last_price)
if klines.is_changing("timestamp"):
frame = klines.data()
if not frame.empty:
print("窗口末行", frame.iloc[-1].to_dict())K 线的最后一行可能还未结束,后续仍会改变。timestamp 变化表示时间窗口发生变化,不是“该末行永远不再修订”的承诺。需要严格完成的 K 线时,应结合周期结束时间、当前行情时间以及策略规则判断。
有限等待
import time
updated = api.wait_update(deadline=time.time() + 5)
if updated:
print(quote.data())
else:
print("本次等待到期,未提交更新")deadline 是 Unix 秒的绝对时间,不能直接传 5 表示五秒。一次返回 False 不意味着账户断开,也不重置恢复预算。
下一步
阅读 更新模型 与 接口参考,再使用 完整示例。只读行情程序无需创建 CtpAccount。如果没有配置交易账户,不要调用资金或持仓观察接口。