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 的分数，最后加权汇总。

常用的指标和我的权重设置：

均线系统的评分逻辑：如果 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/9，RSI 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 apscheduler

• pandas和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 启动流程与定时任务

系统启动后，主循环按以下节奏运行：

1. 开盘前（9:00）：数据采集 Agent 拉取昨日收盘数据和最新新闻，技术面、基本面、情绪分析 Agent 各自计算评分。
2. 开盘后（9:30-15:00）：每 5 分钟触发一次决策流程，决策 Agent 汇总各方评分，生成交易信号，风控 Agent 审核，执行 Agent 模拟下单。
3. 收盘后（15:30）：风控 Agent 生成当日报告，数据采集 Agent 更新缓存。

def main(): bus = MessageBus() account = Account(initial_capital=1000000) 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", hour=9, minute=0) scheduler.add_job(intraday_task, "interval", minutes=5, start_date="09:30", end_date="15:00") scheduler.add_job(post_market_task, "cron", hour=15, minute=30) 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(f"Agent {agent.name} 处理消息失败: {e}", exc_info=True)

原因三：重复订阅。如果 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 常见问题速查表

独家避坑技巧：每次修改参数后，先用最近一年的历史数据跑一遍回测，对比修改前后的收益、回撤、交易次数。如果交易次数暴增但收益没提升，说明参数改坏了。这个习惯帮我避免了很多次“拍脑袋调参”带来的损失。

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 allocations

5.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，只要能跑通“拉数据→出信号→模拟下单”这个流程，后面加什么都是在这个骨架上扩展。反过来，如果一开始就设计一个完美架构，很可能写到一半就卡住了，因为你想解决的问题太多，反而一个都解决不好。