简介本资源是一套面向本科毕业设计与量化交易初学者的Python量化交易系统实现聚焦于天勤SDK接口集成与实盘策略框架搭建解决从策略开发、数据接入到自动下单的全流程实践难题。压缩包共58个文件含34个核心Python源码覆盖行情订阅、信号生成、订单管理、风控逻辑等模块、15个编译后pyc文件、5个XML配置及日志模板、1个conf配置文件整体仅44KB轻量易读结构清晰便于理解各模块职责与调用关系。目前已有133人学习下载适合具备基础Python编程能力、希望快速掌握国内主流量化平台开发范式的高校学生与个人开发者。读者可直接运行调试策略模块复现回测与模拟交易流程获取完整的工程目录结构、标准化配置方式及天勤SDK典型用法示例为毕设答辩或实盘策略迭代提供可靠支撑。 最近有不少朋友在折腾Python量化交易发现大家问得最多的不是策略怎么写而是“数据从哪来”、“怎么接到实盘”。市面上的数据接口要么收费要么文档写得让人头大真正能拿过来就跑的方案并不多。我自己用天勤SDK搭过一套完整的期货量化交易系统从行情获取、策略回测到模拟盘部署都跑通了这篇就把整个过程整理出来包括环境配置、核心代码、关键原理和踩过的坑希望能给准备入坑的朋友省点时间。这套东西能做什么简单说用Python写策略通过天勤SDK连接期货市场的实时行情和交易通道实现自动化下单。无论你是刚学Python想找个实战项目练手还是已经会写策略但不知道怎么接到市场上这篇都值得看完。我尽量把每一步的原理也讲清楚而不是甩一堆代码让你自己猜。1. 项目背景与整体设计思路1.1 为什么选天勤SDK而不是自己写接口做量化交易第一步就是解决数据源和交易通道的问题。早期的做法是自己连期货公司的CTP接口CTP是上期技术提供的交易系统接口性能很强但直接用CTP开发有几个痛点一是官方文档偏底层全是C风格的接口定义Python封装需要自己处理很多细节二是行情和交易是两套通道要自己维护连接状态、重连逻辑、数据拼接工作量不小三是CTP的账户体系、风控规则都暴露在底层新手很容易在资金管理和报单逻辑上出问题。天勤SDK相当于把这一层全都包掉了。它提供的是Python原生的API底层自动帮你处理行情订阅、断线重连、合约信息拉取、订单状态同步这些脏活累活。你只需要关心策略逻辑本身。而且天勤对个人开发者是免费的这点很重要——很多量化平台收费模式是按月订阅或者按交易量抽成对还没稳定盈利的个人来说负担挺重。还有一个很实际的原因天勤SDK支持从仿真账户到实盘账户的无缝切换。你可以在模拟盘上把策略跑熟再把账户配置改成实盘代码几乎不用动。这个特性在开发阶段特别香因为量化策略的成熟必须经过大量实盘环境下的测试不是回测跑得好就行。1.2 量化交易系统的模块拆解一个能稳定运行的量化交易系统不只是“写个策略”那么简单。我把它拆成五个模块来理解数据模块负责订阅行情、维护K线序列、处理历史数据。这是系统的基础数据质量直接决定策略表现。策略模块根据市场数据产生交易信号。常见的有均线突破、MACD背离、布林带回归等核心是信号生成的逻辑。风控模块控制仓位、设置止损止盈、限制单笔亏损、防止过度交易。这层是保护资金的关键很多人忽略这层结果策略没问题钱没了。执行模块把策略信号转化为实际订单处理报单、撤单、成交回报。执行效率决定了滑点成本。监控模块实时记录系统运行状态、仓位变化、订单异常出了问题能第一时间发现。天勤SDK主要在数据模块和执行模块上帮你省力但策略和风控还是得自己写。这也是这篇博文的重点——怎么把天勤SDK的能力和自研策略结合起来形成一套完整的系统。2. 环境准备与核心概念2.1 Python环境与天勤SDK安装先说环境。我用的Python版本是3.8理论上3.7以上都能跑但建议不要用太新的版本有些依赖库可能还没跟上。安装天勤SDK只需要一条命令pip install tqsdk安装过程很简单但有一个容易踩的坑天勤SDK依赖websockets、pandas、numpy这些库如果你之前装过比较老的版本可能会冲突。建议在干净的虚拟环境里装比如用conda create -n quant python3.8创建一个独立环境再装tqsdk这样不会影响你其他的Python项目。装完之后还需要一个天勤账号。去天勤的官网注册一下在代码里登录的时候需要用到账号和密码。仿真盘和实盘用的是同一套鉴权机制不需要额外申请什么权限。2.2 理解核心对象TqApi、TqAuth和行情引用使用天勤SDK第一行代码通常是这样的from tqsdk import TqApi, TqAuth api TqApi(authTqAuth(你的账号, 你的密码))这个TqApi对象是整个系统的核心它管理着与服务器之间的所有通信。创建这个对象之后它会自动建立连接、启动后台线程处理数据收发。TqApi有几个重要的初始化参数我建议重点关注auth天勤账号认证信息必须传入否则连不上。account账户配置。不传时默认是仿真账号传入TqKq()快期模拟或TqAccount(期货公司代码, 账户名, 密码)实盘账户来切换交易环境。auth参数的认证方式和早期版本相比有了调整现在必须显式传入TqAuth对象老代码里直接传username/password的方式已经不行了。创建了TqApi之后你还需要获取行情引用。这里要理解天勤SDK的一个设计哲学它不主动推送数据而是等你调用wait_update()来更新数据。这种“拉模式”让策略的时间节奏由你自己控制和事件驱动的框架相比更容易理解。最简单的用法是订阅一个合约的行情quote api.get_quote(SHFE.rb2310)这个quote对象就是螺纹钢2310合约的实时行情引用。每次调用api.wait_update()之后quote对象上的属性比如last_price、bid_price_1、ask_price_1就会更新到最新值。2.3 理解K线数据与数据序列实时tick行情适合做高频和日内交易但大多数策略需要的是K线。天勤SDK提供了get_kline_serial()方法可以直接获取K线序列klines api.get_kline_serial(SHFE.rb2310, duration_seconds60, data_length200)这个方法返回的是一个类似DataFrame的对象包含open、high、low、close、volume、open_oi等字段其中open_oi是开盘持仓量。duration_seconds参数控制K线周期60就是1分钟线也可以设成24小时线86400秒等。需要注意的一个细节get_kline_serial()返回的K线是不断更新的当一根K线还没结束时close会随着实时价格变化。所以在做策略判断的时候要判断当前K线是否已经结束通常用klines.iloc[-1][id]是否变化来判断或者看klines.iloc[-1][datetime]对应的时间。# 判断是否出现新K线 if klines.iloc[-1][id] ! last_kline_id: # 新K线出现执行策略逻辑 last_kline_id klines.iloc[-1][id]3. 系统实现从行情到策略信号3.1 连接行情并实现一个基础的数据中枢行情数据是整个交易系统的血液。在天勤SDK里多个合约的行情可以同时订阅你只需要多次调用get_quote()或get_kline_serial()就行。比如我想同时跟踪螺纹钢和热卷的行情quotes {} quotes[rb] api.get_quote(SHFE.rb2310) quotes[hc] api.get_quote(SHFE.hc2310)然后在一个循环里不断调用api.wait_update()每轮循环代表一次数据更新。这里有一个性能相关的注意点wait_update()默认是阻塞等待直到有数据更新所以循环不会空转浪费CPU。我一般会维护一个“数据中枢”模块把所有订阅的行情、K线统一管理起来策略模块只从数据中枢拿数据不直接调用SDK。这样以后想换数据源或加新的合约只需要改数据中枢一个文件。class DataCenter: def __init__(self, api, symbols): self.api api self.klines {} for symbol in symbols: self.klines[symbol] api.get_kline_serial(symbol, duration_seconds60, data_length300) def update(self): self.api.wait_update() def get_latest_klines(self, symbol, n10): return self.klines[symbol].iloc[-n:]3.2 写一个简单的趋势跟踪策略有数据之后就可以写策略了。我拿一个相对简单但有效的双均线策略来演示。核心逻辑是当短期均线上穿长期均线时做多下穿时做空。这个策略虽然古老但作为框架的示例非常合适因为它能完整展示数据获取、信号生成、订单执行的链路。def generate_signal(kline_data): close_prices kline_data[close] ma_short close_prices.rolling(window5).mean() ma_long close_prices.rolling(window20).mean() if ma_short.iloc[-2] ma_long.iloc[-2] and ma_short.iloc[-1] ma_long.iloc[-1]: return BUY elif ma_short.iloc[-2] ma_long.iloc[-2] and ma_short.iloc[-1] ma_long.iloc[-1]: return SELL else: return None注意我这里用了iloc[-2]和iloc[-1]来比较前一根和当前这根K线的均线关系这样能避免使用尚未收盘的K线数据。如果不注意这个策略的信号会频繁闪烁回测表现和实盘表现会差距很大。3.3 订单生命周期与管理生成信号之后就要把信号变成实际订单。天勤SDK的insert_order()方法用来报单order api.insert_order( symbolSHFE.rb2310, directionBUY, offsetOPEN, volume1, limit_pricequote.limit_up # 用涨停价作为限价表示市价买入 )这个接口有几个参数需要重点理解direction买单传BUY卖单传SELL。offset开仓传OPEN平仓传CLOSE或CLOSETODAY平今仓。limit_price限价单的价格。如果想尽快成交可以传一个偏离现价较多的价格比如涨跌停价。volume手数注意期货是乘数合约一手对应的实际价值远大于名义价格。insert_order()返回一个Order对象你可以通过它的属性来跟踪订单状态if order.status FINISHED: # 订单已成交 if order.trade_price: print(成交价, order.trade_price) elif order.status REJECTED: # 单子被拒 print(拒单原因, order.message)在实际的系统中我建议在订单下单后设置一个超时机制如果一段时间内没有完全成交就撤单避免单子挂在那边占着资金。天勤SDK提供了api.cancel_order()来撤单api.cancel_order(order)3.4 风控模块量化系统的保命符很多新手第一次写量化系统满脑子都是策略怎么赚大钱忽略了风控。我自己的经验是风控模块的重要性远大于策略本身。一个能稳定运行的系统必须要有这几层风控第一层是单笔止损。每次开仓之前就定好止损价一旦价格触及止损位无条件平仓。这个逻辑不能放在策略信号里要独立成模块因为策略信号有时会失效。# 持仓管理类 class PositionManager: def __init__(self, api, symbol, stop_loss_price): self.api api self.symbol symbol self.stop_loss_price stop_loss_price self.position 0 def check_stop_loss(self, current_price): if self.position 0 and current_price self.stop_loss_price: # 触发止损平多仓 self.close_position(SELL, abs(self.position)) elif self.position 0 and current_price self.stop_loss_price: # 触发止损平空仓 self.close_position(BUY, abs(self.position))第二层是总仓位控制。单一品种的持仓市值不要超过总资金的一定比例比如20%。这样即使某个品种出现极端行情也不至于伤筋动骨。第三层是异常检测。如果策略连续触发止损或者连续开仓频繁就要停下来检查是不是策略逻辑出了bug或者市场状态和策略假设不符。我一般会在系统里加一个简单的计数器和冷却时间if trade_count 10: # 10笔交易内触发过多次止损强制停止交易 self.trading_enabled False print(触发熔断机制禁止继续交易)4. 回测框架与模拟盘部署4.1 用历史数据做初步验证策略写完之后直接上模拟盘之前建议先用历史数据做一轮回测。天勤SDK可以获取历史K线数据用get_kline_serial()传入data_length比较大的值就行。但我的习惯是单独写一个简单的回测框架把策略逻辑和数据获取解耦这样回测和实盘共用同一套策略类。一个基础的回测框架至少要包括这些部分数据回放按月/按日滑动历史K线拼成连续的行情流。撮合逻辑模拟订单成交考虑滑点、手续费。账户记录记录每一笔交易的盈亏、手续费、资金曲线。统计指标计算总收益率、最大回撤、夏普比率、胜率等。我自己写过一个不到200行的回测框架核心就是循环遍历每根K线调用策略的generate_signal()函数然后撮合成交并记录。代码不复杂但很实用特别是能帮你在几分钟内过滤掉明显不靠谱的策略想法。def backtest(strategy, kline_data): account 1000000 # 初始资金 position 0 trades [] for i in range(20, len(kline_data)): bar kline_data.iloc[:i1] signal strategy.generate_signal(bar) current_price bar[close].iloc[-1] if signal BUY and position 0: position 1 trades.append({price: current_price, position: position}) elif signal SELL and position 1: position 0 trades.append({price: current_price, position: position}) # 计算收益 return trades这个框架虽然简单但它逼着你去思考一个问题策略信号在什么时候触发、按什么价格成交、是否考虑手续费和滑点。如果回测时不考虑手续费实盘你很可能发现赚的还不够交手续费。4.2 从回测到仿真交易的关键切换回测数据漂亮不代表实盘能赚钱。我见过太多策略在回测里年化收益翻倍一上实盘就连续亏损。这里面的核心原因就是回测和实盘的环境差异。在回测里数据是完整的、静态的策略可以“回头看”所有数据来做决策。但在实盘里数据是逐步到来的策略必须在每一根K线收盘时立刻做出决定不能有一丝犹豫。这就是为什么要用前面代码里iloc[-2]和iloc[-1]判断K线闭合后才动手的原因。天勤SDK的模拟盘环境是过渡到实盘的最好桥梁。把系统从回测切换到仿真只需要改几个关键配置TqApi的account参数不传默认就是仿真环境。仿真环境的数据和实盘几乎一样下单的流程也完全一致只是不涉及真实资金。在仿真环境里可以测试自动交易逻辑的稳定性、断线重连、异常处理这些回测里测不到的东西。我在仿真环境里跑了大概两周验证了几件事策略在每天的成交情况是否符合预期、系统连续运行两天以上内存会不会泄漏、断网之后重连是否正常。这些在回测里根本没机会验证。4.3 模拟盘常见问题为什么我的单子没有成交在模拟盘上一个特别容易遇到的问题就是“为什么我下了单子但一直没成交”。排除代码逻辑错误之外最常见的原因是仿真环境的对手盘深度和实盘有差异。期货市场的对手盘来自其他真实交易者但仿真盘上的对手盘是模拟撮合的流动性分布和真实市场不完全一致。如果你的策略是挂限价单等成交在仿真盘上可能要等更久。所以做高频或者日内策略的时候建议用对手价比如买一卖一价来报单而不是用对盘价。这里也顺便说一下怎么处理未成交单# 如果3秒内没有成交就撤单重发 import time order api.insert_order(...) start_time time.time() while order.status ! FINISHED: api.wait_update() if time.time() - start_time 3: api.cancel_order(order) # 重新发单 break5. 常见问题与排查技巧实录5.1 连接失败与鉴权错误天勤SDK最常遇到的第一个问题就是鉴权失败。旧版代码里直接传账号密码就能用但新版必须要用TqAuth对象。如果你遇到了AuthError或者PermissionDenied之类的错误先检查这两点一是账号密码是否正确注意天勤的账号密码和期货公司的账号是两套体系不要搞混。二是检查你有没有在TqApi初始化时显式传入TqAuth对象。如果以前写过老代码可以直接照着下面的写法改api TqApi(authTqAuth(username, password))如果是连接超时的问题通常有两种可能本地网络环境问题或者天勤服务器暂时不可用。可以先用浏览器访问一下天勤官网看是否能正常打开排除本地网络故障。5.2 数据不更新或者K线异常wait_update()超时也是个常见的坑。默认情况下wait_update()会一直阻塞等待数据更新但如果你的策略逻辑处理时间太长会导致数据积压行情延时越来越大。解决办法是在wait_update()里加超时参数while True: api.wait_update(timeout1) # 处理策略逻辑这样每秒钟至少会回来一次不会因为某个瞬间的网络阻塞而卡死。如果发现处理一次循环的时间超过了几百毫秒你就要优化代码了比如把计算量大的操作放到另一条线程去执行。K线异常最常见的原因是合约代码写错。天勤的合约代码格式是交易所代码.合约号比如SHFE.rb2310中SHFE是上期所rb是螺纹钢2310是交割月份。如果你写成了SHFE.rb10系统会报合约不存在。可以先用api.get_quote()试探一下看能否正常返回行情。5.3 关于“人狗大作战”那种刷单策略在量化里能不能用最近看到热搜里有什么“人狗大作战python代码”之类的词很多人把刷单策略和量化交易混为一谈。我的观点很明确那种高频报单刷手续费的模式在期货市场会很快被手续费吃掉利润而且容易被交易所认定为异常交易行为。量化交易的核心是系统化的信号生成和风险控制不是靠无脑刷单。如果你刚开始学量化我建议优先把精力放在这两件事上一是把数据管好确保策略拿到的数据是准确的、及时的二是把风控做好确保单笔亏损可控、总回撤可控。策略本身能不能赚钱反而是在这两个基座之上慢慢迭代出来的。5.4 性能优化多品种监控下的延迟控制当你把系统扩展到多个品种时性能问题就会显现出来。我自己遇到过的情况是同时监控10个合约的1分钟K线单个合约平均每次数据处理超过了1秒导致整体循环延迟到了2-3秒这对日内策略来说是不可接受的。分析下来主要瓶颈有两点一是大量的pandas滚动窗口计算耗时二是日志输出阻塞了主循环。针对这两个问题我的优化手段很直接用numpy数组代替pandas来做滚动窗口计算速度提升明显。天勤SDK返回的K线序列本质上就是numpy的二维数组配合pandas按列访问很方便但滚动均线的计算可以基于numpy的convolve实现。日志输出改到独立线程或者用队列缓冲避免高频打印卡住主循环。import numpy as np def fast_ma(closes, window): return np.convolve(closes, np.ones(window)/window, modevalid)这个优化听起来不起眼但在行情密集的时候省下的几十毫秒可能就是成交价格的差别。5.5 Python相关环境问题速查表最后整理一个环境常见问题速查表方便你按图索骥排查问题现象可能原因解决方法pip install tqsdk报错Python版本过旧升级到Python 3.7连接服务器超时本地网络问题检查网络重试合约代码无效代码格式错误使用SHFE.rb2310格式鉴权失败账号密码错误注册天勤账号正确传入TqAuthK线序列中NaN值合约停牌或数据缺失用fillna()处理或丢弃异常段系统运行一段时间后卡死内存泄漏或连接断开检查是否有死循环尝试重建Api对象6. 从这套系统还能延伸出去的方向量化交易系统的搭建之路没有终点。上面的系统能跑通之后你可以根据自己的兴趣和资金实力往这几个方向延伸第一从单品种扩展到多品种、多周期。把数据中枢的订阅列表增加策略模块增加品种参数和周期参数然后做一个简单的多策略组合分散风险。第二引入更深入的信号源。除了价格和成交量你还可以接入持仓量、资金流向、现货数据等丰富策略的维度。第三做一个Web监控面板。用Flask或Streamlit把系统的资金曲线、持仓、最近交易记录展示出来方便远程查看系统状态。第四把策略的调参和回测自动化。用itertools或者optuna做参数搜索自动寻找更优的参数组合。这个需要谨慎参数寻优很容易过拟合但可以作为一个研究工具来用。我个人在写完这套系统后最大的收获不是赚了多少钱而是理解了量化交易系统的核心难点到底在哪——不是策略信号怎么生成而是怎么让整个系统在真实市场环境下稳定、可靠、可维护地运行。希望这篇文章能给正在折腾的朋友一些参考少走几步弯路。本文还有配套的精品资源点击获取
Python量化交易实战：用天勤SDK搭建期货自动化交易系统
简介本资源是一套面向本科毕业设计与量化交易初学者的Python量化交易系统实现聚焦于天勤SDK接口集成与实盘策略框架搭建解决从策略开发、数据接入到自动下单的全流程实践难题。压缩包共58个文件含34个核心Python源码覆盖行情订阅、信号生成、订单管理、风控逻辑等模块、15个编译后pyc文件、5个XML配置及日志模板、1个conf配置文件整体仅44KB轻量易读结构清晰便于理解各模块职责与调用关系。目前已有133人学习下载适合具备基础Python编程能力、希望快速掌握国内主流量化平台开发范式的高校学生与个人开发者。读者可直接运行调试策略模块复现回测与模拟交易流程获取完整的工程目录结构、标准化配置方式及天勤SDK典型用法示例为毕设答辩或实盘策略迭代提供可靠支撑。 最近有不少朋友在折腾Python量化交易发现大家问得最多的不是策略怎么写而是“数据从哪来”、“怎么接到实盘”。市面上的数据接口要么收费要么文档写得让人头大真正能拿过来就跑的方案并不多。我自己用天勤SDK搭过一套完整的期货量化交易系统从行情获取、策略回测到模拟盘部署都跑通了这篇就把整个过程整理出来包括环境配置、核心代码、关键原理和踩过的坑希望能给准备入坑的朋友省点时间。这套东西能做什么简单说用Python写策略通过天勤SDK连接期货市场的实时行情和交易通道实现自动化下单。无论你是刚学Python想找个实战项目练手还是已经会写策略但不知道怎么接到市场上这篇都值得看完。我尽量把每一步的原理也讲清楚而不是甩一堆代码让你自己猜。1. 项目背景与整体设计思路1.1 为什么选天勤SDK而不是自己写接口做量化交易第一步就是解决数据源和交易通道的问题。早期的做法是自己连期货公司的CTP接口CTP是上期技术提供的交易系统接口性能很强但直接用CTP开发有几个痛点一是官方文档偏底层全是C风格的接口定义Python封装需要自己处理很多细节二是行情和交易是两套通道要自己维护连接状态、重连逻辑、数据拼接工作量不小三是CTP的账户体系、风控规则都暴露在底层新手很容易在资金管理和报单逻辑上出问题。天勤SDK相当于把这一层全都包掉了。它提供的是Python原生的API底层自动帮你处理行情订阅、断线重连、合约信息拉取、订单状态同步这些脏活累活。你只需要关心策略逻辑本身。而且天勤对个人开发者是免费的这点很重要——很多量化平台收费模式是按月订阅或者按交易量抽成对还没稳定盈利的个人来说负担挺重。还有一个很实际的原因天勤SDK支持从仿真账户到实盘账户的无缝切换。你可以在模拟盘上把策略跑熟再把账户配置改成实盘代码几乎不用动。这个特性在开发阶段特别香因为量化策略的成熟必须经过大量实盘环境下的测试不是回测跑得好就行。1.2 量化交易系统的模块拆解一个能稳定运行的量化交易系统不只是“写个策略”那么简单。我把它拆成五个模块来理解数据模块负责订阅行情、维护K线序列、处理历史数据。这是系统的基础数据质量直接决定策略表现。策略模块根据市场数据产生交易信号。常见的有均线突破、MACD背离、布林带回归等核心是信号生成的逻辑。风控模块控制仓位、设置止损止盈、限制单笔亏损、防止过度交易。这层是保护资金的关键很多人忽略这层结果策略没问题钱没了。执行模块把策略信号转化为实际订单处理报单、撤单、成交回报。执行效率决定了滑点成本。监控模块实时记录系统运行状态、仓位变化、订单异常出了问题能第一时间发现。天勤SDK主要在数据模块和执行模块上帮你省力但策略和风控还是得自己写。这也是这篇博文的重点——怎么把天勤SDK的能力和自研策略结合起来形成一套完整的系统。2. 环境准备与核心概念2.1 Python环境与天勤SDK安装先说环境。我用的Python版本是3.8理论上3.7以上都能跑但建议不要用太新的版本有些依赖库可能还没跟上。安装天勤SDK只需要一条命令pip install tqsdk安装过程很简单但有一个容易踩的坑天勤SDK依赖websockets、pandas、numpy这些库如果你之前装过比较老的版本可能会冲突。建议在干净的虚拟环境里装比如用conda create -n quant python3.8创建一个独立环境再装tqsdk这样不会影响你其他的Python项目。装完之后还需要一个天勤账号。去天勤的官网注册一下在代码里登录的时候需要用到账号和密码。仿真盘和实盘用的是同一套鉴权机制不需要额外申请什么权限。2.2 理解核心对象TqApi、TqAuth和行情引用使用天勤SDK第一行代码通常是这样的from tqsdk import TqApi, TqAuth api TqApi(authTqAuth(你的账号, 你的密码))这个TqApi对象是整个系统的核心它管理着与服务器之间的所有通信。创建这个对象之后它会自动建立连接、启动后台线程处理数据收发。TqApi有几个重要的初始化参数我建议重点关注auth天勤账号认证信息必须传入否则连不上。account账户配置。不传时默认是仿真账号传入TqKq()快期模拟或TqAccount(期货公司代码, 账户名, 密码)实盘账户来切换交易环境。auth参数的认证方式和早期版本相比有了调整现在必须显式传入TqAuth对象老代码里直接传username/password的方式已经不行了。创建了TqApi之后你还需要获取行情引用。这里要理解天勤SDK的一个设计哲学它不主动推送数据而是等你调用wait_update()来更新数据。这种“拉模式”让策略的时间节奏由你自己控制和事件驱动的框架相比更容易理解。最简单的用法是订阅一个合约的行情quote api.get_quote(SHFE.rb2310)这个quote对象就是螺纹钢2310合约的实时行情引用。每次调用api.wait_update()之后quote对象上的属性比如last_price、bid_price_1、ask_price_1就会更新到最新值。2.3 理解K线数据与数据序列实时tick行情适合做高频和日内交易但大多数策略需要的是K线。天勤SDK提供了get_kline_serial()方法可以直接获取K线序列klines api.get_kline_serial(SHFE.rb2310, duration_seconds60, data_length200)这个方法返回的是一个类似DataFrame的对象包含open、high、low、close、volume、open_oi等字段其中open_oi是开盘持仓量。duration_seconds参数控制K线周期60就是1分钟线也可以设成24小时线86400秒等。需要注意的一个细节get_kline_serial()返回的K线是不断更新的当一根K线还没结束时close会随着实时价格变化。所以在做策略判断的时候要判断当前K线是否已经结束通常用klines.iloc[-1][id]是否变化来判断或者看klines.iloc[-1][datetime]对应的时间。# 判断是否出现新K线 if klines.iloc[-1][id] ! last_kline_id: # 新K线出现执行策略逻辑 last_kline_id klines.iloc[-1][id]3. 系统实现从行情到策略信号3.1 连接行情并实现一个基础的数据中枢行情数据是整个交易系统的血液。在天勤SDK里多个合约的行情可以同时订阅你只需要多次调用get_quote()或get_kline_serial()就行。比如我想同时跟踪螺纹钢和热卷的行情quotes {} quotes[rb] api.get_quote(SHFE.rb2310) quotes[hc] api.get_quote(SHFE.hc2310)然后在一个循环里不断调用api.wait_update()每轮循环代表一次数据更新。这里有一个性能相关的注意点wait_update()默认是阻塞等待直到有数据更新所以循环不会空转浪费CPU。我一般会维护一个“数据中枢”模块把所有订阅的行情、K线统一管理起来策略模块只从数据中枢拿数据不直接调用SDK。这样以后想换数据源或加新的合约只需要改数据中枢一个文件。class DataCenter: def __init__(self, api, symbols): self.api api self.klines {} for symbol in symbols: self.klines[symbol] api.get_kline_serial(symbol, duration_seconds60, data_length300) def update(self): self.api.wait_update() def get_latest_klines(self, symbol, n10): return self.klines[symbol].iloc[-n:]3.2 写一个简单的趋势跟踪策略有数据之后就可以写策略了。我拿一个相对简单但有效的双均线策略来演示。核心逻辑是当短期均线上穿长期均线时做多下穿时做空。这个策略虽然古老但作为框架的示例非常合适因为它能完整展示数据获取、信号生成、订单执行的链路。def generate_signal(kline_data): close_prices kline_data[close] ma_short close_prices.rolling(window5).mean() ma_long close_prices.rolling(window20).mean() if ma_short.iloc[-2] ma_long.iloc[-2] and ma_short.iloc[-1] ma_long.iloc[-1]: return BUY elif ma_short.iloc[-2] ma_long.iloc[-2] and ma_short.iloc[-1] ma_long.iloc[-1]: return SELL else: return None注意我这里用了iloc[-2]和iloc[-1]来比较前一根和当前这根K线的均线关系这样能避免使用尚未收盘的K线数据。如果不注意这个策略的信号会频繁闪烁回测表现和实盘表现会差距很大。3.3 订单生命周期与管理生成信号之后就要把信号变成实际订单。天勤SDK的insert_order()方法用来报单order api.insert_order( symbolSHFE.rb2310, directionBUY, offsetOPEN, volume1, limit_pricequote.limit_up # 用涨停价作为限价表示市价买入 )这个接口有几个参数需要重点理解direction买单传BUY卖单传SELL。offset开仓传OPEN平仓传CLOSE或CLOSETODAY平今仓。limit_price限价单的价格。如果想尽快成交可以传一个偏离现价较多的价格比如涨跌停价。volume手数注意期货是乘数合约一手对应的实际价值远大于名义价格。insert_order()返回一个Order对象你可以通过它的属性来跟踪订单状态if order.status FINISHED: # 订单已成交 if order.trade_price: print(成交价, order.trade_price) elif order.status REJECTED: # 单子被拒 print(拒单原因, order.message)在实际的系统中我建议在订单下单后设置一个超时机制如果一段时间内没有完全成交就撤单避免单子挂在那边占着资金。天勤SDK提供了api.cancel_order()来撤单api.cancel_order(order)3.4 风控模块量化系统的保命符很多新手第一次写量化系统满脑子都是策略怎么赚大钱忽略了风控。我自己的经验是风控模块的重要性远大于策略本身。一个能稳定运行的系统必须要有这几层风控第一层是单笔止损。每次开仓之前就定好止损价一旦价格触及止损位无条件平仓。这个逻辑不能放在策略信号里要独立成模块因为策略信号有时会失效。# 持仓管理类 class PositionManager: def __init__(self, api, symbol, stop_loss_price): self.api api self.symbol symbol self.stop_loss_price stop_loss_price self.position 0 def check_stop_loss(self, current_price): if self.position 0 and current_price self.stop_loss_price: # 触发止损平多仓 self.close_position(SELL, abs(self.position)) elif self.position 0 and current_price self.stop_loss_price: # 触发止损平空仓 self.close_position(BUY, abs(self.position))第二层是总仓位控制。单一品种的持仓市值不要超过总资金的一定比例比如20%。这样即使某个品种出现极端行情也不至于伤筋动骨。第三层是异常检测。如果策略连续触发止损或者连续开仓频繁就要停下来检查是不是策略逻辑出了bug或者市场状态和策略假设不符。我一般会在系统里加一个简单的计数器和冷却时间if trade_count 10: # 10笔交易内触发过多次止损强制停止交易 self.trading_enabled False print(触发熔断机制禁止继续交易)4. 回测框架与模拟盘部署4.1 用历史数据做初步验证策略写完之后直接上模拟盘之前建议先用历史数据做一轮回测。天勤SDK可以获取历史K线数据用get_kline_serial()传入data_length比较大的值就行。但我的习惯是单独写一个简单的回测框架把策略逻辑和数据获取解耦这样回测和实盘共用同一套策略类。一个基础的回测框架至少要包括这些部分数据回放按月/按日滑动历史K线拼成连续的行情流。撮合逻辑模拟订单成交考虑滑点、手续费。账户记录记录每一笔交易的盈亏、手续费、资金曲线。统计指标计算总收益率、最大回撤、夏普比率、胜率等。我自己写过一个不到200行的回测框架核心就是循环遍历每根K线调用策略的generate_signal()函数然后撮合成交并记录。代码不复杂但很实用特别是能帮你在几分钟内过滤掉明显不靠谱的策略想法。def backtest(strategy, kline_data): account 1000000 # 初始资金 position 0 trades [] for i in range(20, len(kline_data)): bar kline_data.iloc[:i1] signal strategy.generate_signal(bar) current_price bar[close].iloc[-1] if signal BUY and position 0: position 1 trades.append({price: current_price, position: position}) elif signal SELL and position 1: position 0 trades.append({price: current_price, position: position}) # 计算收益 return trades这个框架虽然简单但它逼着你去思考一个问题策略信号在什么时候触发、按什么价格成交、是否考虑手续费和滑点。如果回测时不考虑手续费实盘你很可能发现赚的还不够交手续费。4.2 从回测到仿真交易的关键切换回测数据漂亮不代表实盘能赚钱。我见过太多策略在回测里年化收益翻倍一上实盘就连续亏损。这里面的核心原因就是回测和实盘的环境差异。在回测里数据是完整的、静态的策略可以“回头看”所有数据来做决策。但在实盘里数据是逐步到来的策略必须在每一根K线收盘时立刻做出决定不能有一丝犹豫。这就是为什么要用前面代码里iloc[-2]和iloc[-1]判断K线闭合后才动手的原因。天勤SDK的模拟盘环境是过渡到实盘的最好桥梁。把系统从回测切换到仿真只需要改几个关键配置TqApi的account参数不传默认就是仿真环境。仿真环境的数据和实盘几乎一样下单的流程也完全一致只是不涉及真实资金。在仿真环境里可以测试自动交易逻辑的稳定性、断线重连、异常处理这些回测里测不到的东西。我在仿真环境里跑了大概两周验证了几件事策略在每天的成交情况是否符合预期、系统连续运行两天以上内存会不会泄漏、断网之后重连是否正常。这些在回测里根本没机会验证。4.3 模拟盘常见问题为什么我的单子没有成交在模拟盘上一个特别容易遇到的问题就是“为什么我下了单子但一直没成交”。排除代码逻辑错误之外最常见的原因是仿真环境的对手盘深度和实盘有差异。期货市场的对手盘来自其他真实交易者但仿真盘上的对手盘是模拟撮合的流动性分布和真实市场不完全一致。如果你的策略是挂限价单等成交在仿真盘上可能要等更久。所以做高频或者日内策略的时候建议用对手价比如买一卖一价来报单而不是用对盘价。这里也顺便说一下怎么处理未成交单# 如果3秒内没有成交就撤单重发 import time order api.insert_order(...) start_time time.time() while order.status ! FINISHED: api.wait_update() if time.time() - start_time 3: api.cancel_order(order) # 重新发单 break5. 常见问题与排查技巧实录5.1 连接失败与鉴权错误天勤SDK最常遇到的第一个问题就是鉴权失败。旧版代码里直接传账号密码就能用但新版必须要用TqAuth对象。如果你遇到了AuthError或者PermissionDenied之类的错误先检查这两点一是账号密码是否正确注意天勤的账号密码和期货公司的账号是两套体系不要搞混。二是检查你有没有在TqApi初始化时显式传入TqAuth对象。如果以前写过老代码可以直接照着下面的写法改api TqApi(authTqAuth(username, password))如果是连接超时的问题通常有两种可能本地网络环境问题或者天勤服务器暂时不可用。可以先用浏览器访问一下天勤官网看是否能正常打开排除本地网络故障。5.2 数据不更新或者K线异常wait_update()超时也是个常见的坑。默认情况下wait_update()会一直阻塞等待数据更新但如果你的策略逻辑处理时间太长会导致数据积压行情延时越来越大。解决办法是在wait_update()里加超时参数while True: api.wait_update(timeout1) # 处理策略逻辑这样每秒钟至少会回来一次不会因为某个瞬间的网络阻塞而卡死。如果发现处理一次循环的时间超过了几百毫秒你就要优化代码了比如把计算量大的操作放到另一条线程去执行。K线异常最常见的原因是合约代码写错。天勤的合约代码格式是交易所代码.合约号比如SHFE.rb2310中SHFE是上期所rb是螺纹钢2310是交割月份。如果你写成了SHFE.rb10系统会报合约不存在。可以先用api.get_quote()试探一下看能否正常返回行情。5.3 关于“人狗大作战”那种刷单策略在量化里能不能用最近看到热搜里有什么“人狗大作战python代码”之类的词很多人把刷单策略和量化交易混为一谈。我的观点很明确那种高频报单刷手续费的模式在期货市场会很快被手续费吃掉利润而且容易被交易所认定为异常交易行为。量化交易的核心是系统化的信号生成和风险控制不是靠无脑刷单。如果你刚开始学量化我建议优先把精力放在这两件事上一是把数据管好确保策略拿到的数据是准确的、及时的二是把风控做好确保单笔亏损可控、总回撤可控。策略本身能不能赚钱反而是在这两个基座之上慢慢迭代出来的。5.4 性能优化多品种监控下的延迟控制当你把系统扩展到多个品种时性能问题就会显现出来。我自己遇到过的情况是同时监控10个合约的1分钟K线单个合约平均每次数据处理超过了1秒导致整体循环延迟到了2-3秒这对日内策略来说是不可接受的。分析下来主要瓶颈有两点一是大量的pandas滚动窗口计算耗时二是日志输出阻塞了主循环。针对这两个问题我的优化手段很直接用numpy数组代替pandas来做滚动窗口计算速度提升明显。天勤SDK返回的K线序列本质上就是numpy的二维数组配合pandas按列访问很方便但滚动均线的计算可以基于numpy的convolve实现。日志输出改到独立线程或者用队列缓冲避免高频打印卡住主循环。import numpy as np def fast_ma(closes, window): return np.convolve(closes, np.ones(window)/window, modevalid)这个优化听起来不起眼但在行情密集的时候省下的几十毫秒可能就是成交价格的差别。5.5 Python相关环境问题速查表最后整理一个环境常见问题速查表方便你按图索骥排查问题现象可能原因解决方法pip install tqsdk报错Python版本过旧升级到Python 3.7连接服务器超时本地网络问题检查网络重试合约代码无效代码格式错误使用SHFE.rb2310格式鉴权失败账号密码错误注册天勤账号正确传入TqAuthK线序列中NaN值合约停牌或数据缺失用fillna()处理或丢弃异常段系统运行一段时间后卡死内存泄漏或连接断开检查是否有死循环尝试重建Api对象6. 从这套系统还能延伸出去的方向量化交易系统的搭建之路没有终点。上面的系统能跑通之后你可以根据自己的兴趣和资金实力往这几个方向延伸第一从单品种扩展到多品种、多周期。把数据中枢的订阅列表增加策略模块增加品种参数和周期参数然后做一个简单的多策略组合分散风险。第二引入更深入的信号源。除了价格和成交量你还可以接入持仓量、资金流向、现货数据等丰富策略的维度。第三做一个Web监控面板。用Flask或Streamlit把系统的资金曲线、持仓、最近交易记录展示出来方便远程查看系统状态。第四把策略的调参和回测自动化。用itertools或者optuna做参数搜索自动寻找更优的参数组合。这个需要谨慎参数寻优很容易过拟合但可以作为一个研究工具来用。我个人在写完这套系统后最大的收获不是赚了多少钱而是理解了量化交易系统的核心难点到底在哪——不是策略信号怎么生成而是怎么让整个系统在真实市场环境下稳定、可靠、可维护地运行。希望这篇文章能给正在折腾的朋友一些参考少走几步弯路。本文还有配套的精品资源点击获取
