企业官网建设流程全解析

做量化这几年，我一直被一件事折磨：策略思路不复杂，复杂的是每天围绕策略的杂活。导数据、算指标、看消息、盯持仓、复盘归因，每一样都耗时，每一样都容易出错。后来我干脆把这些工作抽象成一套多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.py

core目录放的是与业务无关的基础设施，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_update，execution_agent订阅signal_generated，report_agent订阅daily_close。这样整个数据流就是单向的，谁都不需要知道数据是从哪来的。

3.3 核心Agent的代码骨架示例

我拿signal_agent来展示一个Agent的基本结构。它内部会维护一个策略列表，每次收到行情事件时，遍历所有策略计算信号，然后把通过的信号发到消息总线。

class SignalAgent: def __init__(self, bus, strategies=None): 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, trigger="date", run_date=f"{date_info['date']} 09:00:00") scheduler.add_job(post_market_job, trigger="date", run_date=f"{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类问题时，我最常用的方法不是看完整日志，而是把消息总线上的事件流转过程单独打成一个trace文件，每个事件产生时间、消费时间、处理结果都记录下来。这样出问题时顺着trace文件扫一遍，基本上能很快定位到是哪个Agent拖慢了整个链路或者没有正确响应。

另外想提醒一点，Agent自动化跑起来之后，千万不要完全不管。我的做法是每晚收盘后快速扫一遍当天的告警日志，每周做一次策略触发率和成交率的统计。自动化只是把重复劳动压缩了，真正对系统健康度的掌控还是得靠人的判断力。QuantBot这种盘前、盘中、盘后闭环的架构，最大的价值不是让你“躺着赚钱”，而是让你把时间花在策略本身的分析上，而不是被无穷无尽的运维杂活淹没。

这是我在实际操作中体会最深的一点：闭环不是终点，而是让你有时间继续迭代策略的起点。