1. 从零搭建多 Agent 股票分析系统的整体设计思路1.1 为什么选择多 Agent 架构而不是单体脚本最开始动手的时候我其实想得很简单拉数据、算指标、给个买卖信号一个 Python 脚本几百行就搞定了。但真正跑起来才发现问题——一个脚本里塞满了数据获取、技术面计算、情绪判断、仓位管理、下单执行改一处逻辑就牵一发动全身调试的时候根本分不清是哪个环节出的错。后来我换了个思路把整个分析流程拆成几个独立的角色每个角色只干一件事角色之间通过标准化的消息传递来协作。这就是多 Agent 架构的核心逻辑。打个比方单体脚本就像一个全能选手什么都会但什么都不精多 Agent 就像一个小团队有人负责看财报有人负责盯盘面有人负责管风险最后有个“队长”来拍板。具体拆成了这么几个角色数据采集 Agent负责拉取行情数据、财务数据、新闻资讯做初步清洗技术面分析 Agent基于 K 线、均线、MACD、RSI 等指标给出技术面评分基本面分析 Agent处理财报数据计算估值指标给出基本面评分情绪分析 Agent抓取新闻和公告做文本情绪打分风控 Agent独立于其他 Agent专门盯着仓位、回撤、集中度决策 Agent汇总各方意见输出最终交易指令执行 Agent负责模拟下单、记录持仓、更新账户状态这么拆的好处非常明显每个 Agent 可以独立测试、独立替换。比如你觉得情绪分析用关键词匹配太粗糙想换成大模型来做只需要替换那一个 Agent 的实现其他部分完全不受影响。1.2 消息总线与状态管理的关键取舍Agent 之间怎么通信这是架构设计里最容易踩坑的地方。我试过三种方案第一种是直接函数调用A 调 B 的方法。简单是简单但耦合太死而且没法做异步和并行。第二种是用消息队列每个 Agent 订阅自己关心的消息。解耦是解耦了但引入了一个额外的中间件本地跑的时候还得起个服务太重了。第三种是我最终采用的方案内存消息总线 共享状态字典。所有 Agent 注册到总线上通过publish和subscribe来收发消息同时有一个全局的context字典存放共享数据比如当前持仓、账户余额、历史信号。这个方案的好处是零依赖、启动快、调试方便直接 print 就能看到消息流转。class MessageBus: def __init__(self): self.subscribers {} self.history [] def subscribe(self, topic, agent): self.subscribers.setdefault(topic, []).append(agent) def publish(self, topic, payload): msg {topic: topic, payload: payload, ts: time.time()} self.history.append(msg) for agent in self.subscribers.get(topic, []): agent.on_message(msg)状态管理上有个原则我踩过坑之后才明白风控相关的状态必须独立存储不能被其他 Agent 修改。一开始我把持仓信息放在共享 context 里结果执行 Agent 更新持仓的时候风控 Agent 读到的是中间状态导致风控判断出错。后来我把持仓和账户状态单独抽出来只允许执行 Agent 写入其他 Agent 只读问题就解决了。1.3 模拟交易引擎的设计要点既然是模拟交易就得尽量贴近真实交易环境否则回测出来的结果没有参考价值。我在模拟引擎里重点处理了这几件事滑点模拟真实交易中你的成交价和预期价总有偏差。我用了固定比例滑点加随机扰动的方案买入时成交价 信号价 × (1 滑点率)卖出时成交价 信号价 × (1 - 滑点率)滑点率默认设 0.1%可以根据标的流动性调整。手续费计算按成交金额的万分之三收取最低 5 元卖出时额外收千分之一的印花税。这些参数都做成可配置的方便不同市场切换。成交延迟信号发出后不是立即成交而是延迟一个 tick 或固定秒数模拟真实的下单到成交的时间差。部分成交当委托量超过当前模拟盘口深度时只成交一部分剩余部分挂单等待或撤单。这些细节看起来不起眼但实测下来对策略收益的影响能到 5% 到 10%绝对不能忽略。2. 核心 Agent 的实现细节与实操要点2.1 数据采集 Agent稳定获取数据的三个关键数据采集看起来最简单实际上是最容易出问题的环节。我总结了三个关键点第一多数据源冗余。不要只依赖一个数据接口主接口挂了要有备用接口。我的做法是封装一个DataFetcher类内部维护一个数据源列表按优先级依次尝试任何一个成功就返回。class DataFetcher: def __init__(self, sources): self.sources sources # 按优先级排序的数据源列表 def get_kline(self, symbol, start, end): for source in self.sources: try: data source.fetch_kline(symbol, start, end) if data is not None and len(data) 0: return self._validate(data) except Exception as e: log.warning(f数据源 {source.name} 失败: {e}) continue raise DataUnavailableError(symbol)第二数据校验不能省。拿到数据后必须做基本校验价格是否为正、最高价是否大于等于最低价、成交量是否非负、时间序列是否连续。我遇到过一次数据源返回了价格为 0 的记录导致后续所有指标计算全部异常排查了半天才发现是数据源的问题。第三缓存策略要合理。历史 K 线数据不会变可以永久缓存当日数据需要定时刷新。我用的是本地 SQLite 做缓存按symbol date建索引查询速度很快也方便离线回测。实操心得数据采集 Agent 一定要加详细的日志记录每次请求的数据源、耗时、返回条数。出问题的时候日志就是你的救命稻草。2.2 技术面分析 Agent指标计算与信号生成技术面分析 Agent 的核心任务是把原始 K 线数据转化成可比较的评分。我的做法是计算多个指标每个指标给出 -1 到 1 的分数最后加权汇总。常用的指标和我的权重设置指标权重说明均线系统0.255日、20日、60日均线的排列关系MACD0.20金叉死叉、柱状线变化RSI0.15超买超卖判断布林带0.15价格相对位置成交量0.15量价配合情况波动率0.10ATR 相对历史水平均线系统的评分逻辑如果 5日 20日 60日给 1如果 5日 20日 60日给 -1其他情况按排列程度给中间值。这个逻辑简单但有效能过滤掉大部分震荡行情。MACD 的评分我做了改进不只看金叉死叉还看柱状线的斜率和持续时间。金叉后柱状线持续放大给高分金叉后柱状线很快缩小给低分甚至负分。这样能避免被假金叉骗进去。def score_macd(macd_line, signal_line, histogram): score 0 # 金叉死叉基础分 if macd_line signal_line: score 0.5 else: score - 0.5 # 柱状线趋势加分 if len(histogram) 3: if histogram[-1] histogram[-2] histogram[-3]: score 0.3 elif histogram[-1] histogram[-2] histogram[-3]: score - 0.3 # 零轴位置加分 if macd_line 0: score 0.2 else: score - 0.2 return max(-1, min(1, score))注意事项技术指标参数不要过度优化。我试过用网格搜索找最优参数结果在回测集上表现很好一到实盘就拉胯。后来固定用经典参数MACD 12/26/9RSI 14布林带 20/2反而更稳定。2.3 情绪分析 Agent从新闻到情绪分数情绪分析这块我走过不少弯路。最开始用关键词匹配简单粗暴出现“利好”“增长”“突破”就加分出现“亏损”“下滑”“风险”就减分。效果一般因为很多新闻是中性表述而且反讽、否定句处理不了。后来换成用预训练的情感分析模型效果好了一些但有两个问题一是模型对金融领域术语不敏感二是推理速度慢几十条新闻要跑好几秒。最终的方案是关键词 规则 轻量模型三层过滤第一层关键词快速筛选把明显无关的新闻过滤掉比如体育、娱乐类。第二层规则处理识别否定词、程度副词调整情绪强度。“大幅增长”比“增长”分数高“不及预期”要反转。第三层对剩下的新闻用轻量模型打分取平均作为最终情绪分。def analyze_sentiment(news_list): scores [] for news in news_list: if not is_finance_related(news.title): continue base_score keyword_score(news.title news.content) base_score apply_negation_rules(base_score, news.content) if abs(base_score) 0.3: # 模糊的用模型再判 base_score model_score(news.content) scores.append(base_score) if not scores: return 0.0 return sum(scores) / len(scores)情绪分的更新频率我设的是每小时一次因为新闻是持续产生的但太频繁也没必要毕竟情绪对价格的影响是渐进的。2.4 风控 Agent独立才是硬道理风控 Agent 是整个系统里我最看重的部分。原因很简单分析和决策可以错但风控不能失效。一旦风控被绕过一次极端行情就可能把之前的利润全部吐回去。风控 Agent 独立运行不参与信号生成只做三件事事前风控在决策 Agent 输出交易指令后、执行 Agent 下单前检查这笔交易是否合规。检查项包括单笔交易金额是否超过账户总资产的 20%、单只股票持仓是否超过总资产的 30%、当前总仓位是否超过 80%、当日亏损是否超过 5%。事中风控持仓期间持续监控如果单只股票亏损超过 8%触发止损信号如果盈利超过 20%触发止盈信号如果大盘指数单日跌幅超过 3%触发减仓信号。事后风控每日收盘后统计当日交易计算最大回撤、夏普比率、胜率等指标生成风控报告。class RiskAgent: def __init__(self, config): self.max_single_position config.get(max_single_position, 0.30) self.max_total_position config.get(max_total_position, 0.80) self.stop_loss config.get(stop_loss, -0.08) self.take_profit config.get(take_profit, 0.20) def pre_trade_check(self, order, account): if order.amount account.total_value * 0.20: return False, 单笔交易金额超限 new_position account.get_position_value(order.symbol) order.amount if new_position account.total_value * self.max_single_position: return False, 单只股票持仓超限 if account.total_position_value order.amount account.total_value * self.max_total_position: return False, 总仓位超限 return True, 通过实操心得风控参数不要设得太紧否则会频繁触发导致交易成本上升也不要太松否则形同虚设。我的经验是先用历史数据回测找到能过滤掉大部分极端亏损又不影响正常交易的参数区间然后取中间值。3. 完整实操流程从启动到交易的全链路3.1 环境准备与依赖安装整个系统基于 Python 3.10 开发核心依赖不多尽量用标准库和轻量级第三方库。pip install pandas numpy requests apschedulerpandas和numpy用于数据处理和指标计算requests用于数据采集apscheduler用于定时任务调度不需要额外的消息队列、数据库或 Web 框架本地跑一个进程就够了。如果你想做可视化可以加matplotlib或plotly但这不是必须的。项目目录结构stock_agent/ ├── main.py # 入口 ├── bus.py # 消息总线 ├── agents/ │ ├── data_agent.py │ ├── tech_agent.py │ ├── fund_agent.py │ ├── sentiment_agent.py │ ├── risk_agent.py │ ├── decision_agent.py │ └── exec_agent.py ├── engine/ │ ├── simulator.py # 模拟交易引擎 │ └── account.py # 账户与持仓 ├── config.yaml # 配置文件 └── data/ # 缓存数据3.2 配置文件与参数设置所有可调参数集中在config.yaml里方便修改和版本管理。data: symbols: [000001, 600036, 300750] kline_period: daily cache_dir: ./data tech: ma_periods: [5, 20, 60] macd: [12, 26, 9] rsi_period: 14 boll_period: 20 sentiment: news_count: 20 update_interval: 3600 risk: max_single_position: 0.30 max_total_position: 0.80 stop_loss: -0.08 take_profit: 0.20 max_daily_loss: -0.05 trade: initial_capital: 1000000 commission_rate: 0.0003 min_commission: 5 stamp_tax: 0.001 slippage: 0.001参数设置有个原则先保守后激进。初始资金设 100 万单只股票最多 30%总仓位最多 80%止损 8%止盈 20%。这些参数不是拍脑袋定的是回测了多个参数组合后选出来的相对稳健的值。3.3 启动流程与定时任务系统启动后主循环按以下节奏运行开盘前9:00数据采集 Agent 拉取昨日收盘数据和最新新闻技术面、基本面、情绪分析 Agent 各自计算评分。开盘后9:30-15:00每 5 分钟触发一次决策流程决策 Agent 汇总各方评分生成交易信号风控 Agent 审核执行 Agent 模拟下单。收盘后15:30风控 Agent 生成当日报告数据采集 Agent 更新缓存。def main(): bus MessageBus() account Account(initial_capital1000000) simulator Simulator(account, config) data_agent DataAgent(bus, config) tech_agent TechAgent(bus, config) fund_agent FundAgent(bus, config) sentiment_agent SentimentAgent(bus, config) risk_agent RiskAgent(bus, config) decision_agent DecisionAgent(bus, config) exec_agent ExecAgent(bus, config, simulator) scheduler BlockingScheduler() scheduler.add_job(pre_market_task, cron, hour9, minute0) scheduler.add_job(intraday_task, interval, minutes5, start_date09:30, end_date15:00) scheduler.add_job(post_market_task, cron, hour15, minute30) scheduler.start()3.4 一次完整的交易决策过程拿一个实际场景来走一遍。假设当前持仓有某银行股 20 万账户总资产 100 万现金 50 万。第一步数据采集。数据 Agent 拉取了该股票最近 60 个交易日的 K 线以及最近 20 条相关新闻。第二步技术面分析。5 日线上穿 20 日线MACD 金叉且柱状线连续三天放大RSI 58 处于中性偏强布林带价格在中轨上方。技术面综合评分 0.65。第三步基本面分析。市盈率 6.5 倍低于行业平均净利润同比增长 8%股息率 4.2%。基本面评分 0.55。第四步情绪分析。最近新闻以中性偏正面为主有一条关于分红提升的公告。情绪评分 0.30。第五步决策。决策 Agent 按权重汇总技术面 0.4、基本面 0.35、情绪 0.25。综合评分 0.65×0.4 0.55×0.35 0.30×0.25 0.26 0.19 0.075 0.525。超过买入阈值 0.4生成买入信号建议买入金额为账户总资产的 5%即 5 万元。第六步风控审核。风控 Agent 检查单笔 5 万未超过 20 万上限该股票现有持仓 20 万加 5 万后 25 万未超过 30 万上限总仓位 50 万加 5 万后 55 万未超过 80 万上限。审核通过。第七步执行。执行 Agent 按当前价格加滑点模拟成交更新持仓和现金记录交易日志。整个过程从数据采集到成交完成大约 2 到 3 秒其中大部分时间花在数据请求上。如果数据有缓存1 秒内就能完成。4. 常见问题与排查技巧实录4.1 Agent 之间消息丢失或重复处理这是多 Agent 系统最常见的问题。表现是某个 Agent 没收到消息或者同一条消息被处理了两次。原因一订阅时机不对。如果 Agent 在消息发布之后才订阅自然收不到。解决办法是在系统启动时统一注册所有 Agent确保订阅关系在第一条消息发布前就建立好。原因二消息处理异常导致中断。某个 Agent 处理消息时抛异常如果没有捕获后续消息就不会被处理。我的做法是在消息总线的分发逻辑里加 try-except单个 Agent 出错不影响其他 Agent。def publish(self, topic, payload): msg {topic: topic, payload: payload, ts: time.time()} self.history.append(msg) for agent in self.subscribers.get(topic, []): try: agent.on_message(msg) except Exception as e: log.error(fAgent {agent.name} 处理消息失败: {e}, exc_infoTrue)原因三重复订阅。如果 Agent 被多次注册到同一个 topic同一条消息会被处理多次。解决办法是在subscribe里做去重检查。排查技巧消息总线维护一个history列表记录所有发布过的消息。出问题的时候直接查 history 就能看到消息有没有发出去、发了几次、被哪些 Agent 消费了。4.2 模拟交易结果与预期偏差大回测的时候收益很好模拟盘跑起来却差很多通常有这几个原因滑点设置不合理。如果滑点设得太小模拟成交价过于理想实际收益会被高估。我的经验是流动性好的大盘股滑点设 0.05% 到 0.1%小盘股设 0.2% 到 0.5%。手续费漏算。很多人只算佣金忘了印花税和过户费。卖出时印花税千分之一是硬成本频繁交易的话这部分能吃掉不少利润。成交延迟没模拟。信号发出后立即按当前价成交这在实盘中是不可能的。加一个 1 到 3 秒的延迟或者按下一个 tick 的价格成交结果会更真实。幸存者偏差。回测的股票池如果只选了现在还在的股票那些退市的、被并购的都没算进去收益自然偏高。解决办法是尽量用历史成分股数据或者至少意识到这个偏差的存在。4.3 风控频繁触发导致无法交易风控太严会导致两种情况要么一直不让买要么刚买就触发止损。止损设得太紧。8% 的止损对于波动大的股票来说可能正常回调就触发了。解决办法是根据股票的 ATR平均真实波幅动态设置止损比如止损位 入场价 - 2×ATR。仓位限制太死。如果单只股票上限 10%总仓位上限 30%那基本只能买两三只股票分散度不够。我的建议是单只 20% 到 30%总仓位 70% 到 80%留一定现金应对极端情况。大盘风控误伤。如果大盘跌 3% 就减仓那在震荡市里会频繁买卖。可以改成大盘连续两日累计跌幅超过 5% 才触发或者用个股相对大盘的超额跌幅来判断。4.4 常见问题速查表问题现象可能原因排查方法解决方案Agent 收不到消息订阅时机晚于发布查消息总线 history启动时统一注册同一消息处理多次重复订阅打印订阅列表subscribe 去重指标计算结果异常数据含空值或零值检查原始数据数据校验 填充模拟收益虚高滑点/手续费漏算对比实盘成本补全成本项风控频繁触发参数过紧统计触发频率放宽参数或动态调整系统运行变慢历史消息堆积查内存占用定期清理 history定时任务不执行时间配置错误查调度器日志检查 cron 表达式独家避坑技巧每次修改参数后先用最近一年的历史数据跑一遍回测对比修改前后的收益、回撤、交易次数。如果交易次数暴增但收益没提升说明参数改坏了。这个习惯帮我避免了很多次“拍脑袋调参”带来的损失。5. 系统扩展与进阶方向5.1 接入大模型增强分析与决策当前的分析 Agent 主要基于规则和传统指标如果你想进一步提升分析深度可以接入大语言模型来做几件事财报摘要与关键信息提取。把财报原文丢给模型让它提取营收、利润、毛利率、现金流等关键数据并总结管理层讨论中的重点。这比写正则表达式去匹配要灵活得多。新闻深度解读。不只是判断正负面而是让模型分析新闻对公司的具体影响路径。比如“某公司获得大额订单”模型可以进一步分析这个订单占公司年营收的比例、对利润的贡献、是否具有可持续性。决策解释生成。决策 Agent 输出交易信号后让模型生成一段人类可读的解释说明为什么买、为什么卖、主要依据是什么。这对复盘和信任建立很有帮助。接入方式建议用 API 调用不要本地部署模型除非你有很强的 GPU 资源。API 调用按 token 计费成本可控而且模型更新及时。5.2 多策略并行与组合管理单个策略很难适应所有市场环境。你可以让系统同时运行多个策略每个策略有自己的 Agent 组合和参数最后做一个组合管理。比如趋势策略均线 MACD适合单边行情均值回归策略布林带 RSI适合震荡行情事件驱动策略新闻情绪 公告适合突发事件组合管理 Agent 根据各策略的历史表现动态分配资金。表现好的策略多分钱表现差的少分钱。分配算法可以用简单的加权平均也可以用更复杂的凯利公式或风险平价。def allocate_capital(strategies, total_capital): # 按夏普比率加权分配 sharpes [max(0.1, s.sharpe_ratio) for s in strategies] total_sharpe sum(sharpes) allocations {} for s, sharpe in zip(strategies, sharpes): allocations[s.name] total_capital * sharpe / total_sharpe return allocations5.3 实盘对接的注意事项如果你想把模拟系统对接实盘有几个地方必须改订单状态管理。模拟盘下单立即成交实盘有“已报”“部成”“已成”“已撤”等多种状态需要轮询或订阅回报来处理。网络异常处理。实盘环境网络可能不稳定下单请求可能超时。必须做幂等设计避免重复下单。每笔订单带唯一 ID超时后先查询订单状态再决定是否重发。资金和持仓同步。实盘的资金和持仓以券商为准本地记录只是缓存。每次交易前先同步一次避免本地状态和实际不符。风控前置。实盘的风控必须在券商层面也设一道比如通过券商的 API 设置单笔限额、日限额。本地风控再完善也可能因为程序 bug 失效券商层面的硬限制是最后一道防线。重要提示实盘对接前务必用模拟盘跑至少一个月覆盖各种市场行情。我见过太多人模拟盘跑了一周觉得没问题就上实盘结果遇到极端行情直接爆仓。模拟盘和实盘的差距比你想象的大得多。5.4 性能优化与监控当股票池扩大到几十只、Agent 数量增加到十几个的时候系统性能会成为瓶颈。几个优化方向数据缓存分层。热数据当日行情放内存温数据近 60 日 K 线放本地文件冷数据历史财报放数据库。查询时按层级依次查找。指标计算向量化。用 pandas 的向量化操作替代循环速度能提升几十倍。比如计算均线用df[close].rolling(20).mean()而不是 for 循环。异步数据采集。多只股票的数据请求用asyncio或线程池并发执行总耗时从 O(n) 降到 O(1) 级别。监控告警。关键指标要实时监控Agent 响应时间、消息队列长度、账户回撤、当日盈亏。超过阈值就发告警。我用的是最简单的方案写日志 定时检查日志关键词发现异常就发邮件。这套系统我从最初的一个脚本慢慢迭代到现在七八个 Agent 协作的规模中间踩了无数的坑。最大的体会是不要追求一步到位先跑通最小闭环再逐步增加复杂度。最开始哪怕只有一个数据 Agent 和一个决策 Agent只要能跑通“拉数据→出信号→模拟下单”这个流程后面加什么都是在这个骨架上扩展。反过来如果一开始就设计一个完美架构很可能写到一半就卡住了因为你想解决的问题太多反而一个都解决不好。