aaron/stock-ai-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加港股支持

Browse Source
This commit is contained in:
aaron 2026-02-05 18:23:22 +08:00
parent 24e8a5c1a2
commit 68f38d4062
13 changed files with 613 additions and 3359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

378
backend/app/agent/core.py
View File

@ -1,378 +0,0 @@
"""
AI Agent核心
基于LangChain的股票分析Agent
"""
import re
import json
from typing import Dict, Any, Optional
from app.config import get_settings
from app.agent.context import ContextManager
from app.agent.skill_manager import skill_manager
from app.skills.market_data import MarketDataSkill
from app.skills.technical_analysis import TechnicalAnalysisSkill
from app.skills.fundamental import FundamentalSkill
from app.skills.visualization import VisualizationSkill
from app.utils.logger import logger
class StockAnalysisAgent:
"""股票分析Agent"""
def __init__(self):
"""初始化Agent"""
self.context_manager = ContextManager()
self.settings = get_settings()
# 注册技能
self._register_skills()
# 初始化LLM（简化版，使用规则匹配）
# 在实际部署时，这里应该集成智谱AI GLM-4
self.use_llm = bool(self.settings.zhipuai_api_key)
logger.info("Stock Analysis Agent初始化完成")
def _register_skills(self):
"""注册所有技能"""
skill_manager.register(MarketDataSkill())
skill_manager.register(TechnicalAnalysisSkill())
skill_manager.register(FundamentalSkill())
skill_manager.register(VisualizationSkill())
logger.info("技能注册完成")
async def process_message(
self,
message: str,
session_id: str,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
处理用户消息
Args:
message: 用户消息
session_id: 会话ID
user_id: 用户ID
Returns:
响应结果
"""
logger.info(f"处理消息: {message[:50]}...")
# 保存用户消息
self.context_manager.add_message(session_id, "user", message)
# 意图识别和技能调用
intent = self._recognize_intent(message)
logger.info(f"识别意图: {intent}")
# 执行技能
result = await self._execute_intent(intent, message)
# 生成响应
response = self._generate_response(intent, result)
# 保存助手响应
self.context_manager.add_message(
session_id,
"assistant",
response["message"],
metadata=response.get("metadata")
)
return response
def _recognize_intent(self, message: str) -> Dict[str, Any]:
"""
识别用户意图（简化版规则匹配）
Args:
message: 用户消息
Returns:
意图字典
"""
message_lower = message.lower()
# 提取股票代码
stock_code = self._extract_stock_code(message)
# 行情查询
if any(keyword in message_lower for keyword in ["行情", "价格", "涨跌", "实时", "quote"]):
return {
"type": "market_data",
"skill": "market_data",
"params": {
"stock_code": stock_code,
"data_type": "quote"
}
}
# K线查询
if any(keyword in message_lower for keyword in ["k线", "kline", "走势", "图表"]):
return {
"type": "visualization",
"skill": "visualization",
"params": {
"stock_code": stock_code,
"chart_type": "candlestick"
}
}
# 技术分析
if any(keyword in message_lower for keyword in ["技术", "指标", "macd", "rsi", "kdj", "均线", "ma"]):
return {
"type": "technical_analysis",
"skill": "technical_analysis",
"params": {
"stock_code": stock_code,
"indicators": ["ma", "macd", "rsi"]
}
}
# 基本面
if any(keyword in message_lower for keyword in ["基本面", "公司", "行业", "信息"]):
return {
"type": "fundamental",
"skill": "fundamental",
"params": {
"stock_code": stock_code
}
}
# 默认：行情查询
if stock_code:
return {
"type": "market_data",
"skill": "market_data",
"params": {
"stock_code": stock_code,
"data_type": "quote"
}
}
# 无法识别
return {
"type": "unknown",
"skill": None,
"params": {}
}
def _extract_stock_code(self, message: str) -> Optional[str]:
"""
从消息中提取股票代码
Args:
message: 用户消息
Returns:
股票代码或None
"""
from app.utils.stock_names import search_stock_by_name
# 匹配6位数字
pattern = r'\b\d{6}\b'
matches = re.findall(pattern, message)
if matches:
return matches[0]
# 使用股票名称数据库搜索
# 提取可能的股票名称（2-6个汉字）
chinese_pattern = r'[\u4e00-\u9fa5]{2,6}'
chinese_words = re.findall(chinese_pattern, message)
for word in chinese_words:
code = search_stock_by_name(word)
if code:
logger.info(f"识别股票名称: {word} -> {code}")
return code
return None
async def _execute_intent(self, intent: Dict[str, Any], message: str) -> Dict[str, Any]:
"""
执行意图对应的技能
Args:
intent: 意图字典
message: 原始消息
Returns:
执行结果
"""
if intent["type"] == "unknown":
return {
"success": False,
"error": "无法理解您的问题，请提供股票代码或明确的查询意图"
}
skill_name = intent["skill"]
params = intent["params"]
if not params.get("stock_code"):
return {
"success": False,
"error": "请提供股票代码（6位数字）"
}
# 执行技能
result = await skill_manager.execute_skill(skill_name, **params)
return result
def _generate_response(self, intent: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]:
"""
生成响应消息
Args:
intent: 意图
result: 执行结果
Returns:
响应字典
"""
if not result.get("success", True):
return {
"message": f"抱歉，{result.get('error', '处理失败')}",
"metadata": {
"type": "error"
}
}
data = result.get("data", result)
# 根据意图类型生成不同响应
if intent["type"] == "market_data":
return self._format_market_data_response(data)
elif intent["type"] == "technical_analysis":
return self._format_technical_response(data)
elif intent["type"] == "fundamental":
return self._format_fundamental_response(data)
elif intent["type"] == "visualization":
return self._format_visualization_response(data)
else:
return {
"message": "查询完成",
"metadata": {
"type": "data",
"data": data
}
}
def _format_market_data_response(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化行情数据响应"""
if "error" in data:
return {
"message": f"查询失败：{data['error']}",
"metadata": {"type": "error"}
}
if "kline_data" in data:
kline_data = data["kline_data"]
message = f"已获取K线数据，共{len(kline_data)}条记录"
return {
"message": message,
"metadata": {
"type": "kline",
"data": kline_data
}
}
# 实时行情
message = f"""
【{data.get('name', '股票')}】({data.get('ts_code', '')})
交易日期：{data.get('trade_date', '')}
最新价：{data.get('close', 0):.2f}
涨跌额：{data.get('change', 0):.2f}
涨跌幅：{data.get('pct_chg', 0):.2f}%
开盘价：{data.get('open', 0):.2f}
最高价：{data.get('high', 0):.2f}
最低价：{data.get('low', 0):.2f}
成交量：{data.get('vol', 0):.0f}手
成交额：{data.get('amount', 0):.0f}千元
""".strip()
return {
"message": message,
"metadata": {
"type": "quote",
"data": data
}
}
def _format_technical_response(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化技术分析响应"""
if "error" in data:
return {
"message": f"分析失败：{data['error']}",
"metadata": {"type": "error"}
}
indicators = data.get("indicators", {})
message_parts = [f"【{data.get('stock_code', '')}】技术指标：\n"]
if "ma" in indicators:
ma = indicators["ma"]
message_parts.append(f"均线：MA5={ma.get('ma5')}, MA10={ma.get('ma10')}, MA20={ma.get('ma20')}")
if "macd" in indicators:
macd = indicators["macd"]
message_parts.append(f"MACD：DIF={macd.get('dif')}, DEA={macd.get('dea')}, MACD={macd.get('macd')}")
if "rsi" in indicators:
rsi = indicators["rsi"]
message_parts.append(f"RSI：RSI6={rsi.get('rsi6')}, RSI12={rsi.get('rsi12')}, RSI24={rsi.get('rsi24')}")
return {
"message": "\n".join(message_parts),
"metadata": {
"type": "technical",
"data": data
}
}
def _format_fundamental_response(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化基本面响应"""
if "error" in data:
return {
"message": f"查询失败：{data['error']}",
"metadata": {"type": "error"}
}
message = f"""
【{data.get('name', '股票')}】基本信息
股票代码：{data.get('ts_code', '')}
所属地域：{data.get('area', '')}
所属行业：{data.get('industry', '')}
上市市场：{data.get('market', '')}
上市日期：{data.get('list_date', '')}
""".strip()
return {
"message": message,
"metadata": {
"type": "fundamental",
"data": data
}
}
def _format_visualization_response(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化可视化响应"""
if "error" in data:
return {
"message": f"生成图表失败：{data['error']}",
"metadata": {"type": "error"}
}
return {
"message": f"已生成{data.get('stock_code', '')}的K线图",
"metadata": {
"type": "chart",
"data": data
}
}
# 创建全局Agent实例
stock_agent = StockAnalysisAgent()

377
backend/app/agent/enhanced_agent.py
View File

@ -1,377 +0,0 @@
"""
增强版Agent - 集成LLM智能分析
"""
import re
import json
from typing import Dict, Any, Optional
from app.config import get_settings
from app.agent.context import ContextManager
from app.agent.skill_manager import skill_manager
from app.skills.market_data import MarketDataSkill
from app.skills.technical_analysis import TechnicalAnalysisSkill
from app.skills.fundamental import FundamentalSkill
from app.skills.visualization import VisualizationSkill
from app.services.llm_service import llm_service
from app.utils.logger import logger
from app.utils.stock_names import search_stock_by_name, get_stock_name
class EnhancedStockAgent:
"""增强版股票分析Agent（集成LLM）"""
def __init__(self):
"""初始化Agent"""
self.context_manager = ContextManager()
self.settings = get_settings()
# 注册技能
self._register_skills()
# 检查LLM是否可用
self.use_llm = bool(self.settings.zhipuai_api_key) and llm_service.client is not None
if self.use_llm:
logger.info("Enhanced Agent初始化完成（LLM模式）")
else:
logger.info("Enhanced Agent初始化完成（规则模式）")
def _register_skills(self):
"""注册所有技能"""
skill_manager.register(MarketDataSkill())
skill_manager.register(TechnicalAnalysisSkill())
skill_manager.register(FundamentalSkill())
skill_manager.register(VisualizationSkill())
logger.info("技能注册完成")
async def process_message(
self,
message: str,
session_id: str,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
处理用户消息（增强版）
Args:
message: 用户消息
session_id: 会话ID
user_id: 用户ID
Returns:
响应结果
"""
logger.info(f"处理消息: {message[:50]}...")
# 保存用户消息
self.context_manager.add_message(session_id, "user", message)
# 提取股票代码
stock_code = self._extract_stock_code(message)
# 使用LLM或规则识别意图
if self.use_llm:
intent = await self._recognize_intent_with_llm(message, stock_code)
else:
intent = self._recognize_intent_with_rules(message, stock_code)
logger.info(f"识别意图: {intent}")
# 执行技能
result = await self._execute_intent(intent, message)
# 生成响应（使用LLM增强）
response = await self._generate_response(intent, result, stock_code)
# 保存助手响应
self.context_manager.add_message(
session_id,
"assistant",
response["message"],
metadata=response.get("metadata")
)
return response
async def _recognize_intent_with_llm(
self,
message: str,
stock_code: Optional[str]
) -> Dict[str, Any]:
"""使用LLM识别意图"""
try:
llm_result = llm_service.analyze_intent(message)
intent_type = llm_result.get("type", "unknown")
confidence = llm_result.get("confidence", 0)
# 如果置信度太低，回退到规则模式
if confidence < 0.5:
logger.info("LLM置信度低，回退到规则模式")
return self._recognize_intent_with_rules(message, stock_code)
# 构建意图
intent = {
"type": intent_type,
"confidence": confidence,
"skill": self._map_intent_to_skill(intent_type),
"params": {"stock_code": stock_code} if stock_code else {}
}
return intent
except Exception as e:
logger.error(f"LLM意图识别失败: {e}")
return self._recognize_intent_with_rules(message, stock_code)
def _recognize_intent_with_rules(
self,
message: str,
stock_code: Optional[str]
) -> Dict[str, Any]:
"""使用规则识别意图（原有逻辑）"""
message_lower = message.lower()
# 行情查询
if any(keyword in message_lower for keyword in ["行情", "价格", "涨跌", "实时", "quote"]):
return {
"type": "market_data",
"skill": "market_data",
"params": {
"stock_code": stock_code,
"data_type": "quote"
}
}
# K线查询
if any(keyword in message_lower for keyword in ["k线", "kline", "走势", "图表"]):
return {
"type": "visualization",
"skill": "visualization",
"params": {
"stock_code": stock_code,
"chart_type": "candlestick"
}
}
# 技术分析
if any(keyword in message_lower for keyword in ["技术", "指标", "macd", "rsi", "kdj", "均线", "ma"]):
return {
"type": "technical_analysis",
"skill": "technical_analysis",
"params": {
"stock_code": stock_code,
"indicators": ["ma", "macd", "rsi"]
}
}
# 基本面
if any(keyword in message_lower for keyword in ["基本面", "公司", "行业", "信息"]):
return {
"type": "fundamental",
"skill": "fundamental",
"params": {
"stock_code": stock_code
}
}
# 默认：行情查询
if stock_code:
return {
"type": "market_data",
"skill": "market_data",
"params": {
"stock_code": stock_code,
"data_type": "quote"
}
}
# 无法识别
return {
"type": "unknown",
"skill": None,
"params": {}
}
def _map_intent_to_skill(self, intent_type: str) -> Optional[str]:
"""将意图类型映射到技能名称"""
mapping = {
"market_data": "market_data",
"technical_analysis": "technical_analysis",
"fundamental": "fundamental",
"visualization": "visualization"
}
return mapping.get(intent_type)
def _extract_stock_code(self, message: str) -> Optional[str]:
"""从消息中提取股票代码"""
# 匹配6位数字
pattern = r'\b\d{6}\b'
matches = re.findall(pattern, message)
if matches:
return matches[0]
# 使用股票名称数据库搜索
chinese_pattern = r'[\u4e00-\u9fa5]{2,6}'
chinese_words = re.findall(chinese_pattern, message)
for word in chinese_words:
code = search_stock_by_name(word)
if code:
logger.info(f"识别股票名称: {word} -> {code}")
return code
return None
async def _execute_intent(self, intent: Dict[str, Any], message: str) -> Dict[str, Any]:
"""执行意图对应的技能"""
if intent["type"] == "unknown":
return {
"success": False,
"error": "无法理解您的问题，请提供股票代码或明确的查询意图"
}
skill_name = intent["skill"]
params = intent["params"]
if not params.get("stock_code"):
return {
"success": False,
"error": "请提供股票代码或股票名称"
}
# 执行技能
result = await skill_manager.execute_skill(skill_name, **params)
return result
async def _generate_response(
self,
intent: Dict[str, Any],
result: Dict[str, Any],
stock_code: Optional[str]
) -> Dict[str, Any]:
"""生成响应消息（使用LLM增强）"""
if not result.get("success", True):
return {
"message": f"抱歉，{result.get('error', '处理失败')}",
"metadata": {"type": "error"}
}
data = result.get("data", result)
# 基础格式化
base_response = self._format_response_basic(intent, data)
# 如果启用LLM，添加智能分析
if self.use_llm and stock_code and intent["type"] == "technical_analysis":
try:
stock_name = get_stock_name(stock_code) or stock_code
llm_summary = llm_service.generate_analysis_summary(
stock_code, stock_name, data
)
base_response["message"] += f"\n\n【AI分析】\n{llm_summary}"
except Exception as e:
logger.error(f"LLM分析生成失败: {e}")
return base_response
def _format_response_basic(self, intent: Dict[str, Any], data: Dict[str, Any]) -> Dict[str, Any]:
"""基础响应格式化（原有逻辑）"""
if "error" in data:
return {
"message": f"查询失败：{data['error']}",
"metadata": {"type": "error"}
}
intent_type = intent["type"]
if intent_type == "market_data":
return self._format_market_data(data)
elif intent_type == "technical_analysis":
return self._format_technical(data)
elif intent_type == "fundamental":
return self._format_fundamental(data)
elif intent_type == "visualization":
return self._format_visualization(data)
else:
return {
"message": "查询完成",
"metadata": {"type": "data", "data": data}
}
def _format_market_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化行情数据"""
if "kline_data" in data:
kline_data = data["kline_data"]
message = f"已获取K线数据，共{len(kline_data)}条记录"
return {
"message": message,
"metadata": {"type": "kline", "data": kline_data}
}
message = f"""
【{data.get('name', '股票')}】({data.get('ts_code', '')})
交易日期：{data.get('trade_date', '')}
最新价：{data.get('close', 0):.2f}
涨跌额：{data.get('change', 0):.2f}
涨跌幅：{data.get('pct_chg', 0):.2f}%
开盘价：{data.get('open', 0):.2f}
最高价：{data.get('high', 0):.2f}
最低价：{data.get('low', 0):.2f}
成交量：{data.get('vol', 0):.0f}手
成交额：{data.get('amount', 0):.0f}千元
""".strip()
return {
"message": message,
"metadata": {"type": "quote", "data": data}
}
def _format_technical(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化技术分析"""
indicators = data.get("indicators", {})
message_parts = [f"【{data.get('stock_code', '')}】技术指标：\n"]
if "ma" in indicators:
ma = indicators["ma"]
message_parts.append(f"均线：MA5={ma.get('ma5')}, MA10={ma.get('ma10')}, MA20={ma.get('ma20')}")
if "macd" in indicators:
macd = indicators["macd"]
message_parts.append(f"MACD：DIF={macd.get('dif')}, DEA={macd.get('dea')}, MACD={macd.get('macd')}")
if "rsi" in indicators:
rsi = indicators["rsi"]
message_parts.append(f"RSI：RSI6={rsi.get('rsi6')}, RSI12={rsi.get('rsi12')}, RSI24={rsi.get('rsi24')}")
return {
"message": "\n".join(message_parts),
"metadata": {"type": "technical", "data": data}
}
def _format_fundamental(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化基本面"""
message = f"""
【{data.get('name', '股票')}】基本信息
股票代码：{data.get('ts_code', '')}
所属地域：{data.get('area', '')}
所属行业：{data.get('industry', '')}
上市市场：{data.get('market', '')}
上市日期：{data.get('list_date', '')}
""".strip()
return {
"message": message,
"metadata": {"type": "fundamental", "data": data}
}
def _format_visualization(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""格式化可视化"""
return {
"message": f"已生成{data.get('stock_code', '')}的K线图",
"metadata": {"type": "chart", "data": data}
}
# 创建全局Agent实例
enhanced_agent = EnhancedStockAgent()

47
backend/app/agent/question_analyzer.py
View File

@ -139,7 +139,7 @@ class QuestionAnalyzer:
请分析以下维度：
1. **问题类型**
- stock_analysis: 针对**特定单只股票**的分析（如"贵州茅台怎么样"、"分析比亚迪"、"AAPL走势"）
- stock_analysis: 针对**特定单只股票**的分析（如"贵州茅台怎么样"、"分析比亚迪"、"AAPL走势"、"阿里巴巴美股"）
**注意**：如果用户问的是"板块"、"行业"、"概念股"等，这不是stock_analysis，而是market_overview
- market_overview: 市场整体分析、行业板块分析、投资机会（如"最近有什么投资机会"、"商业航天板块怎么样"、"新能源行业走势"、"现在适合买股票吗"）
- knowledge: 金融知识问答（如"什么是MACD"、"如何看K线图"）
@ -150,7 +150,34 @@ class QuestionAnalyzer:
- 如果提到"板块"、"行业"、"概念"、"赛道"、"领域" → market_overview
- 如果问"哪些股票"、"什么机会" → market_overview
2. **用户关注维度**（如果是stock_analysis）
2. **股票识别**（如果是stock_analysis，这是最重要的部分）
请识别用户提到的股票，并返回准确的股票代码：
**A股代码格式**：6位数字
- 上海主板：600xxx、601xxx、603xxx、605xxx
- 深圳主板：000xxx、001xxx
- 创业板：300xxx、301xxx
- 科创板：688xxx
- 常见示例：贵州茅台→600519，比亚迪→002594，宁德时代→300750
**美股代码格式**：1-5位大写字母
- 常见示例：苹果→AAPL，特斯拉→TSLA，微软→MSFT，谷歌→GOOGL
- 中概股美股：阿里巴巴美股→BABA，京东美股→JD，拼多多→PDD，百度美股→BIDU，网易美股→NTES，哔哩哔哩美股→BILI
**港股代码格式**：4-5位数字加.HK后缀
- 常见示例：腾讯→0700.HK，阿里巴巴港股→9988.HK，美团→3690.HK，小米→1810.HK，京东港股→9618.HK，百度港股→9888.HK，网易港股→9999.HK，哔哩哔哩港股→9626.HK
- 注意：港股代码需要包含.HK后缀
**市场判断**：
- 如果用户明确说"美股"、"纳斯达克"、"纽交所" → 美股
- 如果用户明确说"港股"、"香港"、"恒生" → 港股
- 对于同时在多地上市的公司（如阿里巴巴、京东、百度等）：
- 用户说"美股"或没有明确指定 → 返回美股代码（如BABA）
- 用户说"港股" → 返回港股代码（如9988.HK）
- 纯港股公司（如腾讯、美团、小米）→ 港股
- 默认情况下，中国公司优先考虑A股市场
3. **用户关注维度**（如果是stock_analysis）
分析用户想了解哪些方面：
- price_trend: 价格走势、涨跌情况、最新价格
- technical: 技术指标（MACD、RSI、均线、KDJ等）
@ -159,28 +186,28 @@ class QuestionAnalyzer:
- money_flow: 资金流向、主力动向、大单流入流出
- risk: 风险分析、风险提示、投资风险
3. **时间范围**
4. **时间范围**
- short_term: 短期（1-2周）- 如"短期走势"、"近期表现"
- medium_term: 中期（1-3月）- 如"中期趋势"、"未来一个月"
- long_term: 长期（半年以上）- 如"长期投资"、"适合长期持有吗"
4. **分析深度**
5. **分析深度**
- quick: 快速查看（只需要基本信息，如"价格多少"）
- standard: 标准分析（常规分析，如"怎么样"、"分析一下"）
- deep: 深度分析（全面详细，如"全面分析"、"深度研究"）
5. **特定关注点**
6. **特定关注点**
提取用户明确提到的关注点，如：
- "支撑位在哪"
- "盈利能力如何"
- "适合长期持有吗"
- "有没有金叉"
6. **上下文引用**
7. **上下文引用**
- 是否引用了之前的对话（"这只股票"、"它"、"那技术面呢"）
- 是否要求对比分析（"和上次相比"、"对比一下"）
7. **用户风格**
8. **用户风格**
- tone: professional（专业，使用专业术语）/ casual（随意，通俗易懂）
- detail_level: brief（简洁，简短回答）/ detailed（详细，详细分析）
@ -188,9 +215,9 @@ class QuestionAnalyzer:
{{
"type": "问题类型",
"target": {{
"stock_code": "股票代码（如有，只返回纯数字代码，如600519或002594，不要包含市场标识）",
"stock_name": "股票名称（如有，只返回公司名称，如贵州茅台或比亚迪）",
"market": "A股/美股"
"stock_code": "股票代码（A股返回6位数字如600519，美股返回大写字母如BABA，港股返回带.HK后缀如0700.HK）",
"stock_name": "股票/公司名称（如贵州茅台、阿里巴巴、腾讯）",
"market": "A股/美股/港股"
}},
"dimensions": {{
"price_trend": true/false,

175
backend/app/agent/skill_planner.py
View File

@ -8,53 +8,85 @@ from app.utils.logger import logger
class SkillPlanner:
"""智能技能规划器 - 根据问题意图动态选择技能"""
# 维度到技能的映射
DIMENSION_SKILL_MAP = {
# A股维度到技能的映射
A_STOCK_DIMENSION_SKILL_MAP = {
'price_trend': {
'required': ['market_data', 'brave_search'], # brave_search 必需
'required': ['market_data', 'brave_search'],
'optional': []
},
'technical': {
'required': ['market_data', 'technical_analysis', 'brave_search'], # brave_search 必需
'required': ['market_data', 'technical_analysis', 'brave_search'],
'optional': ['visualization']
},
'fundamental': {
'required': ['fundamental', 'brave_search'], # brave_search 必需
'required': ['fundamental', 'brave_search'],
'optional': []
},
'valuation': {
'required': ['advanced_data', 'brave_search'], # brave_search 必需
'required': ['advanced_data', 'brave_search'],
'optional': []
},
'money_flow': {
'required': ['advanced_data', 'brave_search'], # brave_search 必需
'required': ['advanced_data', 'brave_search'],
'optional': []
},
'risk': {
'required': ['technical_analysis', 'advanced_data', 'brave_search'], # brave_search 必需
'required': ['technical_analysis', 'advanced_data', 'brave_search'],
'optional': []
},
'news': { # 新闻维度
'news': {
'required': ['brave_search'],
'optional': []
}
}
# 技能依赖关系
# 美股/港股维度到技能的映射（使用 yfinance）
INTL_STOCK_DIMENSION_SKILL_MAP = {
'price_trend': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'technical': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'fundamental': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'valuation': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'money_flow': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'risk': {
'required': ['us_stock_analysis', 'brave_search'],
'optional': []
},
'news': {
'required': ['brave_search'],
'optional': []
}
}
# 技能依赖关系（仅 A 股）
SKILL_DEPENDENCIES = {
'technical_analysis': ['market_data'], # 技术分析依赖行情数据
'visualization': ['market_data'], # 可视化依赖行情数据
'technical_analysis': ['market_data'],
'visualization': ['market_data'],
}
# 技能优先级（数字越小优先级越高）
SKILL_PRIORITY = {
'market_data': 1, # 最高优先级
'market_data': 1,
'fundamental': 1,
'brave_search': 1, # 新闻搜索也是高优先级
'brave_search': 1,
'us_stock_analysis': 1,
'technical_analysis': 2,
'advanced_data': 2,
'visualization': 3, # 最低优先级
'us_stock_analysis': 1
'visualization': 3,
}
# 分析深度策略
@ -70,7 +102,7 @@ class SkillPlanner:
'use_cache': True
},
'deep': {
'max_skills': None, # 无限制
'max_skills': None,
'include_optional': True,
'use_cache': False
}
@ -88,23 +120,27 @@ class SkillPlanner:
intent: 问题意图（来自QuestionAnalyzer）
Returns:
SkillExecutionPlan: {
'skills': [
{
'name': 'market_data',
'params': {...},
'priority': 1,
'required': True,
'reason': '用户关注价格走势'
},
...
],
'execution_strategy': 'parallel' | 'sequential',
'cache_strategy': 'use' | 'bypass'
}
SkillExecutionPlan
"""
# 获取市场类型
target = intent.get('target', {})
market = target.get('market', 'A股')
stock_code = target.get('stock_code', '')
stock_name = target.get('stock_name', '')
# 根据市场类型选择不同的技能映射
if market in ('美股', '港股'):
return self._plan_intl_stock_skills(intent, market, stock_code, stock_name)
else:
return self._plan_a_stock_skills(intent)
def _plan_a_stock_skills(self, intent: Dict[str, Any]) -> Dict[str, Any]:
"""规划 A 股技能"""
# 1. 根据维度映射技能
skills = self._map_dimensions_to_skills(intent.get('dimensions', {}))
skills = self._map_dimensions_to_skills(
intent.get('dimensions', {}),
self.A_STOCK_DIMENSION_SKILL_MAP
)
# 2. 根据分析深度调整
depth = intent.get('analysis_depth', 'standard')
@ -113,13 +149,11 @@ class SkillPlanner:
# 3. 解析依赖关系
skills = self._resolve_dependencies(skills)
# 4. 去重
# 4. 去重并排序
skills = list(set(skills))
# 5. 排序（按优先级）
sorted_skills = self._sort_by_priority(skills)
# 6. 构建执行计划
# 5. 构建执行计划
plan = {
'skills': [
{
@ -135,18 +169,77 @@ class SkillPlanner:
'cache_strategy': 'use' if self.DEPTH_STRATEGY[depth]['use_cache'] else 'bypass'
}
logger.info(f"技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
logger.info(f"[A股] 技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
return plan
def _map_dimensions_to_skills(self, dimensions: Dict[str, bool]) -> List[str]:
def _plan_intl_stock_skills(self, intent: Dict[str, Any], market: str, stock_code: str, stock_name: str) -> Dict[str, Any]:
"""规划美股/港股技能"""
# 1. 根据维度映射技能
skills = self._map_dimensions_to_skills(
intent.get('dimensions', {}),
self.INTL_STOCK_DIMENSION_SKILL_MAP
)
# 2. 确保至少有 us_stock_analysis
if 'us_stock_analysis' not in skills:
skills.append('us_stock_analysis')
# 3. 去重并排序
skills = list(set(skills))
sorted_skills = self._sort_by_priority(skills)
# 4. 构建执行计划
depth = intent.get('analysis_depth', 'standard')
plan = {
'skills': [
{
'name': skill,
'params': self._get_intl_skill_params(skill, stock_code, stock_name),
'priority': self.SKILL_PRIORITY.get(skill, 5),
'required': skill == 'us_stock_analysis',
'reason': self._get_intl_skill_reason(skill, market)
}
for skill in sorted_skills
],
'execution_strategy': 'parallel',
'cache_strategy': 'use' if self.DEPTH_STRATEGY[depth]['use_cache'] else 'bypass'
}
logger.info(f"[{market}] 技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
return plan
def _get_intl_skill_params(self, skill_name: str, stock_code: str, stock_name: str) -> Dict[str, Any]:
"""获取美股/港股技能参数"""
if skill_name == 'us_stock_analysis':
return {
'symbol': stock_code,
'analysis_type': 'comprehensive'
}
elif skill_name == 'brave_search':
return {
'query': f'{stock_name} 最新动态 财报',
'search_type': 'news',
'count': 5,
'freshness': 'pw'
}
return {}
def _get_intl_skill_reason(self, skill_name: str, market: str) -> str:
"""获取美股/港股技能调用原因"""
if skill_name == 'us_stock_analysis':
return f'获取{market}基础数据和技术指标'
elif skill_name == 'brave_search':
return '获取最新市场资讯和舆情'
return '提供分析数据'
def _map_dimensions_to_skills(self, dimensions: Dict[str, bool], skill_map: Dict) -> List[str]:
"""将用户关注维度映射到技能"""
skills = []
for dimension, enabled in dimensions.items():
if enabled and dimension in self.DIMENSION_SKILL_MAP:
mapping = self.DIMENSION_SKILL_MAP[dimension]
if enabled and dimension in skill_map:
mapping = skill_map[dimension]
skills.extend(mapping['required'])
# 默认也添加可选技能（特别是 brave_search）
skills.extend(mapping['optional'])
return skills

511
backend/app/agent/smart_agent.py
View File

@ -1393,42 +1393,11 @@ MA60：{f"{ma['ma60']:.2f}" if ma['ma60'] else '计算中'}
if keyword.isupper() and keyword.isalpha() and len(keyword) <= 5:
return True
# 常见美股公司名称映射
us_stock_names = {
"苹果": "AAPL",
"特斯拉": "TSLA",
"微软": "MSFT",
"谷歌": "GOOGL",
"亚马逊": "AMZN",
"Meta": "META",
"脸书": "META",
"英伟达": "NVDA",
"奈飞": "NFLX",
"网飞": "NFLX",
"迪士尼": "DIS",
"可口可乐": "KO",
"麦当劳": "MCD",
"星巴克": "SBUX",
"耐克": "NKE",
"波音": "BA",
"英特尔": "INTC",
"AMD": "AMD",
"高通": "QCOM",
"推特": "TWTR",
"优步": "UBER",
"Uber": "UBER",
"Airbnb": "ABNB",
"爱彼迎": "ABNB",
}
if keyword in us_stock_names:
return True
return False
def _get_us_stock_symbol(self, keyword: str) -> str:
"""
获取美股代码
获取美股代码（同步版本，用于兼容）
Args:
keyword: 股票关键词
@ -1440,51 +1409,45 @@ MA60：{f"{ma['ma60']:.2f}" if ma['ma60'] else '计算中'}
if keyword.isupper() and keyword.isalpha():
return keyword
# 中文名称映射
us_stock_names = {
"苹果": "AAPL",
"特斯拉": "TSLA",
"微软": "MSFT",
"谷歌": "GOOGL",
"亚马逊": "AMZN",
"Meta": "META",
"脸书": "META",
"英伟达": "NVDA",
"奈飞": "NFLX",
"网飞": "NFLX",
"迪士尼": "DIS",
"可口可乐": "KO",
"麦当劳": "MCD",
"星巴克": "SBUX",
"耐克": "NKE",
"波音": "BA",
"英特尔": "INTC",
"AMD": "AMD",
"高通": "QCOM",
"推特": "TWTR",
"优步": "UBER",
"Uber": "UBER",
"Airbnb": "ABNB",
"爱彼迎": "ABNB",
}
return us_stock_names.get(keyword, keyword.upper())
# 默认返回大写形式
return keyword.upper()
async def _handle_us_stock(self, keyword: str, message: str) -> Dict[str, Any]:
"""
处理美股查询（使用 skill_planner）
处理美股查询（兼容旧接口，内部调用 _handle_us_stock_with_code）
Args:
keyword: 股票关键词
keyword: 股票关键词（可能是代码或名称）
message: 用户消息
Returns:
分析结果
"""
# 获取美股代码
symbol = self._get_us_stock_symbol(keyword)
# 如果是大写字母，直接作为代码使用
if keyword.isupper() and keyword.isalpha() and len(keyword) <= 5:
return await self._handle_us_stock_with_code(keyword, keyword, message)
logger.info(f"处理美股查询: {keyword} -> {symbol}")
# 否则需要通过 QuestionAnalyzer 重新分析获取代码
# 这种情况理论上不应该发生，因为 QuestionAnalyzer 应该已经返回了代码
logger.warning(f"_handle_us_stock 收到非代码格式的关键词: {keyword}")
return {
"message": f"抱歉，无法识别美股 \"{keyword}\"。请直接输入美股代码（如 BABA、AAPL、TSLA）进行查询。",
"metadata": {"type": "error"}
}
async def _handle_us_stock_with_code(self, symbol: str, stock_name: str, message: str) -> Dict[str, Any]:
"""
处理美股查询（使用已知的股票代码）
Args:
symbol: 美股代码（如 BABA、AAPL）
stock_name: 股票名称
message: 用户消息
Returns:
分析结果
"""
logger.info(f"处理美股查询: {stock_name} -> {symbol}")
try:
# 1. 使用 QuestionAnalyzer 分析问题意图
@ -1498,28 +1461,17 @@ MA60：{f"{ma['ma60']:.2f}" if ma['ma60'] else '计算中'}
if 'target' not in intent:
intent['target'] = {}
intent['target']['stock_code'] = symbol
intent['target']['stock_name'] = keyword
intent['target']['stock_name'] = stock_name
intent['target']['market'] = '美股'
logger.info(f"美股问题意图分析: dimensions={intent.get('dimensions')}")
# 2. 使用 SkillPlanner 规划技能（包括 brave_search）
# 2. 使用 SkillPlanner 规划技能（会自动识别美股并使用正确的技能）
plan = self.skill_planner.plan_skills(intent)
# 3. 将 us_stock_analysis 添加到技能列表（如果不存在）
skill_names = [s['name'] for s in plan['skills']]
if 'us_stock_analysis' not in skill_names:
plan['skills'].insert(0, {
'name': 'us_stock_analysis',
'params': {'symbol': symbol, 'analysis_type': 'comprehensive'},
'priority': 1,
'required': True,
'reason': '获取美股基础数据'
})
logger.info(f"美股技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
# 4. 执行技能规划
# 3. 执行技能规划
execution_results = await skill_manager.execute_plan(
plan=plan,
stock_code=symbol
@ -1539,7 +1491,7 @@ MA60：{f"{ma['ma60']:.2f}" if ma['ma60'] else '计算中'}
# 6. 整合数据
all_data = {
"symbol": symbol,
"name": keyword,
"name": stock_name,
**us_stock_data,
"news": execution_results['results'].get("brave_search") # 新增：新闻数据
}
@ -1566,6 +1518,162 @@ MA60：{f"{ma['ma60']:.2f}" if ma['ma60'] else '计算中'}
"metadata": {"type": "error"}
}
async def _handle_hk_stock_with_code(self, symbol: str, stock_name: str, message: str) -> Dict[str, Any]:
"""
处理港股查询（使用已知的股票代码）
Args:
symbol: 港股代码（如 0700.HK、9988.HK）
stock_name: 股票名称
message: 用户消息
Returns:
分析结果
"""
logger.info(f"处理港股查询: {stock_name} -> {symbol}")
try:
# 1. 使用 QuestionAnalyzer 分析问题意图
intent = await self.question_analyzer.analyze_question(
question=message,
context=[],
session_id=""
)
# 确保 intent 包含股票信息
if 'target' not in intent:
intent['target'] = {}
intent['target']['stock_code'] = symbol
intent['target']['stock_name'] = stock_name
intent['target']['market'] = '港股'
logger.info(f"港股问题意图分析: dimensions={intent.get('dimensions')}")
# 2. 使用 SkillPlanner 规划技能（会自动识别港股并使用正确的技能）
plan = self.skill_planner.plan_skills(intent)
logger.info(f"港股技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
# 3. 执行技能规划
execution_results = await skill_manager.execute_plan(
plan=plan,
stock_code=symbol
)
if execution_results['errors']:
logger.warning(f"港股技能执行有错误: {execution_results['errors']}")
# 5. 检查 us_stock_analysis 是否成功
hk_stock_data = execution_results['results'].get('us_stock_analysis')
if not hk_stock_data or 'error' in hk_stock_data:
return {
"message": f"抱歉，未找到港股 {symbol}。请确认股票代码是否正确。\n\n提示：港股代码格式为数字加.HK后缀，如 0700.HK（腾讯）、9988.HK（阿里巴巴）等。",
"metadata": {"type": "error"}
}
# 6. 整合数据
all_data = {
"symbol": symbol,
"name": stock_name,
"market": "港股",
**hk_stock_data,
"news": execution_results['results'].get("brave_search")
}
# 7. 使用LLM分析港股数据
if self.use_llm:
analysis = await self._llm_hk_stock_analysis(all_data, message)
else:
analysis = self._format_us_stock_data(all_data)
return {
"message": analysis,
"metadata": {
"type": "hk_stock_analysis",
"data": all_data,
"plan": plan
}
}
except Exception as e:
logger.error(f"港股查询失败: {e}")
return {
"message": f"查询港股 {symbol} 时出错：{str(e)}",
"metadata": {"type": "error"}
}
async def _llm_hk_stock_analysis(self, data: Dict[str, Any], user_message: str) -> str:
"""使用LLM分析港股数据"""
from datetime import datetime
current_time = datetime.now().strftime("%Y-%m-%d %H:%M")
# 提取关键数据（使用 us_stock_service 返回的扁平结构）
symbol = data.get('symbol', '')
name = data.get('name', symbol)
technical = data.get('technical_indicators', {})
news = data.get('news', [])
# 构建数据摘要
data_summary = f"""
【港股数据】{name}（{symbol}）
查询时间：{current_time}
【基本信息】
- 公司名称：{name}
- 行业：{data.get('industry', '未知')}
- 板块：{data.get('sector', '未知')}
- 市值：{data.get('market_cap', '未知')}
【最新行情】
- 当前价格：{data.get('current_price', '未知')}
- 今日涨跌：{data.get('change_percent', '未知')}%
- 52周最高：{data.get('52_week_high', '未知')}
- 52周最低：{data.get('52_week_low', '未知')}
【估值指标】
- 市盈率(PE)：{data.get('pe_ratio', '未知')}
- 市净率(PB)：{data.get('pb_ratio', '未知')}
- 股息率：{data.get('dividend_yield', '未知')}
【技术指标】
- MA5：{technical.get('ma5', '未知')}
- MA20：{technical.get('ma20', '未知')}
- RSI：{technical.get('rsi', '未知')}
- MACD：{technical.get('macd', '未知')}
"""
# 添加新闻摘要
if news:
data_summary += "\n【相关新闻】\n"
for i, item in enumerate(news[:3], 1):
if isinstance(item, dict):
title = item.get('title', '')
data_summary += f"{i}. {title}\n"
prompt = f"""你是一个专业的港股分析师。请根据以下数据，回答用户的问题。
{data_summary}
用户问题：{user_message}
请提供专业、客观的分析，包括：
1. 直接回答用户的问题
2. 基于数据的分析和判断
3. 潜在的风险提示
注意：港股以港币计价，交易时间为港交所交易时段。"""
try:
result = await self._call_llm_async(
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=2000
)
return result or self._format_us_stock_data(data)
except Exception as e:
logger.error(f"LLM港股分析失败: {e}")
return self._format_us_stock_data(data)
async def _llm_us_stock_analysis(self, data: Dict[str, Any], user_message: str) -> str:
"""使用LLM分析美股数据"""
from datetime import datetime
@ -1911,9 +2019,20 @@ RSI：{technical.get('rsi', 0):.2f if technical.get('rsi') else '计算中'}
yield f"分析{stock_name}时出错：{str(e)}"
async def _handle_us_stock_stream(self, keyword: str, message: str):
"""流式处理美股分析（使用 skill_planner）"""
symbol = self._get_us_stock_symbol(keyword)
logger.info(f"[智能模式-流式] 美股查询: {keyword} -> {symbol}")
"""流式处理美股分析（兼容旧接口）"""
# 如果是大写字母，直接作为代码使用
if keyword.isupper() and keyword.isalpha() and len(keyword) <= 5:
async for chunk in self._handle_us_stock_stream_with_code(keyword, keyword, message):
yield chunk
return
# 否则报错
logger.warning(f"_handle_us_stock_stream 收到非代码格式的关键词: {keyword}")
yield f"抱歉，无法识别美股 \"{keyword}\"。请直接输入美股代码（如 BABA、AAPL、TSLA）进行查询。"
async def _handle_us_stock_stream_with_code(self, symbol: str, stock_name: str, message: str):
"""流式处理美股分析（使用已知的股票代码）"""
logger.info(f"[智能模式-流式] 美股查询: {stock_name} -> {symbol}")
try:
# 1. 使用 QuestionAnalyzer 分析问题意图
@ -1927,28 +2046,17 @@ RSI：{technical.get('rsi', 0):.2f if technical.get('rsi') else '计算中'}
if 'target' not in intent:
intent['target'] = {}
intent['target']['stock_code'] = symbol
intent['target']['stock_name'] = keyword
intent['target']['stock_name'] = stock_name
intent['target']['market'] = '美股'
logger.info(f"[流式] 美股问题意图分析: dimensions={intent.get('dimensions')}")
# 2. 使用 SkillPlanner 规划技能（包括 brave_search）
# 2. 使用 SkillPlanner 规划技能（会自动识别美股并使用正确的技能）
plan = self.skill_planner.plan_skills(intent)
# 3. 将 us_stock_analysis 添加到技能列表（如果不存在）
skill_names = [s['name'] for s in plan['skills']]
if 'us_stock_analysis' not in skill_names:
plan['skills'].insert(0, {
'name': 'us_stock_analysis',
'params': {'symbol': symbol, 'analysis_type': 'comprehensive'},
'priority': 1,
'required': True,
'reason': '获取美股基础数据'
})
logger.info(f"[流式] 美股技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
# 4. 执行技能规划
# 3. 执行技能规划
execution_results = await skill_manager.execute_plan(
plan=plan,
stock_code=symbol
@ -1966,7 +2074,7 @@ RSI：{technical.get('rsi', 0):.2f if technical.get('rsi') else '计算中'}
# 6. 整合数据
all_data = {
"symbol": symbol,
"name": keyword,
"name": stock_name,
**us_stock_data,
"news": execution_results['results'].get("brave_search") # 新增：新闻数据
}
@ -1994,6 +2102,147 @@ RSI：{technical.get('rsi', 0):.2f if technical.get('rsi') else '计算中'}
logger.error(traceback.format_exc())
yield f"查询美股 {symbol} 时出错：{str(e)}"
async def _handle_hk_stock_stream_with_code(self, symbol: str, stock_name: str, message: str):
"""流式处理港股分析（使用已知的股票代码）"""
logger.info(f"[智能模式-流式] 港股查询: {stock_name} -> {symbol}")
try:
# 1. 使用 QuestionAnalyzer 分析问题意图
intent = await self.question_analyzer.analyze_question(
question=message,
context=[],
session_id=""
)
# 确保 intent 包含股票信息
if 'target' not in intent:
intent['target'] = {}
intent['target']['stock_code'] = symbol
intent['target']['stock_name'] = stock_name
intent['target']['market'] = '港股'
logger.info(f"[流式] 港股问题意图分析: dimensions={intent.get('dimensions')}")
# 2. 使用 SkillPlanner 规划技能（会自动识别港股并使用正确的技能）
plan = self.skill_planner.plan_skills(intent)
logger.info(f"[流式] 港股技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
logger.info(f"[流式] 港股技能规划完成: {[s['name'] for s in plan['skills']]}, 策略: {plan['execution_strategy']}")
# 4. 执行技能规划
execution_results = await skill_manager.execute_plan(
plan=plan,
stock_code=symbol
)
if execution_results['errors']:
logger.warning(f"[流式] 港股技能执行有错误: {execution_results['errors']}")
# 5. 检查 us_stock_analysis 是否成功
hk_stock_data = execution_results['results'].get('us_stock_analysis')
if not hk_stock_data or 'error' in hk_stock_data:
yield f"抱歉，未找到港股 {symbol}。请确认股票代码是否正确。\n\n提示：港股代码格式为数字加.HK后缀，如 0700.HK（腾讯）、9988.HK（阿里巴巴）等。"
return
# 6. 整合数据
all_data = {
"symbol": symbol,
"name": stock_name,
"market": "港股",
**hk_stock_data,
"news": execution_results['results'].get("brave_search")
}
# 7. 使用智能模式的动态prompt生成
if self.use_llm:
# 构建港股数据的动态prompt
prompt = self._build_hk_stock_dynamic_prompt(all_data, symbol, message)
# 流式生成
stream = llm_service.chat_stream(
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=2500
)
for chunk in stream:
yield chunk
else:
yield self._format_us_stock_data(all_data)
except Exception as e:
logger.error(f"港股查询失败: {e}")
import traceback
logger.error(traceback.format_exc())
yield f"查询港股 {symbol} 时出错：{str(e)}"
def _build_hk_stock_dynamic_prompt(self, data: Dict[str, Any], symbol: str, user_message: str) -> str:
"""构建港股分析的动态prompt"""
from datetime import datetime
current_time = datetime.now().strftime("%Y-%m-%d %H:%M")
# 提取关键数据（使用 us_stock_service 返回的扁平结构）
name = data.get('name', symbol)
technical = data.get('technical_indicators', {})
news = data.get('news', [])
# 构建数据摘要
data_summary = f"""
【港股数据】{name}（{symbol}）
查询时间：{current_time}
【基本信息】
- 公司名称：{name}
- 行业：{data.get('industry', '未知')}
- 板块：{data.get('sector', '未知')}
- 市值：{data.get('market_cap', '未知')}
【最新行情】
- 当前价格：{data.get('current_price', '未知')}
- 今日涨跌：{data.get('change_percent', '未知')}%
- 52周最高：{data.get('52_week_high', '未知')}
- 52周最低：{data.get('52_week_low', '未知')}
【估值指标】
- 市盈率(PE)：{data.get('pe_ratio', '未知')}
- 市净率(PB)：{data.get('pb_ratio', '未知')}
- 股息率：{data.get('dividend_yield', '未知')}
【技术指标】
- MA5：{technical.get('ma5', '未知')}
- MA20：{technical.get('ma20', '未知')}
- RSI：{technical.get('rsi', '未知')}
- MACD：{technical.get('macd', '未知')}
"""
# 添加新闻摘要
if news:
data_summary += "\n【相关新闻】\n"
if isinstance(news, dict) and 'results' in news:
news_list = news.get('results', [])
else:
news_list = news if isinstance(news, list) else []
for i, item in enumerate(news_list[:3], 1):
if isinstance(item, dict):
title = item.get('title', '')
data_summary += f"{i}. {title}\n"
prompt = f"""你是一个专业的港股分析师。请根据以下数据，回答用户的问题。
{data_summary}
用户问题：{user_message}
请提供专业、客观的分析，包括：
1. 直接回答用户的问题
2. 基于数据的分析和判断
3. 潜在的风险提示
注意：港股以港币(HKD)计价，交易时间为港交所交易时段（北京时间9:30-12:00, 13:00-16:00）。"""
return prompt
def _format_news_section(self, news_data: Dict[str, Any]) -> str:
"""
格式化新闻数据为统一格式
@ -2364,31 +2613,22 @@ MACD：{f"{technical.get('macd'):.4f}" if technical.get('macd') else '计算中'
stock_name = target.get('stock_name')
market = target.get('market', 'A股')
# 如果没有股票代码，尝试匹配
if not stock_code and stock_name:
# 检测是否为美股
is_us_stock = self._is_us_stock(stock_name, market)
if is_us_stock:
return await self._handle_us_stock(stock_name, message)
# A股匹配
stock_info = await self._match_stock_with_llm(stock_name)
if not stock_info:
return {
"message": f"抱歉，未找到股票\"{stock_name}\"。请确认名称或代码是否正确。",
"metadata": {"type": "error"}
}
stock_code = stock_info['code']
stock_name = stock_info['name']
# QuestionAnalyzer 应该已经返回了股票代码
if not stock_code:
return {
"message": "抱歉，我没有识别到您提到的股票。请提供更明确的股票代码或名称。",
"message": f"抱歉，我没有识别到您提到的股票「{stock_name or ''}」。请提供更明确的股票代码或名称。",
"metadata": {"type": "error"}
}
# 根据市场类型处理
if market == '美股':
# 美股处理
return await self._handle_us_stock_with_code(stock_code, stock_name or stock_code, message)
if market == '港股':
# 港股处理（使用 yfinance，与美股类似）
return await self._handle_hk_stock_with_code(stock_code, stock_name or stock_code, message)
logger.info(f"[智能模式] 分析股票: {stock_name}({stock_code})")
# 1. 技能规划
@ -2665,30 +2905,25 @@ MACD：{f"{technical.get('macd'):.4f}" if technical.get('macd') else '计算中'
stock_name = target.get('stock_name')
market = target.get('market', 'A股')
# 检测是否为美股
is_us_stock = market == '美股' or self._is_us_stock(stock_name or stock_code or '', market)
# QuestionAnalyzer 应该已经返回了股票代码
if not stock_code:
yield f"抱歉，我没有识别到您提到的股票「{stock_name or ''}」。请提供更明确的股票代码或名称。"
return
# 如果是美股，直接使用美股处理流程
if is_us_stock:
async for chunk in self._handle_us_stock_stream(stock_name or stock_code, message):
# 根据市场类型处理
if market == '美股':
# 美股处理流程
async for chunk in self._handle_us_stock_stream_with_code(stock_code, stock_name or stock_code, message):
yield chunk
return
if market == '港股':
# 港股处理流程（使用 yfinance，与美股类似）
async for chunk in self._handle_hk_stock_stream_with_code(stock_code, stock_name or stock_code, message):
yield chunk
return
# A股处理流程
# 如果没有股票代码，尝试匹配
if not stock_code and stock_name:
stock_info = await self._match_stock_with_llm(stock_name)
if not stock_info:
yield f"抱歉，未找到股票\"{stock_name}\"。请确认名称或代码是否正确。"
return
stock_code = stock_info['code']
stock_name = stock_info['name']
if not stock_code:
yield "抱歉，我没有识别到您提到的股票。请提供更明确的股票代码或名称。"
return
logger.info(f"[智能模式-流式] 分析股票: {stock_name}({stock_code})")
# 1. 技能规划

2126
backend/app/agent/smart_agent.py.bak
View File

File diff suppressed because it is too large Load Diff

10
backend/app/skills/us_stock_skill.py
View File

@ -1,5 +1,5 @@
"""
美股分析技能
美股/港股分析技能
"""
from typing import Dict, Any
from app.skills.base import BaseSkill, SkillParameter
@ -8,17 +8,17 @@ from app.utils.logger import logger
class USStockSkill(BaseSkill):
"""美股分析技能"""
"""美股/港股分析技能（使用 yfinance）"""
def __init__(self):
super().__init__()
self.name = "us_stock_analysis"
self.description = "分析美股（如 AAPL, TSLA, MSFT 等），获取实时行情、技术指标、基本面数据"
self.description = "分析美股（如 AAPL, TSLA）和港股（如 0700.HK, 9988.HK），获取实时行情、技术指标、基本面数据"
self.parameters = [
SkillParameter(
name="symbol",
type="string",
description="美股代码（如 AAPL, TSLA, MSFT）",
description="股票代码（美股如 AAPL, TSLA；港股如 0700.HK, 9988.HK）",
required=True
),
SkillParameter(
@ -51,7 +51,7 @@ class USStockSkill(BaseSkill):
"error": "请提供美股代码"
}
logger.info(f"开始分析美股: {symbol}, 类型: {analysis_type}")
logger.info(f"开始分析股票: {symbol}, 类型: {analysis_type}")
if analysis_type == "basic":
# 基本信息

254
backend/app/utils/stock_names.py
View File

@ -1,254 +0,0 @@
"""
股票名称映射数据库
包含常见A股股票的名称到代码的映射
"""
from typing import Optional
# 常见A股股票名称映射（按行业分类）
STOCK_NAME_MAP = {
# 白酒
"贵州茅台": "600519",
"茅台": "600519",
"五粮液": "000858",
"泸州老窖": "000568",
"山西汾酒": "600809",
"洋河股份": "002304",
# 银行
"工商银行": "601398",
"工行": "601398",
"建设银行": "601939",
"建行": "601939",
"农业银行": "601288",
"农行": "601288",
"中国银行": "601988",
"中行": "601988",
"交通银行": "601328",
"交行": "601328",
"招商银行": "600036",
"招行": "600036",
"兴业银行": "601166",
"浦发银行": "600000",
"民生银行": "600016",
"光大银行": "601818",
"平安银行": "000001",
"宁波银行": "002142",
# 保险
"中国平安": "601318",
"平安": "601318",
"中国人寿": "601628",
"中国太保": "601601",
"新华保险": "601336",
# 证券
"中信证券": "600030",
"中信": "600030",
"海通证券": "600837",
"国泰君安": "601211",
"华泰证券": "601688",
"广发证券": "000776",
"招商证券": "600999",
"东方证券": "600958",
# 科技
"中兴通讯": "000063",
"中兴": "000063",
"立讯精密": "002475",
"京东方A": "000725",
"京东方": "000725",
"TCL科技": "000100",
"海康威视": "002415",
"大华股份": "002236",
"科大讯飞": "002230",
"讯飞": "002230",
"紫光国微": "002049",
"中芯国际": "688981",
"韦尔股份": "603501",
# 新能源汽车
"比亚迪": "002594",
"宁德时代": "300750",
"宁德": "300750",
"长城汽车": "601633",
"长城": "601633",
"上汽集团": "600104",
"上汽": "600104",
"广汽集团": "601238",
"广汽": "601238",
"吉利汽车": "00175", # 港股
"理想汽车": "02015", # 港股
"小鹏汽车": "09868", # 港股
"蔚来": "09866", # 港股
# 医药
"恒瑞医药": "600276",
"恒瑞": "600276",
"药明康德": "603259",
"迈瑞医疗": "300760",
"迈瑞": "300760",
"片仔癀": "600436",
"云南白药": "000538",
"白药": "000538",
"爱尔眼科": "300015",
"智飞生物": "300122",
# 消费
"伊利股份": "600887",
"伊利": "600887",
"海天味业": "603288",
"海天": "603288",
"格力电器": "000651",
"格力": "000651",
"美的集团": "000333",
"美的": "000333",
"海尔智家": "600690",
"海尔": "600690",
"老板电器": "002508",
# 地产
"万科A": "000002",
"万科": "000002",
"保利发展": "600048",
"保利": "600048",
"招商蛇口": "001979",
"金地集团": "600383",
"金地": "600383",
# 能源
"中国石油": "601857",
"中石油": "601857",
"中国石化": "600028",
"中石化": "600028",
"中国神华": "601088",
"神华": "601088",
"陕西煤业": "601225",
"长江电力": "600900",
"三峡能源": "600905",
# 通信
"中国移动": "600941",
"移动": "600941",
"中国电信": "601728",
"电信": "601728",
"中国联通": "600050",
"联通": "600050",
"中国卫通": "601698",
"卫通": "601698",
# 航空航天
"中国国航": "601111",
"国航": "601111",
"南方航空": "600029",
"南航": "600029",
"东方航空": "600115",
"东航": "600115",
"中国卫星": "600118",
"航天科技": "000901",
# 钢铁
"宝钢股份": "600019",
"宝钢": "600019",
"河钢股份": "000709",
"河钢": "000709",
"鞍钢股份": "000898",
"鞍钢": "000898",
# 有色金属
"紫金矿业": "601899",
"紫金": "601899",
"中国铝业": "601600",
"中铝": "601600",
"江西铜业": "600362",
"江铜": "600362",
"洛阳钼业": "603993",
# 化工
"万华化学": "600309",
"万华": "600309",
"华鲁恒升": "600426",
"恒力石化": "600346",
"荣盛石化": "002493",
# 电力设备
"隆基绿能": "601012",
"隆基": "601012",
"阳光电源": "300274",
"通威股份": "600438",
"通威": "600438",
"特变电工": "600089",
# 军工
"中航沈飞": "600760",
"沈飞": "600760",
"中航西飞": "000768",
"西飞": "000768",
"中国船舶": "600150",
"中船": "600150",
"航发动力": "600893",
"航天发展": "000547",
# 互联网
"腾讯控股": "00700", # 港股
"腾讯": "00700",
"阿里巴巴": "09988", # 港股
"阿里": "09988",
"美团": "03690", # 港股
"京东": "09618", # 港股
"拼多多": "PDD", # 美股
"百度": "09888", # 港股
"网易": "09999", # 港股
"小米集团": "01810", # 港股
"小米": "01810",
# 指数
"上证指数": "000001",
"上证": "000001",
"沪指": "000001",
"深证成指": "399001",
"深成指": "399001",
"创业板指": "399006",
"创业板": "399006",
"科创50": "000688",
"沪深300": "000300",
"中证500": "000905",
"中证1000": "000852",
}
def search_stock_by_name(name: str) -> Optional[str]:
"""
根据股票名称搜索代码
Args:
name: 股票名称或简称
Returns:
股票代码或None
"""
# 精确匹配
if name in STOCK_NAME_MAP:
return STOCK_NAME_MAP[name]
# 模糊匹配（包含关系）
for stock_name, code in STOCK_NAME_MAP.items():
if name in stock_name or stock_name in name:
return code
return None
def get_stock_name(code: str) -> Optional[str]:
"""
根据代码获取股票名称
Args:
code: 股票代码
Returns:
股票名称或None
"""
for name, stock_code in STOCK_NAME_MAP.items():
if stock_code == code:
return name
return None

25
backend/run.sh
View File

@ -34,14 +34,11 @@ try:
from app.services.tushare_service import tushare_service
print(" ✓ Tushare服务")
from app.utils.stock_names import search_stock_by_name
print(" ✓ 股票名称库")
from app.services.llm_service import llm_service
print(" ✓ LLM服务")
from app.agent.enhanced_agent import enhanced_agent
print(" ✓ 增强版Agent")
from app.agent.smart_agent import smart_agent
print(" ✓ 智能Agent")
print("\n所有模块导入成功！")
@ -76,24 +73,6 @@ if not settings.zhipuai_api_key:
print("⚠️ 警告: 智谱AI Key未配置，将使用规则模式（无AI分析）")
EOF
echo ""
echo "3. 测试股票名称识别..."
python3 << 'EOF'
from app.utils.stock_names import search_stock_by_name
test_cases = [
("中国卫通", "601698"),
("贵州茅台", "600519"),
("比亚迪", "002594"),
("宁德时代", "300750")
]
for name, expected in test_cases:
result = search_stock_by_name(name)
status = "✓" if result == expected else "❌"
print(f" {status} {name} -> {result}")
EOF
echo ""
echo "================================"
echo "检查完成！准备启动..."

2
backend/test_import.sh
View File

@ -20,7 +20,7 @@ python3 -c "from app.services.cache_service import cache_service; print('✓ 缓
echo ""
echo "测试Agent..."
python3 -c "from app.agent.core import stock_agent; print('✓ Agent初始化成功')"
python3 -c "from app.agent.smart_agent import smart_agent; print('✓ Agent初始化成功')"
echo ""
echo "所有测试通过！可以启动应用了。"

39
frontend/css/style.css
View File

@ -415,6 +415,45 @@ html, body {
color: var(--text-tertiary);
}
/* Share Image Question Section */
.share-image-question {
margin-bottom: 20px;
padding: 16px 20px;
background: rgba(0, 255, 65, 0.05);
border: 1px solid rgba(0, 255, 65, 0.2);
border-radius: 6px;
}
.share-image-question-label {
font-size: 12px;
font-weight: 500;
color: var(--accent);
margin-bottom: 8px;
text-transform: uppercase;
letter-spacing: 1px;
}
.share-image-question-text {
font-size: 16px;
font-weight: 500;
color: var(--text-primary);
line-height: 1.5;
}
/* Share Image Answer Section */
.share-image-answer {
margin-bottom: 0;
}
.share-image-answer-label {
font-size: 12px;
font-weight: 500;
color: var(--text-secondary);
margin-bottom: 12px;
text-transform: uppercase;
letter-spacing: 1px;
}
/* Markdown Styles */
.markdown h1,
.markdown h2,

9
frontend/index.html
View File

@ -62,7 +62,7 @@
</svg>
</div>
<h1>AI 金融智能体</h1>
<p class="welcome-subtitle">支持 A股 + 美股双市场分析</p>
<p class="welcome-subtitle">支持 A股 · 美股 · 港股 三大市场分析</p>
<div class="guide-section">
<div class="example-queries">
@ -70,12 +70,13 @@
<button class="example-btn" @click="sendExample('比亚迪怎么样')">比亚迪怎么样</button>
<button class="example-btn" @click="sendExample('分析特斯拉')">分析特斯拉</button>
<button class="example-btn" @click="sendExample('苹果股票')">苹果股票</button>
<button class="example-btn" @click="sendExample('英伟达股票怎么样')">英伟达股票怎么样</button>
<button class="example-btn" @click="sendExample('港股腾讯')">港股腾讯</button>
<button class="example-btn" @click="sendExample('分析小米')">分析小米</button>
</div>
</div>
<div class="welcome-footer">
<p>💬 输入股票名称或代码开始分析</p>
<p>💬 输入股票名称或代码，支持 A股/美股/港股</p>
</div>
</div>
@ -130,7 +131,7 @@
<textarea
v-model="userInput"
@keydown.enter.exact.prevent="sendMessage"
placeholder="输入股票名称或代码..."
placeholder="输入股票名称或代码（A股/美股/港股）..."
rows="1"
:disabled="loading"
ref="textarea"

19
frontend/js/app.js
View File

@ -369,12 +369,18 @@ createApp({
try {
this.showNotification('正在生成分享图...');
// 获取用户提问（前一条消息）
let userQuestion = '';
if (index > 0 && this.messages[index - 1].role === 'user') {
userQuestion = this.messages[index - 1].content;
}
// 创建临时容器
const container = document.createElement('div');
container.className = 'share-image-container';
container.style.left = '-9999px';
// 构建分享图内容
// 构建分享图内容（包含用户提问和AI回答）
container.innerHTML = `
<div class="share-image-header">
<svg width="24" height="24" viewBox="0 0 24 24" fill="none">
@ -382,7 +388,16 @@ createApp({
</svg>
<div class="share-image-logo">Tradus｜AI 金融智能体</div>
</div>
<div class="share-image-content">${marked.parse(content)}</div>
${userQuestion ? `
<div class="share-image-question">
<div class="share-image-question-label">提问</div>
<div class="share-image-question-text">${userQuestion}</div>
</div>
` : ''}
<div class="share-image-answer">
<div class="share-image-answer-label">AI 分析</div>
<div class="share-image-content">${marked.parse(content)}</div>
</div>
<div class="share-image-footer">
由 AI 智能分析生成 | 仅供参考，不构成投资建议
</div>