trader

交易员智能体

概述

交易员智能体是 TradingAgents 框架的核心决策组件，负责综合分析师报告和研究员辩论结果，制定最终的交易决策。交易员智能体具备专业的交易知识和风险意识，能够在复杂的市场环境中做出明智的投资决策。

交易员架构

基础交易员类

class Trader:
    """交易员智能体 - 负责最终交易决策"""
    
    def __init__(self, llm, config):
        self.llm = llm
        self.config = config
        self.trading_style = config.get("trading_style", "balanced")
        self.risk_tolerance = config.get("risk_tolerance", "medium")
        self.memory = TradingMemory()
        self.position_manager = PositionManager()
        
    def make_decision(self, analysis_data: Dict) -> Dict:
        """制定交易决策"""
        
        # 1. 综合分析所有输入
        comprehensive_analysis = self.synthesize_analysis(analysis_data)
        
        # 2. 评估市场条件
        market_assessment = self.assess_market_conditions(analysis_data)
        
        # 3. 制定交易策略
        trading_strategy = self.develop_trading_strategy(
            comprehensive_analysis, market_assessment
        )
        
        # 4. 确定仓位大小
        position_size = self.calculate_position_size(trading_strategy)
        
        # 5. 设置风险管理参数
        risk_parameters = self.set_risk_parameters(trading_strategy)
        
        # 6. 生成最终决策
        final_decision = self.generate_final_decision(
            trading_strategy, position_size, risk_parameters
        )
        
        # 7. 更新交易记忆
        self.memory.update_decision(final_decision)
        
        return final_decision

核心功能模块

1. 分析综合模块

def synthesize_analysis(self, analysis_data: Dict) -> Dict:
    """综合分析所有输入数据"""
    
    # 提取各类分析结果
    analyst_reports = analysis_data.get("analyst_reports", {})
    research_consensus = analysis_data.get("research_consensus", {})
    market_data = analysis_data.get("market_data", {})
    
    # 分析师报告权重
    analyst_weights = self.config.get("analyst_weights", {
        "fundamentals": 0.3,
        "technical": 0.3,
        "news": 0.2,
        "social": 0.2
    })
    
    # 计算加权分析评分
    weighted_scores = {}
    total_score = 0
    
    for analyst_type, weight in analyst_weights.items():
        if analyst_type in analyst_reports:
            score = analyst_reports[analyst_type].get("overall_score", 0.5)
            weighted_scores[analyst_type] = score * weight
            total_score += weighted_scores[analyst_type]
    
    # 研究员共识影响
    consensus_impact = self._assess_consensus_impact(research_consensus)
    
    # 综合评估
    synthesis = {
        "analyst_scores": weighted_scores,
        "total_analyst_score": total_score,
        "consensus_impact": consensus_impact,
        "adjusted_score": self._adjust_score_with_consensus(total_score, consensus_impact),
        "confidence_level": self._calculate_overall_confidence(analyst_reports, research_consensus),
        "key_factors": self._extract_key_factors(analyst_reports, research_consensus)
    }
    
    return synthesis

def _assess_consensus_impact(self, research_consensus: Dict) -> Dict:
    """评估研究员共识的影响"""
    
    consensus_strength = research_consensus.get("consensus_level", 0.5)
    recommendation = research_consensus.get("recommendation", "neutral")
    
    # 共识强度影响权重
    if consensus_strength > 0.8:
        impact_weight = 0.3  # 高共识，高影响
    elif consensus_strength > 0.6:
        impact_weight = 0.2  # 中等共识，中等影响
    else:
        impact_weight = 0.1  # 低共识，低影响
    
    # 推荐方向影响
    direction_impact = {
        "谨慎乐观": 0.15,
        "谨慎悲观": -0.15,
        "中性观望": 0.0
    }.get(recommendation, 0.0)
    
    return {
        "strength": consensus_strength,
        "weight": impact_weight,
        "direction": direction_impact,
        "net_impact": direction_impact * impact_weight
    }