QuantBot 这个项目其实是我把自己做量化两年多踩过的坑重新整理了一遍之后才决定认真搭起来的。以前我也是写一堆零散脚本今天用 A 脚本抓数据明天用 B 脚本跑策略后天手动在交易软件里点几下下单晚上再对着 Excel 表格复盘。时间一长人已经变成系统里最不稳定的环节了——数据漏了没发现交易信号来了人没在电脑前复盘的时候又根本没有完整的决策日志可以追溯。QuantBot 就是想把自己这个「人类不稳定因素」从交易链路里尽量抽离出来做成一个多 AI-Agent 协作的量化工作台把盘前、盘中、盘后三个阶段重新组织成一条自动化的闭环流水线。这篇文章会把整个项目的设计思路、各 Agent 的职责拆分、核心代码片段、以及我实盘落地过程中踩过的坑全部整理出来。不吹不黑这个方案适合已经具备基础量化能力、但还停留在“脚本手动下单”阶段的人参考也适合想把 vn.py、qlib、backtrader 这些工具整合成一套自动化系统的人做个对照。1. 项目整体设计为什么一定要多 Agent 协作1.1 从一个“能用”的脚本到一个“靠谱”的系统最早我做自动化交易的时候用的是一种很朴素的写法把所有逻辑塞进一个 while 循环里拉行情、算指标、判断买卖、下单全部写到同一个文件里。单看代码也不算乱但运行一段时间就会发现问题某个行情源偶尔断流数据更新慢了策略立刻做出错误判断但日志里完全看不出来。策略参数想调一下结果影响了下单模块的状态整个系统要重启。晚上想看复盘报告发现白天只记录了成交价格和数量决策依据根本没存下来。这些问题本质上都不是“代码 bug”而是架构问题。交易链路天然是分阶段、多职责的数据采集、信号生成、风险过滤、交易执行、绩效归因各管一块。如果全部耦合在一起任何一个环节的波动都会传导到整个系统。而且回落的时候你根本没法定位问题出在哪一步。所以我后来对 QuantBot 的定位非常明确它是一个多 AI-Agent 协作的工作台而不是一个“自动交易脚本”。所谓 Agent在这里就是一个拥有明确职责边界、能独立接收输入并产出结果、还可以与其他模块通信的自治单元。策略 Agent 只负责生成信号风控 Agent 只负责否决和拦截执行 Agent 只负责把指令变成订单。每个 Agent 都可以单独调试、单独升级、单独回滚。1.2 盘前、盘中、盘后为什么要拆成三块拆三个阶段不是拍脑袋设计的而是交易市场本身的节奏决定的。盘前的核心是“准备”。这个时间窗口里数据源最安静计算资源最充足你可以放心地做全市场扫描、模型推理、生成候选池然后输出一份「今天大概怎么交易」的计划。盘中的核心是“响应”。行情实时在变每一秒都可能有新信息此时要求的是低延迟、高可靠绝不能在这个阶段跑重计算或者模型重训练。而盘后的核心是“学习”。交易结束后所有的成交、行情、Agent 决策都有了完整记录可以用来复盘、归因、迭代策略甚至重新训练模型把当天的经验沉淀到系统里。这个划分还有个附带好处每个阶段的失败影响范围被天然隔离了。如果盘前模型跑崩了最多是没有候选池盘中系统可以进入保守模式如果盘中执行模块出问题盘后复盘照样能跑把问题记录下来。不会出现“一个异常把整天数据毁掉”的情况。1.3 个人量化者为什么要关心 AI-Agent很多人一听 AI-Agent 就以为是上大模型、写提示词。其实在量化场景里AI-Agent 更常见的形式是“带状态、会决策的自动化模块”。就拿我自己的经验说单靠一个模型预测涨跌是做不成闭环的。你需要一个模块去判断“今天该不该交易”一个模块决定“下多大仓位”一个模块在盘中盯住股票池的变化还要有一个模块在收盘后把当天的交易决策重新梳理一遍。这些模块每个单独拎出来都不复杂但拼在一起就需要一套协作机制。Agent 化的价值就在这里它让每条业务逻辑都能独立演化也让它们之间的交互有明确协议而不是像面条一样纠缠在一起。再加上现在 LLM 的能力也被我引入了复盘环节。以前写复盘报告要自己从日志里翻数据现在可以让模型 Agent 读取当天的决策快照生成一份叙述性的总结和建议。这个后面会详细说。2. 系统架构与 Agent 职责拆分2.1 Agent 角色说明QuantBot 当前一共有七个 Agent分别覆盖数据、策略、风控、执行、记录、复盘和调度每个 Agent 都是独立进程通过事件总线通信。Agent 名称核心职责主要输入主要输出Data Agent行情数据清洗、复权、对齐、落库交易所原始数据、第三方行情源标准化 DataFrame、行情快照Strategy Agent策略信号生成、候选池打分特征矩阵、模型权重带分数的交易建议列表Risk Agent仓位校验、黑名单过滤、熔断控制当前持仓、账户资金、策略信号放行/拒绝/降仓指令Execution Agent下单、撤单、重试、滑点控制风控通过后的交易指令订单状态、成交回报Record Agent全链路日志、决策快照、成交归档各 Agent 状态、行情、订单结构化事件日志Review Agent盘后绩效归因、报告生成、策略迭代建议Record Agent 的日志、账户曲线复盘报告、参数建议Scheduler Agent管理以上 Agent 的生命周期和运行节奏定时器、交易日历启动/停止事件这套分工我试过很多版本一开始只有三四个 Agent后来发现日志和数据整理如果不单独抽出来复盘模块根本没法写。Record Agent 单独化之后整个系统的可观测性提升了一个量级。2.2 通信机制与事件流为什么不用一般函数调用Agent 之间如果用函数直接调用写起来很快但架构会迅速腐化。比如 Strategy Agent 调用了 Execution Agent 的下单函数那风控逻辑放在哪里如果风控也想调用下单那执行这边不就乱了QuantBot 用的是事件总线模式Agent 之间不直接依赖对方而是往总线上发布事件、订阅自己关心的事件。流程大概是盘前策略跑完Strategy Agent 发一个StrategySignal事件。Risk Agent 订阅这个事件校验后发一个RiskApprovedSignal或RiskRejectedSignal。Execution Agent 只订阅RiskApprovedSignal然后执行下单完成后发一个OrderFilledEvent。Record Agent 订阅所有事件全部归档。这样每个 Agent 都只需要知道自己和事件总线之间的关系不需要知道其他 Agent 是谁、什么时候上线、什么时候崩溃。哪个环节挂了重启那个 Agent 就行不牵连别的模块。我最初想用 Redis Stream 做事件总线后来为了部署简单直接用 Python 的asyncio.Queue加一个中心调度器实测是够用的。如果未来要横向扩展再考虑换消息中间件。技术选型不求最新但求整个链路最简单可靠。2.3 技术栈选型回测、实盘和数据层各用什么我在 QuantBot 里的技术栈分三层逻辑清晰。数据层使用pandasclickhouse本地开发时用 SQLite 也能代替负责存储行情、特征和日志。策略层回测我优先用qlib它自带因子和模型训练的工具链处理未来函数、数据对齐之类的细节比我自己造轮子稳得多实盘信号计算则用numpypandas手写避免引入重框架拖慢盘中速度。执行层基于自己封装的一个下单模块底层直连券商的 CTP 或 HTTP API。实盘初期不要直接上全自动最好的方式是先跑“信号提示 人工确认”稳定后再逐渐放权。对了如果你是刚开始搭不建议一上来就写自己的回测框架。qlib、backtrader 这些现成工具已经足够优秀先用熟再考虑要不要替换。3. 盘前模块把数据、模型和风险全部预制好3.1 数据准备链路从原始行情到特征矩阵盘前第一件事是把今天要用的数据全部准备好。这一步做得越细盘中就越轻松。我每天开盘前大概跑四条数据管线日线数据更新做前复权处理剔除停牌、上市未满 60 天的次新股。分钟级数据回补保证策略回测和实盘使用同一套数据源。基本面快照更新用于基本面过滤主要针对财务暴雷股。舆情数据压缩把新闻、公告、社交媒体上的热词浓缩成情绪分作为辅助因子。重点说下复权。很多新手在这个地方踩坑以为股票价格直接用就行。但除权除息之后价格是断档的指标计算会出问题。我习惯用前复权这样历史价格会被统一调整到当前视角技术指标比较稳定。回测和实盘必须保持一致否则盘后复盘的时候你会被数据不一致折磨死。代码上核心就是这个流程def build_daily_feature_set(trade_date, universe): daily load_daily_data(universe, start_date300D, end_datetrade_date) # 前复权 daily adjust_price(daily, methodpre) # 过滤停牌、次新股 daily daily[(daily.volume 0) (daily.trade_days 60)] # 生成特征动量、波动率、量价背离、分钟级尾盘强度 features pd.DataFrame(indexdaily.index) features[momentum_20] daily[close].pct_change(20) features[volatility_10] daily[close].pct_change().rolling(10).std() features[volume_ratio] daily[volume] / daily[volume].rolling(20).mean() # 与存量数据拼接按股票代码和日期去重 upsert_features(features) return features这一段看起来简单但真正的坑在adjust_price的实现上。如果使用 qlib 或 vn.py 的复权接口要确认它返回的是否包含所有历史区间还是只返回近期。我遇到过接口只缓存最近 100 天导致 60 日动量特征被截断的问题那会儿排查了整整一天。3.2 模型预热与信号预跑AI 在开盘前先给出候选池盘中实时跑模型不是不行但计算量一大延迟就上去了还容易跟交易模块抢 CPU。更好的做法是盘前把所有模型跑一遍生成一个候选池盘中只对候选池里的标的做增量更新。QuantBot 的盘前信号生成分成三步从特征库取出全市场特征矩阵。加载已经训练好的模型对全市场排序打分。取 Top K 作为当天的候选池交给风险模块过滤。这里说的“模型”不是特指深度学习。我实盘中表现最稳的反而是 LightGBM 和逻辑回归这类可解释模型。深度学习模型做超参数搜索太贵实盘行为也难解释在量化里我更多用它做辅助因子而不是最终决策。代码如下# 盘前预跑生成候选池 def generate_candidates(features, model, top_k20): preds model.predict_proba(features)[:, 1] ranked features.assign(predpreds).sort_values(pred, ascendingFalse) candidates ranked.head(top_k) return candidates.index.tolist() # 加载风控清单 blacklist load_blacklist() trade_plan [c for c in candidates if c not in blacklist]这一步跑完系统已经知道今天大概关注哪些股票、大概什么价格区间可以交易。盘中的任务就变成「验证候选池里哪些信号真正满足技术条件」而不是从头开始选股。3.3 盘前风控清单别等亏了才想起来设防线盘前另一个重要产出是当天的风险边界。我见过不少人把风控逻辑放在盘中去写结果盘中某个信号一激动人脑一热就绕过去了。QuantBot 的风控原则是任何盘中动态调整都不能突破盘前画好的底限。盘前要做的事包括计算当前总资产、可用资金、持仓市值。设定当日最大亏损阈值通常是总资产的 1% 到 2%。设定单一标的最高仓位比例我一般定为总资产的 10% 到 15%。更新黑名单包含财务问题股、当日公告停牌股、已经被标记为风险极高的标的。设定当日最多交易次数防止频繁交易产生的手续费侵蚀收益。这些参数会在盘前明文写入配置文件Risk Agent 启动时加载盘中任何 Agent 都无权直接修改。4. 盘中模块实时信号、交易执行与动态风控4.1 调度循环事件驱动而不是定时轮询盘中系统最核心的部分是事件循环。我不建议用多线程共享变量去处理行情那太容易出线程安全问题。QuantBot 的做法是完全异步的事件驱动模型每个 Agent 都在自己的协程里运行收到事件才响应没有事件就挂起。核心循环大概是这样的async def trading_loop(): scheduler get_scheduler() await scheduler.start() while market_is_open(): event await event_bus.wait_next() if isinstance(event, TickEvent): await data_agent.on_tick(event) elif isinstance(event, StrategySignal): await risk_agent.check(event) elif isinstance(event, RiskApprovedSignal): await execution_agent.execute(event) elif isinstance(event, OrderFilledEvent): await record_agent.archive(event)这个模式的优点是任何一个 Agent 处理出错只要异常被捕获调度循环还能继续跑。代价是代码里到处是异步细节需要一点点训练才能习惯。不过一旦跑顺你会特别爽——加一个新 Agent 只是注册一个新订阅不影响现有逻辑。4.2 交易执行封装限价、重试和滑点控制执行模块是直接跟钱打交道的也是我最谨慎的部分。QuantBot 的执行 Agent 只接受限价单不接受市价单。原因很简单市价单在行情快速变动的时候成交价格可能比预期差很多这种滑点在回测里根本体现不出来。我的执行策略是信号给出后先用当前盘口的价格加一个微小的容忍偏移比如买单价不高于最新价的 0.1%挂限价单。如果 30 秒内没有成交再决定是否撤单重挂。这里有个关键参数——重试次数超过三次就放弃这笔交易回到事件循环里继续等下一个信号。处理撤单和成交回报的时候有一个特别容易踩的坑重复下单。本地进程和券商服务器之间的连接可能会出现“请求已发出但响应没回来”的情况如果客户端因此重试就会在券商那边产生重复订单。我的解决方案是给每个订单生成一个唯一client_order_id券商 API 支持幂等创建的时候用这个 ID 做去重不支持的话就在本地维护一个“待确认订单队列”只有确认某笔订单确实被拒绝后才重下。这块我贴一段下单逻辑的骨架async def execute(self, approved_signal): order_id generate_order_id(signal_idapproved_signal.signal_id) status await broker.submit_order( symbolapproved_signal.symbol, sideapproved_signal.side, priceapproved_signal.ref_price * (1 0.001), quantityapproved_signal.quantity, client_order_idorder_id, ) # 如果超时或状态未知查询后置位不盲目重发 if status UNKNOWN: actual await broker.query_order(order_id) if actual is None: await self.retry_with_backoff(order_id)4.3 动态风控如果 AI 出错谁来踩刹车机器学习模型永远会有误判所以盘中风控不能只是一个“事前过滤”闸门它还必须具备实时熔断能力。QuantBot 的 Risk Agent 在盘中维护了三层防线每分钟检查一次账户总亏损。如果达到盘前设定阈值系统进入只许平仓不许开仓的紧急模式。每笔订单在发出前做单笔限额校验。比如某只票已经买了 8% 仓位又来一个加仓信号Risk Agent 直接拦截。监控策略信号的置信度。如果模型输出的置信度低于预设的最小阈值即使排名再高也不执行。这个模块的代码很简单但设计上有一点值得强调风控 Agent 和策略 Agent 是平级的不是上下级。策略 Agent 有义务提供信号但风控 Agent 有绝对权力拒绝。这种平级关系保证了风控判断不会被“策略当前需要建仓”这种上下文影响。5. 盘后模块复盘、归因与策略自适应更新5.1 数据归档没有完整日志复盘就是空谈收盘之后第一件事是把当天产生的所有数据归档。这里说的数据不止成交记录还包括策略 Agent 当时看到了什么行情、给了什么信号、因为什么条件放弃、Risk Agent 是基于哪条规则放行的、执行 Agent 实际成交价格和信号价格的差异是多少。QuantBot 的 Record Agent 把每个事件都抽象成统一的 JSON 格式同时落一份到本地文件、一份到数据库。这样复盘的时候既能做 SQL 统计分析也能直接打开日志看某个时间点发生了什么。这里分享一个很有用的细节记录行情快照时不光记录当时的价格还要记录该时刻的一些衍生指标比如均线位置、成交量放大比例。因为事后回看的时候如果日志里只有价格你根本不知道策略当时是基于什么形态做判断把特征快照一起存档复盘效率会高很多。5.2 绩效归因把盈亏拆到“策略 vs 运气”层面很多个人量化者盘后只看一个指标今天赚了还是亏了。这个观察维度太粗糙了。QuantBot 的 Review Agent 会做三件事收益拆解算出今天的总收益中有多少来自选股 alpha、多少来自大盘 beta、多少来自交易成本损耗。信号真实性校验回看当天实际成交的信号有多少是模型有效捕捉的有多少是执行阶段因为滑点或少成而明显偏离预期的。决策一致性检查策略发出的信号最终被执行的比例有多高。如果大量信号被风控拦截可能是策略太激进如果信号稀少可能是候选池太小。这些分析不需要太复杂的数学模型用 DataFrame 就能跑出来。但做好之后你会对自己策略的“真实水平”有非常清晰的认知而不是被一两天的大涨大跌冲昏头脑。5.3 轻量 LLM 报告助手从数据到结论的最后一公里量化分析的短板往往是数据有了结论没有。所以我给 QuantBot 加了一个可选的 LLM 复盘 Agent它会把 Record Agent 归档的数据摘要整理成提示词让模型生成一份「今日交易总结 明日注意事项」。注意这条链路的设计原则是LLM 只负责“解释”不负责“决策”。它读取的是统计结果和日志摘要输出的是自然语言报告实际参数调整永远由规则代码完成不允许模型直接改交易参数。这样既利用了 LLM 的文本生成能力又避免了幻觉对交易系统的污染。实际使用中我最常用的一句话提示词类似这样请基于以下信息生成一份中文交易复盘报告今日总收益 X%策略 alpha Y%风控拦截次数 N 最大单笔亏损 M明日关注的黑名单变化列表...模型生成的报告会放到一个专门的复盘文件里。长期积累下来这就是一笔非常有价值的经验资产。5.4 Agent 的状态与记忆系统不是每次开机都失忆我早期版本有个很蠢的问题每次重启系统所有 Agent 的状态都会清零。比如 Risk Agent 昨天发现某只股票有异常波动做了标记今天重启就忘了。后来我引入了一个轻量的“记忆存储”每个 Agent 可以把自己的关键状态写到Redis 或 SQLiteStrategy Agent 记录当前的模型版本、候选池 history。Risk Agent 记录黑名单、当天已触发的告警次数。Data Agent 记录已经同步到哪一天、哪些数据源健康。这样系统重启后每个 Agent 会自动从持久化的状态里恢复。尤其是连续多日的封闭运行时这个记忆机制让系统行为保持连贯不会出现“昨天明明规避了某只股票今天又买进去”这种奇怪的失误。6. 落地过程中的典型问题与排查技巧6.1 未来函数与数据泄露回测惊艳实盘崩盘的第一元凶量化人都知道“未来函数”四个字但真正能严格避免的人很少。我踩过最大的坑是用当天的收盘数据计算信号却在盘中提前使用了这个信号。回测里看起来胜率超高实盘里完全不可复现。解决办法分两层。第一层是数据对齐纪律回测和实盘必须用同一套数据生成逻辑盘前跑特征矩阵时严格只使用截至昨天收盘的数据盘中使用 tick 数据时任何计算都不能融入未来的 bar。第二层是信号延迟检验写一个脚本把信号生成时间点往后移 1 分钟、5 分钟观察收益是否还显著如果收益垮掉说明信号其实依赖了当时来不及交易的信息这就是未来函数泄露。6.2 接口断线与重连不能让网络抖动毁掉交易券商接口偶尔断连是常态。我遇到最危险的一次是盘中接口断连 10 秒重连成功后本地没有任何状态恢复机制导致一笔本该撤掉的订单一直挂着最后成交了一个不利价格。从那以后我在执行 Agent 里加了三个保护每次下单后启动一个 watchdog 定时器如果订单状态超过阈值没有回报主动发起查询。网络断开期间所有新信号丢进一个“pending 队列”重连后不自动执行而是重新过一遍 Risk Agent 校验。断连超过 2 分钟系统自动进入降级模式只保留撤单和卖出能力禁止开新仓。6.3 滑点与冲击成本回测是零成本实盘不是回测系统默认你按收盘价成交那是极度理想化的情况。实盘里滑点主要来自两方面盘口深度不足和下单速度不够快。我常用的控制手段有三种限价单优先买入挂买一价附近卖出挂卖一价附近。把大单拆成多个小单每隔几秒钟分批下单避免一笔打穿盘口。对流动性差的标的做“禁买名单”日均成交额低于某个阈值的股票直接过滤。这类细节会在盘后绩效归因里直接体现出来。如果你发现实际收益和模型预期收益差距很大优先检查滑点而不是怀疑模型本身。6.4 多 Agent 协作中的常见坑死锁、前置条件和重复消费Agent 多了之后协作问题开始变得明显。我整理了一个排查速查表遇到类似问题直接对照现象常见原因解决办法信号事件一直没人处理订阅事件名拼错了或者订阅在启动前未注册检查 Agent 的 startup 函数与事件总线订阅关系风控和策略互相等对方A 等待 B 的事件B 又在等待 A给每个事件处理加超时超时后主动放弃本次协作日志中出现重复订单事件被多个执行 Agent 实例消费确认执行 Agent 是单消费者模式或者用 client_order_id 幂等盘后复盘时缺数据Record Agent 漏订阅了某个事件检查 Record Agent 的订阅列表建议订阅所有类型模型特征和实盘不一致回测和实盘用了不同数据版本确保特征生成代码只保留一套不要维护两套实现6.5 给新手的建议先模拟盘跑一个月再谈实盘最后说一个最实在的建议不要一上来就实盘满仓跑。QuantBot 我最早是在模拟盘上跑了整整一个月覆盖了各种行情场景才敢用最小仓位上实盘。上实盘的顺序可以这样排第一周只发信号人工确认下单。第二周开始自动下单但只用 10% 仓位。第三周加上风控熔断验证紧急模式有效。一个月后根据绩效归因结果决定是否提升仓位。这套流程看起来很慢但它能保证你在系统真正赚钱之前先把系统的“活下来”能力磨扎实。我个人在使用 QuantBot 这小半年的体会是多 AI-Agent 协作的价值不在于它听起来多前沿而在于它强制你把交易系统当成一套需要长期维护的工程来对待。每个 Agent 的职责边界清晰出问题时能快速定位迭代时能只改一个模块而不影响全局。如果你现在也处于“脚本一把梭”的阶段不妨从今天开始把数据、策略、风控、执行四个模块先拆开哪怕不用成熟框架也能感受到架构对系统稳定性的巨大帮助。最后再分享一个小技巧给每个 Agent 的决策都加一个“回放快照”。我后来在复盘时遇到的绝大多数“这个信号为什么会出现”的问题都是靠翻快照解决的。这个习惯一旦养成你会发现自己对策略的理解会比只看净值曲线的人深刻得多。