研究员团队是 TradingAgents 框架中的关键组件，负责对分析师团队的报告进行深度研究和辩论。通过看涨和看跌研究员之间的结构化辩论，系统能够从多个角度评估投资机会，形成更加平衡和全面的投资观点。
class BaseResearcher:
"""所有研究员的基础类"""
def __init__(self, llm, config, stance="neutral"):
self.llm = llm
self.config = config
self.stance = stance # "bullish", "bearish", "neutral"
self.memory = ResearcherMemory()
self.debate_history = []
def research(self, analyst_reports: Dict, context: Dict = None) -> Dict:
"""执行研究分析"""
# 1. 分析师报告解读
interpretation = self.interpret_reports(analyst_reports)
# 2. 立场分析
stance_analysis = self.analyze_from_stance(interpretation)
# 3. 生成研究观点
research_view = self.generate_research_view(stance_analysis)
# 4. 准备辩论要点
debate_points = self.prepare_debate_points(research_view)
return {
"interpretation": interpretation,
"stance_analysis": stance_analysis,
"research_view": research_view,
"debate_points": debate_points,
"confidence": self.calculate_confidence(research_view)
}
def debate(self, opponent_view: Dict, round_number: int) -> Dict:
"""参与辩论"""
# 1. 分析对手观点
opponent_analysis = self.analyze_opponent_view(opponent_view)
# 2. 准备反驳
counter_arguments = self.prepare_counter_arguments(opponent_analysis)
# 3. 强化自己观点
reinforced_view = self.reinforce_own_view(counter_arguments)
# 4. 生成辩论回应
debate_response = self.generate_debate_response(
counter_arguments, reinforced_view, round_number
)
# 5. 更新辩论历史
self.debate_history.append({
"round": round_number,
"opponent_view": opponent_view,
"response": debate_response
})
return debate_response
class BullResearcher(BaseResearcher):
"""看涨研究员 - 从乐观角度评估投资机会"""
def __init__(self, llm, config):
super().__init__(llm, config, stance="bullish")
专业特点:
- 积极寻找增长机会
- 强调正面催化剂
- 关注上涨潜力
- 挑战悲观观点
分析重点:
- 收入增长驱动因素
- 市场扩张机会
- 竞争优势分析
- 估值吸引力
- 技术创新潜力
- 管理层执行力
def analyze_from_stance(self, interpretation: Dict) -> Dict:
"""从看涨角度分析"""
bullish_factors = []
growth_opportunities = []
positive_catalysts = []
# 基本面看涨因素
fundamental_data = interpretation.get("fundamental_analysis", {})
if fundamental_data.get("revenue_growth", 0) > 0.1: # 10%以上增长
bullish_factors.append("强劲的收入增长")
if fundamental_data.get("profit_margin_trend") == "improving":
bullish_factors.append("利润率改善趋势")
# 技术面看涨信号
technical_data = interpretation.get("technical_analysis", {})
if technical_data.get("trend_direction") == "uptrend":
bullish_factors.append("技术面上升趋势")
if technical_data.get("momentum") == "positive":
bullish_factors.append("正面动量信号")
# 新闻面积极因素
news_data = interpretation.get("news_analysis", {})
if news_data.get("sentiment_score", 0) > 0.6:
positive_catalysts.append("积极的新闻情绪")
# 识别增长机会
growth_opportunities = self._identify_growth_opportunities(interpretation)
return {
"bullish_factors": bullish_factors,
"growth_opportunities": growth_opportunities,
"positive_catalysts": positive_catalysts,
"upside_potential": self._calculate_upside_potential(interpretation),
"risk_mitigation": self._identify_risk_mitigation_factors(interpretation)
}
def _identify_growth_opportunities(self, interpretation: Dict) -> List[str]:
"""识别增长机会"""
opportunities = []
# 市场扩张机会
if interpretation.get("market_size_growth", 0) > 0.05:
opportunities.append("市场规模持续扩张")
# 新产品/服务机会
if interpretation.get("new_product_pipeline"):
opportunities.append("丰富的新产品管线")
# 国际化机会
if interpretation.get("international_expansion"):
opportunities.append("国际市场扩张潜力")
# 技术创新机会
if interpretation.get("innovation_score", 0) > 0.7:
opportunities.append("技术创新领先优势")
return opportunities
def prepare_debate_points(self, research_view: Dict) -> Dict:
"""准备辩论要点"""
return {
"main_arguments": [
"基本面数据显示强劲增长潜力",
"技术指标支持上涨趋势",
"市场情绪积极向好",
"估值仍有上升空间"
],
"supporting_evidence": research_view.get("bullish_factors", []),
"growth_thesis": research_view.get("growth_opportunities", []),
"risk_responses": self._prepare_risk_responses(research_view),
"target_scenarios": self._develop_bull_scenarios(research_view)
}
def generate_debate_response(self, counter_args: Dict, reinforced_view: Dict, round_num: int) -> Dict:
"""生成辩论回应"""
response_strategy = self._determine_response_strategy(round_num)
if response_strategy == "aggressive":
return self._aggressive_response(counter_args, reinforced_view)
elif response_strategy == "defensive":
return self._defensive_response(counter_args, reinforced_view)
else:
return self._balanced_response(counter_args, reinforced_view)
def _aggressive_response(self, counter_args: Dict, reinforced_view: Dict) -> Dict:
"""积极进攻型回应"""
return {
"response_type": "aggressive",
"main_points": [
"对手过度关注短期风险，忽视长期价值",
"市场恐慌情绪创造了绝佳买入机会",
"基本面改善趋势不可逆转",
"当前估值明显低估真实价值"
],
"evidence_reinforcement": reinforced_view.get("strengthened_evidence", []),
"opponent_weaknesses": self._identify_opponent_weaknesses(counter_args),
"confidence_boost": "高度确信看涨观点的正确性"
}
def _defensive_response(self, counter_args: Dict, reinforced_view: Dict) -> Dict:
"""防守型回应"""
return {
"response_type": "defensive",
"risk_acknowledgment": "承认存在一定风险，但风险可控",
"mitigation_strategies": [
"分散投资降低单一风险",
"设置合理止损位",
"关注基本面变化",
"动态调整仓位"
],
"qualified_optimism": "在风险可控前提下保持乐观",
"evidence_reaffirmation": reinforced_view.get("core_evidence", [])
}
class BearResearcher(BaseResearcher):
"""看跌研究员 - 从悲观角度评估投资风险"""
def __init__(self, llm, config):
super().__init__(llm, config, stance="bearish")
专业特点:
- 专注风险识别
- 质疑乐观假设
- 关注下跌风险
- 挑战看涨观点
分析重点:
- 潜在风险因素
- 估值过高风险
- 竞争威胁分析
- 宏观经济风险
- 行业周期性风险
- 公司治理问题
def analyze_from_stance(self, interpretation: Dict) -> Dict:
"""从看跌角度分析"""
bearish_factors = []
risk_factors = []
negative_catalysts = []
# 基本面风险因素
fundamental_data = interpretation.get("fundamental_analysis", {})
if fundamental_data.get("debt_to_equity", 0) > 0.6:
bearish_factors.append("高负债比率风险")
if fundamental_data.get("profit_margin_trend") == "declining":
bearish_factors.append("利润率下降趋势")
# 技术面看跌信号
technical_data = interpretation.get("technical_analysis", {})
if technical_data.get("trend_direction") == "downtrend":
bearish_factors.append("技术面下降趋势")
if technical_data.get("volume_trend") == "declining":
bearish_factors.append("成交量萎缩信号")
# 新闻面负面因素
news_data = interpretation.get("news_analysis", {})
if news_data.get("sentiment_score", 0) < 0.4:
negative_catalysts.append("负面新闻情绪")
# 识别风险因素
risk_factors = self._identify_risk_factors(interpretation)
return {
"bearish_factors": bearish_factors,
"risk_factors": risk_factors,
"negative_catalysts": negative_catalysts,
"downside_potential": self._calculate_downside_potential(interpretation),
"valuation_concerns": self._assess_valuation_risks(interpretation)
}
def _identify_risk_factors(self, interpretation: Dict) -> List[Dict]:
"""识别风险因素"""
risks = []
# 市场风险
if interpretation.get("market_volatility", 0) > 0.3:
risks.append({
"type": "market_risk",
"description": "市场波动性过高",
"severity": "high",
"probability": 0.7
})
# 行业风险
if interpretation.get("industry_headwinds"):
risks.append({
"type": "industry_risk",
"description": "行业面临逆风",
"severity": "medium",
"probability": 0.6
})
# 公司特定风险
if interpretation.get("company_specific_issues"):
risks.append({
"type": "company_risk",
"description": "公司特定问题",
"severity": "high",
"probability": 0.8
})
return risks
def prepare_debate_points(self, research_view: Dict) -> Dict:
"""准备辩论要点"""
return {
"main_arguments": [
"估值过高，缺乏安全边际",
"基本面恶化趋势明显",
"技术指标显示下跌风险",
"宏观环境不利于增长"
],
"risk_evidence": research_view.get("risk_factors", []),
"valuation_concerns": research_view.get("valuation_concerns", []),
"scenario_analysis": self._develop_bear_scenarios(research_view),
"contrarian_points": self._prepare_contrarian_arguments(research_view)
}
def _assess_valuation_risks(self, interpretation: Dict) -> Dict:
"""评估估值风险"""
valuation_data = interpretation.get("fundamental_analysis", {})
concerns = []
# P/E 比率分析
pe_ratio = valuation_data.get("pe_ratio", 0)
industry_avg_pe = valuation_data.get("industry_avg_pe", 0)
if pe_ratio > industry_avg_pe * 1.5:
concerns.append("P/E比率显著高于行业平均")
# P/B 比率分析
pb_ratio = valuation_data.get("pb_ratio", 0)
if pb_ratio > 3.0:
concerns.append("P/B比率过高，存在泡沫风险")
# 增长率与估值匹配度
growth_rate = valuation_data.get("growth_rate", 0)
peg_ratio = pe_ratio / max(growth_rate * 100, 1)
if peg_ratio > 2.0:
concerns.append("PEG比率过高，增长不足以支撑估值")
return {
"concerns": concerns,
"overvaluation_risk": "high" if len(concerns) >= 2 else "medium",
"fair_value_estimate": self._calculate_conservative_fair_value(valuation_data),
"downside_protection": self._assess_downside_protection(valuation_data)
}
class DebateManager:
"""辩论管理器 - 协调研究员之间的辩论"""
def __init__(self, config):
self.config = config
self.max_rounds = config.get("max_debate_rounds", 3)
self.consensus_threshold = config.get("consensus_threshold", 0.8)
def conduct_debate(self, bull_researcher: BullResearcher,
bear_researcher: BearResearcher,
analyst_reports: Dict) -> Dict:
"""主持辩论过程"""
# 初始研究
bull_initial = bull_researcher.research(analyst_reports)
bear_initial = bear_researcher.research(analyst_reports)
debate_history = []
current_round = 1
while current_round <= self.max_rounds:
print(f"=== 辩论第 {current_round} 轮 ===")
# 看涨方辩论
bull_response = bull_researcher.debate(bear_initial, current_round)
# 看跌方辩论
bear_response = bear_researcher.debate(bull_initial, current_round)
# 记录辩论
round_record = {
"round": current_round,
"bull_response": bull_response,
"bear_response": bear_response,
"consensus_level": self._calculate_consensus(bull_response, bear_response)
}
debate_history.append(round_record)
# 检查是否达成共识或需要继续
if self._should_continue_debate(round_record, current_round):
current_round += 1
# 更新观点用于下一轮
bull_initial = bull_response
bear_initial = bear_response
else:
break
# 生成最终共识
final_consensus = self._generate_consensus(debate_history, bull_initial, bear_initial)
return {
"debate_history": debate_history,
"final_consensus": final_consensus,
"total_rounds": current_round,
"consensus_quality": self._assess_consensus_quality(final_consensus)
}
def _calculate_consensus(self, bull_view: Dict, bear_view: Dict) -> float:
"""计算共识水平"""
# 提取关键观点
bull_confidence = bull_view.get("confidence", 0.5)
bear_confidence = bear_view.get("confidence", 0.5)
# 计算观点差异
confidence_diff = abs(bull_confidence - bear_confidence)
# 共识水平 = 1 - 观点差异
consensus_level = 1.0 - confidence_diff
return max(0.0, min(1.0, consensus_level))
def _should_continue_debate(self, round_record: Dict, current_round: int) -> bool:
"""判断是否继续辩论"""
# 达到最大轮次
if current_round >= self.max_rounds:
return False
# 达成足够共识
if round_record["consensus_level"] >= self.consensus_threshold:
return False
# 观点分歧仍然较大，继续辩论
return True
def _generate_consensus(self, debate_history: List[Dict],
final_bull: Dict, final_bear: Dict) -> Dict:
"""生成最终共识"""
# 综合双方观点
consensus_points = []
# 提取共同认可的要点
bull_factors = set(final_bull.get("main_points", []))
bear_factors = set(final_bear.get("main_points", []))
common_points = bull_factors.intersection(bear_factors)
consensus_points.extend(list(common_points))
# 平衡风险和机会
opportunities = final_bull.get("growth_opportunities", [])
risks = final_bear.get("risk_factors", [])
# 生成平衡的投资建议
if len(opportunities) > len(risks):
recommendation = "谨慎乐观"
confidence = 0.6
elif len(risks) > len(opportunities):
recommendation = "谨慎悲观"
confidence = 0.4
else:
recommendation = "中性观望"
confidence = 0.5
return {
"consensus_points": consensus_points,
"balanced_view": {
"opportunities": opportunities[:3], # 前3个机会
"risks": risks[:3], # 前3个风险
},
"recommendation": recommendation,
"confidence": confidence,
"key_factors_to_watch": self._identify_key_factors(final_bull, final_bear)
}
class DebateQualityAssessor:
"""辩论质量评估器"""
def assess_debate_quality(self, debate_result: Dict) -> Dict:
"""评估辩论质量"""
quality_metrics = {
"depth_score": self._assess_depth(debate_result),
"balance_score": self._assess_balance(debate_result),
"evidence_quality": self._assess_evidence_quality(debate_result),
"logical_consistency": self._assess_logical_consistency(debate_result),
"consensus_quality": self._assess_consensus_quality(debate_result)
}
overall_quality = sum(quality_metrics.values()) / len(quality_metrics)
return {
"quality_metrics": quality_metrics,
"overall_quality": overall_quality,
"quality_grade": self._assign_quality_grade(overall_quality),
"improvement_suggestions": self._generate_improvement_suggestions(quality_metrics)
}
研究员团队通过结构化的辩论机制，确保投资决策考虑了多个角度和潜在风险，提高了决策的质量和可靠性。