做量化这几年我一直被一件事折磨策略思路不复杂复杂的是每天围绕策略的杂活。导数据、算指标、看消息、盯持仓、复盘归因每一样都耗时每一样都容易出错。后来我干脆把这些工作抽象成一套多AI-Agent协作的工作台起了个名字叫QuantBot目标是让整个交易流程形成一个从盘前到盘后不依赖人肉盯盘的自动化闭环。这篇文章就把这套工作台的设计思路、模块怎么拆、Agent怎么协作、落地时踩过的坑全都摊开讲一遍适合那些已经写过一点量化策略、想往工程化和自动化方向走的同学参考。先说清楚QuantBot到底是什么。它不是一个策略库也不是一个简单的回测框架而是一个把AI-Agent当作用户围绕交易日历自动调度任务的系统。盘前自动聚合数据、生成当日观察清单盘中根据价格和市场状态触发信号、执行策略逻辑盘后自动拉取成交记录、计算指标、生成复盘报告然后把当天的数据和结论反馈给策略模块做下一次优化。整个链路以时间为驱动Agent之间通过消息队列通信互不阻塞。1. 项目整体设计思路交易日常是怎么被拆成Agent任务的1.1 先理解手工交易流程里的“隐形成本”很多人觉得量化的核心是策略和模型实际上等你真正跑起来就会发现最大的成本是流程组织。手工操作时你每天要在数据终端、行情软件、交易接口、Excel之间来回切换任何一步断了后面的决策就跟着受影响。QuantBot的设计起点不是怎么写策略而是先把一名交易员每天的动作完整列一遍再把这些动作抽象成可并行的任务单元。整体拆下来交易日常大致可以分成三个阶段。盘前拉取隔夜外盘、新闻舆情、个股公告计算昨日收盘后需要关注的标的池跑一遍候补策略的指标条件输出当日观察列表。盘中按固定频率轮询行情、更新持仓浮动盈亏、触发策略信号后生成委托指令同时监控风控阈值。盘后下载当日成交、核对持仓、结算收益、绘制资金曲线、把每一次信号触发的原因和结果写成结构化文本存入数据库。这三个阶段的任务类型差异很大用一套单体程序硬写也能跑但每一处改动都会牵扯到其他模块我在早期版本里就吃过这种亏。所以QuantBot在设计上做了一个决定把不同阶段的任务交给不同职能的Agent去干Agent之间不直接调函数而是通过事件总线交换消息。盘前任务完成之后会发布一个“观察池已生成”的消息盘中任务收到消息后才开始对接下来的轮询逻辑做初始化。这样做的最大好处是单个Agent的内部逻辑可以随便改不会影响上下游任务。1.2 为什么选AI-Agent协作而不是回调式流程其实第一版QuantBot用的是回调式流程盘中模块直接import盘前模块的数据类盘后模块又直接读取盘中模块的全局变量。问题出在异常处理上只要某一个环节因为数据缺失抛异常整个链路就中断了而且不好定位是哪个环节出的错。后来我意识到量化工作流本质上是一个多角色协作的过程数据源、信号生成器、风控器、执行器、记录器各自负责一块它们之间需要的是解耦不是强引用。用AI-Agent的好处有三个。第一Agent天然有一个“会话边界”你只需要定义清楚它接收什么消息、输出什么消息内部的实现细节可以被完全封装。第二Agent可以根据消息内容动态选择处理逻辑比如盘中Agent收到不同标的的事件消息时可以自行决定是触发信号计算还是直接进入风控校验不需要主流程做一大串条件判断。第三Agent可以独立启停和升级这在实际运维中很实用某个Agent出问题时可以只重启它而不用把整个进程都拉起来。在QuantBot里每个Agent的输入输出都被定义成标准化的JSON结构内部则是一段核心逻辑加一个可选的LLM辅助层。LLM不是用来做交易的而是用来处理那些传统规则很难覆盖的内容比如把新闻文本转成对持仓影响的定性结论或者把复盘数据整理成自然语言报告。策略计算本身是确定性的规则和模型代码不依赖LLM的随机性。1.3 盘前、盘中、盘后闭环的数据流怎么串起来为了让整个闭环可观察、可追踪我给每个Agent的消息都加了一个trace_id这条id会从盘前一直贯穿到盘后。盘前Agent生成观察池时会创建trace_id盘中Agent处理行情事件时把同一个trace_id带过去盘后Agent在写复盘记录时也引用它。这样在查询某天某个标的为什么被买入或没有被买入时只需要按trace_id拉出整条链路的事件日志。具体的数据流大概是这样的盘前任务先从天级行情库和新闻API拉数据通过因子计算模块生成一个候选池再把这些候选池数据写入Redis缓存同时发布一个watchlist_ready事件。盘中Agent订阅以tick_为前缀的事件每收到一条或者累计到一定时间窗口后从Redis里取该标的基本信息结合实时价格计算买入条件触发之后把信号写入订单队列等待执行模块确认。执行回报一旦推回来Agent会再更新一轮持仓缓存并在内存里维护一份当日已操作记录避免重复发单。盘后Agent则是在收盘信号之后统一触发它会把当日订单流水、成交回报、持仓快照拉出来用统一的计算函数算出当日盈亏、累计收益、回撤等指标再调用LLM服务生成一段文字总结。这份总结和结构化指标最终会被写入一个按日期分表的SQLite或者PostgreSQL库。到这里一整天的交易循环才真正闭合。2. 核心模块拆解五个Agent各自负责什么2.1 行情与数据Agent底层数据的质量决定上层策略的生死行情与数据Agent是整个QuantBot里接触外部依赖最多的角色。它在盘前负责连接数据服务商接口拉取日线、分钟线和实时快照在盘中则通过WebSocket接收持续推送的行情流。选择这个模块作为独立Agent主要是考虑到数据层可能会频繁切换数据源比如从A股Level-1换到Level-2或者接入外盘行情。如果不做封装切换带来的改动会扩散到所有上层模块那绝对是一场灾难。这个Agent在设计上有两个关键点数据对齐和异常标注。数据对齐是指把不同来源的数据统一成一套字段规范比如时间统一成UTC时间戳价格统一成浮点成交量统一成手。异常标注是指在拉取过程中发现数据缺失、停牌、涨跌停、长时间无成交等情况时不做主观填值而是把这些状态写进数据的meta字段里。这样做看似在给自己找麻烦但在后续信号生成阶段非常有用策略Agent可以直接读取meta来避开无效行情。我在实际项目里遇到过一个典型问题某只股票盘中长期停牌如果只看价格字段会误以为没有满足触发条件导致本该暂停处理的标的进入了信号计算。后来我在数据Agent里加入了“最新状态”字段盘中轮询时先判断state是否为可交易再做后续计算这类误触发才被彻底解决。2.2 策略信号Agent把规则和模型统一成可解释的信号输出策略信号Agent是整个工作台里最像“大脑”的模块。它接收标准化后的行情数据按照用户预先注册的策略列表逐一计算信号。这里的重点不是策略本身有多复杂而是如何保证多个策略的输出格式一致性。QuantBot里每个策略最终都要返回一个统一的Signal对象包含标的代码、方向、预期持仓周期、置信度、触发原因、关联指标快照。这样的好处是无论你写的是简单的均线交叉策略还是用了机器学习的分类模型下游风控和执行模块都不需要做任何区分只要按Signal协议处理就可以。这个Agent同时也要处理策略之间的优先级关系。我的做法是给每个策略注册时分配一个priority字段默认100数值越小越优先。如果同一标的同时被多个策略触发会先按优先级排序再由一个条件判断器决定是否合并订单。比如底仓策略和短线策略同时给了一个买入信号条件判断器会检查当前仓位如果底仓策略已经持仓就只执行短线策略超出的那部分仓位。还有一个容易被忽略的细节策略信号Agent要维护一个“信号去重”的窗口。盘中行情频繁波动时同一时刻可能产生大量重复信号。我在模块里用了一个cache记录最近60秒内已经处理过的策略标的组合在这个窗口内同样的信号会被直接丢弃避免重复下单。2.3 执行与风控Agent在下单前做最后一道闸门执行与风控Agent是QuantBot里最严格的一个模块它的核心任务不是赚钱而是不亏不该亏的钱。所有策略信号在生成委托指令之前都必须经过这个Agent的校验通道。校验通道由一系列规则组成每一个规则都是一个独立的检查函数按顺序执行任何一个规则不通过就会拒绝本次委托然后写入日志。我在风控规则列表里开了这几个默认项最大单笔买入金额、单标的最大持仓比例、账户总回撤阈值、日内最大亏损阈值、单标的日内累计买入次数限制。其中账户总回撤阈值用的是阶梯式设计比如回撤小于5%时不限制回撤5%-8%时降低单笔金额回撤超过8%时暂停新开仓。这么做比单一固定阈值更合理因为行情波动大的时候固定阈值很容易被一次正常的波动扫掉导致后续策略直接全部失效。执行模块本身负责把通过校验的委托指令发送到交易接口。发送之前还会做一次价格偏移检查也就是把最新成交价和信号触发时的参考价做对比如果偏离幅度超过约定阈值就延迟执行或者直接放弃。这个设计主要是应对滑点风险特别是用小周期分钟线的时候质量差的实时价格经常会和信号计算时的价格差出一大截。2.4 复盘与报告Agent让每天的亏损和盈利都有迹可循复盘与报告Agent大概是QuantBot里最早体现出“闭环”价值的模块。以前做量化的人很多都懒得做复盘顶多看一眼资金曲线。但实际上能否把某个策略在某一天、某只股票上为什么赚或为什么亏回答清楚直接决定了策略迭代的效率。这个Agent在收盘后自动启动把当天的委托记录、成交记录、行情快照和策略信号拉出来分标的、分策略地聚合统计输出一份结构化复盘数据。如果要生成自然语言报告这个Agent还可以调用LLM接口通过一个prompt模板把结构化数据渲染成完整的日度复盘报告。prompt模板里会包含各标的的收益率、信号触发次数、成交均价与信号价的偏移、持仓时间等字段要求模型用客观、简洁的语言总结。这里千万要注意不要把LLM的结论当作分析依据它的输出只适合汇报和存档不适合参与决策。盘后分析里还有一个容易被忽视的点信号归因。简单地记录“某策略在某天发出了信号”是不够的需要记录当时触发信号的关键指标值快照比如均线交叉点位、当日成交量、RSI数值等。这样以后复盘时不管是人工查看还是用代码批量计算都能准确还原出当日交易系统的决策依据。QuantBot在信号Agent里强制要求每个Signal都带指标快照这个设计在复盘阶段价值非常大。2.5 调度与监控Agent整个自动化闭环的“幕后总管”调度与监控Agent不直接参与交易逻辑它是所有Agent的宿主。它负责三件事按时间表触发Agent任务、监控每个Agent的运行状态、处理异常后的恢复流程。时间调度部分我一开始用的是简单的cron表达式后来发现交易日历和节假日会让cron很难维护就改成了一个交易日历配置表每晚自动读取下一个交易日的时间节点。监控功能则是每30秒对各个Agent做一次心跳检查如果某个Agent连续几次没有响应就尝试自动重启并且把重启事件写入日志。如果重启后仍然失败就会通过Webhook推送告警到手机避免因为Agent进程挂掉而错过一整天的交易。这个调度器还负责管理所有Agent之间的消息路由在启动时会读取一份Agent注册表把每个Agent能处理的消息类型和订阅关系加载进来。在实际运行中这个模块最关键的优化是“消息积压处理”。某一天行情特别剧烈时盘中Agent产生的行情事件量可能暴涨如果不做限流消息队列会被打爆。我的解决方案是在治理层做了按标的维度的令牌桶限流每个标的每秒最多处理若干条行情事件超出部分直接丢弃因为对于大多数分钟级策略来说高频冗余数据并不会改变最终信号结果丢了也不可惜。3. 实操过程从0到1搭一个最小可运行的QuantBot闭环3.1 技术栈选型和项目目录结构QuantBot本身不绑定特定语言但我个人建议用Python来做前期原型因为数据处理的生态太成熟了等真的跑稳定了再考虑用Go或Rust重写高频模块也不迟。这里给出一个我在项目中实际使用的技术栈组合调度框架用APScheduler消息队列用Redis的Stream结构数据存储用SQLite和PostgreSQL盘中行情通过WebSocket接入回测和数据处理用pandas和numpy。项目目录我会按Agent维度拆成模块而不是按函数维度横向铺开这样每个Agent都能独立测试。目录结构大致长这样quantbot/ ├── agents/ │ ├── data_agent.py │ ├── signal_agent.py │ ├── execution_agent.py │ ├── risk_agent.py │ └── report_agent.py ├── core/ │ ├── message_bus.py │ ├── scheduler.py │ └── config.py ├── strategies/ │ ├── base.py │ ├── ma_cross.py │ └── ml_model.py ├── storage/ │ ├── models.py │ └── repository.py └── main.pycore目录放的是与业务无关的基础设施agents目录是每个Agent的独立实现strategies目录里放策略类storage里放数据库模型。main.py是整个程序的入口负责读取配置、初始化消息总线、注册所有Agent并启动调度器。3.2 消息总线和事件定义怎么写消息总线是QuantBot能解耦的关键。我在core/message_bus.py里实现了一个简单的发布订阅模式底层用Redis Stream来做持久化和消费组管理。为什么要用Redis Stream而不是直接用Redis Pub/Sub因为Pub/Sub的消息是即发即弃消费者掉线时消息就丢了对于交易系统来说这是致命的。Stream则会把消息保存在Redis里消费者可以按组拉取保证每条事件至少被处理一次。消息格式我统一用JSON每个事件至少包含这些字段{ event_id: uuid字符串全局唯一, trace_id: 从盘前到盘后贯穿使用的链路id, event_type: watchlist_ready / tick_update / signal_generated / order_filled / daily_close, timestamp: 事件产生时的UTC时间戳, payload: { # 业务数据不同事件类型有不同结构 } }Agent在订阅事件时只需要声明自己关心哪些event_type。比如signal_agent订阅watchlist_ready和tick_updateexecution_agent订阅signal_generatedreport_agent订阅daily_close。这样整个数据流就是单向的谁都不需要知道数据是从哪来的。3.3 核心Agent的代码骨架示例我拿signal_agent来展示一个Agent的基本结构。它内部会维护一个策略列表每次收到行情事件时遍历所有策略计算信号然后把通过的信号发到消息总线。class SignalAgent: def __init__(self, bus, strategiesNone): self.bus bus self.strategies strategies or [] self.processed_cache {} # 用于信号去重 async def handle_tick(self, event): tick event[payload] symbol tick[symbol] # 60秒内已处理过该标的策略则跳过 cache_key f{symbol}:{tick[minute_key]} if cache_key in self.processed_cache: return signals [] for strategy in self.strategies: signal strategy.generate(tick) if signal and self._pass_dup_check(symbol, strategy.name): signals.append(signal.to_dict()) for sig in signals: await self.bus.publish(signal_generated, { trace_id: event[trace_id], signal: sig }) self.processed_cache[cache_key] True def _pass_dup_check(self, symbol, strategy_name): # 用短周期缓存去重 return True注意这里的processed_cache只是一个简单的内存缓存真实项目中我会用Redis带TTL的key来实现这样即使Agent重启也不会丢失去重状态。策略类本身继承一个base类要求实现generate方法输入是标准化tick输出是Signal对象或None。风控Agent的骨架类似但它订阅的是signal_generated事件校验通过之后才会发布order_request事件。执行模块再订阅order_request去调用券商接口。这样一整条链路上每个Agent都是被动驱动逻辑清晰排查问题时只需要看消息在哪个环节停了。3.4 定时调度盘前、盘中、盘后怎么自动触发调度这一块我用了APScheduler但配合交易日历做了二次封装。核心思路是不要直接写固定的cron时间而是先加载交易日历然后根据当前日期确定当天的盘前时间、开盘时间、收盘时间。盘前任务通常设置在开盘前30-60分钟比如9:30开盘我就设置在9:00跑盘前数据聚合。盘中任务不是靠cron触发的而是由行情推送驱动所以盘中其实不需要调度器做什么只要保证数据Agent的WebSocket连接是健康的。盘后任务则设置在收盘后5-10分钟主要等券商那边把当天最终清算数据生成完毕。交易日历的维护我会用一个简单表格字段包括日期、是否交易日、开盘时间、收盘时间。每年初手动更新一次或者从交易所官网自动拉取。调度器启动时读取全年度日历然后生成对应任务的触发时间列表。from apscheduler.schedulers.asyncio import AsyncIOScheduler scheduler AsyncIOScheduler() async def pre_market_job(): await bus.publish(daily_start, {...}) async def post_market_job(): await bus.publish(daily_close, {...}) for date_info in calendar.get_trading_days(): if date_info[is_trading]: scheduler.add_job(pre_market_job, triggerdate, run_datef{date_info[date]} 09:00:00) scheduler.add_job(post_market_job, triggerdate, run_datef{date_info[date]} 15:10:00)这里有个细节用APScheduler的date触发器比cron好维护因为它天然支持一次性的精确调度不需要担心节假日和周末。而且如果系统在任务触发时才启动date触发器会直接跳过已经过去的时间点避免补跑。3.5 策略注册和参数配置怎么组织因为Agent是完全解耦的所以策略的注册集中在一个config文件里启动时统一加载。config.py里会用一个字典定义每个策略的启用状态、参数、适用标的池和优先级。STRATEGY_CONFIG { ma_cross: { enable: True, params: {fast_window: 5, slow_window: 20}, symbol_filter: [600.SH, 000.SZ], priority: 100, max_position_pct: 0.1 }, momentum_ml: { enable: True, params: {model_path: models/lgb_model.pkl, top_k: 3}, symbol_filter: [all], priority: 90, max_position_pct: 0.05 } }参数配置不要写死在策略代码里否则每次调参都要重新部署。QuantBot的策略基类会接收params字典在初始化时注册为实例属性这样想调整窗口周期或者模型路径只改配置重启Agent即可不用动代码。4. 常见问题与排查技巧实战中踩过的坑4.1 Agent之间信息不同步信号Agent算出的价格和执行Agent实际拿到的价格不一致这是我在实盘模拟中遇到最多的问题。原因是行情数据有多个来源不同Agent订阅的数据快照可能来自不同的推送批次导致同一时间点的价格出现十几秒的偏差。我的解决办法是在消息总线上传递数据时把数据的时间戳作为关键字段执行Agent在收到订单请求后先校验信号时间和当前时间的差值如果超过设定阈值比如30秒就直接拒绝这笔订单不做滑点补偿。另外信号Agent里面不用全局最新行情而是在策略触发时把当时的行情快照连同Signal一起发送给下游。这样下游拿到的价格是确定性的不依赖接收时刻的行情状态。这个调整之后因为价格不一致导致的无效订单基本清零。4.2 盘中Agent重复触发同一个信号高频行情下同一个均线金叉信号可能在几分钟内反复出现如果不做去重执行Agent收到多条相同方向的信号就会重复加仓。这个问题我前面提过解决方案就是信号去重缓存但要注意去重的时间窗口不能拍脑袋定。时间窗口设太短起不到作用设太长又可能错过真实的二次加仓机会。我的建议是按策略逻辑来定如果是短线策略去重窗口设为bar周期的2倍比如基于5分钟K线的策略就设10分钟如果是日线级策略盘中信号基本只在开盘后出现一次设60分钟足够。关键是在配置里给每个策略单独配置dedup_window参数而不是全局统一。4.3 盘后复盘报告里出现了异常数据复盘报告是每天自动生成的如果当天行情数据或者成交数据有问题报告里就会出现明显的异常值比如收益率突然变成几千个百分点。刚开始我以为是指标计算函数写错了后来排查发现是数据Agent在盘中拉取行情时把某些复权因子的跳变也当成正常价格记录了。解决思路是在数据写入时增加一个合理性过滤单笔价格跳动超过前值一定倍数时自动标记为异常数据不参与指标计算。同时复盘Agent生成报告前会先对数据做一次完整性检查如果发现某个标的的成交记录和行情记录不匹配就在报告里注明“该标的数据不完整”而不是给出一个虚假的归因结论。4.4 常见问题速查表现象可能原因排查步骤解决方案Agent进程频繁挂掉行情推送量过大内存占用飙升查看Agent内存曲线和日志定位哪个循环在积累数据加入批量处理限制单次处理条数增加内存上限信号一直不触发消息事件类型未匹配检查Agent订阅的event_type和发布方是否一致在消息总线打日志追踪每条消息的路由结果下单被拒但日志没有错误风控规则静默拦截查看risk_agent日志里的reject_reason字段在风控日志中增加明确的拒绝原因和规则ID盘后报告数据错乱多Agent同时写入同一个数据库表检查写库操作是否用了独立事务按日期分表或加写锁避免并发写覆盖定时任务没有触发交易日历配置遗漏节假日检查calendar表当天是否有记录用交易所公布的节假日表定期同步排查Agent类问题时我最常用的方法不是看完整日志而是把消息总线上的事件流转过程单独打成一个trace文件每个事件产生时间、消费时间、处理结果都记录下来。这样出问题时顺着trace文件扫一遍基本上能很快定位到是哪个Agent拖慢了整个链路或者没有正确响应。另外想提醒一点Agent自动化跑起来之后千万不要完全不管。我的做法是每晚收盘后快速扫一遍当天的告警日志每周做一次策略触发率和成交率的统计。自动化只是把重复劳动压缩了真正对系统健康度的掌控还是得靠人的判断力。QuantBot这种盘前、盘中、盘后闭环的架构最大的价值不是让你“躺着赚钱”而是让你把时间花在策略本身的分析上而不是被无穷无尽的运维杂活淹没。这是我在实际操作中体会最深的一点闭环不是终点而是让你有时间继续迭代策略的起点。