最近三周时间我搭了一个基于 DeepSeek 的股票趋势分析多 Agent 系统。不是简单拿大模型预测股价而是把基金经理那套投研工作流拆成六个角色每个角色由一个 DeepSeek 实例扮演分别负责数据清洗、指标计算、情绪打分、趋势研判、风险控制最后汇总成一份带数据支撑的分析报告。项目跑通的那一刻挺有成就感但更多的收获是在踩坑里总结出来的。这篇文章写给两类人。一类是想用大模型做真实业务的人尤其是对 Agent 多智能体架构感兴趣、却不知道从哪下手的开发者另一类是懂点 Python、想给自己的行情数据分析流程加一层“自动解读”的投资者。我不谈什么宏大概念直接讲清楚为什么这么做、怎么拆 Agent、提示词怎么设计、代码怎么落以及那些文档里不会写的坑。如果你指望这个系统能预测明天涨停那可以关掉了——趋势分析解决的是“现在处于什么状态、概率上偏向哪边”不是算命。1. 先想清楚为什么单模型做不了股票趋势分析1.1 大模型在金融场景的四个硬伤拿一个大模型直接问“这只股票未来趋势如何”表面上能收到一段流畅回答实际根本没法用。我把它总结成四个硬伤。第一没有实时数据入口。模型的训练语料有截止时间它对“今天这根 K 线”一无所知。你问它它只能基于历史记忆编一个“看起来合理”的回答而金融分析恰恰是时效性大于一切。第二上下文窗口是有限的。哪怕模型支持 128K 上下文把 300 天行情、公告、新闻全塞进去也不现实塞进去之后还会出现“中间遗忘”现象——模型对开头和结尾记得清楚中间的数据基本没被认真对待。第三输出不可控。你问趋势它给你写三百字小作文没有结构化字段下游程序没法直接消费。第四也是更要命的单模型没有自校验机制。它回答错了没有人反驳它、纠正它错误会一路贯彻到底。1.2 多 Agent 架构的本质是投研团队我决定用多 Agent 的起因特别朴素你见过哪个基金经理既亲自盯盘又自己算指标还要自己刷公告、做风控的不可能。现实里的投研是一套严密分工——研究员找数据、分析师算技术指标、宏观组看消息面、风控部门设红线最后基金经理拍板。多 Agent 就是把这套分工搬进代码里。每个 Agent 只干一件窄活提示词变得极其简单幻觉空间被压缩每个 Agent 的输入输出都是结构化 JSON数据来源可回溯任意一个 Agent 出了错上游数据不会连带崩盘替换和调试成本极低。实践里最大的感受是Agent 拆得越细单次调用越便宜结果越稳。你把一个复杂问题拆成六步每步只需要两三百 Token 的推理量模型犯错的概率远比让它一口气推理两千 Token 要低。1.3 为什么基座选 DeepSeek选 DeepSeek 做基座模型我有四个具体的理由。一是中文理解能力在同级别模型里靠前A股公告、新闻、研报这种满是中文缩略语的文本处理起来自然。二是推理成本足够低多 Agent 系统一天跑几十次分析Token 成本几乎可以忽略。三是它对 JSON 结构化输出的支持做得比较稳配合提示词约束能稳定产出可解析的结果。四是它既支持官方 API 也支持本地部署如果对数据隐私敏感可以完全离线跑这一点对金融场景很关键。架构上我没有绑定某个特定模型。所有 Agent 只跟“统一的 HTTP 接口”通信换底座模型只改一处配置。这套设计带来的好处在后面的实操部分会体现出来。2. 系统整体设计六个 Agent 如何分工协作2.1 Agent 角色表与职责边界整个系统我拆成了六个 Agent每个角色定位非常明确。先看总览表Agent 名称核心职责关键输入结构化输出数据采集 Agent拉取并清洗行情数据股票代码、日期范围干净的 K 线 DataFrame技术指标 Agent计算常用技术指标K 线数据MACD、RSI、KDJ、布林带当前值及变化情绪因子 Agent分析公开消息与公告倾向新闻/公告文本情绪标签与情绪分-100 到 100趋势研判 Agent综合技术面给出方向判断指标摘要、近期统计特征看多/震荡/看空 置信度风险控制 Agent排查异常波动与风险项波动率、回撤、估值分位风险等级低/中/高 触发明细汇总决策 Agent整合全部结果输出研报前五个 Agent 的结构化 JSON综合研报 最终倾向这里要强调一个原则每个 Agent 的职责必须互斥。技术指标 Agent 不能去评情绪情绪因子 Agent 也不能碰行情数据。边界一旦模糊Agent 之间的输出就会互相污染后面做仲裁的时候你会想骂人。2.2 协作流程先串行再并行最后汇聚整个执行链路是固定的。第一步必须是数据采集 Agent下游所有 Agent 都依赖它产出的干净 K 线。数据到位之后技术指标 Agent 和情绪因子 Agent 可以并行执行——前者吃 K 线后者吃公告和新闻文本两者没有依赖关系。再往后趋势研判 Agent 同时接收技术指标的输出和近期统计特征风险控制 Agent 也并行读数据做排查。最后汇总决策 Agent 把趋势研判、风险控制、情绪因子三份结果收拢按既定规则生成最终研报。用代码表达就是有依赖关系的串行 await没有依赖关系的用任务并发。这里不依赖任何重型编排框架Python 自带异步能力就够用。Agent 数量少于十个真没必要上 LangGraph 之类的工具自己维护一个简单调度器反而更可控。2.3 消息协议Agent 之间只认 JSONAgent 之间的通信协议是整个系统的骨架。我定义的统一消息格式是{ agent: trend_analysis, status: success, data: { direction: 震荡偏多, confidence: 0.63, reason: MACD金叉但RSI处于中性区布林带收口短期方向不明 }, meta: { model: deepseek-v3, latency_ms: 1800, created_at: 2025-01-15T14:30:00 } }优点在于下游 Agent 完全不关心上游是用哪个模型算的、花了多长时间只要data字段符合约定就能继续往下走。这给了你极大的替换自由——今天用 DeepSeek明天想换别的模型或者干脆把一个 Agent 改成规则引擎下游零感知。这个设计是整套系统里我认为最值钱的部分。3. 提示词工程和趋势打分多 Agent 系统的灵魂3.1 给每个 Agent 写“岗位说明书”模型本身没有角色感角色感完全靠提示词塑造。我给每个 Agent 都写了一段“岗位说明书”固定结构是这样的角色定位你是谁负责什么。例如“你是技术指标分析员只负责解读指标数据”。任务边界你该做什么、不该做什么。明确写上“不要预测未来价格不要给出买卖建议”。输入说明告诉模型你会收到什么格式的数据。输出要求强制 JSON 格式字段名、字段类型全部给定。拒答逻辑数据不足时怎么处理必须明确回复“数据不足”不能编造。拿趋势研判 Agent 的提示词举例你是趋势研判分析师。你只会收到技术指标Agent产出的结构化JSON数据包括MACD、RSI、KDJ、布林带的最新值及其近20日变化。 你的任务基于这些指标给出趋势方向判断看多/震荡/看空并给出0到1之间的置信度。 规则 1. 只输出JSON不要任何解释性文字。 2. 不要预测具体涨跌幅不要给出买卖建议。 3. 如果输入数据缺失或异常请在json中输出{direction:unknown,confidence:0}。 输出格式 {direction:看多或震荡或看空,confidence:0-1之间的小数,reason:不超过80字的判断理由}这里角标里写的 80 字其实是个实用技巧——限制理由长度能防止模型写小作文很多人会忽略提示词里的“篇幅预算”而它直接关系到 Token 成本和输出的稳定度。3.2 JSON 输出如何保持稳定模型输出不是天然合法 JSON这是所有大模型应用都会遇到的坑。三层防护下来我的 JSON 解析成功率从最初的八成提升到接近百分之百。第一层在请求参数里打开 JSON 模式。DeepSeek 的接口支持将返回格式固定为 JSON object请求体里带上response_format{type: json_object}并在提示词里强制声明“只输出 JSON”。这两者配合模型就基本不会跑偏。第二层在解析层做容错。即使设置了 JSON 模式偶发情况下也会在 JSON 前后多出几个字符。我的做法是用正则取出第一对花括号包裹的完整片段再交给 JSON 解析器。如果解析失败自动重试一次重试时把 temperature 降到 0通常都能过。第三层在后端做数据校验。解析出来的 JSON 必须通过 Pydantic 校验字段类型不对、缺失直接判定该次调用失败。为什么这么做因为一个字段缺失的 JSON 流进下游比解析失败更隐蔽危害更大。3.3 打分设计档位越细输出越不稳定我早期犯过一个典型错误让模型给趋势强度打 1 到 10 分。结果输出波动大得离谱同一组数据上午 8 分下午 3 分完全没有商业可用性。后来我意识到大模型的“连续数值输出一致性”本来就很差它对非整数、多档位的数字缺乏稳定的锚点。现在我的打分体系就三条趋势方向只给“看多/震荡/看空”三分类加置信度情绪分用一个连续值但给出明确锚点规则结合公告文本里的用词倾向正面词、负面词-上下限 ±100风险等级固定低/中/高三档。档位越粗模型越容易稳定表达而置信度字段才是真正体现模型“犹豫程度”的地方。3.4 上下文裁剪不要让 Agent 看全量数据这是整个项目里 Token 成本下降最明显的一步。早期我图省事把 120 天 K 线全塞给趋势研判 Agent让它自己看。结果不只是贵效果还差——模型根本没有能力从几百行数字里“看出”趋势它只是在做模式补全。后来我把策略改成“统计摘要 指标结论”。传给趋势研判 Agent 的只有最近 5 日开盘收盘价概览、近 20 日涨跌幅分布、均线位置关系、以及技术指标 Agent 算出的 MACD/RSI/KDJ 当前数值和变化方向。数据量从几百行压缩到几行摘要Token 消耗降到原来的六分之一判断质量反而更稳。原理很简单你给模型喂的是人处理过的信息而不是让它替你做特征工程。4. 从零搭建核心代码与调度流程实测4.1 工程结构整个项目没有用重型框架标准 Python 工程。目录结构如下stock_agent/ ├── agents/ │ ├── base.py # Agent 基类接口封装、重试、JSON解析 │ ├── data_fetcher.py # 数据采集 Agent │ ├── indicator.py # 技术指标 Agent │ ├── sentiment.py # 情绪因子 Agent │ ├── trend.py # 趋势研判 Agent │ ├── risk.py # 风险控制 Agent │ └── aggregator.py # 汇总决策 Agent ├── core/ │ ├── llm.py # DeepSeek 接口封装 │ └── schema.py # Pydantic 数据校验模型 ├── scripts/ │ └── run_pipeline.py # 调度入口 ├── output/ # 生成的 HTML 研报 └── requirements.txt这是我比较推荐的粒度。每个 Agent 独立文件核心调度脚本只干一件事按照依赖图把六个 Agent 串起来。你看着结构简单但维护起来非常舒服改一个 Agent 不牵动其他逻辑。4.2 行情数据获取与清洗数据源我用的是开源行情库拉取前复权日线数据。前复权这一步非常关键因为 A 股除权除息会导致 K 线跳空不复权的数据算出来的 MACD 全是假的。import pandas as pd def load_kline(symbol: str, start_date: str, end_date: str) - pd.DataFrame: # df ak.stock_zh_a_hist(symbolsymbol, perioddaily, # start_datestart_date, end_dateend_date, # adjustqfq) # qfq 前复权 kline pd.DataFrame({ date: pd.to_datetime(df[日期]), open: df[开盘], high: df[最高], low: df[最低], close: df[收盘], volume: df[成交量], }).sort_values(date).reset_index(dropTrue) # 剔除停牌产生的空行 kline kline.dropna(subset[open, close]) # 去除重复日期保留最后一个交易记录 kline kline[~kline[date].duplicated(keeplast)] return kline注意停牌日的处理。很多免费接口停牌日会返回全 0 或者空值直接 dropna 会把这些行丢掉但你要意识到丢掉之后均线计算用的是“仅交易日”这是正确的做法。另外如果后期要做分钟级分析停牌处理逻辑还得再复杂一些。4.3 技术指标计算指标计算我直接用 pandas 自己写没引额外的技术分析库。原因一是可控二是参数心里有数。这里给出 MACD 和 RSI 的核心实现def calc_macd(close: pd.Series, fast12, slow26, signal9): ema_fast close.ewm(spanfast, adjustFalse).mean() ema_slow close.ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() hist (dif - dea) * 2 return dif.iloc[-1], dea.iloc[-1], hist.iloc[-1] def calc_rsi(close: pd.Series, period14): delta close.diff() gain delta.clip(lower0).rolling(windowperiod).mean() loss (-delta.clip(upper0)).rolling(windowperiod).mean() rs gain / loss rsi 100 - (100 / (1 rs)) return rsi.iloc[-1]ewm(adjustFalse)是这里容易踩的细节。如果用默认的adjustTrue计算出的 EMA 序列前几个值会和自己算的 MACD 对不上因为调整式的加权方式不同。做技术指标统一用adjustFalse这也是各种行情软件里 MACD 的标准算法。4.4 DeepSeek 接口封装调用层我封装了一个统一函数。这里要说清楚我用的 HTTP 接口风格兼容业界通用格式所以换成其他模型只需要改地址和密钥。import requests import json import time API_URL https://api.deepseek.com/chat/completions API_KEY your-api-key def chat_json(system_prompt: str, user_content: str, temperature: float 0.2): payload { model: deepseek-chat, messages: [ {role: system, content: system_prompt}, {role: user, content: user_content} ], temperature: temperature, response_format: {type: json_object}, max_tokens: 1024, } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } for attempt in range(3): try: resp requests.post(API_URL, jsonpayload, headersheaders, timeout30) resp.raise_for_status() data resp.json() content data[choices][0][message][content] return parse_json_safe(content) except Exception as e: if attempt 2: raise e time.sleep(2 ** attempt)temperature 我固定取 0.2。为什么不是 0实测 0 虽然稳定但略显死板推理类任务 0.2 既有随机性又不至于跑飞。对分析报告这种场景稳定比新颖更重要0.2 是平衡点。4.5 多 Agent 调度核心调度层是所有 Agent 串起来的地方。核心要点是两个并行任务的控制以及并发数上限。import asyncio from agents.data_fetcher import DataFetcherAgent from agents.indicator import IndicatorAgent from agents.sentiment import SentimentAgent from agents.trend import TrendAgent from agents.risk import RiskAgent from agents.aggregator import AggregatorAgent semaphore asyncio.Semaphore(3) # 限制同时进行的 API 请求数量 async def run_analysis(symbol: str): data_agent DataFetcherAgent() kline await data_agent.run(symbol) indicator_agent IndicatorAgent() sentiment_agent SentimentAgent() indicator_task asyncio.create_task(indicator_agent.run(kline)) sentiment_task asyncio.create_task(sentiment_agent.run(symbol)) indicator_result, sentiment_result await asyncio.gather(indicator_task, sentiment_task) trend_agent TrendAgent() risk_agent RiskAgent() trend_task asyncio.create_task(trend_agent.run(indicator_result)) risk_task asyncio.create_task(risk_agent.run(kline, indicator_result)) trend_result, risk_result await asyncio.gather(trend_task, risk_task) aggregator AggregatorAgent() report await aggregator.run(trend_result, risk_result, sentiment_result) return report信号量在这里非常关键。没有它六个 Agent 同时发起请求接口很容易触发限流。实测下来并发数压到 3 时速度和稳定性平衡最好。另外注意asyncio.gather的使用位置——只有没有依赖关系的任务才能放进去并行有依赖的任务必须先拿到上游结果。4.6 输出生成投研日报最后一步是把六个 Agent 的结论渲染成一份人类可读的 HTML 报告。我用 matplotlib 画 K 线和 MACD 图然后把 Agent 结论以结构化卡片的形式放在图表下方再导出为一个自包含 HTML 文件。这样做的原因是实用性。直接看六个 Agent 的 JSON 输出你得来回对照太反人类。渲染成一张报告页K 线在左、指标结论在右、风险提示在最下方扫一眼就能掌握核心信息。报告里明确标注“仅供参考不构成投资建议”这既是职业操守也是必要的风险提示。5. 跑起来看看实测效果与单模型对比5.1 一次完整分析实例为了不构成变相荐股我这里不放具体股票代码用“某消费类蓝筹股”举例。整个流程跑完大约 15 秒具体输出如下。数据采集 Agent 成功拉了近 120 个交易日的日线无缺失。技术指标 Agent 给出MACD 刚形成金叉DIF 上穿 DEA柱状图由负转正RSI(14) 位于 52中性偏强KDJ 的 K 值上穿 D 值布林带收口价格沿中轨运行。情绪因子 Agent 扫描了最近一周公开公告识别到“回购”“分红”等偏正面词情绪分 32情绪标签“偏正面”。趋势研判 Agent 结合统计摘要给出的结论是“震荡偏多”置信度 0.63理由是 MACD 金叉但 RSI 未进入强势区布林带尚未开口方向需要进一步确认。风险控制 Agent 给出风险等级“中”提示项包括近 20 日最大回撤 4.2%、当前价格正处于一年估值分位 78% 的高位区间。汇总决策 Agent 综合四份输入最终输出“中期趋势震荡偏多短期存在回调压力估值偏高建议保持谨慎观察。”手动对照这组结论我认为它是合理的因为逻辑可以回溯。趋势偏多来自 MACD 和 KDJ估值压力来自风险控制 Agent 的分位提示情绪偏正面来自公告条款。每一步都有依据没有一句是模型凭空编出来的。5.2 与单模型直接提问的对比我也做了对照组同一个股票代码直接把近 120 日 K 线以文本格式粘贴给 DeepSeek问“分析一下趋势”。结果有参考价值但参考价值非常有限。从三个维度对比差距很明显对比维度单模型直接提问多 Agent 系统信息时效性模型看不到实时行情只能泛泛而谈数据 Agent 实时拉取分析基于当下 K 线结构化程度自由文本需要人脑二次提炼六个维度分层 JSON可直接入库存档可解释性回答“感觉要涨”无法追溯依据每个结论都能回溯到具体指标数据实践中我体会最深的是可解释性。单模型的回答是黑盒它说“技术面转好”你根本不知道它看了哪根 K 线。多 Agent 系统里趋势研判 Agent 说“MACD 金叉”这个结论来自技术指标 Agent 的计算结果而计算结果来自明确公式和数据。出了问题可以一路排查到底这在金融场景里是底线要求。5.3 必须泼的冷水这系统有用但必须说清楚它的边界。它能做的是“对已有数据的模式识别和概率判断”不是未来预测。突发事件、政策变化、财务造假这种黑天鹅模型捕捉不到。情绪因子 Agent 依赖公开文本公开文本本身就有滞后性和选择性。回测环节也容易过拟合。你说“这个信号在金叉之后上涨概率高”听起来很美但换个时间段可能完全相反。我的建议是趋势分析结果用于辅助复盘、辅助完善自己的交易系统可以实盘决策必须结合自身的风控规则任何模型输出都不能作为买卖依据。合规上也要严格遵守“不构成投资建议”的要求。6. 踩坑记录与常见问题排查6.1 并发请求触发限流与超时症状多个 Agent 同时发起请求接口频繁返回 429或者直接超时。排查后发现原因有两个。一是同时发起的请求数超过了接口配额二是网络层没有设置合理的超时时间默认 requests 库会一直等待。解决方案就是加信号量限制并发数到 3同时给每次请求设置 30 秒超时并做指数退避重试。重试间隔按 1 秒、2 秒、4 秒递增最多重试三轮。6.2 上下文窗口爆掉接口直接报 400这是我早期犯的最大错误。把 120 天 K 线全部转成文本喂给趋势研判 Agent数据量大到接近上下文极限接口直接拒绝服务。解决办法就是用 3.4 节里说的上下文裁剪策略。只传最近 5 日价格概览、近 20 日统计特征和指标计算结论。调整后单次调用的 Token 数从 4000 多降到 700 出头成本大约是原来的六分之一模型推理的准确率反而更稳定。这个教训值得每个做大模型应用的人记住喂给模型的数据不是越多越好喂“提炼后的信息”才是关键。6.3 多个 Agent 意见冲突怎么仲裁系统早期跑出来的报告经常自相矛盾趋势研判 Agent 说“看多”风险控制 Agent 说“高风险”两条结论并列在报告里读者不知道信哪个。后来我设计了仲裁规则优先级明确。第一风险控制 Agent 拥有一票否决权只要风险等级为“高”最终倾向直接降到“谨慎”不管趋势 Agent 怎么说。第二置信度低于 0.6 的 Agent 意见不参与最终加权平均。第三如果两个 Agent 意见直接冲突汇总 Agent 要在报告里如实写出分歧点不强行和稀泥。这套规则的逻辑是风控永远是第一位的宁可错过不能踩雷。趋势判断只是一种“倾向参考”而风险是底线。6.4 除权除息导致 K 线跳空这个问题非常隐蔽。某天技术指标 Agent 突然报出 MACD 柱状图瞬间放大 10 倍我差点以为模型算错了。查了半天发现是当天股票除权股价直接从 23 元跳到 18 元不复权的数据让均线系统瞬间失真。解决方法是统一使用前复权数据。前复权价格会持续调整历史价格以匹配当前价值但要注意除权日前后的成交量、市值类数据不用复权处理。另外如果涉及跨年分析复权方式的选择会影响历史信号回测的准确性统一标准非常重要。6.5 模型返回不是合法 JSON即使开了 JSON 模式偶尔仍会返回类似好的以下是JSON{...}这种带前缀的文本。一开始我的解析逻辑直接抛异常整条链路失败。后来改成两步兜底先用正则从返回文本里提取第一对花括号的完整内容再交给json.loads。如果还失败就让这一轮 Agent 用 temperature0 重试一次。两轮基本能覆盖全部异常情况。如果仍然失败直接把这次 Agent 调用的结果标记为 failed由汇总层决策怎么处理不要让异常中断整个管道。做完这一步整条管道从数据获取到报告生成的稳定可靠才算真正具备了可以定时调度的条件。最后分享一点我个人的核心体会。多 Agent 的价值不在“Agent”这个词本身在于把任务拆到足够小、足够明确让大模型在擅长的“语言理解和生成”范围内发挥而把精确计算、数据检索这些它不擅长的事交给传统代码。用 DeepSeek 做底座是因为它便宜且稳定但真正让系统能跑的是每一层的结构化协议和数据清洗逻辑。下一步我给这个系统做了定时任务扩展每天早上开盘前自动跑一次分析把结果推送到本地服务的 Web 页面。如果你想在现有行情分析流程里加入大模型“自动解读”能力我建议你先不要追求复杂框架按这套思路从两三个 Agent 开始做起跑通之后再逐步加角色。Agent 越多协调成本越大没有充分的理由别轻易加人。
