多智能体交易系统：TradingAgents 开源架构深度解析与行业预判

# ===== 辩论协议：Agent 冲突解决 =====
class DebateProtocol:
    def __init__(self, max_rounds=3, threshold=0.7):
        self.max_rounds = max_rounds
        self.threshold = threshold
    def run_debate(self, agent_a, agent_b):
        for r in range(1, self.max_rounds + 1):
            agent_a["confidence"] *= (1 - self._impact(agent_b))
            agent_b["confidence"] *= (1 - self._impact(agent_a))
            if abs(agent_a["confidence"] - agent_b["confidence"]) < self.threshold:
                return {"result": "consensus", "rounds": r}
        return {"result": "deadlock", "requires_human_review": True}
    def _impact(self, agent):
        return min(0.3, len(agent.get("arguments", [])) * 0.1)
# result = DebateProtocol().run_debate(analyst, risk_mgr)
# print(f"辩论结果: {result['result']}")

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Any, Optional
import json
import time

# ===== 消息格式定义 =====
class MessageType(Enum):
    RESEARCH = "research"       # 研究报告
    ANALYSIS = "analysis"       # 分析意见
    RISK = "risk"               # 风险评级
    TRADE = "trade"             # 交易指令
    DEBATE = "debate"           # 辩论发言

@dataclass
class AgentMessage:
    """Agent 之间的结构化消息"""
    msg_type: MessageType
    sender: str
    content: str
    confidence: float            # 0.0 ~ 1.0
    evidence: List[str]          # 支撑结论的关键证据
    timestamp: float = field(default_factory=time.time)

# ===== Agent 角色定义 =====
@dataclass
class AgentRole:
    """Agent 角色：定义职责、工具和输出格式"""
    name: str
    system_prompt: str
    tools: List[str]
    output_format: Dict[str, Any]

# 四个核心角色
ROLES = {
    "researcher": AgentRole(
        name="Researcher",
        system_prompt="""你是金融研究员。你的职责：
1. 收集目标股票的市场数据和新闻
2. 过滤无关信息，提取关键发现
3. 输出结构化研究报告，包含置信度评估""",
        tools=["get_price", "get_news", "get_financials"],
        output_format={"findings": [], "confidence": 0.0, "key_metrics": {}}
    ),
    "analyst": AgentRole(
        name="Analyst",
        system_prompt="""你是投资分析师。你的职责：
1. 基于研究报告进行多维度分析
2. 从基本面、技术面、情绪面给出评分
3. 输出综合投资建议（买入/卖出/持有）""",
        tools=["technical_analysis", "sentiment_analysis"],
        output_format={"recommendation": "", "score": 0.0, "reasoning": ""}
    ),
    "risk_manager": AgentRole(
        name="RiskManager",
        system_prompt="""你是风险管理官。你的职责：
1. 评估交易的风险等级
2. 检查风险敞口、VaR、最大回撤
3. 如果风险超限，否决交易或建议减仓""",
        tools=["check_exposure", "calculate_var", "check_concentration"],
        output_format={"risk_level": "", "max_position": 0.0, "warnings": []}
    ),
    "trader": AgentRole(
        name="Trader",
        system_prompt="""你是投资经理。你的职责：
1. 综合各方意见做出最终决策
2. 平衡收益目标与风险约束
3. 生成交易指令（方向、仓位、价格）""",
        tools=["place_order", "check_portfolio"],
        output_format={"action": "", "position_size": 0.0, "stop_loss": 0.0}
    ),
}

