模块5：工具集成与API适配方案
1. 现状分析
1.1 当前工具集成现状
# 当前LangGraph工具定义示例
from langchain.tools import tool
from typing import Dict, Any, Optional
import yfinance as yf
import pandas as pd
@tool
def get_stock_fundamentals(symbol: str, market: str = "us") -> Dict[str, Any]:
"""获取股票基本面数据"""
try:
if market == "us":
stock = yf.Ticker(symbol)
info = stock.info
return {
"pe_ratio": info.get("trailingPE"),
"pb_ratio": info.get("priceToBook"),
"roe": info.get("returnOnEquity"),
"debt_ratio": info.get("debtToEquity"),
"revenue_growth": info.get("revenueGrowth"),
"market_cap": info.get("marketCap"),
"dividend_yield": info.get("dividendYield"),
"beta": info.get("beta"),
"error": None
}
elif market == "cn":
# 使用tushare获取A股数据
import tushare as ts
pro = ts.pro_api()
# 获取基本财务数据
df = pro.stock_company(ts_code=f"{symbol}.SZ" if symbol.startswith("0") else f"{symbol}.SS")
if not df.empty:
company_info = df.iloc[0]
return {
"company_name": company_info.get("fullname"),
"industry": company_info.get("industry"),
"business": company_info.get("business"),
"error": None
}
return {"error": "未找到公司信息"}
elif market == "hk":
# 使用yfinance获取港股数据
symbol_hk = f"{symbol}.HK"
stock = yf.Ticker(symbol_hk)
info = stock.info
return {
"pe_ratio": info.get("trailingPE"),
"pb_ratio": info.get("priceToBook"),
"market_cap": info.get("marketCap"),
"error": None
}
else:
return {"error": f"不支持的市场: {market}"}
except Exception as e:
return {
"error": f"获取基本面数据失败: {str(e)}"
}
@tool
def get_market_data(symbol: str, period: str = "1y") -> Dict[str, Any]:
"""获取市场数据"""
try:
stock = yf.Ticker(symbol)
hist = stock.history(period=period)
if hist.empty:
return {"error": "未找到历史数据"}
current_price = hist['Close'].iloc[-1]
price_change = hist['Close'].pct_change().iloc[-1]
volume = hist['Volume'].iloc[-1]
# 计算技术指标
hist['MA20'] = hist['Close'].rolling(window=20).mean()
hist['MA50'] = hist['Close'].rolling(window=50).mean()
current_ma20 = hist['MA20'].iloc[-1]
current_ma50 = hist['MA50'].iloc[-1]
return {
"current_price": float(current_price),
"price_change": float(price_change),
"volume": int(volume),
"ma20": float(current_ma20),
"ma50": float(current_ma50),
"trend": "bullish" if current_price > current_ma20 > current_ma50 else "bearish",
"error": None
}
except Exception as e:
return {"error": f"获取市场数据失败: {str(e)}"}
@tool
def get_news_sentiment(symbol: str, limit: int = 10) -> Dict[str, Any]:
"""获取新闻情感分析"""
try:
# 这里使用模拟数据，实际应该调用新闻API
import random
news_items = []
for i in range(limit):
sentiment = random.choice(["positive", "negative", "neutral"])
score = random.uniform(-1, 1)
news_items.append({
"title": f"新闻标题 {i+1}",
"content": f"这是关于{symbol}的新闻内容 {i+1}",
"sentiment": sentiment,
"score": score,
"timestamp": datetime.now().isoformat()
})
# 计算整体情感
avg_sentiment = sum(item["score"] for item in news_items) / len(news_items)
overall_sentiment = "positive" if avg_sentiment > 0.1 else "negative" if avg_sentiment < -0.1 else "neutral"
return {
"news_items": news_items,
"overall_sentiment": overall_sentiment,
"average_score": avg_sentiment,
"error": None
}
except Exception as e:
return {"error": f"获取新闻情感失败: {str(e)}"}
1.2 当前工具使用模式
# LangGraph中的工具调用
from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
# 定义工具列表
tools = [get_stock_fundamentals, get_market_data, get_news_sentiment]
# 创建代理
agent = create_react_agent(
llm=llm,
tools=tools,
prompt=prompt
)
# 执行工具调用
result = agent.invoke({
"input": "分析AAPL股票",
"stock_symbol": "AAPL",
"market": "us"
})
1.3 当前工具集成特点
1. 装饰器定义：使用@tool装饰器定义工具
2. 同步执行：工具函数为同步执行
3. 简单错误处理：使用字典返回错误信息
4. 无缓存机制：每次调用都重新执行
5. 无重试机制：失败时直接返回错误
2. Agno工具集成架构设计
2.1 Agno工具基类设计
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, List, Union
from pydantic import BaseModel, Field
from datetime import datetime
import asyncio
import logging
from functools import wraps
import hashlib
import json
logger = logging.getLogger(__name__)
class ToolResult(BaseModel):
"""工具执行结果"""
success: bool = Field(..., description="是否成功")
data: Optional[Dict[str, Any]] = Field(None, description="结果数据")
error: Optional[str] = Field(None, description="错误信息")
execution_time: float = Field(..., description="执行时间（秒）")
cached: bool = Field(default=False, description="是否来自缓存")
timestamp: datetime = Field(default_factory=datetime.now, description="时间戳")
metadata: Dict[str, Any] = Field(default_factory=dict, description="元数据")
class ToolMetadata(BaseModel):
"""工具元数据"""
name: str = Field(..., description="工具名称")
description: str = Field(..., description="工具描述")
version: str = Field(default="1.0.0", description="版本")
author: str = Field(default="", description="作者")
category: str = Field(default="general", description="类别")
tags: List[str] = Field(default_factory=list, description="标签")
parameters: Dict[str, Any] = Field(default_factory=dict, description="参数定义")
return_type: str = Field(default="dict", description="返回类型")
timeout: int = Field(default=30, description="超时时间（秒）")
retry_count: int = Field(default=3, description="重试次数")
retry_delay: float = Field(default=1.0, description="重试延迟（秒）")
cache_ttl: int = Field(default=300, description="缓存TTL（秒）")
rate_limit: Optional[int] = Field(None, description="速率限制（次/分钟）")
class BaseAgnoTool(ABC):
"""Agno工具基类"""
def __init__(self, metadata: ToolMetadata):
self.metadata = metadata
self.logger = logging.getLogger(f"{__name__}.{metadata.name}")
self._cache = {}
self._rate_limiter = RateLimiter(
max_calls=metadata.rate_limit or 1000,
time_window=60
)
@abstractmethod
async def execute_async(self, **kwargs) -> ToolResult:
"""异步执行工具"""
pass
def execute_sync(self, **kwargs) -> ToolResult:
"""同步执行工具"""
try:
# 检查速率限制
if not self._rate_limiter.check_rate_limit():
return ToolResult(
success=False,
error="速率限制超出，请稍后重试",
execution_time=0.0
)
# 检查缓存
cache_key = self._generate_cache_key(**kwargs)
if cache_key in self._cache:
cached_result = self._cache[cache_key]
if datetime.now() - cached_result.timestamp < timedelta(seconds=self.metadata.cache_ttl):
cached_result.cached = True
self.logger.info(f"工具 {self.metadata.name} 使用缓存结果")
return cached_result
# 执行工具
start_time = datetime.now()
# 运行异步函数
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
asyncio.wait_for(
self.execute_async(**kwargs),
timeout=self.metadata.timeout
)
)
finally:
loop.close()
# 更新执行时间
result.execution_time = (datetime.now() - start_time).total_seconds()
# 缓存结果
if result.success:
self._cache[cache_key] = result
return result
except asyncio.TimeoutError:
return ToolResult(
success=False,
error=f"工具执行超时（{self.metadata.timeout}秒）",
execution_time=self.metadata.timeout
)
except Exception as e:
self.logger.error(f"工具 {self.metadata.name} 执行失败: {str(e)}")
return ToolResult(
success=False,
error=f"工具执行失败: {str(e)}",
execution_time=0.0
)
def _generate_cache_key(self, **kwargs) -> str:
"""生成缓存键"""
# 排序参数以确保一致性
sorted_kwargs = sorted(kwargs.items())
param_str = json.dumps(sorted_kwargs, sort_keys=True, default=str)
return hashlib.md5(f"{self.metadata.name}:{param_str}".encode()).hexdigest()
def clear_cache(self):
"""清除缓存"""
self._cache.clear()
self.logger.info(f"工具 {self.metadata.name} 缓存已清除")
def get_cache_stats(self) -> Dict[str, Any]:
"""获取缓存统计"""
return {
"cache_size": len(self._cache),
"tool_name": self.metadata.name,
"cache_ttl": self.metadata.cache_ttl
}
class RateLimiter:
"""速率限制器"""
def __init__(self, max_calls: int, time_window: int):
self.max_calls = max_calls
self.time_window = time_window
self.calls = []
self.lock = asyncio.Lock()
def check_rate_limit(self) -> bool:
"""检查速率限制"""
now = datetime.now()
# 清理过期的调用记录
self.calls = [
call_time for call_time in self.calls
if (now - call_time).total_seconds() < self.time_window
]
# 检查是否超出限制
if len(self.calls) >= self.max_calls:
return False
# 记录当前调用
self.calls.append(now)
return True
def get_remaining_calls(self) -> int:
"""获取剩余调用次数"""
now = datetime.now()
# 清理过期的调用记录
self.calls = [
call_time for call_time in self.calls
if (now - call_time).total_seconds() < self.time_window
]
return max(0, self.max_calls - len(self.calls))
2.2 具体工具实现
# 股票基本面工具
class StockFundamentalsTool(BaseAgnoTool):
"""股票基本面分析工具"""
def __init__(self):
metadata = ToolMetadata(
name="get_stock_fundamentals",
description="获取股票基本面数据，包括PE、PB、ROE等关键指标",
version="2.0.0",
category="financial",
tags=["stocks", "fundamentals", "financial"],
parameters={
"symbol": {
"type": "string",
"description": "股票代码",
"required": True
},
"market": {
"type": "string",
"description": "市场（us/hk/cn）",
"required": False,
"default": "us"
}
},
timeout=30,
retry_count=3,
cache_ttl=600, # 10分钟缓存
rate_limit=100 # 每分钟100次
)
super().__init__(metadata)
async def execute_async(self, symbol: str, market: str = "us") -> ToolResult:
"""异步执行基本面分析"""
try:
self.logger.info(f"获取 {symbol} 在 {market} 市场的基本面数据")
if market == "us":
return await self._get_us_fundamentals(symbol)
elif market == "cn":
return await self._get_cn_fundamentals(symbol)
elif market == "hk":
return await self._get_hk_fundamentals(symbol)
else:
return ToolResult(
success=False,
error=f"不支持的市场: {market}",
execution_time=0.0
)
except Exception as e:
self.logger.error(f"获取基本面数据失败: {str(e)}")
return ToolResult(
success=False,
error=f"获取基本面数据失败: {str(e)}",
execution_time=0.0
)
async def _get_us_fundamentals(self, symbol: str) -> ToolResult:
"""获取美股基本面数据"""
try:
import yfinance as yf
stock = yf.Ticker(symbol)
info = stock.info
# 等待数据获取完成
await asyncio.sleep(0.1)
data = {
"pe_ratio": info.get("trailingPE"),
"pb_ratio": info.get("priceToBook"),
"roe": info.get("returnOnEquity"),
"debt_ratio": info.get("debtToEquity"),
"revenue_growth": info.get("revenueGrowth"),
"market_cap": info.get("marketCap"),
"dividend_yield": info.get("dividendYield"),
"beta": info.get("beta"),
"eps": info.get("trailingEps"),
"book_value": info.get("bookValue"),
"price_to_sales": info.get("priceToSalesTrailing12Months"),
"enterprise_value": info.get("enterpriseValue"),
"profit_margin": info.get("profitMargins"),
"operating_margin": info.get("operatingMargins"),
"return_on_assets": info.get("returnOnAssets"),
"current_ratio": info.get("currentRatio"),
"quick_ratio": info.get("quickRatio"),
"debt_to_equity": info.get("debtToEquity"),
"free_cash_flow": info.get("freeCashflow"),
"operating_cash_flow": info.get("operatingCashflow"),
"total_cash": info.get("totalCash"),
"total_debt": info.get("totalDebt"),
"total_revenue": info.get("totalRevenue"),
"gross_profits": info.get("grossProfits"),
"net_income": info.get("netIncomeToCommon")
}
# 过滤掉None值
data = {k: v for k, v in data.items() if v is not None}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"source": "yfinance",
"symbol": symbol,
"market": "us",
"data_points": len(data)
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取美股基本面数据失败: {str(e)}",
execution_time=0.0
)
async def _get_cn_fundamentals(self, symbol: str) -> ToolResult:
"""获取A股基本面数据"""
try:
import tushare as ts
pro = ts.pro_api()
# 获取公司基本信息
company_df = pro.stock_company(
ts_code=f"{symbol}.SZ" if symbol.startswith("0") else f"{symbol}.SS"
)
# 获取财务指标
financial_df = pro.fina_indicator(ts_code=symbol, period="20231231")
# 获取最新股价信息
daily_df = pro.daily_basic(ts_code=symbol, trade_date="20241231")
data = {}
if not company_df.empty:
company_info = company_df.iloc[0]
data.update({
"company_name": company_info.get("fullname"),
"industry": company_info.get("industry"),
"business": company_info.get("business"),
"area": company_info.get("area"),
"chairman": company_info.get("chairman"),
"manager": company_info.get("manager"),
"reg_capital": company_info.get("reg_capital"),
"setup_date": company_info.get("setup_date"),
"province": company_info.get("province"),
"city": company_info.get("city")
})
if not financial_df.empty:
financial_info = financial_df.iloc[0]
data.update({
"pe_ratio": financial_info.get("pe"),
"pb_ratio": financial_info.get("pb"),
"roe": financial_info.get("roe"),
"debt_ratio": financial_info.get("debt_to_assets"),
"revenue_growth": financial_info.get("or_yoy"),
"net_profit_growth": financial_info.get("netprofit_yoy"),
"gross_margin": financial_info.get("grossprofit_margin"),
"net_margin": financial_info.get("netprofit_margin"),
"current_ratio": financial_info.get("current_ratio"),
"quick_ratio": financial_info.get("quick_ratio"),
"eps": financial_info.get("eps"),
"bps": financial_info.get("bps"),
"roe_dt": financial_info.get("roe_dt"),
"roa": financial_info.get("roa"),
"roa_dt": financial_info.get("roa_dt")
})
if not daily_df.empty:
daily_info = daily_df.iloc[0]
data.update({
"current_price": daily_info.get("close"),
"market_cap": daily_info.get("total_mv"),
"circ_mv": daily_info.get("circ_mv"),
"turnover_rate": daily_info.get("turnover_rate"),
"volume_ratio": daily_info.get("volume_ratio"),
"pe_ttm": daily_info.get("pe_ttm"),
"pb": daily_info.get("pb"),
"ps_ttm": daily_info.get("ps_ttm"),
"dv_ttm": daily_info.get("dv_ttm")
})
# 过滤掉None值
data = {k: v for k, v in data.items() if v is not None}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"source": "tushare",
"symbol": symbol,
"market": "cn",
"data_points": len(data)
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取A股基本面数据失败: {str(e)}",
execution_time=0.0
)
async def _get_hk_fundamentals(self, symbol: str) -> ToolResult:
"""获取港股基本面数据"""
try:
import yfinance as yf
symbol_hk = f"{symbol}.HK"
stock = yf.Ticker(symbol_hk)
info = stock.info
data = {
"pe_ratio": info.get("trailingPE"),
"pb_ratio": info.get("priceToBook"),
"market_cap": info.get("marketCap"),
"dividend_yield": info.get("dividendYield"),
"beta": info.get("beta"),
"eps": info.get("trailingEps"),
"book_value": info.get("bookValue"),
"price_to_sales": info.get("priceToSalesTrailing12Months"),
"company_name": info.get("longName"),
"sector": info.get("sector"),
"industry": info.get("industry"),
"website": info.get("website"),
"long_business_summary": info.get("longBusinessSummary")
}
# 过滤掉None值
data = {k: v for k, v in data.items() if v is not None}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"source": "yfinance",
"symbol": symbol,
"market": "hk",
"data_points": len(data)
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取港股基本面数据失败: {str(e)}",
execution_time=0.0
)
# 市场数据工具
class MarketDataTool(BaseAgnoTool):
"""市场数据获取工具"""
def __init__(self):
metadata = ToolMetadata(
name="get_market_data",
description="获取市场数据，包括价格、成交量、技术指标等",
version="2.0.0",
category="market",
tags=["stocks", "market", "technical", "price"],
parameters={
"symbol": {
"type": "string",
"description": "股票代码",
"required": True
},
"period": {
"type": "string",
"description": "时间周期（1d/5d/1mo/3mo/6mo/1y/2y/5y/10y/ytd/max）",
"required": False,
"default": "1y"
},
"interval": {
"type": "string",
"description": "时间间隔（1m/2m/5m/15m/30m/60m/90m/1h/1d/5d/1wk/1mo/3mo）",
"required": False,
"default": "1d"
}
},
timeout=30,
retry_count=3,
cache_ttl=300, # 5分钟缓存
rate_limit=200 # 每分钟200次
)
super().__init__(metadata)
async def execute_async(self, symbol: str, period: str = "1y", interval: str = "1d") -> ToolResult:
"""异步执行市场数据获取"""
try:
self.logger.info(f"获取 {symbol} 的市场数据，周期: {period}，间隔: {interval}")
import yfinance as yf
stock = yf.Ticker(symbol)
hist = stock.history(period=period, interval=interval)
if hist.empty:
return ToolResult(
success=False,
error="未找到历史数据",
execution_time=0.0
)
# 等待数据获取完成
await asyncio.sleep(0.1)
# 计算各种指标
current_price = float(hist['Close'].iloc[-1])
open_price = float(hist['Open'].iloc[-1])
high_price = float(hist['High'].iloc[-1])
low_price = float(hist['Low'].iloc[-1])
volume = int(hist['Volume'].iloc[-1])
# 价格变化
price_change = current_price - open_price
price_change_pct = (price_change / open_price) * 100 if open_price > 0 else 0
# 技术指标
hist['SMA5'] = hist['Close'].rolling(window=5).mean()
hist['SMA10'] = hist['Close'].rolling(window=10).mean()
hist['SMA20'] = hist['Close'].rolling(window=20).mean()
hist['SMA50'] = hist['Close'].rolling(window=50).mean()
hist['SMA200'] = hist['Close'].rolling(window=200).mean()
# 计算RSI
hist['RSI'] = self._calculate_rsi(hist['Close'])
# 计算MACD
hist['MACD'], hist['MACD_Signal'] = self._calculate_macd(hist['Close'])
# 计算布林带
hist['BB_Upper'], hist['BB_Middle'], hist['BB_Lower'] = self._calculate_bollinger_bands(hist['Close'])
# 获取最新值
sma5 = float(hist['SMA5'].iloc[-1]) if not pd.isna(hist['SMA5'].iloc[-1]) else None
sma10 = float(hist['SMA10'].iloc[-1]) if not pd.isna(hist['SMA10'].iloc[-1]) else None
sma20 = float(hist['SMA20'].iloc[-1]) if not pd.isna(hist['SMA20'].iloc[-1]) else None
sma50 = float(hist['SMA50'].iloc[-1]) if not pd.isna(hist['SMA50'].iloc[-1]) else None
sma200 = float(hist['SMA200'].iloc[-1]) if not pd.isna(hist['SMA200'].iloc[-1]) else None
rsi = float(hist['RSI'].iloc[-1]) if not pd.isna(hist['RSI'].iloc[-1]) else None
macd = float(hist['MACD'].iloc[-1]) if not pd.isna(hist['MACD'].iloc[-1]) else None
macd_signal = float(hist['MACD_Signal'].iloc[-1]) if not pd.isna(hist['MACD_Signal'].iloc[-1]) else None
bb_upper = float(hist['BB_Upper'].iloc[-1]) if not pd.isna(hist['BB_Upper'].iloc[-1]) else None
bb_middle = float(hist['BB_Middle'].iloc[-1]) if not pd.isna(hist['BB_Middle'].iloc[-1]) else None
bb_lower = float(hist['BB_Lower'].iloc[-1]) if not pd.isna(hist['BB_Lower'].iloc[-1]) else None
# 趋势判断
trend = "neutral"
if current_price and sma20 and sma50:
if current_price > sma20 > sma50:
trend = "bullish"
elif current_price < sma20 < sma50:
trend = "bearish"
# 支撑阻力位（简化计算）
support_level = float(hist['Low'].tail(10).min()) if len(hist) >= 10 else low_price
resistance_level = float(hist['High'].tail(10).max()) if len(hist) >= 10 else high_price
data = {
"current_price": current_price,
"open_price": open_price,
"high_price": high_price,
"low_price": low_price,
"volume": volume,
"price_change": price_change,
"price_change_pct": price_change_pct,
"trend": trend,
"support_level": support_level,
"resistance_level": resistance_level,
"technical_indicators": {
"sma5": sma5,
"sma10": sma10,
"sma20": sma20,
"sma50": sma50,
"sma200": sma200,
"rsi": rsi,
"macd": macd,
"macd_signal": macd_signal,
"bb_upper": bb_upper,
"bb_middle": bb_middle,
"bb_lower": bb_lower
}
}
# 过滤掉None值
data = {k: v for k, v in data.items() if v is not None}
if "technical_indicators" in data:
data["technical_indicators"] = {k: v for k, v in data["technical_indicators"].items() if v is not None}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"symbol": symbol,
"period": period,
"interval": interval,
"data_points": len(hist),
"indicators_calculated": len(data.get("technical_indicators", {}))
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取市场数据失败: {str(e)}",
execution_time=0.0
)
def _calculate_rsi(self, prices, period: int = 14) -> pd.Series:
"""计算RSI指标"""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def _calculate_macd(self, prices, fast: int = 12, slow: int = 26, signal: int = 9) -> tuple:
"""计算MACD指标"""
ema_fast = prices.ewm(span=fast).mean()
ema_slow = prices.ewm(span=slow).mean()
macd = ema_fast - ema_slow
macd_signal = macd.ewm(span=signal).mean()
return macd, macd_signal
def _calculate_bollinger_bands(self, prices, period: int = 20, std_dev: int = 2) -> tuple:
"""计算布林带"""
sma = prices.rolling(window=period).mean()
std = prices.rolling(window=period).std()
upper_band = sma + (std * std_dev)
lower_band = sma - (std * std_dev)
return upper_band, sma, lower_band
# 新闻情感分析工具
class NewsSentimentTool(BaseAgnoTool):
"""新闻情感分析工具"""
def __init__(self):
metadata = ToolMetadata(
name="get_news_sentiment",
description="获取新闻并进行情感分析",
version="2.0.0",
category="sentiment",
tags=["news", "sentiment", "nlp"],
parameters={
"symbol": {
"type": "string",
"description": "股票代码",
"required": True
},
"limit": {
"type": "int",
"description": "新闻数量限制",
"required": False,
"default": 10
},
"language": {
"type": "string",
"description": "语言（cn/en）",
"required": False,
"default": "cn"
}
},
timeout=45,
retry_count=2,
cache_ttl=1800, # 30分钟缓存
rate_limit=50 # 每分钟50次
)
super().__init__(metadata)
async def execute_async(self, symbol: str, limit: int = 10, language: str = "cn") -> ToolResult:
"""异步执行新闻情感分析"""
try:
self.logger.info(f"获取 {symbol} 的新闻情感分析，语言: {language}")
if language == "cn":
return await self._get_cn_news_sentiment(symbol, limit)
else:
return await self._get_en_news_sentiment(symbol, limit)
except Exception as e:
self.logger.error(f"获取新闻情感失败: {str(e)}")
return ToolResult(
success=False,
error=f"获取新闻情感失败: {str(e)}",
execution_time=0.0
)
async def _get_cn_news_sentiment(self, symbol: str, limit: int) -> ToolResult:
"""获取中文新闻情感"""
try:
# 这里使用模拟数据，实际应该调用新闻API
import random
news_items = []
sentiment_words = {
"positive": ["上涨", "增长", "盈利", "利好", "突破", "创新高", "强劲", "优秀"],
"negative": ["下跌", "亏损", "利空", "暴雷", "跌停", "风险", "警告", "下滑"],
"neutral": ["持平", "稳定", "正常", "波动", "调整", "震荡", "观望"]
}
for i in range(limit):
sentiment = random.choice(["positive", "negative", "neutral"])
score = random.uniform(-1, 1)
# 根据情感选择关键词
if sentiment == "positive":
keywords = random.sample(sentiment_words["positive"], 3)
score = random.uniform(0.3, 1.0)
elif sentiment == "negative":
keywords = random.sample(sentiment_words["negative"], 3)
score = random.uniform(-1.0, -0.3)
else:
keywords = random.sample(sentiment_words["neutral"], 3)
score = random.uniform(-0.3, 0.3)
title = f"{symbol} {' '.join(keywords[:2])}，市场反应{keywords[2]}"
content = f"据最新消息，{symbol}相关股票出现{keywords[0]}情况，分析师认为这可能导致{keywords[1]}，投资者应{keywords[2]}。"
news_items.append({
"title": title,
"content": content,
"sentiment": sentiment,
"score": score,
"confidence": random.uniform(0.6, 0.95),
"source": f"财经媒体{i+1}",
"timestamp": (datetime.now() - timedelta(hours=random.randint(1, 48))).isoformat(),
"url": f"https://example.com/news/{symbol}/{i+1}"
})
# 计算整体情感
avg_sentiment = sum(item["score"] for item in news_items) / len(news_items)
overall_sentiment = "positive" if avg_sentiment > 0.1 else "negative" if avg_sentiment < -0.1 else "neutral"
# 情感统计
sentiment_stats = {
"positive": len([item for item in news_items if item["sentiment"] == "positive"]),
"negative": len([item for item in news_items if item["sentiment"] == "negative"]),
"neutral": len([item for item in news_items if item["sentiment"] == "neutral"])
}
data = {
"news_items": news_items,
"overall_sentiment": overall_sentiment,
"average_score": avg_sentiment,
"sentiment_stats": sentiment_stats,
"total_articles": len(news_items),
"analysis_timestamp": datetime.now().isoformat()
}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"symbol": symbol,
"language": "cn",
"articles_analyzed": len(news_items),
"sentiment_distribution": sentiment_stats
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取中文新闻情感失败: {str(e)}",
execution_time=0.0
)
async def _get_en_news_sentiment(self, symbol: str, limit: int) -> ToolResult:
"""获取英文新闻情感"""
try:
# 这里使用模拟数据，实际应该调用英文新闻API
import random
news_items = []
sentiment_words = {
"positive": ["surge", "growth", "profit", "breakthrough", "strong", "excellent", "rally"],
"negative": ["decline", "loss", "risk", "warning", "plunge", "weak", "concern"],
"neutral": ["stable", "flat", "normal", "steady", "unchanged"]
}
for i in range(limit):
sentiment = random.choice(["positive", "negative", "neutral"])
score = random.uniform(-1, 1)
# 根据情感选择关键词
if sentiment == "positive":
keywords = random.sample(sentiment_words["positive"], 3)
score = random.uniform(0.3, 1.0)
elif sentiment == "negative":
keywords = random.sample(sentiment_words["negative"], 3)
score = random.uniform(-1.0, -0.3)
else:
keywords = random.sample(sentiment_words["neutral"], 3)
score = random.uniform(-0.3, 0.3)
title = f"{symbol} shows {keywords[0]} momentum amid {keywords[1]} market conditions"
content = f"Latest market analysis indicates that {symbol} is experiencing {keywords[0]} trends. Analysts suggest this could lead to {keywords[1]} outcomes for investors. Market participants remain {keywords[2]} about future prospects."
news_items.append({
"title": title,
"content": content,
"sentiment": sentiment,
"score": score,
"confidence": random.uniform(0.6, 0.95),
"source": f"Financial News Source {i+1}",
"timestamp": (datetime.now() - timedelta(hours=random.randint(1, 48))).isoformat(),
"url": f"https://example.com/news/{symbol}/{i+1}"
})
# 计算整体情感
avg_sentiment = sum(item["score"] for item in news_items) / len(news_items)
overall_sentiment = "positive" if avg_sentiment > 0.1 else "negative" if avg_sentiment < -0.1 else "neutral"
# 情感统计
sentiment_stats = {
"positive": len([item for item in news_items if item["sentiment"] == "positive"]),
"negative": len([item for item in news_items if item["sentiment"] == "negative"]),
"neutral": len([item for item in news_items if item["sentiment"] == "neutral"])
}
data = {
"news_items": news_items,
"overall_sentiment": overall_sentiment,
"average_score": avg_sentiment,
"sentiment_stats": sentiment_stats,
"total_articles": len(news_items),
"analysis_timestamp": datetime.now().isoformat()
}
return ToolResult(
success=True,
data=data,
execution_time=0.0,
metadata={
"symbol": symbol,
"language": "en",
"articles_analyzed": len(news_items),
"sentiment_distribution": sentiment_stats
}
)
except Exception as e:
return ToolResult(
success=False,
error=f"获取英文新闻情感失败: {str(e)}",
execution_time=0.0
)
## 4. 迁移实施计划
### 4.1 迁移步骤
#### 第一阶段：基础架构准备（1-2周）
1. **环境搭建**
- 安装Agno框架依赖
- 配置异步执行环境
- 设置日志和监控系统
2. **基础类实现**
- 实现BaseAgnoTool基类
- 实现ToolResult和ToolMetadata
- 实现工具管理器AgnoToolManager
3. **核心工具迁移**
- 迁移股票基本面工具
- 迁移市场数据工具
- 迁移新闻情感工具
#### 第二阶段：高级功能实现（2-3周）
1. **错误处理与重试**
- 实现ToolErrorHandler
- 实现RetryExecutor
- 集成到增强工具类
2. **API适配层**
- 实现APIResponseAdapter
- 实现UnifiedAPIClient
- 测试不同API的兼容性
3. **性能优化**
- 实现缓存机制
- 实现速率限制
- 添加性能监控
#### 第三阶段：集成测试（1-2周）
1. **单元测试**
- 测试每个工具的功能
- 测试错误处理机制
- 测试重试逻辑
2. **集成测试**
- 测试工具管理器
- 测试批量执行
- 测试异步执行
3. **性能测试**
- 测试并发性能
- 测试缓存效果
- 测试错误恢复
#### 第四阶段：生产部署（1周）
1. **灰度发布**
- 部分流量切换到新系统
- 监控性能和错误率
- 收集用户反馈
2. **全量切换**
- 修复发现的问题
- 完善文档
- 全量部署
### 4.2 回滚策略
#### 回滚条件
- 错误率超过5%
- 响应时间增加超过50%
- 核心功能不可用
- 数据准确性问题
#### 回滚步骤
1. 立即停止新系统流量
2. 切换回LangGraph系统
3. 检查数据一致性
4. 分析问题原因
5. 修复后重新部署
#### 回滚验证器
## 5. 性能对比与优化
### 5.1 性能指标对比
| 指标 | LangGraph (当前) | Agno (目标) | 改进幅度 |
|------|------------------|-------------|----------|
| 工具执行时间 | 500ms | 300ms | -40% |
| 并发处理能力 | 10 req/s | 50 req/s | +400% |
| 缓存命中率 | 0% | 60% | +60% |
| 错误恢复时间 | 5s | 1s | -80% |
| 内存使用 | 100MB | 80MB | -20% |
| CPU使用率 | 70% | 50% | -28% |
### 5.2 持续优化建议
1. **缓存优化**
- 实现智能缓存策略
- 添加缓存预热机制
- 优化缓存失效策略
2. **异步优化**
- 优化事件循环使用
- 减少上下文切换
- 实现连接池
3. **监控优化**
- 添加详细性能指标
- 实现实时监控
- 设置自动告警
4. **工具优化**
- 优化API调用顺序
- 实现批量API调用
- 添加工具依赖管理
2.3 工具管理器
class AgnoToolManager:
"""Agno工具管理器"""
def __init__(self):
self.tools: Dict[str, BaseAgnoTool] = {}
self.categories: Dict[str, List[str]] = {}
self.logger = logging.getLogger(__name__)
self._initialize_default_tools()
def register_tool(self, tool: BaseAgnoTool) -> bool:
"""注册工具"""
try:
tool_name = tool.metadata.name
if tool_name in self.tools:
self.logger.warning(f"工具 {tool_name} 已存在，将被覆盖")
self.tools[tool_name] = tool
# 添加到类别映射
category = tool.metadata.category
if category not in self.categories:
self.categories[category] = []
if tool_name not in self.categories[category]:
self.categories[category].append(tool_name)
self.logger.info(f"工具 {tool_name} 注册成功")
return True
except Exception as e:
self.logger.error(f"工具注册失败: {str(e)}")
return False
def unregister_tool(self, tool_name: str) -> bool:
"""注销工具"""
try:
if tool_name not in self.tools:
self.logger.warning(f"工具 {tool_name} 不存在")
return False
tool = self.tools[tool_name]
category = tool.metadata.category
# 从工具字典中移除
del self.tools[tool_name]
# 从类别映射中移除
if category in self.categories:
if tool_name in self.categories[category]:
self.categories[category].remove(tool_name)
# 如果类别为空，删除类别
if not self.categories[category]:
del self.categories[category]
self.logger.info(f"工具 {tool_name} 注销成功")
return True
except Exception as e:
self.logger.error(f"工具注销失败: {str(e)}")
return False
def get_tool(self, tool_name: str) -> Optional[BaseAgnoTool]:
"""获取工具"""
return self.tools.get(tool_name)
def get_tools_by_category(self, category: str) -> List[BaseAgnoTool]:
"""按类别获取工具"""
tool_names = self.categories.get(category, [])
return [self.tools[name] for name in tool_names if name in self.tools]
def get_all_tools(self) -> List[BaseAgnoTool]:
"""获取所有工具"""
return list(self.tools.values())
def get_tool_names(self) -> List[str]:
"""获取所有工具名称"""
return list(self.tools.keys())
def get_categories(self) -> List[str]:
"""获取所有类别"""
return list(self.categories.keys())
def execute_tool(self, tool_name: str, **kwargs) -> ToolResult:
"""执行工具"""
tool = self.get_tool(tool_name)
if not tool:
return ToolResult(
success=False,
error=f"工具 {tool_name} 不存在",
execution_time=0.0
)
return tool.execute_sync(**kwargs)
async def execute_tool_async(self, tool_name: str, **kwargs) -> ToolResult:
"""异步执行工具"""
tool = self.get_tool(tool_name)
if not tool:
return ToolResult(
success=False,
error=f"工具 {tool_name} 不存在",
execution_time=0.0
)
return await tool.execute_async(**kwargs)
def execute_tools_batch(self, tool_calls: List[Dict[str, Any]]) -> List[ToolResult]:
"""批量执行工具"""
results = []
for tool_call in tool_calls:
tool_name = tool_call.get("tool_name")
parameters = tool_call.get("parameters", {})
result = self.execute_tool(tool_name, **parameters)
results.append(result)
return results
async def execute_tools_batch_async(self, tool_calls: List[Dict[str, Any]]) -> List[ToolResult]:
"""异步批量执行工具"""
tasks = []
for tool_call in tool_calls:
tool_name = tool_call.get("tool_name")
parameters = tool_call.get("parameters", {})
task = self.execute_tool_async(tool_name, **parameters)
tasks.append(task)
return await asyncio.gather(*tasks)
def get_tool_info(self, tool_name: str) -> Optional[Dict[str, Any]]:
"""获取工具信息"""
tool = self.get_tool(tool_name)
if not tool:
return None
return {
"metadata": tool.metadata.dict(),
"cache_stats": tool.get_cache_stats()
}
def get_tools_info(self) -> Dict[str, Dict[str, Any]]:
"""获取所有工具信息"""
info = {}
for tool_name in self.get_tool_names():
tool_info = self.get_tool_info(tool_name)
if tool_info:
info[tool_name] = tool_info
return info
def clear_all_caches(self):
"""清除所有工具缓存"""
for tool in self.tools.values():
tool.clear_cache()
self.logger.info("所有工具缓存已清除")
def _initialize_default_tools(self):
"""初始化默认工具"""
try:
# 注册基本面分析工具
self.register_tool(StockFundamentalsTool())
# 注册市场数据工具
self.register_tool(MarketDataTool())
# 注册新闻情感工具
self.register_tool(NewsSentimentTool())
self.logger.info("默认工具初始化完成")
except Exception as e:
self.logger.error(f"默认工具初始化失败: {str(e)}")
3. 迁移挑战与解决方案
3.1 异步执行转换
#### 挑战
- LangGraph工具是同步执行
- Agno工具需要异步支持
- 需要兼容现有同步代码
class AsyncToolAdapter:
"""异步工具适配器"""
@staticmethod
def sync_to_async(sync_func):
"""将同步函数转换为异步函数"""
@wraps(sync_func)
async def async_wrapper(*args, **kwargs):
# 在线程池中运行同步函数
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, sync_func, *args, **kwargs)
return async_wrapper
@staticmethod
def async_to_sync(async_func):
"""将异步函数转换为同步函数"""
@wraps(async_func)
def sync_wrapper(*args, **kwargs):
# 运行异步函数
try:
loop = asyncio.get_event_loop()
if loop.is_running():
# 如果事件循环已在运行，创建新循环
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
try:
return new_loop.run_until_complete(async_func(*args, **kwargs))
finally:
new_loop.close()
else:
return loop.run_until_complete(async_func(*args, **kwargs))
except RuntimeError:
# 没有事件循环，创建新的
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(async_func(*args, **kwargs))
finally:
loop.close()
return sync_wrapper
# 迁移适配器
class ToolMigrationAdapter:
"""工具迁移适配器"""
def __init__(self, agno_tool_manager: AgnoToolManager):
self.agno_manager = agno_tool_manager
self.async_adapter = AsyncToolAdapter()
def convert_langgraph_tool(self, langgraph_tool) -> BaseAgnoTool:
"""转换LangGraph工具到Agno工具"""
class ConvertedTool(BaseAgnoTool):
def __init__(self, original_tool):
# 提取工具信息
tool_name = getattr(original_tool, 'name', original_tool.__name__)
tool_description = getattr(original_tool, 'description', '转换的工具')
metadata = ToolMetadata(
name=tool_name,
description=tool_description,
version="1.0.0",
category="migrated",
tags=["migrated", "langgraph"]
)
super().__init__(metadata)
self.original_tool = original_tool
async def execute_async(self, **kwargs) -> ToolResult:
"""异步执行原始工具"""
try:
# 将异步转换为同步
sync_func = self.async_adapter.async_to_sync(self.original_tool)
result = sync_func(**kwargs)
# 转换结果格式
if isinstance(result, dict):
if "error" in result and result["error"]:
return ToolResult(
success=False,
error=result["error"],
execution_time=0.0
)
else:
return ToolResult(
success=True,
data=result,
execution_time=0.0
)
else:
return ToolResult(
success=True,
data={"result": result},
execution_time=0.0
)
except Exception as e:
return ToolResult(
success=False,
error=f"工具执行失败: {str(e)}",
execution_time=0.0
)
return ConvertedTool(langgraph_tool)
def migrate_tools(self, langgraph_tools: List) -> List[BaseAgnoTool]:
"""批量迁移工具"""
migrated_tools = []
for tool in langgraph_tools:
try:
agno_tool = self.convert_langgraph_tool(tool)
migrated_tools.append(agno_tool)
self.logger.info(f"工具 {tool.__name__} 迁移成功")
except Exception as e:
self.logger.error(f"工具 {tool.__name__} 迁移失败: {str(e)}")
return migrated_tools
3.2 错误处理与重试机制
#### 挑战
- LangGraph工具简单错误处理
- 需要更完善的错误处理
- 需要自动重试机制
class ToolErrorHandler:
"""工具错误处理器"""
def __init__(self):
self.logger = logging.getLogger(__name__)
self.error_patterns = {
"rate_limit": ["rate limit", "too many requests", "quota exceeded"],
"network": ["connection", "timeout", "network", "unreachable"],
"authentication": ["unauthorized", "forbidden", "authentication", "api key"],
"data": ["not found", "invalid", "missing", "empty"],
"service": ["service unavailable", "maintenance", "error 500", "internal error"]
}
def classify_error(self, error_message: str) -> str:
"""分类错误类型"""
error_message_lower = error_message.lower()
for error_type, patterns in self.error_patterns.items():
for pattern in patterns:
if pattern in error_message_lower:
return error_type
return "unknown"
def should_retry(self, error_type: str, retry_count: int, max_retries: int) -> bool:
"""判断是否应该重试"""
if retry_count >= max_retries:
return False
# 某些错误类型不应该重试
no_retry_types = ["authentication", "data"]
if error_type in no_retry_types:
return False
# 网络和服务错误应该重试
retry_types = ["network", "service", "rate_limit"]
if error_type in retry_types:
return True
return False
def get_retry_delay(self, error_type: str, retry_count: int) -> float:
"""获取重试延迟"""
base_delays = {
"network": 1.0,
"service": 2.0,
"rate_limit": 5.0,
"unknown": 1.0
}
base_delay = base_delays.get(error_type, 1.0)
# 指数退避
return base_delay * (2 ** retry_count)
def create_error_result(self, error_message: str, error_type: str = None) -> ToolResult:
"""创建错误结果"""
if not error_type:
error_type = self.classify_error(error_message)
return ToolResult(
success=False,
error=error_message,
execution_time=0.0,
metadata={
"error_type": error_type,
"retry_suggested": self.should_retry(error_type, 0, 3)
}
)
class RetryExecutor:
"""重试执行器"""
def __init__(self, error_handler: ToolErrorHandler):
self.error_handler = error_handler
self.logger = logging.getLogger(__name__)
async def execute_with_retry(
self,
func,
max_retries: int = 3,
*args,
**kwargs
) -> ToolResult:
"""带重试的执行"""
for attempt in range(max_retries + 1):
try:
self.logger.info(f"执行尝试 {attempt + 1}/{max_retries + 1}")
# 执行函数
result = await func(*args, **kwargs)
# 如果成功，直接返回
if result.success:
if attempt > 0:
result.metadata["retry_attempts"] = attempt
result.metadata["retry_successful"] = True
return result
# 如果失败，检查是否应该重试
error_message = result.error or "未知错误"
error_type = result.metadata.get("error_type", "unknown")
if not self.error_handler.should_retry(error_type, attempt, max_retries):
self.logger.info(f"错误类型 {error_type} 不需要重试")
return result
# 计算重试延迟
retry_delay = self.error_handler.get_retry_delay(error_type, attempt)
self.logger.info(f"将在 {retry_delay} 秒后重试")
await asyncio.sleep(retry_delay)
except Exception as e:
error_message = str(e)
error_type = self.error_handler.classify_error(error_message)
self.logger.error(f"执行失败: {error_message} (类型: {error_type})")
if not self.error_handler.should_retry(error_type, attempt, max_retries):
return self.error_handler.create_error_result(error_message, error_type)
# 计算重试延迟
retry_delay = self.error_handler.get_retry_delay(error_type, attempt)
self.logger.info(f"将在 {retry_delay} 秒后重试")
await asyncio.sleep(retry_delay)
# 所有重试都失败
final_error = f"所有 {max_retries + 1} 次尝试都失败"
self.logger.error(final_error)
return ToolResult(
success=False,
error=final_error,
execution_time=0.0,
metadata={
"retry_attempts": max_retries + 1,
"retry_successful": False,
"last_error": error_message,
"last_error_type": error_type
}
)
# 增强的基础工具类
class EnhancedBaseAgnoTool(BaseAgnoTool):
"""增强的Agno工具基类，包含重试和错误处理"""
def __init__(self, metadata: ToolMetadata):
super().__init__(metadata)
self.error_handler = ToolErrorHandler()
self.retry_executor = RetryExecutor(self.error_handler)
async def execute_async(self, **kwargs) -> ToolResult:
"""异步执行，带重试机制"""
return await self.retry_executor.execute_with_retry(
self._execute_with_error_handling,
max_retries=self.metadata.retry_count,
**kwargs
)
async def _execute_with_error_handling(self, **kwargs) -> ToolResult:
"""执行并处理错误"""
try:
result = await self._do_execute(**kwargs)
# 如果结果已经是ToolResult，直接返回
if isinstance(result, ToolResult):
return result
# 否则包装成ToolResult
return ToolResult(
success=True,
data=result if isinstance(result, dict) else {"result": result},
execution_time=0.0
)
except Exception as e:
error_message = str(e)
error_type = self.error_handler.classify_error(error_message)
self.logger.error(f"工具 {self.metadata.name} 执行失败: {error_message}")
return ToolResult(
success=False,
error=error_message,
execution_time=0.0,
metadata={
"error_type": error_type,
"retry_suggested": self.error_handler.should_retry(error_type, 0, self.metadata.retry_count)
}
)
@abstractmethod
async def _do_execute(self, **kwargs) -> Union[Dict[str, Any], ToolResult]:
"""实际执行逻辑，子类需要实现"""
pass
3.3 API适配与兼容性
#### 挑战
- 不同API的响应格式不同
- 需要统一的数据格式
- 向后兼容性
```python class APIResponseAdapter: """API响应适配器""" @staticmethod def adapt_yfinance_response(raw_data: Dict[str, Any]) -> Dict[str, Any]: """适配yfinance响应""" return { "pe_ratio": raw_data.get("trailingPE"), "pb_ratio": raw_data.get("priceToBook"), "roe": raw_data.get("returnOnEquity"), "debt_ratio": raw_data.get("debtToEquity"), "revenue_growth": raw_data.get("revenueGrowth"), "market_cap": raw_data.get("marketCap"), "dividend_yield": raw_data.get("dividendYield"), "beta": raw_data.get("beta"), "eps": raw_data.get("trailingEps"), "book_value": raw_data.get("bookValue"), "price_to_sales": raw_data.get("priceToSalesTrailing12Months"), "enterprise_value": raw_data.get("enterpriseValue"), "profit_margin": raw_data.get("profitMargins"), "operating_margin": raw_data.get("operatingMargins"), "return_on_assets": raw_data.get("returnOnAssets"), "current_ratio": raw_data.get("currentRatio"), "quick_ratio": raw_data.get("quickRatio"), "free_cash_flow": raw_data.get("freeCashflow"), "operating_cash_flow": raw_data.get("operatingCashflow"), "total_cash": raw_data.get("totalCash"), "total_debt": raw_data.get("totalDebt"), "total_revenue": raw_data.get("totalRevenue"), "gross_profits": raw_data.get("grossProfits"), "net_income": raw_data.get("netIncomeToCommon") } @staticmethod def adapt_tushare_response(raw_data: Dict[str, Any]) -> Dict[str, Any]: """适配tushare响应""" return { "company_name": raw_data.get("fullname"), "industry": raw_data.get("industry"), "business": raw_data.get("business"), "area": raw_data.get("area"), "pe_ratio": raw_data.get("pe"), "pb_ratio": raw_data.get("pb"), "roe": raw_data.get("roe"), "debt_ratio": raw_data.get("debt_to_assets"), "revenue_growth": raw_data.get("or_yoy"), "net_profit_growth": raw_data.get("netprofit_yoy"), "gross_margin": raw_data.get("grossprofit_margin"), "net_margin": raw_data.get("netprofit_margin"), "current_ratio": raw_data.get("current_ratio"), "quick_ratio": raw_data.get("quick_ratio"), "eps": raw_data.get("eps"), "bps": raw_data.get("bps"), "market_cap": raw_data.get("total_mv"), "turnover_rate": raw_data.get("turnover_rate") } @staticmethod def adapt_alpha_vantage_response(raw_data: Dict[str, Any]) -> Dict[str, Any]: """适配Alpha Vantage响应""" return { "pe_ratio": raw_data.get("PERatio"), "pb_ratio": raw_data.get("PriceToBookRatio"), "roe": raw_data.get("ReturnOnEquityTTM"), "debt_ratio": raw_data.get("DebtToEquityRatio"), "revenue_growth": raw_data.get("RevenueGrowth"), "market_cap": raw_data.get("MarketCapitalization"), "dividend_yield": raw_data.get("DividendYield"), "beta": raw_data.get("Beta"), "eps": raw_data.get("EarningsPerShare"), "book_value": raw_data.get("BookValue"), "price_to_sales": raw_data.get("PriceToSalesRatio"), "profit_margin": raw_data.get("ProfitMargin"), "operating_margin": raw_data.get("OperatingMarginTTM"), "return_on_assets": raw_data.get("ReturnOnAssetsTTM"), "current_ratio": raw_data.get("CurrentRatio"), "quick_ratio": raw_data.get("QuickRatio") }
class UnifiedAPIClient: """统一API客户端""" def __init__(self): self.adapters = { "yfinance": APIResponseAdapter.adapt_yfinance_response, "tushare": APIResponseAdapter.adapt_tushare_response, "alpha_vantage": APIResponseAdapter.adapt_alpha_vantage_response } self.logger = logging.getLogger(__name__) async def get_fundamentals(self, symbol: str, market: str, api_source: str = "auto") -> Dict[str, Any]: """获取基本面数据""" try: if api_source == "auto": # 根据市场自动选择API if market == "us": api_source = "yfinance" elif market == "cn": api_source = "tushare" elif market == "hk": api_source = "yfinance" else: raise ValueError(f"不支持的市场: {market}") # 获取原始数据 raw_data = await self._fetch_raw_data(symbol, market, api_source) # 适配响应格式 if api_source in self.adapters: adapted_data = self.adaptersapi_source else: adapted_data = raw_data # 添加元数据 adapted_data["api_source"] = api_source adapted_data["market"] = market adapted_data["symbol"] = symbol adapted_data["timestamp"] = datetime.now().isoformat() return adapted_data except Exception as e: self.logger.error(f"获取基本面数据失败: {str(e)}") raise async def _fetch_raw_data(self, symbol: str, market: str, api_source: str) -> Dict[str, Any]: """获取原始数据""" # 这里实现具体的API调用逻辑 # 为了简化，这里返回模拟数据 if api_source == "yfinance": # 模拟yfinance响应 return { "trailingPE": 15.5, "priceToBook": 2.1, "returnOnEquity": 0.15, "debtToEquity": 0.5, "revenueGrowth": 0.08, "marketCap": 1000000000, "dividendYield":