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)
参数 | 类型 | 默认值 | 描述 |
---|---|---|---|
debug | bool | False | 启用详细日志和调试输出 |
config | dict | None | 用于自定义行为的配置字典 |
核心方法
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.7 | 0.0-2.0 |
智能体行为
参数 | 描述 | 默认值 | 选项 |
---|---|---|---|
max_debate_rounds | 研究员辩论轮数 | 3 | 1-5 |
consensus_threshold | 一致性阈值 | 0.6 | 0.0-1.0 |
risk_tolerance | 风险接受水平 | "moderate" | "low" , "moderate" , "high" |
数据源
参数 | 描述 | 默认值 | 选项 |
---|---|---|---|
online_tools | 使用实时数据 | False | True , False |
data_cache_ttl | 缓存生存时间 | 3600 | 秒数 |
fallback_to_cache | API失败时使用缓存 | True | True , 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))
下一步
- 了解配置详情
- 探索Python使用示例
- 理解数据源