# ===== 协作编排器 =====
class TradingOrchestrator:
    """多智能体交易协作编排器"""
    
    def __init__(self, roles: Dict[str, AgentRole]):
        self.roles = roles
        self.message_log: List[AgentMessage] = []
    
    def execute_chain(self, ticker: str) -> Dict[str, Any]:
        """执行链式协作流程：Researcher → Analyst → Risk → Trader"""
        print(f"=== 开始分析 {ticker} ===")
        
        # Step 1: Researcher 收集信息
        research_msg = self._call_agent("researcher", {
            "ticker": ticker,
            "task": "收集市场数据和新闻"
        })
        self.message_log.append(research_msg)
        print(f"[Researcher] 置信度: {research_msg.confidence:.2f}")
        
        # Step 2: Analyst 基于研究报告做分析
        analysis_msg = self._call_agent("analyst", {
            "research": research_msg.content,
            "ticker": ticker
        })
        self.message_log.append(analysis_msg)
        print(f"[Analyst] 建议: {analysis_msg.content}")
        
        # Step 3: Risk Manager 评估风险
        risk_msg = self._call_agent("risk_manager", {
            "proposed_trade": analysis_msg.content,
            "ticker": ticker
        })
        self.message_log.append(risk_msg)
        print(f"[RiskManager] 风险等级: {risk_msg.content}")
        
        # 检查是否需要辩论
        if self._needs_debate(analysis_msg, risk_msg):
            print("[Debate] 检测到意见冲突，启动辩论流程...")
            debate_result = self._run_debate(
                analysis_msg, risk_msg, max_rounds=3
            )
            print(f"[Debate] 结果: {debate_result}")
        
        # Step 4: Trader 做最终决策
        trade_msg = self._call_agent("trader", {
            "analysis": analysis_msg,
            "risk": risk_msg,
            "ticker": ticker
        })
        self.message_log.append(trade_msg)
        print(f"[Trader] 最终决策: {trade_msg.content}")
        
        return self._parse_trade_decision(trade_msg)
    
    def _needs_debate(self, analysis: AgentMessage, risk: AgentMessage) -> bool:
        """判断是否需要启动辩论流程"""
        buy_signals = ["买入", "buy", "bullish"]
        high_risk = ["高风险", "high", "critical"]
        
        is_buy = any(s in analysis.content.lower() for s in buy_signals)
        is_high_risk = any(r in risk.content.lower() for r in high_risk)
        
        return is_buy and is_high_risk
    
    def _run_debate(self, analyst_msg: AgentMessage, risk_msg: AgentMessage,
                    max_rounds: int = 3) -> str:
        """执行辩论协议"""
        for round_num in range(1, max_rounds + 1):
            print(f"  --- 辩论第 {round_num} 轮 ---")
            # 实际实现中会调用 LLM 进行辩论
            # 这里展示框架逻辑
            pass
        return "compromise"  # 可能的结果: consensus/compromise/deadlock

# ===== 回测框架：验证多智能体策略 =====
# 在实盘之前，必须通过历史回测验证策略有效性

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MultiAgentBacktester:
    """多智能体交易策略回测框架"""
    
    def __init__(self, initial_capital: float = 1_000_000):
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.positions: Dict[str, float] = {}
        self.trade_log: List[Dict] = []
        self.equity_curve: List[float] = []
    
    def run_backtest(
        self,
        data: pd.DataFrame,
        start_date: str,
        end_date: str,
    ) -> Dict[str, float]:
        """运行回测并计算绩效指标"""
        period = data.loc[start_date:end_date]
        
        for date, row in period.iterrows():
            # 模拟多 Agent 决策
            signal = self._simulate_agent_decision(row)
            
            if signal["action"] == "buy" and not self.positions.get(signal["ticker"]):
                self._execute_buy(signal, row)
            elif signal["action"] == "sell" and self.positions.get(signal["ticker"]):
                self._execute_sell(signal, row)
            
            # 记录每日净值
            self.equity_curve.append(self._calculate_equity(row))
        
        return self._calculate_metrics()
    
    def _simulate_agent_decision(self, row: pd.Series) -> Dict:
        """模拟多 Agent 协作决策（简化版）"""
        # 实际实现中会调用真实 Agent
        # 这里用规则模拟
        signals = {}
        
        # Researcher 信号：基于价格和新闻
        signals["research"] = {
            "price_trend": "up" if row["sma_20"] > row["sma_50"] else "down",
            "news_sentiment": row.get("sentiment", 0),
        }
        
        # Analyst 信号：技术指标
        signals["analyst"] = {
            "rsi": row.get("rsi", 50),
            "macd_signal": row.get("macd", 0) > row.get("macd_signal", 0),
        }
        
        # Risk Manager 信号：波动率检查
        signals["risk"] = {
            "volatility": row.get("volatility_20d", 0.2),
            "acceptable": row.get("volatility_20d", 0.2) < 0.35,
        }
        
        # Trader 综合决策
        if (signals["research"]["price_trend"] == "up" 
            and signals["analyst"]["rsi"] < 70
            and signals["risk"]["acceptable"]):
            return {"action": "buy", "ticker": row.get("ticker", "AAPL"),
                    "confidence": 0.75}
        elif signals["analyst"]["rsi"] > 80 or not signals["risk"]["acceptable"]:
            return {"action": "sell", "ticker": row.get("ticker", "AAPL"),
                    "confidence": 0.8}
        
        return {"action": "hold", "ticker": row.get("ticker", "AAPL")}
    
    def _calculate_metrics(self) -> Dict[str, float]:
        """计算回测绩效指标"""
        equity = pd.Series(self.equity_curve)
        returns = equity.pct_change().dropna()
        
        total_return = (equity.iloc[-1] / self.initial_capital) - 1
        annual_return = total_return * (252 / len(returns))
        annual_vol = returns.std() * np.sqrt(252)
        sharpe = annual_return / annual_vol if annual_vol > 0 else 0
        
        # 最大回撤
        peak = equity.cummax()
        drawdown = (equity - peak) / peak
        max_drawdown = drawdown.min()
        
        return {
            "total_return": total_return,
            "annual_return": annual_return,
            "sharpe_ratio": sharpe,
            "max_drawdown": max_drawdown,
            "num_trades": len(self.trade_log),
            "win_rate": self._calculate_win_rate(),
        }
    
    def _calculate_win_rate(self) -> float:
        profitable = sum(1 for t in self.trade_log if t.get("pnl", 0) > 0)
        return profitable / len(self.trade_log) if self.trade_log else 0

