股票投资顾问Agent架构解析
项目概览
这是一个基于多智能体协作和 MCP (Model Context Protocol) 的股票投资分析系统,由三个核心模块组成:
- a-share-mcp-is-just-i-need: A股数据服务MCP服务器
- Financial-MCP-Agent: 多智能体金融分析系统
- 模型训练与测试: 情感分析和风险评估模型
整体架构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54
| Finance/ ├── a-share-mcp-is-just-i-need/ │ ├── mcp_server.py │ └── src/ │ ├── data_source_interface.py │ ├── baostock_data_source.py │ ├── tools/ │ │ ├── stock_market.py │ │ ├── financial_reports.py │ │ ├── indices.py │ │ ├── market_overview.py │ │ ├── macroeconomic.py │ │ ├── date_utils.py │ │ ├── analysis.py │ │ └── news_crawler.py │ └── formatting/ │ └── markdown_formatter.py │ ├── Financial-MCP-Agent/ │ ├── src/ │ │ ├── main.py │ │ ├── agents/ │ │ │ ├── fundamental_agent.py │ │ │ ├── technical_agent.py │ │ │ ├── value_agent.py │ │ │ ├── news_agent.py │ │ │ └── summary_agent.py │ │ ├── tools/ │ │ │ ├── mcp_client.py │ │ │ ├── mcp_config.py │ │ │ └── openrouter_config.py │ │ └── utils/ │ │ ├── state_definition.py │ │ ├── execution_logger.py │ │ ├── logging_config.py │ │ └── llm_clients.py │ ├── logs/ │ └── reports/ │ ├── 模型训练脚本 │ ├── train_qwen_sentiment.py │ ├── train_qwen_risk.py │ ├── test_qwen_sentiment.py │ └── test_risk_model.py │ ├── 数据处理脚本 │ ├── data_process.py │ └── download.py │ ├── 数据目录 │ ├── nasdaq_news_sentiment/ │ └── risk_nasdaq/ │ └── requirements.txt
|
核心架构解析
1. MCP数据服务层 (a-share-mcp-is-just-i-need)
设计理念
采用 MCP协议 提供标准化的金融数据访问接口,通过抽象层实现数据源的可替换性。
核心组件
数据源抽象接口 (FinancialDataSource)
1 2 3 4 5 6 7 8 9 10 11 12
| class FinancialDataSource(ABC): @abstractmethod def get_historical_k_data(...) -> pd.DataFrame """获取历史K线数据""" @abstractmethod def get_stock_basic_info(...) -> pd.DataFrame """获取股票基本信息""" @abstractmethod def get_trade_dates(...) -> pd.DataFrame """获取交易日历"""
|
具体实现 (BaostockDataSource)
- 基于 baostock 库实现A股数据获取
- 支持K线、财报、指数、宏观数据
- 实现了完整的
FinancialDataSource 接口
工具模块注册
每个工具模块通过 register_*_tools() 函数向MCP服务器注册功能:
register_stock_market_tools() - 股票行情工具
register_financial_report_tools() - 财报分析工具
register_index_tools() - 指数追踪工具
register_market_overview_tools() - 市场概览工具
register_macroeconomic_tools() - 宏观经济工具
register_date_utils_tools() - 日期工具
register_analysis_tools() - 分析工具
register_news_crawler_tools() - 新闻爬虫工具
技术栈
- FastMCP: 现代化的MCP服务器框架
- baostock: 免费A股数据接口
- pandas: 数据处理
2. 多智能体分析系统 (Financial-MCP-Agent)
设计理念
采用 LangGraph 构建并行智能体工作流,每个智能体专注于特定分析维度,最终由总结智能体整合结果。
工作流架构
1 2 3 4 5 6 7 8 9 10 11 12
| 用户输入 (股票名称/代码) ↓ start_node (提取股票信息) ↓ ├─→ fundamental_analyst (基本面分析) ├─→ technical_analyst (技术分析) ├─→ value_analyst (估值分析) └─→ news_analyst (新闻分析) ↓ summarizer (综合总结) ↓ 生成分析报告
|
核心智能体
1. 基本面分析智能体 (fundamental_agent.py)
- 分析财务状况(营收、利润、现金流)
- 评估盈利能力和成长性
- 行业地位对比
- 使用MCP工具获取财报数据
- 基于 ReAct Agent 框架实现
2. 技术分析智能体 (technical_agent.py)
- 价格趋势分析
- 技术指标计算(MA、MACD、RSI等)
- 成交量分析
- 支撑阻力位识别
- 基于 ReAct Agent 框架实现
3. 估值分析智能体 (value_agent.py)
- 市盈率、市净率分析
- PEG、DCF等估值模型
- 行业估值对比
- 内在价值评估
- 基于 ReAct Agent 框架实现
4. 新闻分析智能体 (news_agent.py)
- 新闻情感分析
- 风险因素识别
- 重大事件影响评估
- 使用新闻爬虫工具获取最新新闻
- 基于 ReAct Agent 框架实现
5. 综合总结智能体 (summary_agent.py)
- 整合四个维度的分析结果
- 生成投资建议
- 风险提示
- 输出结构化报告
- 使用 Summary Agent 框架
状态管理
使用 AgentState 实现智能体间的数据传递:
1 2 3 4
| class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] data: Annotated[Dict[str, Any], merge_dicts] metadata: Annotated[Dict[str, Any], merge_dicts]
|
自然语言处理
主程序包含复杂的股票信息提取逻辑,支持多种查询格式:
- “分析嘉友国际”
- “帮我看看比亚迪这只股票怎么样”
- “603871 这个股票值得买吗?”
- “茅台(600519)值得投资吗”
提取模式包括20多种正则表达式模式:
- 括号内的股票代码
- 直接的公司名称
- 股票代码+公司名组合
- 智能语义识别
- 复杂句式解析
3. 模型训练与测试模块
3.1 情感分析模型训练 (train_qwen_sentiment.py)
功能: 训练金融新闻情感分析模型
技术架构:
1
| 数据加载 → 提示模板构建 → Tokenization → LoRA微调 → 模型保存
|
核心组件:
数据预处理 (load_and_preprocess_data)
1 2
| df = df[df['Lsa_summary'].notna() & df['sentiment_deepseek'].notna()] df = df[df['sentiment_deepseek'] != 0]
|
提示模板构建 (create_prompt_template)
1 2 3 4 5 6 7 8
| system_prompt = "You are a financial expert... Score from 1 to 5..." conversation = f"""System: {system_prompt}
User: News to Stock Symbol -- AAPL: Apple (AAPL) increase 22% Assistant: 5
User: {user_content} Assistant: {sentiment}"""
|
数据集准备 (prepare_dataset)
- 训练集/验证集分割 (80%/20%)
- Tokenization处理
- 智能损失计算(只对Assistant回答部分计算损失)
LoRA微调配置
1 2 3 4 5 6 7 8
| lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, lora_alpha=32, lora_dropout=0.1, target_modules=["q_proj", "v_proj", "k_proj", "o_proj", "gate_proj", "up_proj", "down_proj"] )
|
评分标准:
- 1: 负面
- 2: 轻微负面
- 3: 中性
- 4: 正面
- 5: 极正面
训练参数:
1 2 3 4 5 6 7 8 9 10 11 12
| TrainingArguments( output_dir="./qwen_sentiment_model", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, warmup_steps=100, learning_rate=2e-5, fp16=True, logging_steps=50, save_steps=500, eval_steps=500 )
|
3.2 风险评估模型训练 (train_qwen_risk.py)
功能: 训练股票风险评估模型
技术架构: 与情感分析模型完全相同的架构,但专注于风险评估
核心差异:
评分标准不同:
- 1: 极低风险
- 2: 低风险
- 3: 中等风险(默认,无明显风险迹象)
- 4: 高风险
- 5: 极高风险
提示词不同:
1 2
| system_prompt = "You are a financial expert specializing in risk assessment... Provide a risk score from 1 to 5..."
|
示例不同:
1 2 3 4 5
| User: Apple (AAPL) increases 22% Assistant: 3
User: Apple (AAPL) price decreased 30% Assistant: 4
|
训练输出: 模型保存在 ./qwen_risk_model
3.3 情感模型测试 (test_qwen_sentiment.py)
功能: 测试训练好的情感分析模型
测试流程:
模型加载
1 2 3 4 5
| def load_trained_sentiment_model(model_path): tokenizer = AutoTokenizer.from_pretrained(model_path) base_model = AutoModelForCausalLM.from_pretrained("/root/code/Finance/Qwen", ...) model = PeftModel.from_pretrained(base_model, model_path) return model, tokenizer
|
预测函数
1 2 3 4 5
| def predict_sentiment(model, tokenizer, text, stock_symbol): prompt = create_sentiment_test_prompt(text, stock_symbol) inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512) outputs = model.generate(**inputs, max_new_tokens=5, do_sample=False) return extract_score(outputs)
|
测试内容:
- 预设测试用例: 10个不同场景的新闻
- 真实数据测试: 从CSV读取真实数据进行测试
- 情感分布测试: 测试模型在5个类别上的表现
测试输出示例:
1 2 3 4 5 6 7 8 9 10 11 12 13
| === 情感分析模型测试结果 ===
测试 1: 新闻: Apple reported strong quarterly earnings with revenue growth of 15% 股票: AAPL 预测情感: 4 (正面)
测试 2: 新闻: Apple faces supply chain disruptions and production delays 股票: AAPL 预测情感: 2 (轻微负面)
整体准确率: 8/10 = 80.0%
|
3.4 风险模型测试 (test_risk_model.py)
功能: 测试训练好的风险评估模型
测试流程: 与情感模型测试类似,但专注于风险评分
测试内容:
测试输出示例:
1 2 3 4 5 6 7 8 9 10 11
| === 风险评估模型测试结果 ===
测试 1: 新闻: Apple reported strong quarterly earnings with revenue growth of 15% 股票: AAPL 预测风险: 2 (低风险)
测试 2: 新闻: Apple faces major supply chain disruptions and production delays 股票: AAPL 预测风险: 4 (高风险)
|
4. 数据处理模块 (data_process.py)
功能: 新闻去重和数据预处理
核心组件:
1 2 3 4 5 6
| class NewsDeduplicator: def __init__(self): self.title_threshold = 0.8 self.content_threshold = 0.75 self.simhash_threshold = 3 self.minhash_permutations = 128
|
关键算法:
标题相似度 - 组合方法
1 2 3 4
| def title_similarity(title1, title2): edit_sim = self.edit_distance(title1, title2) cos_sim = self.text_to_tfidf_vector([title1, title2]) return (edit_sim + cos_sim) / 2
|
正文重合度 - MinHash
1 2 3 4 5 6
| def content_overlap(content1, content2): shingles1 = self.get_shingles(content1) shingles2 = self.get_shingles(content2) sig1 = self.minhash_signature(shingles1) sig2 = self.minhash_signature(shingles2) return self.jaccard_similarity_minhash(sig1, sig2)
|
语义相似度 - SimHash
1 2 3 4
| def semantic_similarity(content1, content2): hash1 = self.simhash(content1) hash2 = self.simhash(content2) return self.hamming_distance(hash1, hash2)
|
去重判断 - 三重阈值
1 2 3 4 5 6
| def is_duplicate(item1, item2): return ( self.title_similarity(title1, title2) > 0.8 and self.content_overlap(content1, content2) > 0.75 and self.semantic_similarity(content1, content2) <= 3 )
|
处理流程:
1 2
| CSV文件 → 加载数据 → Unicode归一化 → 计算相似度 → 三重阈值判断 → 去重处理 → 保存JSONL
|
5. 完整程序架构图
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41
| ┌─────────────────────────────────────────────────────────┐ │ Finance Project │ └─────────────────────────────────────────────────────────┘ │ ┌───────────────────┼───────────────────┐ │ │ │ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ 数据处理层 │ │ 模型训练层 │ │ Agent应用层 │ ├──────────────┤ ├──────────────┤ ├──────────────┤ │ │ │ │ │ │ │ data_process │ │ train_qwen_ │ │ Financial- │ │ │ │ _sentiment │ │ MCP-Agent │ │ download.py │ │ │ │ │ │ │ │ train_qwen_ │ │ main.py │ │ NewsDedu- │ │ _risk │ │ │ │ plicator │ │ │ │ agents/ │ │ │ │ test_qwen_ │ │ │ │ - 去重策略 │ │ _sentiment │ │ - 基本面 │ │ - 相似度计算 │ │ │ │ - 技术面 │ │ - 数据清洗 │ │ test_risk_ │ │ - 估值面 │ │ │ │ _model │ │ - 新闻面 │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────┐ │ a-share-mcp-server │ │ (数据服务层) │ ├─────────────────────────────────────────────────────┤ │ - stock_market - financial_reports │ │ - indices - market_overview │ │ - macroeconomic - analysis │ │ - date_utils - news_crawler │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Baostock API │ │ (A股数据源) │ └─────────────────────────────────────────────────────┘
|
6. 数据流程图
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
| 原始新闻数据 (CSV) ↓ data_process.py ├─ load_and_preprocess_data() ├─ unicode_normalize() ├─ title_similarity() ├─ content_overlap() ├─ semantic_similarity() └─ deduplicate() ↓ 清洗后的数据 (JSONL/CSV) ↓ 训练集 / 验证集 (80%/20%) ↓ train_qwen_sentiment.py / train_qwen_risk.py ├─ create_prompt_template() ├─ prepare_dataset() ├─ tokenize_function() ├─ create_model_and_tokenizer() ├─ LoRA微调 └─ save_model() ↓ 训练好的模型 ↓ test_qwen_sentiment.py / test_risk_model.py ├─ load_trained_model() ├─ predict() └─ evaluate()
|
7. 关键技术详解
7.1 损失计算策略
在训练过程中,采用智能损失计算策略:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
| def tokenize_function(examples): labels = tokenized['input_ids'].clone() for i, text in enumerate(examples['text']): assistant_marker = "Assistant: " last_assistant_pos = text.rfind(assistant_marker) if last_assistant_pos != -1: input_part = text[:last_assistant_pos + len(assistant_marker)] input_part_tokens = tokenizer.encode(input_part, add_special_tokens=False) mask_length = len(input_part_tokens) labels[i, :mask_length] = -100 actual_length = (input_ids != pad_token_id).sum().item() if actual_length < len(input_ids): labels[i, actual_length:] = -100 tokenized['labels'] = labels return tokenized
|
优势:
- 只对模型生成部分计算损失
- 避免对System、User等固定部分计算损失
- 提高训练效率
7.2 新闻去重算法
三重阈值策略:
1 2 3 4 5 6 7 8 9 10 11
| def is_duplicate(item1, item2): title_sim = self.title_similarity(title1, title2) content_sim = self.content_overlap(content1, content2) semantic_dist = self.semantic_similarity(content1, content2) return (title_sim > 0.8 and content_sim > 0.75 and semantic_dist <= 3)
|
各算法特点:
- 编辑距离: 字符级别的相似度
- TF-IDF余弦相似度: 词向量级别的相似度
- MinHash: 快速估计Jaccard相似度
- SimHash: 语义级别的指纹匹配
8. 部署架构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39
| ┌─────────────────────────────────────────┐ │ 用户终端 (CLI) │ └────────────┬────────────────────────────┘ │ ↓ ┌─────────────────────────────────────────┐ │ Financial-MCP-Agent │ │ (多智能体工作流引擎) │ │ │ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │基本面│ │技术面│ │估值面│ │新闻面│ │ │ │Agent │ │Agent │ │Agent │ │Agent │ │ │ └───┬──┘ └───┬──┘ └───┬──┘ └───┬──┘ │ │ └───────┴───────┴───────┴───────┘ │ │ │ │ │ ↓ │ │ 总结Agent │ └───────────────────┬───────────────────────┘ │ MCP Protocol ↓ ┌─────────────────────────────────────────┐ │ a-share-mcp-is-just-i-need │ │ (MCP数据服务器) │ │ │ │ ┌──────────┐ ┌──────────┐ │ │ │数据接口层│ │工具注册器│ │ │ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ ↓ ↓ │ │ ┌────────────────────────┐ │ │ │ Baostock数据源 │ │ │ └────────────────────────┘ │ └───────────────────┬──────────────────────┘ │ ↓ ┌────────────────┐ │ Baostock API │ │ (免费A股数据) │ └────────────────┘
|
9. 依赖管理
核心依赖
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21
| # 智能体工作流 langgraph==0.6.6 # 智能体工作流编排 python-dotenv==1.1.1 # 环境变量管理 langchain-openai==0.3.30 # OpenAI API集成 langchain-core==0.3.74 # LangChain核心 langchain-mcp-adapters==0.1.9 # MCP协议适配
# 模型训练 transformers==4.51.3 # HuggingFace模型 peft==0.17.0 # 参数高效微调 (LoRA) huggingface-hub==0.34.4 # 模型仓库
# 数据处理 baostock==0.9.1 # A股数据接口 pandas # 数据处理 numpy # 数值计算 scikit-learn # TF-IDF、余弦相似度 jieba # 中文分词
# 其他 uv==0.8.12 # 包管理器
|
10. 使用示例
命令行调用
1 2 3 4 5 6
| python src/main.py --command "分析嘉友国际"
python src/main.py
|
训练模型
1 2 3 4 5
| python train_qwen_sentiment.py
python train_qwen_risk.py
|
测试模型
1 2 3 4 5
| python test_qwen_sentiment.py
python test_risk_model.py
|
数据处理
11. 查询示例
系统支持多种自然语言查询方式:
- 简单查询: “分析嘉友国际”
- 描述性查询: “帮我看看比亚迪这只股票怎么样”
- 代码查询: “603871 这个股票值得买吗?”
- 格式化查询: “茅台(600519)值得投资吗”
- 复杂查询: “我想了解一下腾讯的投资价值”
- 细分查询: “给我分析一下宁德时代的财务状况”
12. 输出内容
系统会生成结构化的分析报告,包含:
- 基本信息 - 股票代码、公司名称、行业分类
- 基本面分析 - 财务指标、盈利能力、成长性
- 技术面分析 - 趋势判断、技术指标信号
- 估值分析 - 估值水平、投资价值
- 新闻分析 - 近期新闻、风险提示
- 综合建议 - 投资评级、风险提示
报告保存在 reports/ 目录,格式为 report_{公司名}_{代码}_{日期}.md
13. 扩展性设计
添加新的数据源
1 2 3 4 5
| class MyDataSource(FinancialDataSource): def get_historical_k_data(self, ...): pass
|
添加新的分析智能体
1 2 3 4 5 6
| from langgraph.graph import StateGraph, END
workflow.add_node("my_analyst", my_agent) workflow.add_edge("start_node", "my_analyst") workflow.add_edge("my_analyst", "summarizer")
|
添加新的MCP工具
1 2 3 4 5
| @register_tool def my_tool(param1: str) -> str: """工具描述""" return result
|
局限性与改进方向
当前局限
- 数据延迟: Baostock数据有1-2天延迟
- 模型依赖: 分析质量依赖所选LLM
- 计算资源: 并行执行需要足够的计算资源
- 模型规模: Qwen模型较大,部署成本高
- 数据量限制: 训练数据仅1000条,可能影响泛化能力
改进方向
- 实时数据: 集成实时行情API
- 多模型集成: 不同智能体使用不同模型
- 缓存机制: 减少重复数据查询
- Web界面: 开发可视化前端
- 回测系统: 添加策略回测功能
- 模型优化: 使用更小更快的模型(如Qwen2.5-7B)
- 数据增强: 扩充训练数据集,提升模型性能
- 知识蒸馏: 将大模型知识蒸馏到小模型
- 增量学习: 支持模型持续学习和更新
总结
这个股票投资顾问Agent系统展示了现代AI在金融分析中的应用:
核心优势
- 架构清晰: MCP协议 + 多智能体工作流
- 可扩展性: 模块化设计,易于扩展
- 实用性: 提供全面的多维度分析
- 标准化: 使用行业标准和协议
- 完整性: 从数据处理、模型训练到应用部署的完整链路
技术亮点
- MCP协议的实践应用
- LangGraph并行智能体编排
- LoRA参数高效微调
- 三重阈值新闻去重算法
- 智能损失计算策略
该项目为个人投资者提供了一个自动化、全面、易用的股票分析工具,同时也为金融AI应用开发提供了一个很好的参考架构。系统结合了传统金融分析方法和现代AI技术,为金融科技领域的发展提供了有价值的探索。