21天搭建ETF量化交易系统今天写到第19天。前18天我们搞定了数据源接入、动量因子计算、历史回测框架、参数寻优整个策略本地跑起来已经像模像样了。但说句实话本地脚本始终不够“顺手”——每次要看持仓得翻命令行想手动干预一笔调仓得改代码更别提把策略逻辑共享给同事实时看了。所以DAY19的目标很明确把轮动系统做成Web版同时打通miniQMT下单接口让信号能真正变成真金白银的ETF委托单。先交代一下今天做完的效果浏览器里能实时看到候选ETF池的动量排名、当前持仓、今日盈亏、调仓信号提示页面上点一下“手动执行调仓”系统会自动算好卖出清单和买入清单逐笔核对后通过miniQMT推给券商交易通道。整个链路从信号到下单耗时基本在2秒以内实盘验证下来稳定可用。这篇文章我把Web轮动系统的完整设计、miniQMT接入的关键细节、以及我在调试过程中踩过的坑全部记录下来照着做你也能在一天内把策略从“能回测”推进到“能下单”。1. DAY19任务拆解策略落地前的最后一块拼图1.1 为什么选择在DAY19接下单接口很多做量化的人容易陷入一个怪圈回测做得无比精致夏普比率、最大回撤、胜率样样好看但一上实盘就抓瞎。原因很简单回测和实盘之间隔着一整套交易执行链路包括信号产生、订单生成、路径检查、报单推送、成交回报、持仓同步。这一整套东西本地脚本不太好解决尤其是当你需要频繁干预或者半自动交易的时候。我在DAY18做完了策略参数固化当时就想下一步一定得让策略“能用”。怎么定义“能用”至少要做到两点第一打开浏览器就能看到策略当前该干什么、为什么这么干第二策略给出的调仓指令能一键推送到券商并看到委托结果。因此DAY19的核心任务就是Web可视化和交易通道打通这俩缺一不可。1.2 今天要完成的四个核心目标如果只用一个词概括今天的开发内容那就是“接线”——把已有的策略内核接到两个输出端口上一个是Web前端一个是券商交易API。我把今天的任务拆成四个子目标每个都在代码层面闭环目标一构建Web轮动系统基础框架。选型上用FastAPI加轻量级前端静态页面不引入重型前端框架降低维护成本让一个Python工程师也能搞定全栈。目标二完成轮动信号的Web可视化。把候选池、动量排名、当前持仓、调仓建议全部展示在页面上。目标三打通miniQMT交易接口。完成登录鉴权、账户初始化、持仓同步确保能查询资金和持仓。目标四把调仓信号转成真实交易指令。实现从“信号列表”到“委托单”的映射再确认下单的完整链路。这四个目标对应的是同一套主流程策略引擎算出调仓信号Web层负责展示和确认交易层负责执行和回报。下面我会按这个顺序详细展开。2. Web版轮动系统让策略可视化2.1 技术选型为什么用FastAPI加轻量前端我见过很多人做量化Web系统一上来就搞前后端分离Vue或者React起一个工程再配Node中间层折腾两三天连后端接口都没调通。对个人量化系统来说这属于过度设计。我的原则是能用最简单的方式解决绝不上复杂架构。FastAPI的优势很明显原生支持异步性能足够自带自动生成API文档而且和Python生态无缝衔接。我前18天的策略代码全部是Python写的直接import就能用不用做跨语言对接。前端我选择用服务端渲染的Jinja2模板加少量原生JavaScript和ECharts画图整个前端只有三个页面文件不构建、不打包、不编译改完保存就能看到效果。整个系统的模块划分大概是这样的core/strategy.py轮动策略核心逻辑包括动量计算、排名、调仓信号生成。core/data_source.py行情数据模块负责拉取ETF日线数据和实时行情。core/portfolio.py持仓管理和目标仓位计算。trader/xt_client.pyminiQMT交易接口的封装所有券商操作都集中在这一个模块。web/app.pyFastAPI主入口路由、接口、页面渲染都在这里。web/templates/前端模板文件。这样的分层保证了策略逻辑和交易通道是解耦的。策略代码负责“想出结果”trader模块负责“执行结果”Web层只是中间展示和交互的壳。万一某天你想换券商只需要重写trader模块策略逻辑完全不用动。2.2 前端页面的核心设计与交互逻辑Web页面我设计成三个Tab每个Tab解决一个问题第一个Tab是“策略总览”最上面显示当前策略状态、上次调仓时间、今日收益、持仓ETF数量中间是持仓明细表最下面是资产曲线。这个页面主要用来看“现在是什么状态”不需要太多操作。第二个Tab是“候选池排名”展示当前ETF池子里所有标的的动量得分排名支持按得分排序、按代码筛选每一行会标注这只ETF相对上一周期的排名变化方便我快速感知市场风格是否发生了切换。排名表下面还有一个对比图用ECharts画几只头部ETF的净值走势对比直观展示为什么策略选它们。第三个Tab是“调仓执行”这是今天的关键交互页面。策略每次计算完后如果发现当前持仓和目标是不同的就会生成一个调仓方案页面上会展示两部分卖出的ETF和买入的ETF每只都标注了代码、名称、操作方向、目标份额和预估金额。这个页面需要人工确认确认后点击“执行调仓”按钮调仓方案才会被推送到交易模块。这个交互设计是有讲究的。全自动交易虽然省事但初期风险太高万一策略出了bug连个拦截机会都没有。保留一个人工确认步骤等于加了一道保险成本只是每次调仓时多点一次按钮。我建议所有刚开始接实盘的兄弟都保留这个“半自动”模式跑一段时间觉得稳定了再考虑全自动。2.3 后端信号计算与状态管理后端除了提供API接口还有一个后台任务在定时执行策略计算。我用的是FastAPI的BackgroundTasks加一个简单的循环调度器每15分钟拉一次行情数据重新计算一次动量排名。计算完以后结果会保存到内存中的状态对象里同时把关键中间结果存成JSON文件这样即使Web服务重启了也能快速恢复到重启前的状态。状态对象的设计我花了点心思。它需要同时承载三类数据策略计算结果、当前实际持仓、最近一次调仓记录。实际持仓的数据来源是miniQMT接口返回的持仓信息策略计算结果则是纯计算得到的目标持仓两者之间的差距就是待调的仓位。代码实现上我用了一个dataclass来定义这个状态关键字段大概是这样dataclass class StrategyStatus: # 策略元信息 strategy_name: str etf_rotation_day19 last_calc_time: datetime None last_trade_time: datetime None # ETF池与动量排名 etf_pool: list field(default_factorylist) momentum_scores: dict field(default_factorydict) ranking: list field(default_factorylist) # 持仓目标与实盘差异 target_positions: dict field(default_factorydict) # 目标持仓: code - shares actual_positions: dict field(default_factorydict) # 实盘持仓: code - shares pending_trades: list field(default_factorylist) # 待执行调仓指令这里有一个容易忽略的细节target_positions和actual_positions必须分开存。很多人会把这两个混在一起导致信号重复触发或者漏触发。比如策略说买入510300共10000份但实际账户里已经有3000份实际只需要买入7000份。如果你不做区分要么重复买入要么漏买都会带来不可控的仓位偏差。这也是我在做题过程中总结出的一个核心原则量化系统里目标状态和实际状态永远是两本账必须时刻对账。3. miniQMT下单接口从鉴权到委托成交3.1 miniQMT是什么为什么选择它做A股量化交易接口方案其实非常少。普通散户能接触到的通道无非就几类网页手动下单、券商手机App、miniQMT、Ptrade或者其他付费量化终端。miniQMT是目前性价比最高的一档它是券商提供的轻量级量化交易接口以Python库xtquant为核心可以在本地跑策略代码直接调用交易接口下单。相比Ptrade全部在云端跑miniQMT更灵活数据是本地拉取策略代码完全自主可控而且没有额外的按年付费成本通常达到一定资产门槛就能开通。miniQMT本质上是一个本地客户端进程它会连接券商的交易服务器然后在本地提供一个动态库xtquant让我们用Python调用。整个过程不涉及任何敏感操作反而非常依赖券商侧的安全校验包括登录时要验证账号密码、动态口令有些券商还绑定了固定的交易终端。整个链路是在合规框架内运行的你放心使用就行。3.2 xtquant的核心API调用流程我把miniQMT的接入代码封装在trader/xt_client.py里整体流程分为四步初始化、连接、订阅账户、查询同步。初始化阶段要做三件事指定miniQMT的安装路径、分配一个会话ID、准备交易账户对象。from xtquant.xttrader import XtQuantTrader from xtquant.xttype import StockAccount from xtquant import xtconstant # 路径指向miniQMT客户端安装目录 qmt_path C:/qmt/stock session_id 10001 account StockAccount(你的资金账号, STOCK) # 初始化交易会话 trader XtQuantTrader(qmt_path, session_id) trader.start() connect_result trader.connect() if connect_result ! 0: raise RuntimeError(f连接miniQMT失败错误码: {connect_result})连接成功之后还需要把交易账户订阅到这个session上这样才能查询资金和持仓# 订阅账户信息 subscribe_result trader.subscribe(account) if subscribe_result ! 0: raise RuntimeError(f订阅账户失败错误码: {subscribe_result}) # 查询账户资金 asset trader.query_stock_asset(account) print(f总资产: {asset.total_asset}, 可用资金: {asset.cash}) # 查询持仓 positions trader.query_stock_positions(account) for pos in positions: if pos.volume 0: print(f{pos.stock_code}: 持仓{pos.volume}股, 可用{pos.can_use_volume}股)这里要注意query_stock_asset和query_stock_positions返回的是xtquant的封装对象字段名长而且有点偏底层建议封装一层to_dict方便后续使用。另外所有查询操作都是同步阻塞的如果行情卡顿或者网络波动可能会等很久。我的做法是设置一个超时机制用线程池去执行这些查询超过3秒就标记为查询失败避免Web接口被拖死。3.3 下单一笔ETF的完整示例买入ETF的下单接口是order_stock核心参数包括证券代码、操作类型、委托量、价格类型、委托价。由于ETF是场内基金交易规则和股票一致最小单位是100份所以下单数量必须是100的整数倍。我以买入华泰柏瑞沪深300ETF510300为例写一个最小可执行的函数def buy_etf(trader, account, stock_code, volume): # 检查下单数量合法性 if volume % 100 ! 0: volume (volume // 100) * 100 print(f下单量校准为100的整数倍: {volume}) # 使用最新价对手价下单避免指定价格导致无法成交 price_type xtconstant.FIX_PRICE # 先查一下当前最新价做一个价格保护 # 这里用xtdata的行情接口快速获取最新价实际代码里我有单独封装 latest_price get_latest_price(stock_code) # 委托价格设置为最新价向上取2个tick模拟对手价 order_price round(latest_price * 1.001, 3) order_id trader.order_stock( account, stock_code, xtconstant.STOCK_BUY, # 买入 volume, # 委托数量 price_type, # 价格类型 order_price, # 委托价格 etf_rotation_buy # 备注标记 ) return order_id完整下单必须包含几个前置校验首先是资金检查买入的总金额不能超过账户可用资金的90%留出手续费缓冲其次是持仓上限检查单一ETF占比不超过总仓位的20%最后是涨停价检查避免在极端行情下追高买入。这些校验逻辑不是在策略代码里写的而是在交易模块下单前统一执行的确保任何来源的信号都要过风控这一关。我在下单前还会做一次“委托预检”就是先把这笔委托的金额、数量在控制台上完整打印出来与目标仓位做对比确认偏差在允许范围才真正调用order_stock。这一步看似冗余实则是实盘交易和回测最大的区别——回测中不存在资金不足、品种停牌、数量不合法这些问题但实盘每个问题都可能让委托直接废掉。4. 轮动信号与实盘交易的对接细节4.1 从信号触发到委托推送的完整链路策略计算和交易执行是两个模块需要有个东西把它们串联起来。我的做法是让Web后端的调仓接口来当这个中间人。完整链路是这样的定时任务每15分钟计算一次动量排名保存到StrategyStatus中。前端轮询“获取调仓方案”接口拿到pending_trades列表。用户在Web页面点击“执行调仓”前端把调仓方案传给后端。后端重建调仓方案和最新计算结果的哈希值做比对防止用户看到的是过期信号。后端逐笔校验调仓指令的完整性和合规性。对每一笔指令调用trader模块执行买入或卖出。所有委托提交完成后刷新持仓并更新数据库中调仓记录。第4步的哈希比对是我加的一个细节保护。为什么需要这个因为轮动信号是动态计算的用户在页面上看到一个调仓方案如果他没立刻执行等了几小时才去点按钮这期间信号可能已经变了。如果不做校验用户等于拿着一张过期的“购物清单”去交易大概率会造成不必要的换手。所以我每个调仓方案生成时会附带一个基于内容和时间戳的哈希值超过30分钟或者排名有变动页面就会提示“调仓方案已过期请重新获取”。4.2 调仓数量计算与零股处理计算调仓数量看着简单实际坑不少。我举个例子策略目标是把510300的仓位调整到总资产的20%当前总资产是100万目标市值就是20万用最新价算出目标份额可能是75610股但你手里只有75600股。这时候系统应该按“保守覆盖”原则向下取整到100的整数倍即75600股这10股的差距作为跟踪误差处理不要去追求完美。如果是卖出也一样目标份额小于当前持仓时先算出差额再向下取整。但有一种特殊情况要注意如果某只ETF被完全踢出持仓池目标份额是0此时即使当前持仓只有150股不是100的整数倍也要一次性全部卖出把零股清干净。零股单独逻辑处理def calc_trade_volume(target_position, actual_position): diff target_position - actual_position if target_position 0: return actual_position # 清仓时卖出全部持仓包括零股 if diff 0: return (diff // 100) * 100 # 加仓只按手买 if diff 0: sell (-diff) // 100 * 100 # 如果卖出后剩下的不足一手直接清仓 if actual_position - sell 100: sell actual_position return -sell这段代码虽然简单但涵盖了ETF交易最核心的逻辑加仓按手买清仓一次清剩余不足一手就全卖。我在网上看过很多量化教程这个细节被一句话带过但实盘里没处理好轻则委托失败重则持仓无法清空次日还要处理零股非常麻烦。4.3 成交回报与订单状态同步下单完成只是开始后面还有一件重要的事确认这笔单子到底成交了没有。miniQMT的order_stock是异步提交函数返回的order_id只是委托编号不代表已经成交。你需要主动查询订单状态或者在绑定回调函数里等待成交回报。我的trade模块里专门写了一个轮询函数委托提交后每500毫秒查一次订单状态最长等待15秒。如果订单状态是已全部成交就更新持仓缓存如果是部分成交继续等待如果显示废单或已成废单记录错误原因并且不再重试。这里分享一个我的经验委托失败不要盲目重试。废单往往是因为价格超范围、品种停牌、资金不足或者权限问题你应该先打印返回的错误码找到具体原因再决定是否重新下单否则可能连续刷废单还有可能被券商风控判定为异常账户。订单状态这块xtquant提供的接口也很方便# 查询委托 order_list trader.query_stock_orders(account) for order in order_list: if order.order_id target_order_id: if order.order_status xtconstant.ORDER_ALL_TRADED: print(f委托{target_order_id}已全部成交: {order.traded_volume}股) elif order.order_status xtconstant.ORDER_PART_TRADED: print(f委托{target_order_id}部分成交: {order.traded_volume}/{order.order_volume}) elif order.order_status xtconstant.ORDER_REJECTED: print(f委托{target_order_id}被拒绝, 错误码: {order.order_error_msg})如果我的主循环里正在跑策略计算同时还要轮询订单状态我就会用一个独立的后台线程来干轮询这件事避免阻塞策略计算线程。这是多线程协调的问题好在Python的threading模块足够用只要注意共享变量加锁就不会出大问题。5. 实战中的坑与排查方法5.1 Web端数据刷新的两个典型问题Web系统上线后我遇到的第一个问题是ECharts图表数据不刷新。我最初是用页面加载时一次性拉取数据然后setOption但策略每15分钟更新一次排名图表却还是旧数据。排查后发现是浏览器缓存导致API请求被浏览器缓存了FastAPI默认不会为GET接口设置no-cache头。解决办法是在路由装饰器里显式指定缓存禁用from fastapi import Response app.get(/api/status) async def get_status(response: Response): response.headers[Cache-Control] no-store, no-cache, must-revalidate, max-age0 response.headers[Pragma] no-cache return strategy_manager.get_status()第二个问题是前端定时轮询和策略计算线程的时序错乱。前端每10秒拉一次排名数据但策略计算可能刚好在计算到一半状态对象的字段还是旧值导致前端偶尔拿到不完整的数据。解决方案是在状态更新时用一个版本号递增的机制读取的时候检查版本号是否稳定不稳定就等下一轮再返回。这其实就是最简单的乐观锁思路虽然实现不复杂但能有效避免脏读。5.2 miniQMT连接不稳定的排查记录miniQMT有个特点是如果你长时间没有调用接口本地进程可能会进入休眠状态再次调用时就会出现连接超时或者返回错误。最初我以为是网络问题排查了很久后来发现是miniQMT客户端自身的特性。我的处理方式有两条路一是在策略主循环里每隔60秒做一次轻量查询比如查询资金保持连接活跃二是写一个断线重连机制当检测到connect返回失败或者查询结果异常时自动重新执行连接、订阅账户的流程。由于每个session ID绑定一次连接重新连接时最好换一个新的session ID避免旧的session产生冲突。还有一个坑是本地进程里最好只保留一个XtQuantTrader实例到处用。如果代码里不小心创建了多个实例去连接同一个客户端会出现串号或者相互踢下线的问题。我刚开始写测试代码时IPython里每次执行都会新建一个trader实例结果miniQMT客户端频繁弹出重新登录的提示就是多个实例抢占连接导致的。后来我把它改成单例模式问题立刻消失。5.3 账户资金和持仓不一致的核对方法实盘交易中最怕的就是程序显示的持仓和券商账户实际持仓对不上。这种情况通常出现在手动交易和程序交易混用的时候——比如你在券商App上手动买了几手程序不知道然后程序计算调仓时就给出了错误的卖出数量。我的解决办法很简单也很有效在下单之前强制从miniQMT接口拉一次最新持仓而不是用内存中的缓存。只有在接口查询失败时才退回到缓存数据并且此时拒绝执行下单操作。也就是说持仓数据必须以交易所记录为准任何本地记录都只能作为参考。这是我在一次调仓差点卖飞之后总结出的规矩从那以后就把这个逻辑写死在交易模块里了。如果你发现系统持仓和实际持仓不一致最直接的办法是写一个小工具拉取实际持仓与目标持仓做对比把差异明细输出到日志和页面上。这样每次调仓前肉眼扫一眼有没有差异一目了然。我在Web页面上专门加了一个“持仓差异”功能区每次调仓前都会高亮显示差异超过100股的标的提醒人工复核。6. 几个适用于所有量化新手的通用建议今天的内容做完了但我还有几句话想多说。很多人一开始做量化系统总想追求“全自动”“无人值守”我觉得这个思路对新手不太友好。交易系统最核心的应该是稳定、可控、可解释。稳定是说跑一个月不崩可控是说任何时刻出问题你都能快速定位并干预可解释是说每一笔交易都能追溯到具体的策略逻辑和信号来源。这三点做到了再谈自动化也不迟。另外我建议你给系统加一个完整的操作日志模块。每次调仓、每笔委托、每个错误提示、每条风控拦截原因都要记录下来。这个日志不仅是排查问题的工具也是你事后复盘策略表现的依据。我DAY19这天的日志文件就有几千行但正是因为这些日志我才能准确复现白天发生的每一个异常场景。今天的内容就是把“信号”和“交易”这最后一公里接通了。后面还有两天我会继续打磨系统包括加上更完善的策略监控告警、优化调仓算法、增加历史交易记录分析模块。到时候再继续更新。