跳到主要内容

TradingAgentsGraph API

TradingAgentsGraph 是 TradingAgents 框架的核心类。它协调多智能体工作流程,并提供运行交易分析的主要接口。

类概述

from tradingagents.graph.trading_graph import TradingAgentsGraph
from tradingagents.default_config import DEFAULT_CONFIG

# 初始化图
ta = TradingAgentsGraph(debug=True, config=DEFAULT_CONFIG.copy())

构造函数参数

TradingAgentsGraph(debug=False, config=None)

参数类型默认值描述
debugboolFalse启用详细日志和调试输出
configdictNone用于自定义行为的配置字典

核心方法

propagate(symbol, date)

运行特定股票和日期交易分析的主要方法。

state, decision = ta.propagate("NVDA", "2024-05-10")

参数:

  • symbol (str): 股票代码(例如:"NVDA"、"AAPL"、"GOOGL")
  • date (str): YYYY-MM-DD 格式的分析日期

返回值:

  • state (dict): 包含所有智能体输出的完整分析状态
  • decision (dict): 最终交易决策和推理

返回值结构

状态对象

状态对象包含每个智能体的中间结果:

{
"fundamentals_analysis": {...}, # 基本面分析
"sentiment_analysis": {...}, # 情绪分析
"news_analysis": {...}, # 新闻分析
"technical_analysis": {...}, # 技术分析
"bullish_research": {...}, # 看涨研究
"bearish_research": {...}, # 看跌研究
"debate_results": {...}, # 辩论结果
"risk_assessment": {...}, # 风险评估
"trader_decision": {...} # 交易员决策
}

决策对象

决策对象包含最终交易建议:

{
"action": "BUY", # BUY(买入)、SELL(卖出)或 HOLD(持有)
"confidence": 8.5, # 置信度评分 0-10
"reasoning": "强劲的基本面和积极的情绪...",
"risk_level": "MODERATE", # LOW(低)、MODERATE(中等)、HIGH(高)
"key_factors": [
"收益增长加速",
"技术突破模式",
"积极的分析师情绪"
],
"risk_warnings": [
"高市场波动性",
"行业集中风险"
],
"agent_consensus": {
"bullish_agents": 3, # 看涨智能体数量
"bearish_agents": 1, # 看跌智能体数量
"neutral_agents": 0 # 中性智能体数量
},
"portfolio_impact": {
"position_size": "2-3%", # 建议仓位大小
"expected_volatility": "15-20%", # 预期波动率
"correlation_risk": "LOW" # 相关性风险
}
}

配置管理

使用默认配置

from tradingagents.default_config import DEFAULT_CONFIG

# 查看默认设置
print(DEFAULT_CONFIG)

# 使用默认配置
ta = TradingAgentsGraph(config=DEFAULT_CONFIG.copy())

自定义配置

# 创建自定义配置
config = DEFAULT_CONFIG.copy()

# 修改特定设置
config["deep_think_llm"] = "gpt-4o-mini"
config["quick_think_llm"] = "gpt-4o-mini"
config["max_debate_rounds"] = 2
config["online_tools"] = True

# 使用自定义配置初始化
ta = TradingAgentsGraph(debug=True, config=config)

配置选项

LLM 设置

参数描述默认值选项
deep_think_llm复杂分析模型"o1-preview""o1-preview", "gpt-4o", "gpt-4o-mini"
quick_think_llm快速响应模型"gpt-4o""gpt-4o", "gpt-4o-mini"
temperature模型创造性水平0.70.0-2.0

智能体行为

参数描述默认值选项
max_debate_rounds研究员辩论轮数31-5
consensus_threshold一致性阈值0.60.0-1.0
risk_tolerance风险接受水平"moderate""low", "moderate", "high"

数据源

参数描述默认值选项
online_tools使用实时数据FalseTrue, False
data_cache_ttl缓存生存时间3600秒数
fallback_to_cacheAPI失败时使用缓存TrueTrue, False

错误处理

常见异常

from tradingagents.exceptions import (
TradingAgentsError,
APIConnectionError,
InvalidSymbolError,
ConfigurationError
)

try:
state, decision = ta.propagate("INVALID", "2024-05-10")
except InvalidSymbolError as e:
print(f"无效的股票代码: {e}")
except APIConnectionError as e:
print(f"API连接失败: {e}")
except TradingAgentsError as e:
print(f"框架错误: {e}")

重试逻辑

import time
from tradingagents.exceptions import APIConnectionError

def robust_analysis(symbol, date, max_retries=3):
for attempt in range(max_retries):
try:
return ta.propagate(symbol, date)
except APIConnectionError:
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # 指数退避
continue
raise

性能优化

批量处理

def analyze_portfolio(symbols, date):
results = {}
for symbol in symbols:
try:
_, decision = ta.propagate(symbol, date)
results[symbol] = decision
except Exception as e:
results[symbol] = {"error": str(e)}
return results

# 分析多只股票
portfolio = ["AAPL", "GOOGL", "MSFT", "NVDA"]
results = analyze_portfolio(portfolio, "2024-05-10")

缓存结果

import pickle
from pathlib import Path

def cached_analysis(symbol, date, cache_dir="./cache"):
cache_path = Path(cache_dir) / f"{symbol}_{date}.pkl"

# 检查缓存
if cache_path.exists():
with open(cache_path, 'rb') as f:
return pickle.load(f)

# 运行分析
state, decision = ta.propagate(symbol, date)

# 保存到缓存
cache_path.parent.mkdir(exist_ok=True)
with open(cache_path, 'wb') as f:
pickle.dump((state, decision), f)

return state, decision

高级用法

自定义智能体配置

# 配置个别智能体行为
config = DEFAULT_CONFIG.copy()
config["agent_configs"] = {
"fundamentals_analyst": {
"focus_areas": ["earnings", "growth", "valuation"],
"analysis_depth": "deep"
},
"technical_analyst": {
"indicators": ["RSI", "MACD", "BB", "SMA"],
"timeframes": ["1d", "1w", "1m"]
}
}

工作流程自定义

# 自定义工作流程步骤
config["workflow_steps"] = [
"data_collection", # 数据收集
"individual_analysis", # 个别分析
"researcher_debate", # 研究员辩论
"risk_assessment", # 风险评估
"final_decision" # 最终决策
]

# 跳过某些步骤
config["skip_steps"] = ["researcher_debate"] # 更快执行

集成示例

与 Pandas 集成

import pandas as pd

def create_analysis_dataframe(symbols, dates):
results = []
for symbol in symbols:
for date in dates:
_, decision = ta.propagate(symbol, date)
results.append({
'symbol': symbol,
'date': date,
'action': decision['action'],
'confidence': decision['confidence'],
'risk_level': decision['risk_level']
})
return pd.DataFrame(results)

# 创建分析 DataFrame
df = create_analysis_dataframe(['AAPL', 'GOOGL'], ['2024-05-01', '2024-05-02'])

与 Jupyter Notebooks 集成

# 在笔记本中显示丰富输出
from IPython.display import display, JSON

state, decision = ta.propagate("NVDA", "2024-05-10")

# 显示格式化的决策
display(JSON(decision))

下一步