# 使用示例
# backtester = MultiAgentBacktester(initial_capital=1_000_000)
# metrics = backtester.run_backtest(data, "2024-01-01", "2025-12-31")
# print(f"年化收益率: {metrics['annual_return']*100:.1f}%")
# print(f"夏普比率: {metrics['sharpe_ratio']:.2f}")
# print(f"最大回撤: {metrics['max_drawdown']*100:.1f}%")
# print(f"胜率: {metrics['win_rate']*100:.1f}%")

# ===== 框架集成示例：将 dexter 研究模块接入 TradingAgents =====
# dexter 提供深度金融研究，TradingAgents 负责多 Agent 决策
# 这是两种框架互补使用的最佳实践

from trading_agents import TradingOrchestrator
from dexter import FinancialResearchAgent

# 初始化 dexter 研究 Agent（深度金融分析专家）
researcher = FinancialResearchAgent(
    model="claude-sonnet-4",
    focus_areas=["fundamental", "macro", "sentiment"],
    output_format="structured_report",  # 输出结构化报告
)

# 初始化 TradingAgents 编排器
orchestrator = TradingOrchestrator()

# 用 dexter 做深度研究，用 TradingAgents 做交易决策
def analyze_and_trade(ticker: str):
    # Step 1: dexter 生成深度研究报告
    report = researcher.generate_report(ticker)
    print(f"dexter 研究报告: {len(report.sections)} 个章节")
    print(f"核心观点: {report.summary}")
    
    # Step 2: 将研究报告注入 TradingAgents 的 Researcher
    orchestrator.inject_research(report.to_dict())
    
    # Step 3: TradingAgents 完成分析→风控→交易决策
    decision = orchestrator.execute_decision_flow(ticker)
    return decision

# 优势：dexter 的研究深度 + TradingAgents 的决策广度
# 比单独使用任一框架都有更好的决策质量

# 使用示例：
# result = analyze_and_trade("AAPL")
# print(f"最终决策: {result['action']}, 仓位: {result['position_size']}")

# ===== 框架集成：dexter 研究 + TradingAgents 决策 =====
from trading_agents import TradingOrchestrator
from dexter import FinancialResearchAgent

# dexter 负责深度金融研究
researcher = FinancialResearchAgent(
    focus_areas=["fundamental", "macro"],
    output_format="structured_report",
)

# TradingAgents 负责多 Agent 协作决策
orchestrator = TradingOrchestrator()

def analyze_and_trade(ticker):
    # Step 1: dexter 生成深度研究报告
    report = researcher.generate_report(ticker)
    print(f"研究报告: {len(report.sections)} 章节")

    # Step 2: 注入 TradingAgents
    orchestrator.inject_research(report.to_dict())

    # Step 3: 完成分析 → 风控 → 决策
    return orchestrator.execute_decision_flow(ticker)

# result = analyze_and_trade("AAPL")
# print(f"最终决策: {result['action']}, 仓位: {result['position_size']}")