★ FEATURED ARTICLE 在开始动笔之前我盯着自己电脑里那个跑了大半年的QMT策略文件夹想了很久。从最早用Python写回测脚本、天天盯着backtrader曲线反复调参到第一次把策略逻辑搬进QMT终端、看到自动化下单指令发出的一瞬间这个过程的跨越远比我想象中大。如果你也正在研究Python量化交易手里有了一套能稳定盈利的策略代码却不知道如何让它真正跑在实盘上那么QMT可能就是你需要的那座桥。这篇文章我会以双均线策略为例从环境搭建、代码设计到实盘模拟的坑完整带你从零搭建第一个极速策略全程附可直接运行的Python代码。1. 为什么选PythonQMT策略从想法到实盘的最后一步1.1 面向交易的开发框架怎么选才不踩坑很多人在接触QMT之前其实已经被各种框架绕晕了。我最早用的是backtrader它作为回测框架确实很好用但回测结束之后紧接着就是一个灵魂问题策略怎么接到真实行情和交易柜台上这时候你会面临几个主流选项。第一类是QMT、PTrade这类券商官方提供的量化终端。它们的共同特点是终端界面里集成了行情、回测、模拟交易和实盘交易同时向用户开放了Python API你可以在终端里直接写Python策略也可以通过miniQMT的方式用自己本地的Python环境连接交易接口。对普通交易者来说这是目前合规门槛最低、最容易跑通实盘的一条路。PTrade我当年也研究过但很多券商对开通条件有资金门槛要求而且策略托管在券商服务器上灵活度低一些。第二类是vn.py这类开源框架。它的问题不是功能不够强而是你需要自己搞定柜台连接、行情源、账户体系等一系列基础设施配置门槛比较高适合有开发团队的专业玩家不太适合想快速跑通第一个策略的个人交易者。第三类就是只做回测不接实盘的框架。这类框架解决的是策略逻辑是否有效的问题它们并不负责把策略送到真实市场。回过头来看PythonQMT的组合之所以值得花时间研究核心原因是它跳过了先开发一套交易系统的前置工作。你用Python写策略逻辑QMT帮你处理行情接收、账户查询、委托报单、持仓管理等券商底层事务。换句话说你只需要专注写策略本身而不是去解决券商柜台的通讯协议问题。1.2 极速不是一个噱头看看延迟花在哪儿标题里提到的极速策略拆开看其实是两类需求的叠加。一类是策略本身运行不能太慢另一类是信号产生后到指令到达柜台之间的链路要足够快。我自己测过一个普通Python策略如果在QMT终端内部运行行情从交易所到策略回调函数的时间通常就在几十毫秒级别。如果通过miniQMT从外部Python程序来订阅行情速度同样很快因为QMT终端本身就承担了行情网关的角色。真正影响延迟的往往是策略代码里的低效逻辑比如在tick回调里做大量不必要的计算、写同步磁盘日志、频繁调用耗时的接口。所以极速策略这个概念落到代码层面其实就是两件事一是把行情数据的处理路径缩短二是把策略计算和下单执行解耦。这个思路我会在后面的完整代码里具体体现出来。2. 搭建环境时最容易被坑的三个点Python、QMT与终端连接2.1 Python版本与依赖不是越新越好如果你已经装了Python 3.12甚至3.13先别急着得意。QMT终端自带的Python环境和部分券商提供的接口包很多是基于Python 3.6到3.9开发的用太新的版本容易出现二进制包不兼容的情况。根据我实际踩坑的经验最稳妥的组合是组件推荐版本说明Python3.8.x / 3.9.x兼容性最好第三方库基本全支持pandas1.3.x - 1.5.x2.0以上部分接口行为有变化QMT示例代码可能跑不通numpy1.21.x - 1.24.x与pandas版本配套backtrader最新即可回测用与QMT无直接依赖冲突安装依赖时我不建议你一股脑把所有库都装一遍。我的做法是先装基础的pandas和numpy等策略代码跑到哪一步报缺库错误再单独pip install那个库这样能减少版本冲突的可能性。2.2 QMT安装后的常见异常client is null这个错误是QMT新手最容易遇到、也最容易搜到的问题。搜索热词里那句qmt终端 client is null我相信很多人都搜过。这个问题的本质是你希望用外部Python程序通过miniQMT方式操作QMT终端但终端并没有真正启动起来或者没有进入到正确的登录状态。我遇到这个报错时排查顺序是这样的确认券商版QMT终端已经手动登录成功不要只打开登录窗口就切出去。确认程序里指定的路径参数指向的是终端安装目录下的userdata_mini文件夹很多人是路径写错。确认调用init时传递的参数正确不同的券商有不同的账户类型和机构类型字段需要填。简单来说miniQMT通信的前提是QMT终端进程必须处于已登录状态。如果你用Python代码去做连接初始化但终端还停在登录界面那么client就是null。很多教程只会扔给你几行连接代码却不说清楚前置条件导致很多人在这里卡了一整天。2.3 极速策略必须清楚的三种运行模式我用QMT这么久的经验是它的运行模式其实可以粗暴地分成三种理解了这三种模式你后续看代码才不会乱。第一种是在QMT终端内部的策略编辑器里编写Python代码然后直接在终端里运行和启动。这种模式胜在简单终端本身就能运行策略文件适合验证基本逻辑但调试起来不太方便而且要打开界面不符合极速的目标。第二种就是miniQMT模式也是我目前最推荐的方式。你完全不用管QMT的图形界面只需要保证终端在后台登录运行然后在你自己的IDE比如VSCode、PyCharm里编写Python程序调用QMT提供的Python API。行情和交易都能走通策略代码在你自己的环境下运行想怎么调试就怎么调试。第三种是模拟盘和实盘的选择。QMT通常支持模拟账号和实盘账号两者在接口使用上几乎一致但模拟盘的成交逻辑和实盘还是有细微差别。关于这部分我会在第5章单独展开。3. 完整实现一个双均线极速策略从订阅行情到自动下单3.1 策略逻辑与代码结构下面这套代码是我在本地跑了大半年的一个简化版双均线策略核心逻辑很清晰快线上穿慢线时买入开仓快线下穿慢线时卖出平仓。我选择它作为第一个完整示例是因为它足够简单能让你把整套流程跑通同时又足够完整包含了订阅行情、维护K线数据、计算信号、执行下单、记录日志这些关键环节。# -*- coding: utf-8 -*- 极速双均线策略示例 运行前提 1. 券商版QMT终端已启动并手动登录 2. 成功安装Python相关依赖 3. 已通过xtquant包成功连接终端 from xtquant import xttrader, xtdata from xtquant.xttype import StockAccount import time import datetime # ---------- 连接交易接口 ---------- ACCOUNT_ID 你的资金账号 ACCOUNT_TYPE STOCK # 下面的路径需要根据你的QMT安装目录调整 SESSION_PATH rD:\国金QMT交易端\userdata_mini # 回调类QMT交易回调会进入这里 class MyXtQuoteCallback: def on_stock_tick(self, tick): # 这里可以做tick级监控本策略暂不依赖 pass # 初始化交易通道 account StockAccount(ACCOUNT_ID, ACCOUNT_TYPE) trader xttrader.XtQuantTrader(SESSION_PATH, my_strategy_session) trader.start() trader.connect() trader.subscribe(account) # 行情模块初始化 xtdata.connect()这段代码是两个模块的初始化一个是交易模块xttrader负责管理账户、委托、持仓另一个是行情模块xtdata负责订阅行情和拉取历史数据。connect成功后如果打印日志不报错说明你已经和QMT终端建立了正常通信。3.2 数据获取与K线组织双均线策略需要的是K线数据而不是原始tick数据。QMT的xtdata模块可以很方便地获取历史K线也可以实时订阅K线。我常用的方式是先用历史数据“预热”出足够多的K线来计算出当前均线状态然后订阅实时行情来不断更新最新K线。# ---------- 获取历史K线 ---------- STOCK_CODE 600000.SH PERIOD 1d COUNT 100 # 拉取最近100根日K # history_data返回的数据结构是 合约代码 - 字段名 - 列表 history_data xtdata.get_market_data_ex( field_list[time, open, high, low, close, volume], stock_list[STOCK_CODE], periodPERIOD, countCOUNT ) # 转成pandas DataFrame便于计算 import pandas as pd def get_kline_df(code, period1d, count100): data xtdata.get_market_data_ex( field_list[time, open, high, low, close, volume], stock_list[code], periodperiod, countcount ) if code in data: df pd.DataFrame(data[code]) df[time] pd.to_datetime(df[time], units) df df.set_index(time) return df return pd.DataFrame() df get_kline_df(STOCK_CODE) print(df.tail())每次获取最新行情时都需要用新数据滚动更新这个DataFrame而不是每次重新拉取全部历史。我实际开发中发现QMT的get_market_data_ex接口返回的是列表形式直接用pandas处理会比较顺手。3.3 信号生成与下单执行拿到K线以后接下来就是双均线的标准计算。这里我用的是经典的5日和20日均线。每次新K线生成时重新计算均线然后判断金叉和死叉。# ---------- 计算均线并产生信号 ---------- FAST_MA 5 SLOW_MA 20 def generate_signal(df): if len(df) SLOW_MA 1: return None df[fast_ma] df[close].rolling(windowFAST_MA).mean() df[slow_ma] df[close].rolling(windowSLOW_MA).mean() prev_fast df[fast_ma].iloc[-2] prev_slow df[slow_ma].iloc[-2] curr_fast df[fast_ma].iloc[-1] curr_slow df[slow_ma].iloc[-1] if prev_fast prev_slow and curr_fast curr_slow: return BUY elif prev_fast prev_slow and curr_fast curr_slow: return SELL return None生成信号后需要根据当前持仓状态决定是否下单。这里有一个非常关键也容易被忽视的点让策略自己维护一份持仓记录不要每次都去查询真实账户。因为查询账户虽然是异步接口但在tick频繁到来时反复查询会造成不必要的耗时和接口压力。当然交易完成后的成交回报回调里可以同步修正持仓记录。# ---------- 持仓状态管理 ---------- position {} # 维护 合约代码 - 持仓数量 def on_trade_response(response): # 成交回调同步更新持仓 if response.order_status 7: # 全部成交 code response.stock_code volume response.traded_volume if response.offset_flag OPEN: position[code] position.get(code, 0) volume elif response.offset_flag CLOSE: position[code] position.get(code, 0) - volume下单接口用的是order_stock这个函数会异步返回委托信息。对于双均线这种低频策略挂单类型通常用固定价格类型就行如果你要做更快级别的交易可以考虑对手价或最新价但实盘滑点会明显增加。# ---------- 委托下单 ---------- def send_order(code, direction, volume, price_typeFIX): direction: BUY 开多, SELL 平多 fixed_price 0 if price_type FIX: # 用一个合理价格例如当前最新价附近 tick xtdata.get_full_tick([code]) if code in tick: fixed_price tick[code][lastPrice] order_id trader.order_stock( accountaccount, stock_codecode, order_typexttrader.STOCK_BUY if direction BUY else xttrader.STOCK_SELL, order_volumevolume, price_typeprice_type, pricefixed_price, strategy_namema_cross_strategy, order_remark ) return order_id3.4 日志与风控检查策略能跑还不够出问题的时候你要能快速定位。我的做法是在策略里加两层防护第一层是日志记录每一次信号、每一次委托、每一次成交第二层是风控检查下单前先判断持仓是否超限、当日是否已经交易过多次、股票是否处于停牌状态等。# ---------- 简单风控与日志 ---------- log_file ma_strategy.log def log(msg): ts datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) with open(log_file, a, encodingutf-8) as f: f.write(f[{ts}] {msg}\n) print(f[{ts}] {msg}) def risk_check(stock_code, volume, direction): # 限制单笔最大下单量 if volume 10000: return False # 总持仓市值规模检查可以在这里扩展 return True每次循环执行策略时流程是固定的更新K线数据 - 生成信号 - 风控检查 - 下单 - 记录日志。这个顺序不要轻易颠倒尤其是信号生成和风控检查的顺序如果先下单再检查出问题的时候就已经来不及了。4. 让策略跑得更快数据、内存与代码三重优化4.1 数据API怎么选get_full_tick、download_history_data还是subscribe很多刚开始用QMT的人分不清数据API之间的区别常常在代码里混用导致性能下降。关于这个我很早以前就搜过量化交易用的数据api最好的是什么其实没有绝对的最好只有最适合当前场景的选择。QMT的xtdata模块核心接口大概可以分为三类API用途性能特点get_full_tick获取某个合约的最新tick快照适合低频轮询如每秒一次或每几秒一次subscribe_quote订阅实时行情主动推送适合策略实时运行主动更新K线download_history_data下载历史数据到本地缓存用于预热历史K线适合启动时调用一次对于双均线策略最合理的组合是启动时用download_history_data拉一段历史数据然后调用subscribe_quote订阅实时行情之后靠回调推送来滚动更新K线。而不是在主循环里反复get_full_tick去轮询那样既慢又容易让数据不连续。4.2 极速策略的代码性能优化从循环到向量化我见过不少人在QMT里跑策略明明信号逻辑是对的但策略就是“卡”。大部分情况下问题出在代码写法上。Python的循环在数据量小的时候看不出问题但如果你在tick回调里处理全市场几千只股票的K线数据性能就会急剧下降。以双均线为例哪段代码是性能瓶颈呢通常是用rolling计算均线这一步。如果你每次只来了一个最新tick就不需要重算全部历史均线只需要用增量的方式更新。比如维护一个全局变量记录上一根K线的均线值新K线到来时用新收盘价和最近几根K线计算新均值而不是对整个DataFrame重跑一遍。# 增量更新均线的思路 def update_ma_online(df, fast_ma_prev, slow_ma_prev, new_close): # 这里示意性地说明思路维护一个滚动窗口列表 window_fast.append(new_close) window_slow.append(new_close) if len(window_fast) FAST_MA: window_fast.pop(0) if len(window_slow) SLOW_MA: window_slow.pop(0) return sum(window_fast)/len(window_fast), sum(window_slow)/len(window_slow)这个例子只是最简化的示意但在真实高频场景下减少DataFrame重算能省下一大截时间。类似地像是把日志写入从同步写改成异步队列也能减少主循环阻塞。4.3 实盘运行时的调度与生命周期管理策略不是写出来就能挂着跑的尤其是在实盘环境下任何异常都可能导致策略停止、错过信号或者重复下单。我自己的策略里有一个main循环负责长时间运行同时用异常捕获来保证程序不会因为意外退出。def main(): trader_ready False while True: try: if not trader_ready: # 连接检查与重连逻辑 pass df get_kline_df(STOCK_CODE) if len(df) 0: signal generate_signal(df) if signal: log(f生成信号: {signal}) if signal BUY and position.get(STOCK_CODE, 0) 0: ok risk_check(STOCK_CODE, 1000, BUY) if ok: send_order(STOCK_CODE, BUY, 1000) elif signal SELL and position.get(STOCK_CODE, 0) 0: ok risk_check(STOCK_CODE, position[STOCK_CODE], SELL) if ok: send_order(STOCK_CODE, SELL, position[STOCK_CODE]) except Exception as e: log(f异常信息: {e}) time.sleep(3) # 根据标的周期调整轮询间隔 if __name__ __main__: main()这里的sleep(3)是双均线日线策略的轮询间隔因为日K不是实时更新的三秒钟轮询一次已经足够。如果是分钟级策略轮询或订阅的节奏就要相应加快。5. 回测、模拟与实盘的坑先用Python跑通再用QMT跑通5.1 回测框架选型backtrader怎么快速验证策略逻辑我一直认为QMT是用来实战的策略逻辑该验证的环节应该在回测框架里先完成。我在标题相关热词里看到backtrader策略说明很多人也在用这个框架。关于backtrader和QMT的关系我的理解是两者不是二选一而是上下游。backtrader适合做策略逻辑验证QMT适合做真实行情接入和实盘交易。下面是一个用backtrader实现双均线策略的最小回测示例逻辑和上面的QMT策略完全一致。这样可以确保你在写QMT代码之前已经对策略参数和逻辑有把握。import backtrader as bt class MaCrossStrategy(bt.Strategy): params ((fast, 5), (slow, 20),) def __init__(self): self.fast_ma bt.indicators.SimpleMovingAverage( self.data.close, periodself.params.fast) self.slow_ma bt.indicators.SimpleMovingAverage( self.data.close, periodself.params.slow) self.cross bt.indicators.CrossOver(self.fast_ma, self.slow_ma) def next(self): if not self.position: if self.cross 0: self.buy() elif self.cross 0: self.close() if __name__ __main__: cerebro bt.Cerebro() # 加载数据这里示例省略了数据源加载过程 cerebro.addstrategy(MaCrossStrategy) cerebro.run()回测中最容易犯的错误是使用了未来函数比如用当天的收盘价发出当天开盘时的信号这在回测里会大幅美化结果。正确的做法是信号只能基于已经确认的历史数据比如当天收盘后计算出的信号对应的是次日的操作。5.2 模拟盘到实盘的差异滑点、手续费与涨跌停QMT的模拟盘和实盘接口上没有区别代码几乎可以无缝切换但成交结果会有本质差异。这一点我在实盘模拟中体会特别深。模拟盘通常按照当前盘口价格撮合不存在滑点也不一定考虑你的委托量对盘口的影响。实盘的一笔大单如果直接砸向对手价常常会吃掉好几个档位成交均价可能比你预期的差不少。另外手续费虽然看着不高但高频策略中手续费会成为吞噬利润的主要因素。还有涨跌停的问题。模拟盘可能会让你在涨停价成功买入但在实盘里涨停板上买单排队极度拥挤你的小额订单大概率是无法成交的。所以策略里必须加入涨跌停状态的判断在无法交易的情况下放弃该信号避免出现“信号显示开仓了但实际持仓并没有变化”的状态错乱。5.3 我的避坑清单数据复权、停牌处理与时间戳最后分享几个我实际踩过的、看再多文档也很难意识到的坑。第一个是复权数据。股票的K线如果不复权分红除权那天会出现价格跳空双均线策略会在除权日产生虚假信号。QMT的xtdata接口可以获取复权因子建议在K线数据里使用后复权价格。第二个是停牌股票的持仓处理。如果持仓股票停牌你的卖出信号会报错或者一直不成交。最好的方式是维护一个停牌股票列表在信号生成阶段自动过滤掉停牌股票。第三个是QMT时间戳单位。get_market_data_ex返回的time字段通常是Unix时间戳单位是秒。如果你把秒级时间戳直接转成字符串会发现时间对不上因为Python的datetime.fromtimestamp默认处理的就是秒。如果你拿到的是毫秒级时间戳就需要先除以1000再做转换。这个细节让我有一次排查了很久日志里的时间整整偏了8小时怎么都对不上。这套流程跑通之后你再去看QMT终端里那些五花八门的策略示例就会发现它们本质上都是这套逻辑的各种变体连接终端、订阅数据、计算信号、下单、记录结果。我自己的体会是Strategy本身不重要重要的是把数据传输、信号计算、委托管理、风控逻辑和异常处理拆干净每一块都能独立工作组合起来才稳定。最后再分享一个我自己的小技巧第一次在QMT里跑通策略时不要急着上实盘资金先用模拟账号跑两到三周期间每天检查日志和成交记录。你可能会发现自己对策略的运行预期和实际情况之间有不少偏差。调整好心态把基础设施磨扎实了再切换到实盘胜率会高很多。
阅读完成 · 觉得有帮助？
