分析师团队是 TradingAgents 框架的核心组成部分，负责从不同维度分析市场数据。每个分析师都专注于特定的分析领域，通过专业化分工确保分析的深度和准确性。
class BaseAnalyst:
"""所有分析师的基础类"""
def __init__(self, llm, config, tools=None):
self.llm = llm
self.config = config
self.tools = tools or []
self.memory = AnalystMemory()
def analyze(self, state: AgentState) -> Dict[str, Any]:
"""执行分析的主要方法"""
# 1. 数据预处理
processed_data = self.preprocess_data(state)
# 2. 执行专业分析
analysis_result = self.perform_analysis(processed_data)
# 3. 生成分析报告
report = self.generate_report(analysis_result)
# 4. 更新记忆
self.memory.update(state.ticker, report)
return report
def preprocess_data(self, state: AgentState) -> Dict:
"""数据预处理 - 子类可重写"""
return state
def perform_analysis(self, data: Dict) -> Dict:
"""执行分析 - 子类必须实现"""
raise NotImplementedError
def generate_report(self, analysis: Dict) -> Dict:
"""生成分析报告 - 子类可重写"""
return analysis
class FundamentalsAnalyst(BaseAnalyst):
"""基本面分析师 - 专注于公司财务和基本面分析"""
专业领域:
- 财务报表分析
- 估值模型计算
- 行业对比分析
- 盈利能力评估
- 财务健康度评估
分析维度:
- 收入增长率
- 利润率趋势
- 资产负债比率
- 现金流状况
- ROE/ROA 指标
- P/E, P/B 估值比率
def perform_analysis(self, data: Dict) -> Dict:
"""执行基本面分析"""
financial_data = data.get("financial_data", {})
company_info = data.get("company_info", {})
analysis = {
"financial_health": self._assess_financial_health(financial_data),
"valuation": self._calculate_valuation(financial_data),
"growth_analysis": self._analyze_growth(financial_data),
"profitability": self._assess_profitability(financial_data),
"liquidity": self._assess_liquidity(financial_data),
"leverage": self._assess_leverage(financial_data)
}
# 综合评分
analysis["overall_score"] = self._calculate_overall_score(analysis)
analysis["recommendation"] = self._generate_recommendation(analysis)
return analysis
def _assess_financial_health(self, financial_data: Dict) -> Dict:
"""评估财务健康度"""
# 计算关键财务比率
current_ratio = financial_data.get("current_assets", 0) / max(
financial_data.get("current_liabilities", 1), 1
)
debt_to_equity = financial_data.get("total_debt", 0) / max(
financial_data.get("shareholders_equity", 1), 1
)
# 评估财务健康度
health_score = 0
if current_ratio > 1.5:
health_score += 0.3
if debt_to_equity < 0.5:
health_score += 0.3
return {
"current_ratio": current_ratio,
"debt_to_equity": debt_to_equity,
"health_score": health_score,
"assessment": "Strong" if health_score > 0.5 else "Weak"
}
分析工具集:
- DCF估值模型
- 比较估值法
- 财务比率分析
- 行业基准对比
- 盈利质量评估
- 现金流分析
数据源:
- 财务报表数据
- 行业平均数据
- 宏观经济指标
- 同行业公司数据
class MarketAnalyst(BaseAnalyst):
"""技术分析师 - 专注于技术指标和价格趋势分析"""
专业领域:
- 技术指标计算
- 趋势识别
- 支撑阻力位分析
- 交易信号生成
- 市场情绪分析
技术指标:
- 移动平均线 (MA, EMA)
- 相对强弱指数 (RSI)
- MACD 指标
- 布林带 (Bollinger Bands)
- 成交量指标
- 动量指标
def perform_analysis(self, data: Dict) -> Dict:
"""执行技术分析"""
price_data = data.get("price_data", {})
volume_data = data.get("volume_data", {})
# 计算技术指标
indicators = self._calculate_indicators(price_data, volume_data)
# 趋势分析
trend_analysis = self._analyze_trend(price_data, indicators)
# 支撑阻力位
support_resistance = self._find_support_resistance(price_data)
# 交易信号
signals = self._generate_signals(indicators, trend_analysis)
analysis = {
"indicators": indicators,
"trend": trend_analysis,
"support_resistance": support_resistance,
"signals": signals,
"momentum": self._assess_momentum(indicators),
"volatility": self._assess_volatility(price_data)
}
# 综合技术评分
analysis["technical_score"] = self._calculate_technical_score(analysis)
analysis["recommendation"] = self._generate_technical_recommendation(analysis)
return analysis
def _calculate_indicators(self, price_data: Dict, volume_data: Dict) -> Dict:
"""计算技术指标"""
prices = price_data.get("close", [])
volumes = volume_data.get("volume", [])
indicators = {}
# RSI 计算
indicators["rsi"] = self._calculate_rsi(prices)
# MACD 计算
indicators["macd"] = self._calculate_macd(prices)
# 移动平均线
indicators["ma_20"] = self._calculate_ma(prices, 20)
indicators["ma_50"] = self._calculate_ma(prices, 50)
# 布林带
indicators["bollinger"] = self._calculate_bollinger_bands(prices)
# 成交量指标
indicators["volume_ma"] = self._calculate_ma(volumes, 20)
return indicators
def _generate_signals(self, indicators: Dict, trend: Dict) -> Dict:
"""生成交易信号"""
signals = {
"buy_signals": [],
"sell_signals": [],
"neutral_signals": []
}
# RSI 信号
rsi = indicators.get("rsi", 50)
if rsi < 30:
signals["buy_signals"].append("RSI超卖")
elif rsi > 70:
signals["sell_signals"].append("RSI超买")
# MACD 信号
macd = indicators.get("macd", {})
if macd.get("signal") == "bullish_crossover":
signals["buy_signals"].append("MACD金叉")
elif macd.get("signal") == "bearish_crossover":
signals["sell_signals"].append("MACD死叉")
# 趋势信号
if trend.get("direction") == "uptrend":
signals["buy_signals"].append("上升趋势")
elif trend.get("direction") == "downtrend":
signals["sell_signals"].append("下降趋势")
return signals
class NewsAnalyst(BaseAnalyst):
"""新闻分析师 - 专注于新闻事件和宏观因素分析"""
专业领域:
- 新闻情感分析
- 事件影响评估
- 宏观经济分析
- 政策影响分析
- 行业动态分析
分析维度:
- 新闻情感极性
- 事件重要性评级
- 影响时间范围
- 市场反应预期
- 风险因素识别
def perform_analysis(self, data: Dict) -> Dict:
"""执行新闻分析"""
news_data = data.get("news_data", [])
economic_data = data.get("economic_data", {})
analysis = {
"sentiment_analysis": self._analyze_sentiment(news_data),
"event_impact": self._assess_event_impact(news_data),
"macro_analysis": self._analyze_macro_factors(economic_data),
"risk_factors": self._identify_risk_factors(news_data),
"catalysts": self._identify_catalysts(news_data)
}
# 综合新闻评分
analysis["news_score"] = self._calculate_news_score(analysis)
analysis["market_impact"] = self._assess_market_impact(analysis)
return analysis
def _analyze_sentiment(self, news_data: List[Dict]) -> Dict:
"""分析新闻情感"""
sentiments = []
weighted_sentiment = 0
total_weight = 0
for news in news_data:
# 计算单条新闻情感
sentiment = self._calculate_news_sentiment(news["content"])
importance = news.get("importance", 1.0)
sentiments.append({
"title": news["title"],
"sentiment": sentiment,
"importance": importance,
"source": news.get("source", "Unknown")
})
# 加权平均
weighted_sentiment += sentiment * importance
total_weight += importance
overall_sentiment = weighted_sentiment / max(total_weight, 1)
return {
"individual_sentiments": sentiments,
"overall_sentiment": overall_sentiment,
"sentiment_distribution": self._calculate_sentiment_distribution(sentiments),
"confidence": self._calculate_sentiment_confidence(sentiments)
}
def _assess_event_impact(self, news_data: List[Dict]) -> Dict:
"""评估事件影响"""
impact_assessment = {
"high_impact_events": [],
"medium_impact_events": [],
"low_impact_events": []
}
for news in news_data:
impact_score = self._calculate_impact_score(news)
event_info = {
"title": news["title"],
"impact_score": impact_score,
"time_horizon": self._estimate_time_horizon(news),
"affected_sectors": self._identify_affected_sectors(news)
}
if impact_score > 0.7:
impact_assessment["high_impact_events"].append(event_info)
elif impact_score > 0.4:
impact_assessment["medium_impact_events"].append(event_info)
else:
impact_assessment["low_impact_events"].append(event_info)
return impact_assessment
class SocialMediaAnalyst(BaseAnalyst):
"""社交媒体分析师 - 专注于社交媒体情绪和舆论分析"""
专业领域:
- 社交媒体情感分析
- 舆论趋势监测
- 热点话题识别
- 投资者情绪评估
- 病毒式传播分析
数据源:
- Reddit 讨论
- Twitter 情感
- 投资论坛
- 新闻评论
- 社交媒体提及
def perform_analysis(self, data: Dict) -> Dict:
"""执行社交媒体分析"""
social_data = data.get("social_data", {})
analysis = {
"sentiment_trends": self._analyze_sentiment_trends(social_data),
"discussion_volume": self._analyze_discussion_volume(social_data),
"key_topics": self._extract_key_topics(social_data),
"influencer_sentiment": self._analyze_influencer_sentiment(social_data),
"viral_content": self._identify_viral_content(social_data)
}
# 综合社交媒体评分
analysis["social_score"] = self._calculate_social_score(analysis)
analysis["crowd_sentiment"] = self._assess_crowd_sentiment(analysis)
return analysis
def _analyze_sentiment_trends(self, social_data: Dict) -> Dict:
"""分析情感趋势"""
reddit_data = social_data.get("reddit", [])
twitter_data = social_data.get("twitter", [])
# 时间序列情感分析
sentiment_timeline = self._build_sentiment_timeline(reddit_data, twitter_data)
# 趋势检测
trend_direction = self._detect_sentiment_trend(sentiment_timeline)
return {
"timeline": sentiment_timeline,
"trend_direction": trend_direction,
"momentum": self._calculate_sentiment_momentum(sentiment_timeline),
"volatility": self._calculate_sentiment_volatility(sentiment_timeline)
}
class AnalysisIntegrator:
"""分析结果整合器"""
def integrate_analyses(self, analyst_reports: Dict) -> Dict:
"""整合所有分析师的报告"""
integrated = {
"fundamental_score": analyst_reports.get("fundamentals", {}).get("overall_score", 0.5),
"technical_score": analyst_reports.get("technical", {}).get("technical_score", 0.5),
"news_score": analyst_reports.get("news", {}).get("news_score", 0.5),
"social_score": analyst_reports.get("social", {}).get("social_score", 0.5)
}
# 加权综合评分
weights = self.config.get("analyst_weights", {
"fundamentals": 0.3,
"technical": 0.3,
"news": 0.2,
"social": 0.2
})
integrated["composite_score"] = sum(
integrated[f"{analyst}_score"] * weights[analyst]
for analyst in weights.keys()
)
# 一致性分析
integrated["consensus_level"] = self._calculate_consensus(integrated)
return integrated
class AnalysisQualityControl:
"""分析质量控制"""
def validate_analysis(self, analysis: Dict, analyst_type: str) -> Tuple[bool, List[str]]:
"""验证分析质量"""
errors = []
# 检查必需字段
required_fields = self._get_required_fields(analyst_type)
for field in required_fields:
if field not in analysis:
errors.append(f"Missing required field: {field}")
# 检查数值范围
if not self._validate_score_ranges(analysis):
errors.append("Score values out of valid range")
# 检查逻辑一致性
if not self._validate_logical_consistency(analysis):
errors.append("Logical inconsistency detected")
return len(errors) == 0, errors
分析师团队通过专业化分工和协作机制，为交易决策提供全面、准确的市场分析，是整个 TradingAgents 系统的重要基础。