这几年AI投资的话题越来越热各种大模型一出来就有人想着能不能让它帮忙选股票、配仓位。我大概从去年年初开始折腾这个方向试了不少方案最后沉淀下来一套基于多智能体Agent协作的股票筛选系统用来做投资组合的初筛和动态跟踪。这篇文章就把这套系统的设计思路、核心实现、以及我在实操中踩过的坑完整写出来希望能够给想做AI量化选股的朋友一些参考。先明确一点这里说的“代理”不是网络代理而是人工智能领域里的智能体Agent。我在做系统设计的时候参考了软件工程里“代理模式”的思想把复杂的选股任务拆分成多个专职代理每个代理只负责一块专业判断最后由一个决策代理汇总并输出投资组合建议。这样做的好处很直接——单一模型什么都干结果往往是四不像而拆分后每个代理的上下文更干净、提示词更聚焦输出质量会明显提升。这套系统能够解决什么问题最直接的就是人工选股的信息过载。A股几千只股票每只都要看财务、看K线、看消息面一个人一天根本看不过来。多智能体系统可以并行扫描全市场把基本面、技术面、风险面、情绪面分开处理然后按统一规则打分排序从几千只股票里筛出值得深入研究的小池子。它适合谁参考如果你有一定的编程基础想给自己的投资流程加上AI辅助或者你想学习多智能体系统在金融场景下的实际落地方法这篇文章都可以帮到你。下面进入正题我尽量把设计逻辑和实现细节都讲透。1. 内容整体设计与思路拆解1.1 为什么选择多智能体架构而不是单个大模型刚开始动工的时候我的想法很简单把股票数据丢给GPT让它直接输出“该不该买”结果很快就被现实教育了。单一Prompt里如果塞入财务指标、技术指标、新闻摘要模型输出的建议经常自相矛盾——有时它过于侧重某一条负面新闻完全忽略掉强劲的财务增长有时又只看PE够低就推荐完全不理会行业景气度在往下走。后期的回测数据显示这种单模型直接跑的方式胜率甚至低于随机买入。后来我仔细想这件事觉得问题不在模型本身而在任务定义太模糊。选股这件事本质上是多个独立判断的叠加公司基本面值不值得关注、当前价位有没有技术面支撑、潜在风险是否可控、市场情绪有没有催化。这些维度需要的数据源不同、分析逻辑不同、评判标准也不同硬塞进一个对话式的Prompt里模型根本没法建立稳定的“判断基线”。参考《软件设计》里“单一职责原则”的思路我决定把选股拆成四个专职智能体加一个决策智能体智能体名称核心职责主要数据来源数据采集代理获取并清洗行情、财务、新闻数据行情API、财报数据库、新闻RSS基本面分析代理评估公司盈利能力、成长性、估值水平财务三大报表、行业对比数据技术面分析代理分析趋势、动量、量价关系日线/周线行情、技术指标风险识别代理识别财务造假、债务风险、波动异常公告、审计意见、波动率数据决策合成代理融合各代理评分输出最终股票池和组合权重各子代理评分结果这样拆分之后每个代理的职责边界非常清晰上下文窗口可以针对性设计打分规则也可以独立调整。某一个代理的判断逻辑需要升级时不会影响到其他模块。这一点在后续迭代中帮了大忙——比如我发现技术面代理的止损逻辑太宽松只需要改它自己的提示词和阈值基本面代理完全不用动。1.2 系统整体流程的三层设计整个系统在运行流程上分成三个层次数据层、分析层、决策层。数据层负责统一纳管所有外部数据。我选择用Apache Airflow定时触发数据同步任务把日线行情、财报数据、公告信息统一写进PostgreSQL。为什么不用ClickHouse之类的列式数据库因为我这个场景的数据量并不大几千只股票的历史日线一天也就几万行PostgreSQL加好索引后查询性能足够。关键是数据分析团队对SQL更熟悉后面做特征工程也方便。分析层是这套系统的核心也是多智能体真正干活的地方。每个分析代理从数据层取数调用大模型进行结构化打分输出格式统一为JSON。我在这个环节特别强调了一点所有的输出都必须包含“依据引用”也就是模型在给出评分时要指出是哪些数据支撑了这个分数。有了这一步后面即使出现误判也可以回溯排查。决策层做两件事一是把各代理的评分按权重合成出综合分二是根据综合分和风险约束条件生成投资组合权重。这一层我只采用了简单的规则引擎加二次优化——现在很多团队一上来就上强化学习我的经验是先用规则跑通全链路等积累了足够多的有效样本再考虑复杂模型否则连“什么数据是有用的”都还没搞清楚强行上强化学习只会得到一个过拟合的黑盒。2. 数据采集与实际预处理细节2.1 数据源的选型与取舍做AI选股数据是地基。这个环节我没有追求数据的绝对全面而是奉行“够用、干净、可追溯”三个原则。行情数据这块我测试过七八个数据源最后长期使用了两个一个开源财经数据接口面向A股免费的比如AkShare用于获取日线行情和每日的技术指标数据另一家是付费的财报数据服务商保证财务数据的规范性。为什么不直接用免费接口拿财务数据因为免费接口的财报字段经常变动字段缺失或单位不统一的情况很常见。财务数据一旦出错基本面代理的评分就是空中楼阁。为省一点数据费而承担逻辑错误风险没必要。还有个细节可能很多人会忽略新闻与公告数据一定要和行情数据在时间上对齐。我在做情绪因子的时候一开始把公告日期和行情日期做了简单的字符串比较结果有大量公告因为日期格式不统一被错误对齐到了错误的交易日。后来统一统一用时间戳存储并在数据入库时校验交易日历问题才彻底解决。时间对齐这种事看起来不起眼但在后续模型训练时一行错位的数据就可能污染整段特征序列。数据清洗的流程我固定为四步去重、去异常值、复权处理、缺失值标记。去重是按股票代码加交易日做唯一性约束去异常值用的是MAD中位数绝对偏差方法比Z-score更稳健适合金融数据这种厚尾分布复权处理统一采用前复权方便技术指标计算缺失值不主动填充而是标记出来让后续的分析代理自行决定是否跳过这只股票。让模型自己决定怎么处理缺数据比我们强行插值效果更好。2.2 让数据代理保持长期稳定的配置经验数据采集代理是所有代理中技术含量最低、但出问题最多的一个。我踩过的坑包括数据源接口改版导致字段名变化、网络超时导致数据不完整、停牌股票逻辑处理错误等等。最后我总结了一套稳妥配置对每个数据源的调用都封装成独立函数对外暴露统一接口内部做重试与降级。一个源挂了自动切换备选源不阻塞主流程。使用消息队列我用的RabbitMQ解耦数据采集与分析任务。数据采完先落库再发布消息触发分析任务。这样即使分析环节大模型超时了数据已经存下来重新跑分析不需要重复采集。对原始数据做快照备份。每天全量保存一次原始拉取结果后面无论特征工程还是评分逻辑怎么改都能用历史快照做回测。我额外在数据层加了一个质量监控脚本每天跑完采集任务后自动检查每个表的数据条数是否在合理区间字段缺失率是否超标有异常就发钉钉告警。没有这套监控很可能模型已经用了两周错数据你还在那里分析为什么推荐结果越来越离谱。3. 多智能体协同的核心实现过程3.1 智能体的定义与生命周期管理直接用LangChain或预设的Agent框架当然省事但真实场景下灵活度往往不够。我最后是基于LangChain的Message机制自己封装了一层轻量级的Agent基类每个分析代理都继承这个基类并实现三个方法gather_context()负责从数据库取数、call_model()负责调用大模型并解析输出、validate_output()负责校验输出合法性。from dataclasses import dataclass, field from typing import Any dataclass class AnalysisResult: agent_name: str stock_code: str score: float reasoning: str evidence: list raw_output: dict field(default_factorydict) class BaseAgent: agent_name base def __init__(self, llm_client, db_client): self.llm llm_client self.db db_client def gather_context(self, stock_code: str, trade_date: str) - dict: raise NotImplementedError def build_prompt(self, context: dict) - str: raise NotImplementedError def parse_response(self, response: str) - dict: raise NotImplementedError def run(self, stock_code: str, trade_date: str) - AnalysisResult: ctx self.gather_context(stock_code, trade_date) prompt self.build_prompt(ctx) raw self.llm.chat(prompt) parsed self.parse_response(raw) self.validate_output(parsed, ctx) return AnalysisResult( agent_nameself.agent_name, stock_codestock_code, scoreparsed[score], reasoningparsed[reasoning], evidenceparsed.get(evidence, []), raw_outputparsed ) def validate_output(self, parsed: dict, ctx: dict) - None: assert 0 parsed[score] 100 assert isinstance(parsed[reasoning], str) assert len(parsed[reasoning]) 20这里有个很关键的设计每个Agent在输出里必须带上evidence字段。比如基本面代理打出75分它要在evidence里写清楚“ROE连续三年维持15%以上”、“PE处于近五年25%分位以下”、“营收增速超过行业均值”这些具体依据。有了这条约束即使某只股票被高估了我也能回头核查是哪个数据支撑了错误判断。没有证据引用的大模型评分在金融场景里基本没有信任价值。Agent的生命周期管理方面我采用每天早上开盘前跑一次全市场筛选、收盘后跑一次复盘更新的节奏。盘中不做实时分析原因是数据不稳定而且大模型的响应延迟在分钟级这个速度根本追不上分时行情。作为辅助决策工具日级别频率已经够用。生命周期状态我存在一张agent_runs表里记录每次运行的开始时间、结束时间、处理的股票数、失败数、Token消耗量方便排查问题。3.2 提示词工程把专业分析流程写进模型上下文Agent框架搭好后决定分析质量上限的就是提示词。我花了很多时间调优各代理的Prompt分享三个核心体会先定义评判维度再让模型打分。基本面分析代理的Prompt里我明确列出了资产负债率、ROE趋势、毛利率变化、经营现金流增速、商誉占比五个维度并且每个维度写明评分标准。比如商誉占比超过净资产30%直接扣到低分档。这样模型不会自己发明一套标准每次打分都有相对稳定的依据。给模型提供参考基准。大模型对“PE是高是低”没有概念除非你告诉它行业PE中位数。我会先算好股票所属行业的PE分位值、营收增速分位值再把这些相对排名写进Prompt。模型看到的不只是绝对值而是它在行业里的相对位置这种打分更贴近真实投资逻辑。实测效果很明显加上行业分位数据后基本面代理的评分与人工研究员的判断一致性大幅提升。强制使用“先分析后下结论”的思维链。用提示词要求模型必须先列证据清单再做推理最后输出分数。这个做法极大减少了模型直接拍脑袋给分的情况。我试过在测试集上对比启用思维链后代理评分的稳定性提高了非常多。当然代价是多消耗一些Token但投资决策这种场景值得这个成本。技术面代理的Prompt有一点特殊我会预先用Python的TA-Lib库算好RSI、MACD、布林带位置、20日动量等指标把指标数值直接传给模型而不是让模型自己去读行情。因为大模型直接读K线数据算指标并不擅长远不如用现成的库精确计算再交给模型做判断。把“计算”和“判断”分开是智能体设计里非常重要的一条经验。3.3 多模型混合使用与成本控制不同的代理任务适合不同的模型。我没有全部使用最顶级的模型而是根据任务复杂度做了分级代理常用模型原因数据采集代理不调用大模型纯规则代码处理基本面分析代理GPT-4级别大模型需要深度推理与多维度权衡技术面分析代理中等规模模型输入指标明确推理路径短风险识别代理本地部署的开源代码模型涉及敏感数据处理且任务结构化决策合成代理GPT-4级别大模型需要综合多方矛盾信息做权衡成本测算上全市场扫描一次约覆盖3000只股票如果全部用顶级大模型单次运行成本大概在150~220人民币之间一天跑两次一个月光模型费用就要上万。混合模型方案能把成本压低到原来的四成左右。技术面和风险识别的任务结构化程度高用本地部署的开源模型完全能胜任便宜量又足。风险识别代理我专门做了本地化部署既能控制成本也规避了一些数据隐私上的顾虑。另外用一个小的技巧全市场扫描并不是所有股票都需要跑完整流程。我先用规则筛选做预过滤——比如排除ST、排除上市不足60日次新股、排除日均成交额低于2000万的流动性不足标的。预过滤能筛掉约35%的股票剩下约2000只再做智能体深度分析省下来的都是真金白银。4. 投资组合生成与动态跟踪4.1 综合评分与持仓权重的计算逻辑四个分析代理各自输出0到100的分数后决策合成代理负责把它们汇总成一个综合分。我采用的不是简单加权平均而是加了一个风险调节项因为实盘里风险识别代理给出的警告信号需要被充分重视。综合评分公式综合分 基本面分×0.35 技术面分×0.25 情绪面分×0.15 风险调节分×0.25 风险调节分 100 - max(0, 100 - 风险识别分) // 风险识别为100表示无风险等等这个公式写出来有点绕。我直接把我实际用的版本列清楚。每个代理输出原始分0-100风险识别分是“风险越低分越高”的格式100代表最安全0代表极度危险。综合分计算时如果风险分低于50会在最终得分上直接扣除一个惩罚项惩罚力度是(50 - 风险分) * 0.3。这样即使基本面再吸引人风险分过低也会被一票否决机制拦住。生成权重时我没有用马科维茨那种需要协方差矩阵的最优化方法——样本量不够时协方差矩阵估计极不稳定。我采用的是“分层等权再加风险预算裁剪”的简化方案。具体流程按综合分从高到低排序取前30只进入初选池。剔除与持仓中已有股票相关性过高相关系数0.85的标的避免组合过度集中在同一行业或同类型股票。剩余标的按等权分配基础权重。对风险分在50以下的标的权重再打五折多出来的权重按分位比例分配给高分股票。单一行业权重上限设为25%超过的部分按比例平均分配到其他行业。这套规则虽然简单但跑下来风险分散效果并不比复杂模型差关键是每一条规则都可解释、可回测、可调参。对个人投资者或小团队来说可解释性往往比理论最优更重要。4.2 每日闭环筛选、验证、再平衡的实际节奏我的系统每天早晨执行一个标准化的闭环流程整个流程从早上8点自动启动8点数据采集代理自动拉取前一交易日行情更新数据库。8点30分四个分析代理并行启动扫描约2000只预筛后的股票。我用的是asyncio并发控制每个代理配置独立的线程池整批扫描约45分钟能跑完。9点30分前决策合成代理输出当日的推荐股票池和调整建议。收盘后复盘任务自动启动对比当日实际涨跌与系统评分的关系计算当日的评分-收益IC值用于判断各代理是否还有预测力。这里面收盘后的复盘容易被忽略但我觉得恰恰是最重要的环节。我每天追踪一个核心指标——IC值即评分与次日收益的Spearman相关系数。如果某段时间基本面代理的IC持续走低比如连续一周逼近0说明这个代理的输出已经不能预测收益了需要检查是数据过期、市场风格切换、还是Prompt出现了问题。没有这个闭环反馈机制系统就只是每天在生成一堆无法验证的评分数字没有长期改进的方向。再平衡频率上我设置为每两周调一次仓除非触发风控阈值才做临时调仓。试过每周调仓交易成本高得离谱而且高频调仓下模型评分的预测误差会被交易成本吞噬。每两周一次对日频数据来说是个比较平衡的选择具体频率可以根据自己的持仓规模和交易成本再做调整。5. 常见问题与排查技巧实录5.1 大模型“幻觉”导致评分失真的识别与治理这是我在整个项目里遇到最严重的问题。某次系统推荐了一只财务数据亮眼的股票我仔细一查才发现基本面代理在evidence里引用了一个我数据库里根本不存在的指标——营收增速32%但实际数据里营收增速只有11%。大模型在生成理由时“顺理成章”地编出了一个看起来合理的数据。我治理幻觉问题用了三道防线第一道防线是结构化输出与规则校验。要求模型输出严格的JSON格式并且每个评分依据必须对应我在evidence字段里给出的具体指标名称列表。如果模型引用的指标不在原数据上下文里判定为幻觉该条评分直接作废进入人工复核队列。实测这个规则能拦截约六成的幻觉输出。第二道防线是独立交叉验证。让基本面代理的两次独立运行复用不同的大模型只要评分差距超过阈值15分就把这只股票标记为“低置信度”权重减半甚至剔除。这个做法成本翻倍但只对高得分股票做所以总开销可控。两个模型结论一致的时候评分置信度会高很多。第三道防线是事后追溯。每次评分结果都保留完整的输入快照和模型原始输出。一旦某只推荐股票暴雷我能精确回放当时模型看到的数据判断是数据本身问题还是模型推理问题。这一点在复盘和迭代上的价值极大。5.2 数据对齐错误与API调用异常的典型排障记录我梳理了一下项目运行半年多来最高频的几类问题整理成一张速查表问题现象可能原因排查方法解决方案某财务指标大量为空原始数据源字段改名比对数据字典检查入库日志封装字段映射层统一做归一化评分结果系统性偏低行业分位基准未更新检查行业分类表更新时间建立行业基准表每日刷新某日全市场分析耗时飙升至3小时大模型API限流触发重试查看API调用日志中的429状态码增加指数退避重试延长超时技术面评分与行情明显背离前复权价格计算失败抽取个股核对复权因子增加复权结果校验及异常值告警情绪面代理推荐了无新闻股票新闻源抓取范围过窄排查RSS源日志接入更多资讯源并添加“无新闻中立”规则API调用这块我特别想多说一句。刚开始用大模型API时并发一高就经常超时导致整个扫描流程卡死。后来我做了三层防护一是所有API调用统一走一个带超时和重试的封装类重试策略用指数退避加抖动jitter避免重试风暴二是用信号量控制并发数稳定在8到12个并发请求实测吞吐和稳定性综合表现最佳三是给每次调用加上业务ID透传一旦某批结果异常可以直接定位到具体是哪次调用、哪个代理、哪只股票出了问题。6. 效果验证与后续迭代方向6.1 回测结果盘点这套系统到底有没有用我在这套系统正式用于辅助实盘之前做了整整四个月的回测验证。回测区间滚动覆盖了过去两年的数据每两周调仓一次对比基准是中证500指数扣除了双边千三的交易成本与千一的滑点。直接说结果全模型评分前30的股票池年化超额收益大约在8%左右最大回撤比基准降低了约5个百分点。考虑到评测周期不够长、样本量有限这个数据不能说明太多问题但至少证明多智能体筛选产生了一定的信息增量。更让我满意的是另一个指标——评分IC均值在0.06到0.09之间。这个数值不算高但在选股领域已经说明评分与未来收益之间存在真实的单调相关关系而不是纯噪声。体验最明显的变化是在2024年下半年到2025年初那段波动很大的行情里——单靠人工复盘很难覆盖全市场的信息变化系统每天自动盯住两千多只股票风险识别代理好几只持仓股发出了债务风险预警都是我再人工核实后发现确实存在问题的案例。提前减仓规避了一部分回撤。这件事让我确定了这套系统的定位不是代替我做决策而是代替我盯盘和做初筛让我能把有限的精力集中在系统标记出来的关键标的上。6.2 我踩过的一些关键的坑整个项目做下来有几个认知层面的坑值得一提。第一个大坑是盲目追求模型复杂度。我最早想用强化学习来学习组合权重为了跑通全流程折腾了将近一个月最后发现由于有效样本量严重不足训练出来的策略连规则策略都跑不赢。后来我把强化学习方案砍掉换成透明规则系统反而稳定下来。先用规则建立基准再考虑复杂模型这条路才是正解。第二个坑是不重视数据版本管理。初期我直接在原始数据上做特征计算后面修改特征逻辑时发现历史评分没法复现因为底层数据已经被覆盖更新了。后来我强制要求每天的原始数据做快照特征版本、代理版本、提示词版本全部登记在案。现在任何一次评分结果都能精确复现这对调试和迭代来说至关重要。第三个坑是过度相信大模型的“金融知识”。大模型确实懂很多金融概念但它训练数据里的金融市场规则可能已经过时了而且它对具体某一只股票、某一个行业的理解通常停留在非常表面的水平。把它当分析师用是误解把它当“规则执行者”用才是正确定位——专业判断规则由人来定义模型负责大规模、高效率地执行这些规则。这套边界想清楚之后整个系统的定位就清晰了。后续的迭代方向我计划把重点放在三块一是引入更加结构化的另类数据比如产业链上下游的景气度变化信号二是做跨时间尺度的多模型融合把周线级别的中周期信号和日线级别的短周期信号区分开避免信号周期错配三是尝试用图神经网络做股票间相关性建模替代目前简单的相关系数阈值法进一步提升组合分散化的效果。最后分享一个我自己实操中的体会做AI投资系统真正决定成败的往往不是模型有多么先进而是数据有多干净、流程有多稳定、反馈闭环有多及时。把数据、规则、可解释性这三件事做扎实了再谈模型升级才有意义。这套多智能体方案目前的产出已经超出我最初的预期但距离“可靠的投资助手”还有很长的路要走。好在有了完整的架构和闭环验证机制后续每一步优化都能在明确的前进方向上推进。