LangGraph智能交易实战：让多个AI智能体像投资团队一样协作（怎么可以错过）ai智能化交易，

1.triangle智能交易

现在的量化交易早就不是简单的技术指标了真正有效的交易系统需要像一个完整的投资团队一样工作——有专门的分析师收集各种数据，有研究员进行深度分析和辩论，有交易员制定具体策略，还有风险管理团队把关问题是，传统的程序很难模拟这种复杂的协作流程。

2.智能交易百科

LangGraph的多智能体架构正好解决了这个问题我们可以构建一个像真实投资公司一样运作的系统，每个汇通网手机版app下载安装苹果智能体负责特定的职能，它们之间可以进行辩论、协商，最终形成完整的投资决策链条整个系统的工作流程是这样的：。

3.智能ai交易系统

首先是数据收集阶段，专门的分析师智能体会从各个维度收集市场情报，包括技术指标、新闻资讯、社交媒体情绪、公司基本面等等然后多头和空头智能体会针对这些数据进行对抗性辩论，这个过程很关键，能够暴露出投资逻辑中的漏洞。

4.智能交易是个什么平台

研究经理会综合双方观点，形成最终的投资策略接下来交易员智能体会把策略转化为具体的执行方案，包括进场时机、仓位大小、止损设置等细节这个方案还要经过风险管理团队的多重审核——激进派、保守派、中性派三个角色会从不同角度评估风险。

5.智能交易软件平汇通网手机版app下载安装苹果台

最后由投资组合经理做出最终决策，系统会自动提取可执行的交易信号整个过程还有学习能力，每次交易结束后，各个智能体都会反思这次决策的得失，把经验存储到长期记忆中，用于优化后续的决策我们来看看具体怎么实现环境准备和基础配置

6.ai智能化交易

做多智能体系统，观测性非常重要LangSmith的追踪功能可以让我们清楚地看到每个智能体的决策过程，这对于调试和优化系统来说是必需的# 首先，确保你已经安装了必要的库 # !pip install -U langchain langgraph langchain_openai tavily-python yfinance finnhub-python stock汇通网手机版app下载安装苹果stats beautifulsoup4 chromadb rich 。

7.智能交易是什么

import os from getpass import getpass # 定义一个辅助函数来安全地设置环境变量def _set_env(var: str): # 如果环境变量尚未设置，提示用户输入

8.智能交易工具

if not os.environ.get(var): os.environ[var] = getpass(f”Enter your {var}: “) # 为我们将使用的服务设置API密钥_set_env(“OPENAI_API_KEY”)

9.ai智能交易系统有哪些

_set_env(“FI汇通网手机版app下载安装苹果NNHUB_API_KEY”) _set_env(“TAVILY_API_KEY”) _set_env(“LANGSMITH_API_KEY”) # 启用LangSmith追踪以完全观察我们的智能体系统

10.智能交易平台安全可靠吗

os.environ[“LANGSMITH_TRACING”] = “true” # 为LangSmith定义项目名称以组织追踪os.environ[“LANGSMITH_PROJECT”] = “Deep-Trading-System”

这里用到了几个关键的API服务：OpenAI负责运行LLM（当然你也可以用其他开源模型），Finnhub提供实时股市数据，Tavil汇通网手机版app下载安装苹果y负责网络搜索和新闻抓取，LangSmith做系统监控系统的核心是一个配置字典，这相当于整个系统的控制面板：。

from pprint import pprint # 为这个notebook运行定义我们的中央配置config = { “results_dir”: “./results”, # LLM设置指定用于不同认知任务的模型

“llm_provider”: “openai”, “deep_think_llm”: “gpt-4o”, # 用于复杂推理和最终决策的强大模型”quick_think_llm”: “gpt-4o-mini”, # 用于数据处理和初始分析的快速、便宜模型

“backend_汇通网手机版app下载安装苹果url”: “https://api.openai.com/v1″, # 辩论和讨论设置控制协作智能体的流程”max_debate_rounds”: 2, # 多头vs空头辩论将有2轮

“max_risk_discuss_rounds”: 1, # 风险团队有1轮辩论”max_recur_limit”: 100, # 智能体循环的安全限制# 工具设置控制数据获取行为”online_tools”: True, # 使用实时API；设置为False以使用缓存数据进行更快、更便宜的运行

“data_cache_dir”: “./data_cache” # 缓存在线数据的目录} # 如果缓存目录不存在则创汇通网手机版app下载安装苹果建它os.makedirs(config[“data_cache_dir”], exist_ok=True)

print(“Configuration dictionary created:”) pprint(config)这个配置里有个比较巧妙的设计：用两套不同的模型gpt-4o用来处理复杂推理和关键决策，gpt-4o-mini用来做数据处理这些相对简单的任务。

这样既保证了决策质量，又控制了成本另外几个参数也很重要：max_debate_rounds控制多空双方辩论的轮数，max_risk_discuss_rounds决定风险团队讨论的深度，online_tools开关可以让我们在实时数据和缓汇通网手机版app下载安装苹果存数据之间切换。

from langchain_openai import ChatOpenAI # 初始化用于高风险推理任务的强大LLMdeep_thinking_llm = ChatOpenAI( model=config[“deep_think_llm”],

base_url=config[“backend_url”], temperature=0.1 ) # 初始化用于常规数据处理的更快、更经济的LLMquick_thinking_llm = ChatOpenAI(

model=config[“quick_think_llm”], base_url=config[“backend_url”]汇通网手机版app下载安装苹果, temperature=0.1 )注意这里的temperature设为0.1，这是因为金融分析需要的是稳定、可重复的输出，而不是太多创造性。

系统状态设计整个多智能体系统的核心是一个共享的状态管理机制。这就像是所有智能体共享的工作台，每个智能体都可以从中读取信息，也可以把自己的分析结果写入其中。

在LangGraph中，这个状态会在所有节点之间传递，记录着整个决策过程的完整信息流我们用Python的TypedDict来定义数据结构，确保类型安全from typing import Annotated, Sequence, List 。

from typing_extensions import 汇通网手机版app下载安装苹果TypedDict from langgraph.graph import MessagesState # 研究团队辩论的状态，作为专用记事本

class InvestDebateState(TypedDict): bull_history: str # 存储多头智能体的论据bear_history: str # 存储空头智能体的论据history: str # 辩论的完整记录

current_response: str # 最近的论据judge_decision: str # 经理的最终决定count: int # 追踪辩论轮数的计数器# 风险管理团队辩论的状态class RiskDebateS汇通网手机版app下载安装苹果tate(TypedDict):

risky_history: str # 激进风险承担者的历史safe_history: str # 保守智能体的历史neutral_history: str # 平衡智能体的历史history: str # 风险讨论的完整记录

latest_speaker: str # 追踪最后发言的智能体current_risky_response: str current_safe_response: str current_neutral_response: str

judge_decision: str # 投资组合经理的最终决定count: int # 风险讨论轮数的计汇通网手机版app下载安装苹果数器这种设计的好处是把不同的辩论过程隔离开来，避免相互干扰history字段会记录完整的辩论过程，count参数帮助系统判断什么时候该结束辩论。

# 将通过整个图传递的主要状态# 它从MessagesState继承，包含聊天历史的messages字段class AgentState(MessagesState): company_of_interest: str # 我们正在分析的股票代码

trade_date: str # 分析的日期sender: str # 追踪哪个智能体最后修改了状态# 每个分析师将填充其自己的报告字段market_report: str sentiment_report:汇通网手机版app下载安装苹果 str

news_report: str fundamentals_report: str # 辩论的嵌套状态investment_debate_state: InvestDebateState investment_plan: str # 来自研究经理的计划

trader_investment_plan: str # 来自交易员的可执行计划risk_debate_state: RiskDebateState final_trade_decision: str # 来自投资组合经理的最终决定

主状态AgentState包含了所有必要的信息：各个分析师的报告、辩论过程、投资计划、最终决策等等这样设计汇通网手机版app下载安装苹果可以清楚地追踪信息从原始数据到最终交易信号的完整流程构建数据获取工具集智能体系统的能力很大程度上取决于它们能够获取什么样的数据。

我们需要构建一套完整的工具，让智能体能够获取股价数据、技术指标、新闻资讯、社交媒体情绪等各种信息

每个工具都用@tool装饰器包装，并且提供清晰的类型注解，这样LLM就能理解每个工具的用途和参数import yfinance as yf from langchain_core.tools import tool 。

@tool def get_yfinance_data( symbol: Annotated[str, “公司的股票代码”], start_date: Ann汇通网手机版app下载安装苹果otated[str, “yyyy-mm-dd格式的开始日期”],

end_date: Annotated[str, “yyyy-mm-dd格式的结束日期”], ) -> str: “””从Yahoo Finance检索给定股票代码的股价数据””” try: ticker = yf.Ticker(symbol.upper()) 。

data = ticker.history(start=start_date, end=end_date) if data.empty: return f”No data found for symbol {symbol} between {start_date} an汇通网手机版app下载安装苹果d {end_date}”

return data.to_csv() except Exception as e: return f”Error fetching Yahoo Finance data: {e}”这里有个小细节：我们返回CSV格式的数据，因为LLM对这种格式的解析能力很强。

技术指标的计算用stockstats库：from stockstats import wrap as stockstats_wrap @tool def get_technical_indicators(

symbol: Annotated[str, “公司的股票代码”], start_date: Annotat汇通网手机版app下载安装苹果ed[str, “yyyy-mm-dd格式的开始日期”], end_date: Annotated[str, “yyyy-mm-dd格式的结束日期”],

) -> str: “””使用stockstats库检索股票的关键技术指标””” try: df = yf.download(symbol, start=start_date, end=end_date, progress=False) 。

if df.empty: return “No data to calculate indicators.” stock_df = stockstats_wrap(df) indicators = stock汇通网手机版app下载安装苹果_df[[macd, rsi_14, boll, boll_ub, boll_lb, close_50_sma, close_200_sma]]

return indicators.tail().to_csv() except Exception as e: return f”Error calculating stockstats indicators: {e}”

注意我们只返回最新的几行数据，这是为了控制传给LLM的信息量，避免不必要的token消耗新闻数据来自Finnhub：import finnhub @tool def get_finnhub_news(ticker: str, star汇通网手机版app下载安装苹果t_date: str, end_date: str) -> str: 。

“””从Finnhub获取日期范围内的公司新闻””” try: finnhub_client = finnhub.Client(api_key=os.environ[“FINNHUB_API_KEY”]) 。

news_list = finnhub_client.company_news(ticker, _from=start_date, to=end_date) news_items = [] for news in news_list[:5]: # 限制为5个结果

news_items.append(f”Headline汇通网手机版app下载安装苹果: {news[headline]}\nSummary: {news[summary]}”) return “\n\n”.join(news_items) if news_items else “No Finnhub news found.”

except Exception as e: return f”Error fetching Finnhub news: {e}”对于更广泛的网络搜索，我们用Tavily：from langchain_community.tools.tavily_search import TavilySearchResults

# 初始化Tavily搜索工具一次我们可以为汇通网手机版app下载安装苹果多个专用工具重用这个实例tavily_tool = TavilySearchResults(max_results=3)@tool def get_social_media_sentiment(ticker: str, trade_date: str) -> str: 。

“””对股票的社交媒体情绪进行实时网络搜索””” query = f”social media sentiment and discussions for {ticker} stock around {trade_date}” 。

return tavily_tool.invoke({“query”: query}) @tool汇通网手机版app下载安装苹果 def get_fundamental_analysis(ticker: str, trade_date: str) -> str:

“””对股票的最近基本面分析进行实时网络搜索””” query = f”fundamental analysis and key financial metrics for {ticker} stock published around {trade_date}” 。

return tavily_tool.invoke({“query”: query})@tool def get_macroeconomic_news(trade_date: str) -> str汇通网手机版app下载安装苹果:

“””对与股市相关的宏观经济新闻进行实时网络搜索””” query = f”macroeconomic news and market trends affecting the stock market on {trade_date}” 。

return tavily_tool.invoke({“query”: query})虽然这些工具都用的是同一个Tavily引擎，但我们给每个工具设定了不同的查询焦点，这样LLM能更准确地选择合适的工具。

最后把所有工具包装到一个类里：# 工具包类将所有定义的工具聚合到一个方便的对象中class Toolkit: def __init__(self, con汇通网手机版app下载安装苹果fig): self.config = config

self.get_yfinance_data = get_yfinance_data self.get_technical_indicators = get_technical_indicators self.get_finnhub_news = get_finnhub_news

self.get_social_media_sentiment = get_social_media_sentiment self.get_fundamental_analysis = get_fundamental_analysis

self.get_macroeco汇通网手机版app下载安装苹果nomic_news = get_macroeconomic_news # 实例化工具包，使所有工具通过这个单一对象可用toolkit = Toolkit(config)

print(f”Toolkit class defined and instantiated with live data tools.”)这样我们就有了一套完整的数据获取能力，可以从多个维度收集市场信息。

实现长期记忆系统：智能体的学习与适应机制在金融市场的复杂环境中，过往经验往往是未来决策的重要指引。我们的多智能体系统需要具备学习和记忆能力，从历史操作中积累经验，在相似情况下应用过往的成功策略。

ChromaDB向量库我们使用汇通网手机版app下载安装苹果ChromaDB实现一个基于向量嵌入的记忆系统，它能够将情境和对应的经验教训存储在高维向量空间中，通过语义相似性检索相关的历史经验：import chromadbfrom openai import OpenAI

class FinancialSituationMemory:”””基于ChromaDB的金融情况记忆系统用于存储和检索智能体在特定情境下的经验教训”””def __init__(self, collection_name, config):

# 初始化ChromaDB客户端和OpenAI客户端self.client = chromadb.Client()self.openai_clie汇通网手机版app下载安装苹果nt = OpenAI(api_key=config[“OPENAI_API_KEY”])

# 为每个智能体创建独立的记忆集合self.situation_collection = self.client.get_or_create_collection(name=collection_name

)def get_embedding(self, text):”””使用OpenAI API生成文本嵌入向量”””response = self.openai_client.embeddings.create(

input=text,model=”text-embedding-3-small”)return汇通网手机版app下载安装苹果 response.data[0].embeddingdef add_situations(self, situations_and_advice):

“””向记忆中添加新情况和建议”””if not situations_and_advice:return# 偏移确保唯一ID（以防稍后添加新数据）offset = self.situation_collection.count()

ids = [str(offset + i) for i, _ in enumerate(situations_and_advice)]# 分离情况及其对应的建议situations = [s for s, r in 汇通网手机版app下载安装苹果situations_and_advice]

recommendations = [r for s, r in situations_and_advice]# 为所有情况生成嵌入embeddings = [self.get_embedding(s) for s in situations]

# 将所有内容存储在Chroma（向量数据库）中self.situation_collection.add(documents=situations,metadatas=[{“recommendation”: rec} for rec in recommendations],

embeddings=embeddi汇通网手机版app下载安装苹果ngs,ids=ids,)def get_memories(self, current_situation, n_matches=1):”””为给定查询检索最相似的过去情况”””

if self.situation_collection.count() == 0:return []# 嵌入新的/当前情况query_embedding = self.get_embedding(current_situation)

# 查询集合中的相似嵌入results = self.situation_collection.query(query_embeddings=[query_embedding],n_res汇通网手机版app下载安装苹果ults=min(n_matches,

self.situation_collection.count()),include=[“metadatas”], # 仅返回建议)# 从匹配中返回提取的建议return [{recommendation: meta[recommendation]} for meta in results[metadatas][0]]

记忆系统的设计理念这个记忆系统的设计体现了几个重要原则：专门化记忆空间：每个智能体拥有独立的记忆集合，确保不同角色的智能体能够积累与其职责相关的专业经验语义检索机制：通过向量嵌入和相似性搜索，系统能够找到语义上相关的历史情况，而不仅仅是关键词汇通网手机版app下载安装苹果匹配。

经验-教训关联：每个记忆条目都包含情境描述和对应的经验教训，为智能体提供完整的学习上下文现在让我们为每个学习型智能体创建专用的记忆实例：# 为每个学习的智能体创建专用记忆实例bull_memory = FinancialSituationMemory(“bull_memory”, config)。

bear_memory = FinancialSituationMemory(“bear_memory”, config)trader_memory = FinancialSituationMemory(“trader_memory”, config)

invest_judge_memory = 汇通网手机版app下载安装苹果FinancialSituationMemory(“invest_judge_memory”, config)risk_manager_memory = FinancialSituationMemory(“risk_manager_memory”, config)

这种专业化的记忆分配确保了每个智能体都能积累与其特定职责相关的经验例如，多头分析师学到的经验（如”在强劲上升趋势中，估值担忧相对不那么重要”）可能与风险管理师的保守策略形成鲜明对比，两者都需要在各自的决策语境中保持独立性。

分析师团队：360度市场情报收集系统基础设施搭建完成后，我们现在要构建系统的第一个核心组件——分析师团队这个团队担汇通网手机版app下载安装苹果负着情报收集的重任，为后续的战略决策提供全方位的市场洞察分析师角色设计理念在现代金融决策中，单一维度的分析往往无法应对市场的复杂性。

因此，我们设计了四个专业化的分析师角色，每个都专注于特定的信息领域：市场分析师：专注于技术分析和价格行为模式社交媒体分析师：捕捉公众情绪和社交网络讨论新闻分析师：分析公司特定和宏观经济新闻基本面分析师：研究公司财务健康状况和内在价值

分析师工厂函数设计为了避免代码重复并确保一致性，我们采用工厂模式创建分析师节点这种设计模式允许我们从通用模板生成具有特定能力的智能体：from langchain_core.prompts import ChatPromptTempla汇通网手机版app下载安装苹果te, MessagesPlaceholder。

# 这个函数是一个工厂，为特定类型的分析师创建LangGraph节点def create_analyst_node(llm, toolkit, system_message, tools, output_field):

“””为分析师智能体创建节点参数：llm: 智能体使用的语言模型实例toolkit: 智能体可用的工具集合system_message: 定义智能体角色和目标的具体指令tools: 此智能体被允许使用的工具包中特定工具的列表

output_field: AgentState中存储此智能体最终报告的键”””# 为分析师智能体定义提示模板汇通网手机版app下载安装苹果prompt = ChatPromptTemplate.from_messages([(“system”,

“You are a helpful AI assistant, collaborating with other assistants.”” Use the provided tools to progress towards answering the question.”

” If you are unable to fully answer, thats OK; another assistant with different tools”” will help where you 汇通网手机版app下载安装苹果left off. Execute what you can to make progress.”

” You have access to the following tools: {tool_names}.\n{system_message}”” For your reference, the current date is {current_date}. The company we want to look at is {ticker}”),

# MessagesPlaceholder允许我们传入对话历史MessagesPlaceholder(variable_name=”messages汇通网手机版app下载安装苹果“),])# 用这个分析师的特定系统消息和工具名称部分填充提示

prompt = prompt.partial(system_message=system_message)prompt = prompt.partial(tool_names=”, “.join([tool.name for tool in tools]))

# 将指定的工具绑定到LLM这告诉LLM它可以调用哪些函数chain = prompt | llm.bind_tools(tools)# 这是将作为图中节点执行的实际函数def analyst_node(state):。

# 用当前状态的数据填充提示的最后部分prompt_wit汇通网手机版app下载安装苹果h_data = prompt.partial(current_date=state[“trade_date”], ticker=state[“company_of_interest”])

# 用状态中的当前消息调用链result = prompt_with_data.invoke(state[“messages”])report = “”# 如果LLM没有调用工具，意味着它已经生成了最终报告

if not result.tool_calls:report = result.content# 返回LLM的响应和最终报告以更新状态return {“messages”: [result], outpu汇通网手机版app下载安装苹果t_field: report}

return analyst_node专业化分析师实现基于工厂函数，我们现在可以创建四个专业化的分析师：1、市场分析师负责技术分析和价格行为研究：# 市场分析师：专注于技术指标和价格行动market_analyst_system_message = “””你是专门分析金融市场的交易助手。

你的角色是选择最相关的技术指标来分析股票的价格行为、动量和波动性你必须使用你的工具获取历史数据，然后生成包含你的发现的报告，包括汇总表”””market_analyst_node = create_analyst_node(

quick_thinking_llm,toolkit,m汇通网手机版app下载安装苹果arket_analyst_system_message,[toolkit.get_yfinance_data, toolkit.get_technical_indicators],

“market_report”)2、社交媒体分析师专注于情绪分析和公众讨论监控：# 社交媒体分析师：衡量公众情绪social_analyst_system_message = “””你是社交媒体分析师你的工作是分析过去一周特定公司的社交媒体帖子和公众情绪。

使用你的工具找到相关讨论，并撰写详细分析报告，包括对交易者的见解和影响，以及汇总表”””social_analyst_node = create_analyst_汇通网手机版app下载安装苹果node(quick_thinking_llm,

toolkit,social_analyst_system_message,[toolkit.get_social_media_sentiment],”sentiment_report”)3、新闻分析师

涵盖公司特定和宏观经济新闻分析：# 新闻分析师：涵盖公司特定和宏观经济新闻news_analyst_system_message = “””你是分析过去一周最新新闻和趋势的新闻研究员撰写关于当前世界状态的综合报告，重点关注与交易和宏观经济相关的内容。

使用你的工具进行全面分析，并提供详细分析，包括汇总表”””news_analyst_node = c汇通网手机版app下载安装苹果reate_analyst_node(quick_thinking_llm,toolkit,news_analyst_system_message,

[toolkit.get_finnhub_news, toolkit.get_macroeconomic_news],”news_report”)4、基本面分析师深入研究公司财务健康状况：# 基本面分析师：深入研究公司的财务健康状况

fundamentals_analyst_system_message = “””你是分析公司基本面信息的研究员撰写关于公司财务状况、内部人员情绪和交易的综合报告，以全面了解其基本面健康状况，包括汇总表。

“””funda汇通网手机版app下载安装苹果mentals_analyst_node = create_analyst_node(quick_thinking_llm,toolkit,fundamentals_analyst_system_message,

[toolkit.get_fundamental_analysis],”fundamentals_report”)ReAct模式：推理与行动的循环这些分析师智能体采用ReAct（推理和行动）模式工作，这不是简单的一次性LLM调用，而是一个动态的推理-执行循环：

思考：分析当前任务和可用信息行动：调用相应的工具获取数据观察：分析工具返回的结果推理：基于新信息调整策略重复：直到完成分析任务这汇通网手机版app下载安装苹果种模式使每个分析师都能够自主地收集信息、分析数据，并生成专业的分析报告，为后续的投资决策提供坚实的信息基础。

构建多头vs空头研究团队：对抗性分析机制数据收集完成后，我们面临一个关键挑战：如何处理潜在冲突的信息在比如说，技术面、情绪面、新闻面都显示强烈的看涨信号，但基本面分析师却警告了”溢价估值”的风险，这是为什么呢？。

这正是对抗性研究团队发挥作用的时刻。

对抗性辩论的设计理念在现实的投资决策中，最好的策略往往来自于充分的辩论和反思通过构建多头和空头两个对立观点的智能体，我们能够：避免确认偏见：每个观点都会受到对方的严格质疑发现逻辑漏洞：对抗性思辨能揭示单一视角的盲点。

压力测试假设：通过辩论验证投汇通网手机版app下载安装苹果资论点的稳健性综合平衡观点：最终形成考虑多方面因素的决策研究员工厂函数设计与分析师类似，我们使用工厂模式创建研究员节点，但这次加入了记忆检索和辩论逻辑：# 这个函数是一个工厂，为研究员智能体（多头或空头）创建LangGraph节点

def create_researcher_node(llm, memory, role_prompt, agent_name):”””为研究员智能体创建节点参数：llm: 智能体使用的语言模型实例memory: 此智能体的长期记忆实例，用于从过去经验中学习

role_prompt: 定义智能体角色（多头或空头）的特定系统提示agent_name: 智能体的名称，用于日汇通网手机版app下载安装苹果志记录和识别论点”””def researcher_node(state):# 首先，将所有分析师报告合并成单一摘要作为上下文

situation_summary = f”””Market Report: {state[market_report]}Sentiment Report: {state[sentiment_report]}

News Report: {state[news_report]}Fundamentals Report: {state[fundamentals_report]}”””# 从过去相似情况中检索相关记忆

past_memories = memory.get_memor汇通网手机版app下载安装苹果ies(situation_summary)past_memory_str = “\n”.join([mem[recommendation] for mem in past_memories])

# 为LLM构造完整提示prompt = f”””{role_prompt}这是当前分析的状态：{situation_summary}对话历史：{state[investment_debate_state][history]}

你对手的最后论点：{state[investment_debate_state][current_response]}从类似过去情况中的反思：{past_memory_str or汇通网手机版app下载安装苹果 没有找到过去的记忆}。

基于所有这些信息，以对话的方式提出你的论点”””# 调用LLM生成论点response = llm.invoke(prompt)argument = f”{agent_name}: {response.content}”。

# 用新论点更新辩论状态debate_state = state[investment_debate_state].copy()debate_state[history] += “\n” + argument

# 更新此智能体（多头或空头）的特定历史if agent_name == Bull Analyst:debate_state[bull_histo汇通网手机版app下载安装苹果ry] += “\n” + argumentelse:debate_state[bear_history] += “\n” + argument

debate_state[current_response] = argumentdebate_state[count] += 1return {“investment_debate_state”: debate_state}

return researcher_node多头与空头智能体创建基于工厂函数，我们创建两个具有对立观点的研究员：# 多头的角色是乐观的，专注于优势和增长bull_prompt = “””你是多头分析师你的目标是论证应该投资这只股票汇通网手机版app下载安装苹果。

专注于增长潜力、竞争优势和报告中的积极指标有效地反驳空头的论点在你的分析中要有逻辑性和说服力，但要承认任何重大风险，同时解释为什么机会超过这些风险”””# 空头的角色是悲观的，专注于风险和弱点 bear_prompt = “””你是空头分析师。

你的目标是论证不应该投资这只股票专注于风险、挑战和负面指标有效地反驳多头的论点在你的分析中要有逻辑性和说服力，突出可能被多头观点忽视的关键风险因素”””# 使用我们的工厂函数创建可调用节点bull_researcher_node = create_researcher_node(quick_thinking_llm, bull_memory, bull汇通网手机版app下载安装苹果_prompt, “Bull Analyst”)

bear_researcher_node = create_researcher_node(quick_thinking_llm, bear_memory, bear_prompt, “Bear Analyst”)

研究经理：辩论的仲裁者辩论需要一个公正的仲裁者来综合双方观点并形成最终决策研究经理承担这一重要角色：def research_manager_node(state):”””研究经理审查整个辩论并产生最终的投资建议。

这是一个复杂的推理任务，需要综合多个观点”””# 收集所有分析报告和辩论历史all_reports = f”””市场报告：{汇通网手机版app下载安装苹果state[market_report]}情绪报告：{state[sentiment_report]}

新闻报告：{state[news_report]}基本面报告：{state[fundamentals_report]}”””debate_history = state[investment_debate_state][history]

# 从过去的决策中检索相关经验situation_summary = all_reports + “\n” + debate_historypast_memories = invest_judge_memory.get_memories(situation_su汇通网手机版app下载安装苹果mmary)

past_memory_str = “\n”.join([mem[recommendation] for mem in past_memories])# 构建综合分析提示prompt = f”””你是资深投资研究经理。

你的任务是审查分析师团队的所有报告和多头空头研究员之间的辩论，然后做出最终的投资决定原始分析报告：{all_reports}研究团队辩论：{debate_history}过去类似情况的经验：{past_memory_str or 没有相关历史经验。

}请提供一个全面的投资建议，包括：1. 投资决定（买入/卖出/持有）及其理由2. 关键风险因素和风险缓解策略3. 建议的持汇通网手机版app下载安装苹果仓规模和时间框架4. 关键的监控指标你的决定应该平衡所有观点，承认不确定性，并提供清晰的行动计划。

“””# 使用更强大的模型进行复杂推理response = deep_thinking_llm.invoke(prompt)# 更新投资辩论状态，标记决定已做出debate_state = state[investment_debate_state].copy()

debate_state[judge_decision] = response.contentreturn {“investment_debate_state”: debate_state,”investment_analysis”: r汇通网手机版app下载安装苹果esponse.content # 也单独存储以便访问

}记忆学习机制研究过程结束后，系统需要从这次决策中学习：def store_investment_learnings(state):”””将本次投资决策的经验存储到各智能体的长期记忆中”””# 构建学习情境

situation = f”””分析目标：{state[company_of_interest]}日期：{state[trade_date]}市场技术面：{state[market_report][:200]}…

基本面状况：{state[fundamentals_report][:200]}…”””# 提取决策要点作为经验教训fi汇通网手机版app下载安装苹果nal_decision = state[investment_debate_state][judge_decision]

# 为不同智能体存储相关经验bull_lesson = f”多头观点在此情况下的有效性：{final_decision[:300]}…”bear_lesson = f”空头观点在此情况下的考虑因素：{final_decision[:300]}…”

judge_lesson = f”综合决策逻辑：{final_decision}”# 存储到对应的记忆系统bull_memory.add_situations([(situation, bull_lesson)])

bear_汇通网手机版app下载安装苹果memory.add_situations([(situation, bear_lesson)])invest_judge_memory.add_situations([(situation, judge_lesson)])

return state辩论工作流程设计为了确保充分的讨论，我们设计一个多轮辩论机制：def should_continue_investment_debate(state):”””决定是否继续投资辩论”””

debate_count = state[investment_debate_state][count]# 确保至少进行2轮辩论，最多4轮if debate_count汇通网手机版app下载安装苹果 < 2:return “continue_debate”

elif debate_count >= 4:return “end_debate”else:# 检查辩论是否达到收敛recent_args = state[investment_debate_state][history].split(\n)[-4:]

# 简单启发式：如果最近的论点开始重复，结束辩论if len(set(recent_args)) < len(recent_args) * 0.7:return “end_debate”else:return “continue_debate”

def route_investment_d汇通网手机版app下载安装苹果ebate(state):”””路由辩论到下一个发言者”””count = state[investment_debate_state][count]# 交替进行：多头开始，然后空头，依此类推

if count % 2 == 0:return “bull_researcher”else:return “bear_researcher”这个对抗性研究系统确保了投资决策不会被单一观点所主导，而是经过充分辩论和多角度分析的深思熟虑的结果。

通过记忆机制，系统还能从每次决策中学习，逐步提高决策质量交易执行与风险管理：多层决策验证机制完成投资策略分析后，我们进入系统的执行阶段这个阶段包含两个关键组件：将策略汇通网手机版app下载安装苹果转化为具体交易方案的交易员智能体，以及从多个风险偏好角度评估方案的

风险管理团队。

交易员智能体：策略到执行的转换器交易员的核心职责是将研究经理的宏观投资策略转化为具体的、可执行的交易指令这个转换过程需要考虑实际的市场条件、时机选择和风险控制参数import functools# 这个函数创建交易员智能体节点。

def create_trader(llm, memory):def trader_node(state, name):# 从长期记忆中检索相关的交易经验situation_summary = f”””

投资策略：{state[investment_analysis]}市场环境：{state[汇通网手机版app下载安装苹果market_report][:200]}…”””past_memories = memory.get_memories(situation_summary)

past_memory_str = “\n”.join([mem[recommendation] for mem in past_memories])# 提示很简单：基于计划创建提案# 关键指令是强制性的最终标签，便于后续解析

prompt = f”””你是交易智能体基于提供的投资计划，创建一个简洁的交易提案你的回应必须以最终交易提案：**买入/持有/卖出**结尾建议的投资计划：{state[investment_analysis]}。汇通网手机版app下载安装苹果

过去类似情况的交易经验：{past_memory_str or 没有相关历史经验}请提供具体的：1. 建议的仓位大小2. 进场时机和条件3. 止损和止盈设置4. 风险控制措施确保你的提案是实际可执行的，并包含具体的数字参数。

“””result = llm.invoke(prompt)# 输出用交易员的计划和发送者标识更新状态return {“trader_investment_plan”: result.content, “sender”: name}

return trader_node多角度风险管理辩论机制风险管理不是单一视角的评估，而是多个具有不同风险偏好的智能体之间的结构化辩论我们设计了汇通网手机版app下载安装苹果三个角色：激进型分析师：追求高回报，愿意承担更大风险保守型分析师。

：优先考虑资本保护，偏好低风险策略中性分析师：提供平衡视角，综合考虑收益与风险# 这个函数是创建风险辩论者节点的工厂def create_risk_debator(llm, role_prompt, agent_name):

def risk_debator_node(state):# 首先，从状态中获取其他两个辩论者的论点risk_state = state[risk_debate_state]opponents_args = []

# 收集其他智能体的观点用于反驳if agent_name != Risky Analyst an汇通网手机版app下载安装苹果d risk_state[current_risky_response]:opponents_args.append(f”激进派观点：{risk_state[current_risky_response]}”)

if agent_name != Safe Analyst and risk_state[current_safe_response]:opponents_args.append(f”保守派观点：{risk_state[current_safe_response]}”)

if agent_name != Neutral Analyst and risk_state[current_neut汇通网手机版app下载安装苹果ral_response]:opponents_args.append(f”中性派观点：{risk_state[current_neutral_response]}”)

# 用交易员的计划、辩论历史和对手的论点构造提示prompt = f”””{role_prompt}这是交易员的计划：{state[trader_investment_plan]}辩论历史：{risk_state[history]}

你的对手的最后论点：{chr(10).join(opponents_args)}从你的角度批评或支持这个计划要具体指出计划中的优点和不足，并提出改进建议”””response = llm.invoke汇通网手机版app下载安装苹果(prompt).content。

# 用新论点更新风险辩论状态new_risk_state = risk_state.copy()new_risk_state[history] += f”\n{agent_name}: {response}”

new_risk_state[latest_speaker] = agent_name# 将响应存储在此智能体的特定字段中if agent_name == Risky Analyst:new_risk_state[current_risky_response] = response

elif agent_name == Safe Analyst:new_ri汇通网手机版app下载安装苹果sk_state[current_safe_response] = responseelse:new_risk_state[current_neutral_response] = response

new_risk_state[count] += 1return {“risk_debate_state”: new_risk_state}return risk_debator_node风险角色定义三个风险智能体的角色定义体现了不同的投资哲学：

# 风险型角色主张最大化回报，即使这意味着更高的风险risky_prompt = “””你是激进型风险分析师你主张高回报机会和大胆策略你认为在明确的机会面前，汇通网手机版app下载安装苹果保守是最大的风险分析交易计划时，重点关注：- 是否充分利用了上涨潜力

– 仓位是否过于保守- 是否错过了时机窗口但也要保持理性，不要盲目激进”””# 安全型角色优先考虑资本保护胜过一切 safe_prompt = “””你是保守型风险分析师你优先考虑资本保护和最小化波动性。

你认为保住本金比追求高收益更重要分析交易计划时，重点关注：- 下跌风险是否得到充分控制- 止损设置是否合理- 仓位大小是否过于激进- 是否考虑了最坏情况的应对方案”””# 中性角色提供平衡、客观的观点neutral_prompt = “””你是中性风险分析师。

你提供平衡的视角，权衡收益和风险你的目标是找到风险调整后回报的最优平汇通网手机版app下载安装苹果衡点分析交易计划时，重点关注：- 风险回报比是否合理- 计划是否考虑了多种市场情况- 是否在机会和风险之间找到了合适的平衡- 计划的实际可执行性”””

风险管理工作流程# 创建交易员节点我们使用functools.partial来预填name参数trader_node_func = create_trader(quick_thinking_llm, trader_memory)。

trader_node = functools.partial(trader_node_func, name=”Trader”)# 使用特定提示创建三个风险辩论者节点risky_node = create_risk_de汇通网手机版app下载安装苹果bator(quick_thinking_llm, risky_prompt, “Risky Analyst”)

safe_node = create_risk_debator(quick_thinking_llm, safe_prompt, “Safe Analyst”) neutral_node = create_risk_debator(quick_thinking_llm, neutral_prompt, “Neutral Analyst”)

def run_risk_management_debate(state, max_rounds=2):”””运行风险管理辩论流程”””risk_s汇通网手机版app下载安装苹果tate = statefor round_num in range(max_rounds):

print(f”— 风险管理辩论第 {round_num + 1} 轮 —“)# 激进分析师先开始risky_result = risky_node(risk_state)risk_state[risk_debate_state] = risky_result[risk_debate_state]

print(f”\n**激进分析师观点：**”)print(risk_state[risk_debate_state][current_risky_response])# 然后是保守分析师safe_r汇通网手机版app下载安装苹果esult = safe_node(risk_state)

risk_state[risk_debate_state] = safe_result[risk_debate_state]print(f”\n**保守分析师观点：**”)print(risk_state[risk_debate_state][current_safe_response])

# 最后是中性分析师neutral_result = neutral_node(risk_state)risk_state[risk_debate_state] = neutral_result[risk_debate_state]

print(f”\n汇通网手机版app下载安装苹果**中性分析师观点：**”)print(risk_state[risk_debate_state][current_neutral_response])return risk_state

投资组合经理：最终决策的权威决策过程的最后一步由投资组合经理智能体负责这个智能体充当公司的最高决策者，它审查交易员的计划和整个风险辩论，然后发布最终的、具有约束力的决策由于这是最关键的步骤，我们使用更强大的推理模型来确保最高质量的决策：。

# 这个函数创建投资组合经理节点def create_portfolio_manager(llm, memory):def portfolio_manager_node(sta汇通网手机版app下载安装苹果te):# 收集完整的决策上下文

situation_summary = f”””原始分析：{state.get(investment_analysis, )}交易员计划：{state[trader_investment_plan]}风险辩论：{state[risk_debate_state][history]}

“””# 从过去的决策中学习past_memories = memory.get_memories(situation_summary)past_memory_str = “\n”.join([mem[recommendation] for mem in past_memories])

#汇通网手机版app下载安装苹果 提示要求基于所有先前工作做出最终、约束性决策prompt = f”””作为投资组合经理，你的决策是最终的审查交易员的计划和风险辩论，提供一个最终的、具有约束力的决策：买入、卖出或持有，以及简要的理由。

交易员的计划：{state[trader_investment_plan]}风险管理辩论：{state[risk_debate_state][history]}过去类似决策的经验：{past_memory_str or 没有相关历史经验。

}请提供：1. 明确的最终决策（买入/卖出/持有）2. 决策理由和关键考虑因素3. 批准的具体执行参数4. 需要监控的关键风险指标你的决策将直接影响公司的投资组汇通网手机版app下载安装苹果合，请确保决策是深思熟虑和负责任的”””response = llm.invoke(prompt).content

# 输出存储在状态的final_trade_decision字段中return {“final_trade_decision”: response}return portfolio_manager_node

# 创建可调用的组合经理节点portfolio_manager_node = create_portfolio_manager(deep_thinking_llm, risk_manager_memory)

决策学习和记忆存储每次决策完成后，系统需要将这次经验存储到长期记忆中：de汇通网手机版app下载安装苹果f store_trading_learnings(state):”””将交易和风险管理的经验存储到长期记忆中”””# 构建完整的决策情境

situation = f”””标的：{state[company_of_interest]}日期：{state[trade_date]}市场状况：{state[market_report][:200]}…

交易计划：{state[trader_investment_plan][:200]}…风险考虑：{state[risk_debate_state][history][:300]}…”””# 提取最终决策作为经验教训

final_decision 汇通网手机版app下载安装苹果= state[final_trade_decision]# 为不同角色存储相关经验trader_lesson = f”交易执行经验：{final_decision[:300]}…”

risk_lesson = f”风险管理要点：{state[risk_debate_state][history][-200:]}…”manager_lesson = f”组合管理决策：{final_decision}”

# 存储到对应的记忆系统trader_memory.add_situations([(situation, trader_lesson)])risk_manager_memory.add_s汇通网手机版app下载安装苹果ituations([(situation, manager_lesson)])

# 为风险辩论的各个角色也存储经验if state[risk_debate_state][current_risky_response]:risk_memory = FinancialSituationMemory(“risk_analyst_memory”, config)

risk_memory.add_situations([(situation, f”激进派观点效果：{final_decision[:200]}…”)])return state这个多层决策验证机制确保了每个交易决策都经过充分的讨论、质疑和汇通网手机版app下载安装苹果验证，从而最大程度地降低决策失误的风险，同时通过记忆机制不断积累和优化决策质量。

LangGraph工作流程整合：智能体协调的艺术我们已经构建了完整的智能体生态系统，现在需要通过LangGraph将它们有机地整合成一个协调的工作流程LangGraph的状态图架构能够优雅地处理复杂的多智能体交互和条件分支。

工作流程架构设计我们的完整工作流程可以分为几个主要阶段：数据收集阶段：四个分析师并行工作策略研究阶段：多头空头辩论和综合分析执行计划阶段：交易员制定具体方案风险评估阶段：多角度风险辩论最终决策阶段：投资组合经理做出约束性决定

信号提取和学习：反思和经验存储

核心工作流程实现from langgrap汇通网手机版app下载安装苹果h.graph import StateGraph, START, ENDfrom typing import Literaldef create_trading_workflow():

“””创建完整的交易决策工作流程”””# 初始化状态图workflow = StateGraph(AgentState)# 添加分析师节点（并行执行）workflow.add_node(“market_analyst”, market_analyst_node)

workflow.add_node(“social_analyst”, social_analyst_node) workflow.add_node(“汇通网手机版app下载安装苹果news_analyst”, news_analyst_node)

workflow.add_node(“fundamentals_analyst”, fundamentals_analyst_node)# 添加研究团队节点workflow.add_node(“bull_researcher”, bull_researcher_node)

workflow.add_node(“bear_researcher”, bear_researcher_node)workflow.add_node(“research_manager”, research_manager_node)

# 添加交易和风险管理节点汇通网手机版app下载安装苹果workflow.add_node(“trader”, trader_node)workflow.add_node(“risky_analyst”, risky_node)workflow.add_node(“safe_analyst”, safe_node)

workflow.add_node(“neutral_analyst”, neutral_node)workflow.add_node(“portfolio_manager”, portfolio_manager_node)

# 添加学习和信号处理节点workflow.add_node(“signal_processor”, signal_汇通网手机版app下载安装苹果processor_node)workflow.add_node(“reflection_learner”, reflection_learner_node)

# 定义工作流程连接workflow.add_edge(START, “market_analyst”)workflow.add_edge(START, “social_analyst”)workflow.add_edge(START, “news_analyst”)

workflow.add_edge(START, “fundamentals_analyst”)# 所有分析师完成后进入研究阶段workflow.add_edge(“mark汇通网手机版app下载安装苹果et_analyst”, “investment_debate_router”)

workflow.add_edge(“social_analyst”, “investment_debate_router”)workflow.add_edge(“news_analyst”, “investment_debate_router”)

workflow.add_edge(“fundamentals_analyst”, “investment_debate_router”)return workflow# 条件路由函数def investment_debate_router(state) -> Liter汇通网手机版app下载安装苹果al[“bull_researcher”, “research_manager”]:

“””路由投资辩论流程”””debate_count = state[investment_debate_state][count]if debate_count == 0:return “bull_researcher” # 多头先开始

elif debate_count < 4: # 最多4轮辩论if debate_count % 2 == 1:return “bear_researcher” # 空头回应else:return “bull_researcher” # 多头继续

else:return “res汇通网手机版app下载安装苹果earch_manager” # 结束辩论，进入决策def risk_debate_router(state) -> Literal[“risky_analyst”, “safe_analyst”, “neutral_analyst”, “portfolio_manager”]:

“””路由风险管理辩论”””risk_count = state[risk_debate_state][count]if risk_count >= 9: # 每个角色最多3轮return “portfolio_manager”

# 循环顺序：激进->保守->中性order = risk_count % 3if orde汇通网手机版app下载安装苹果r == 0:return “risky_analyst”elif order == 1:return “safe_analyst”

else:return “neutral_analyst”信号处理和学习机制交易决策完成后，系统需要提取可执行信号并进行学习：class SignalProcessor:”””负责将最终LLM输出解析为清晰、机器可读的信号”””

def __init__(self, llm):self.llm = llmdef process_signal(self, full_signal: str) -> str:”””从决策文本中提取BUY/SELL/HOLD信号”””

mes汇通网手机版app下载安装苹果sages = [(“system”, “你是一个助手，设计用于从金融报告中提取最终投资决策：卖出、买入或持有只回应单词决策”),(“human”, full_signal),]result = self.llm.invoke(messages).content.strip().upper()。

# 基本验证以确保输出是三个预期信号之一if result in [“BUY”, “SELL”, “HOLD”, “买入”, “卖出”, “持有”]:# 标准化为英文signal_map = {“买入”: “BUY”, “卖出”: “SELL”, “持有”: “HOLD”}

return signal_m汇通网手机版app下载安装苹果ap.get(result, result)return “ERROR_UNPARSABLE_SIGNAL”def signal_processor_node(state):

“””信号处理节点”””processor = SignalProcessor(quick_thinking_llm)final_signal = processor.process_signal(state[final_trade_decision])

return {“final_signal”: final_signal}反思学习系统class Reflector:”””为智能体编排学习过程的类”””def __ini汇通网手机版app下载安装苹果t__(self, llm):

self.llm = llmself.reflection_prompt = “””你是专业的金融分析师审查交易决策/分析、市场背景和财务结果- 首先，根据结果确定决策是否正确- 分析导致成功或失败的最关键因素。

– 最后，制定一个简洁的、一句话的经验教训或启发式方法，可用于改善类似情况下的未来决策市场背景和分析：{situation}结果（盈利/亏损）：{returns_losses}”””def reflect(self, current_state, returns_losses, memory, component_key_func):

“””执行反思并存储经汇通网手机版app下载安装苹果验教训”””situation = f”””报告：{current_state[market_report]} {current_state[sentiment_report]} {current_state[news_report]} {current_state[fundamentals_report]}

决策/分析文本：{component_key_func(current_state)}”””prompt = self.reflection_prompt.format(situation=situation,

returns_losses=returns_losses)result = s汇通网手机版app下载安装苹果elf.llm.invoke(prompt).content# 情况（上下文）和生成的经验教训存储在智能体的记忆中memory.add_situations([(situation, result)])

def reflection_learner_node(state):”””反思学习节点，模拟假设性结果进行学习”””reflector = Reflector(quick_thinking_llm)# 在实际应用中，这里会是真实的交易结果

# 这里我们模拟一个假设性的盈利结果hypothetical_returns = 1000# 为每个有记忆的智能体运行反思过程reflector.reflec汇通网手机版app下载安装苹果t(state, hypothetical_returns, bull_memory,

lambda s: s[investment_debate_state][bull_history])reflector.reflect(state, hypothetical_returns, bear_memory,

lambda s: s[investment_debate_state][bear_history])reflector.reflect(state, hypothetical_returns, trader_memory,

lambda s: s[trader_investment_plan]汇通网手机版app下载安装苹果)reflector.reflect(state, hypothetical_returns, risk_manager_memory,

lambda s: s[final_trade_decision])return {“learning_completed”: True}系统评估：多维度质量验证机制构建一个复杂的AI交易系统后，我们需要严格的评估机制来验证其决策质量。

我们设计了三种互补的评估策略，从不同角度审核系统性能

LLM-as-a-Judge：定性推理评估使用强大的LLM作为公正的评判者，从多个维度评估决策质量：from pydantic import BaseModel, Fieldf汇通网手机版app下载安装苹果rom langchain_core.prompts import ChatPromptTemplate

class Evaluation(BaseModel):”””评估结果的结构化模式”””reasoning_quality: int = Field(description=”推理连贯性和逻辑的1-10分评分”)

evidence_based_score: int = Field(description=”基于报告证据引用的1-10分评分”)actionability_score: int = Field(description=”决策清晰度和可执行性的1-10分评分”)

risk_assess汇通网手机版app下载安装苹果ment: int = Field(description=”风险考虑充分性的1-10分评分”)justification: str = Field(description=”分数的详细理由”)

# 创建评估提示模板evaluator_prompt = ChatPromptTemplate.from_template(“””你是专业的金融审计员请基于提供的”分析师报告”评估”最终交易决策”分析师报告：

{reports}最终交易决策：{final_decision}请从以下维度进行评估：1. 推理质量：逻辑是否清晰、连贯2. 证据支撑：是否充分引用和利用了分析师报告3. 可执行性：决策是否具体、明汇通网手机版app下载安装苹果确、可操作

4. 风险评估：是否充分考虑了潜在风险每个维度给出1-10分的评分，并提供详细的评分理由”””)# 构建评估链evaluator_chain = evaluator_prompt | deep_thinking_llm.with_structured_output(Evaluation)。

def run_llm_evaluation(state):”””执行LLM-as-a-Judge评估”””# 构建分析师报告摘要reports_summary = f”””市场报告：{state[market_report]}

情绪报告：{state[sentiment_report]} 新闻报告：汇通网手机版app下载安装苹果{state[news_report]}基本面报告：{state[fundamentals_report]}”””eval_input = {

“reports”: reports_summary,”final_decision”: state[final_trade_decision]}evaluation_result = evaluator_chain.invoke(eval_input)

print(“—– LLM-as-a-Judge 评估报告 —–“)print(f”推理质量：{evaluation_result.reasoning_quality}/10”)print(f汇通网手机版app下载安装苹果“证据支撑：{

evaluation_result.evidence_based_score}/10″)print(f”可执行性：{evaluation_result.actionability_score}/10″)

print(f”风险评估：{evaluation_result.risk_assessment}/10″)print(f”评估理由：{evaluation_result.justification}”)

return evaluation_resultGround Truth比较：客观性能验证通过比较智能体决策与实际市场表现来评估预测准确性：import yfinance as yf汇通网手机版app下载安装苹果from datetime import datetime, timedelta

def evaluate_ground_truth(ticker, trade_date, signal, holding_period_days=5):”””评估交易信号的实际市场表现”””try:# 解析交易日期并定义评估窗口

start_date = datetime.strptime(trade_date, “%Y-%m-%d”).date()end_date = start_date + timedelta(days=holding_period_days + 3) # 加3天缓冲

# 下载市场数据data 汇通网手机版app下载安装苹果= yf.download(ticker,start=start_date.isoformat(),end=end_date.isoformat(),progress=False

)if len(data) = start_date]

if len(trade_day_data) == 0:return “交易日期无效或市场关闭”open_price = trade_day_data[Open].iloc[0]# 获取持有期结束后的收盘价

if len(trade_day_data) >= holding_period_days:close_price = trade_day_data[Close]汇通网手机版app下载安装苹果.iloc[holding_period_days – 1]

else:close_price = trade_day_data[Close].iloc[-1]# 计算收益率return_pct = ((close_price – open_price) / open_price) * 100

# 评估信号准确性if signal == “BUY”:prediction_correct = return_pct > 0expected_direction = “上涨”elif signal == “SELL”:

prediction_correct = return_pct < 0expected_汇通网手机版app下载安装苹果direction = “下跌”else: # HOLDprediction_correct = abs(return_pct) < 2 # 2%以内算持平

expected_direction = “持平”result = {“ticker”: ticker,”trade_date”: trade_date,”signal”: signal,”open_price”: open_price,

“close_price”: close_price,”return_pct”: return_pct,”prediction_correct”: prediction_correct,”expected_汇通网手机版app下载安装苹果direction”: expected_direction,

“actual_direction”: “上涨” if return_pct > 0 else “下跌” if return_pct < 0 else “持平”}return resultexcept Exception as e:

return f”Ground Truth评估错误：{str(e)}”def run_ground_truth_evaluation(state):”””执行Ground Truth评估”””ticker = state[company_of_interest]

trade_date = state[tra汇通网手机版app下载安装苹果de_date]signal = state.get(final_signal, UNKNOWN)result = evaluate_ground_truth(ticker, trade_date, signal)

if isinstance(result, dict):print(“—– Ground Truth 评估报告 —–“)print(f”股票代码：{result[ticker]}”)print(f”交易日期：{result[trade_date]}”)

print(f”系统信号：{result[signal]} (预期{result[expected_direction]}汇通网手机版app下载安装苹果)”)print(f”实际表现：{result[return_pct]:.2f}% ({result[actual_direction]})”)

print(f”预测准确性：{✓ 正确 if result[prediction_correct] else ✗ 错误}”)# 计算绝对收益（假设$10,000投资）investment = 10000

if signal == “BUY”:profit = investment * (result[return_pct] / 100)print(f”假设收益：${profit:.2f} (投资 ${investment:,})”)

else:print(汇通网手机版app下载安装苹果f”Ground Truth评估失败：{result}”)return result事实核查：信息准确性验证验证系统使用的信息和数据的准确性：def fact_check_analysis(state):

“””对分析报告进行事实核查”””fact_check_results = {}# 检查技术指标的合理性market_report = state[market_report]if RSI in market_report:

# 提取RSI值并检查是否在合理范围内(0-100)import rersi_pattern = rRSI.*?(\d+\.?\d*)rsi_match = re.sear汇通网手机版app下载安装苹果ch(rsi_pattern, market_report)

if rsi_match:rsi_value = float(rsi_match.group(1))fact_check_results[RSI_valid] = 0 <= rsi_value <= 100

# 检查新闻报告的时效性news_report = state[news_report]current_date = datetime.strptime(state[trade_date], “%Y-%m-%d”)

# 简单的时效性检查（在实际应用中，这里会更复杂）recent_keywords = [本周, 最近, 今日, 昨日]f汇通网手机版app下载安装苹果act_check_results[news_timeliness] = any(keyword in news_report for keyword in recent_keywords)

# 检查基本面数据的合理性fundamentals_report = state[fundamentals_report]fact_check_results[fundamentals_coherence] = len(fundamentals_report) > 100 # 基本长度检查

# 检查情绪分析的平衡性sentiment_report = state[sentiment_report]bullis汇通网手机版app下载安装苹果h_words = [看涨, 积极, 乐观, 买入, 上涨]bearish_words = [看跌, 消极, 悲观, 卖出, 下跌]

bullish_count = sum(1 for word in bullish_words if word in sentiment_report)bearish_count = sum(1 for word in bearish_words if word in sentiment_report)

fact_check_results[sentiment_balance] = abs(bullish_count – bearish_count) <= 5pri汇通网手机版app下载安装苹果nt(“—– 事实核查报告 —–“)

for check, result in fact_check_results.items():status = “✓ 通过” if result else “✗ 存疑”print(f”{check}: {status}”)

return fact_check_resultsdef comprehensive_evaluation(state):”””执行综合评估”””print(“=” * 50)print(” 系统综合评估报告”)

print(“=” * 50)# 1. LLM-as-a-Judge评估llm_eval = run_llm_eva汇通网手机版app下载安装苹果luation(state)print(“\n”)# 2. Ground Truth评估

gt_eval = run_ground_truth_evaluation(state)print(“\n”)# 3. 事实核查fact_check = fact_check_analysis(state)

# 综合评分total_score = 0max_score = 0if hasattr(llm_eval, reasoning_quality):total_score += llm_eval.reasoning_quality +

llm_eval.evidence_based_score + llm_汇通网手机版app下载安装苹果eval.actionability_scoremax_score += 30if isinstance(gt_eval, dict) and gt_eval.get(prediction_correct):

total_score += 10max_score += 10fact_checks_passed = sum(1 for result in fact_check.values() if result)

total_score += fact_checks_passed * 2max_score += len(fact_check) * 2final_score = (total_sco汇通网手机版app下载安装苹果re / max_score) * 100 if max_score > 0 else 0

print(f”\n—– 综合评估结果 —–“)print(f”系统综合得分：{final_score:.1f}/100”)if final_score >= 80:print(“评级：优秀 – 系统表现卓越”)

elif final_score >= 60:print(“评级：良好 – 系统表现令人满意”)else:print(“评级：需要改进 – 系统需要进一步优化”)return {“llm_evaluation”: llm_eval,

“ground_truth”: gt_eval,”fa汇通网手机版app下载安装苹果ct_check”: fact_check,”final_score”: final_score}这套多维度评估体系确保了我们能够从不同角度全面评估系统的决策质量，为持续优化提供数据支撑。

系统部署与扩展：从概念到生产完成了核心系统的构建和评估后，我们可以讨论如何将这个多智能体交易系统扩展到生产环境，以及未来可能的改进方向完整的工作流程编排将所有组件整合到最终的LangGraph工作流程中：。

def create_complete_trading_system():”””创建完整的交易决策系统”””# 初始化状态图workflow = StateGraph(AgentState)# 数据收集阶段汇通网手机版app下载安装苹果（并行执行）

workflow.add_node(“market_analyst”, market_analyst_node)workflow.add_node(“social_analyst”, social_analyst_node)

workflow.add_node(“news_analyst”, news_analyst_node)workflow.add_node(“fundamentals_analyst”, fundamentals_analyst_node)

# 投资研究阶段workflow.add_node(“bull_researcher”, bull_researcher_n汇通网手机版app下载安装苹果ode)workflow.add_node(“bear_researcher”, bear_researcher_node)

workflow.add_node(“research_manager”, research_manager_node)# 交易执行阶段workflow.add_node(“trader”, trader_node)

# 风险管理阶段workflow.add_node(“risky_analyst”, risky_node)workflow.add_node(“safe_analyst”, safe_node)workflow.add_node(“neutral_analy汇通网手机版app下载安装苹果st”, neutral_node)

# 最终决策阶段workflow.add_node(“portfolio_manager”, portfolio_manager_node)# 后处理阶段workflow.add_node(“signal_processor”, signal_processor_node)

workflow.add_node(“evaluation”, comprehensive_evaluation_node)workflow.add_node(“reflection_learner”, reflection_learner_node)

# 定义边和条件路由workflow.汇通网手机版app下载安装苹果add_edge(START, “market_analyst”)workflow.add_edge(START, “social_analyst”)workflow.add_edge(START, “news_analyst”)

workflow.add_edge(START, “fundamentals_analyst”)# 分析师完成后的汇聚和路由workflow.add_edge(“market_analyst”, “investment_debate_coordinator”)

workflow.add_edge(“social_analyst”, “investment_debate_汇通网手机版app下载安装苹果coordinator”)workflow.add_edge(“news_analyst”, “

investment_debate_coordinator”)workflow.add_edge(“fundamentals_analyst”, “investment_debate_coordinator”)

# 条件路由workflow.add_conditional_edges(“investment_debate_coordinator”,investment_debate_router,{“bull_researcher”: “bull_researcher”,

“bear_researche汇通网手机版app下载安装苹果r”: “bear_researcher”, “research_manager”: “research_manager”})workflow.add_conditional_edges(

“bull_researcher”,lambda state: “bear_researcher” if state[investment_debate_state][count] < 4 else “research_manager”

)workflow.add_conditional_edges(“bear_researcher”, lambda state: “bull_researcher” if st汇通网手机版app下载安装苹果ate[investment_debate_state][count] < 4 else “research_manager”

)# 研究完成后进入交易阶段workflow.add_edge(“research_manager”, “trader”)# 风险管理辩论路由workflow.add_conditional_edges(

“trader”,lambda state: “risky_analyst”, # 交易员完成后开始风险辩论)workflow.add_conditional_edges(“risky_analyst”,

risk_debate_router,{“safe_analyst汇通网手机版app下载安装苹果“: “safe_analyst”,”neutral_analyst”: “neutral_analyst”, “risky_analyst”: “risky_analyst”,

“portfolio_manager”: “portfolio_manager”})workflow.add_conditional_edges(“safe_analyst”,risk_debate_router,

{“neutral_analyst”: “neutral_analyst”,”risky_analyst”: “risky_analyst”,”safe_analyst”: “safe_analyst”,

“p汇通网手机版app下载安装苹果ortfolio_manager”: “portfolio_manager”})workflow.add_conditional_edges(“neutral_analyst”,risk_debate_router,

{“risky_analyst”: “risky_analyst”,”safe_analyst”: “safe_analyst”,”neutral_analyst”: “neutral_analyst”,

“portfolio_manager”: “portfolio_manager” })# 最终处理流程workflow.add_edge(“portfolio_manager”, 汇通网手机版app下载安装苹果“signal_processor”)

workflow.add_edge(“signal_processor”, “evaluation”)workflow.add_edge(“evaluation”, “reflection_learner”)

workflow.add_edge(“reflection_learner”, END)return workflow.compile()# 运行完整系统def run_trading_system(ticker: str, trade_date: str):

“””运行完整的交易决策系统”””# 编译工作流程trading_system = creat汇通网手机版app下载安装苹果e_complete_trading_system()# 初始化状态initial_state = AgentState(

messages=[HumanMessage(content=f”分析 {ticker} 在 {trade_date} 的交易机会”)],company_of_interest=ticker,trade_date=trade_date,

investment_debate_state=InvestDebateState({history: , current_response: , count: 0,bull_history: , bear_history: , judge_汇通网手机版app下载安装苹果decision:

}),risk_debate_state=RiskDebateState({history: , latest_speaker: , count: 0,current_risky_response: , current_safe_response: ,

current_neutral_response: , risky_history: ,safe_history: , neutral_history: , judge_decision: }))# 执行工作流程

final_state = trading_system.invoke(initial_state)return fi汇通网手机版app下载安装苹果nal_state# 使用示例if __name__ == “__main__”:result = run_trading_system(“NVDA”, “2024-12-06”)

print(f”最终交易信号：{result.get(final_signal, N/A)}”)print(f”系统评估得分：{result.get(evaluation_score, N/A)}”)生产环境考虑

将系统部署到生产环境需要考虑以下关键因素：可靠性和容错性import asynciofrom tenacity import retry, stop_after_attempt, wait_exponenti汇通网手机版app下载安装苹果al

class ProductionTradingSystem:”””生产级交易系统实现”””def __init__(self, config):self.config = configself.max_retries = 3

self.timeout_seconds = 300@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=4, max=10)

)async def run_with_retry(self, ticker: str, trade_date: str):”””带重试机制的系统运行”汇通网手机版app下载安装苹果“”try:return await asyncio.wait_for(

self.run_async(ticker, trade_date),timeout=self.timeout_seconds)except asyncio.TimeoutError:raise Exception(f”系统运行超时 ({self.timeout_seconds}秒)”)

except Exception as e:print(f”系统运行失败，正在重试… 错误: {str(e)}”)raiseasync def run_async(self, ticker: str, trade_date: str):汇通网手机版app下载安装苹果

“””异步运行交易系统”””# 这里实现异步版本的交易系统passdef health_check(self):”””系统健康检查”””checks = {“api_connections”: self._check_api_connections(),

“memory_usage”: self._check_memory_usage(),”model_availability”: self._check_model_availability()}all_healthy = all(checks.values())

return {“status”: “healthy” if all_health汇通网手机版app下载安装苹果y else “unhealthy”, “details”: checks}def _check_api_connections(self):

“””检查外部API连接”””try:# 测试OpenAI APIquick_thinking_llm.invoke(“test”)# 测试其他API…return Trueexcept:return False

def _check_memory_usage(self):”””检查内存使用情况”””import psutilmemory_percent = psutil.virtual_memory().percent

return memory_pe汇通网手机版app下载安装苹果rcent < 85 # 内存使用率低于85%def _check_model_availability(self):”””检查模型可用性”””try:# 简单的模型响应测试

response = quick_thinking_llm.invoke(“Hello”)return len(response.content) > 0except:return False监控和日志

import loggingfrom datetime import datetimeimport jsonclass TradingSystemMonitor:”””交易系统监控类”””def __init__(self)汇通网手机版app下载安装苹果:

self.logger = self._setup_logger()self.metrics = {}def _setup_logger(self):”””设置日志记录器”””logger = logging.getLogger(TradingSystem)

logger.setLevel(logging.INFO)# 文件处理器file_handler = logging.FileHandler(ftrading_system_{datetime.now().strftime(“%Y%m%d”)}.log)

file_handler.setFormatter(logging.Formatter汇通网手机版app下载安装苹果(%(asctime)s – %(name)s – %(levelname)s – %(message)s))logger.addHandler(file_handler)

return loggerdef log_decision(self, ticker, date, decision, confidence, execution_time):”””记录交易决策”””log_entry = {

“timestamp”: datetime.now().isoformat(),”ticker”: ticker,”trade_date”: date,”decision”: decision,”con汇通网手机版app下载安装苹果fidence”: confidence,

“execution_time”: execution_time,”event_type”: “trading_decision”}self.logger.info(json.dumps(log_entry))

def log_error(self, error_type, error_message, context):”””记录错误”””log_entry = {“timestamp”: datetime.now().isoformat(),

“error_type”: error_type,”error_message”: str(error_mes汇通网手机版app下载安装苹果sage),”context”: context,”event_type”: “error”

}self.logger.error(json.dumps(log_entry))def update_metrics(self, metric_name, value):”””更新系统指标”””self.metrics[metric_name] = {

“value”: value,”timestamp”: datetime.now().isoformat()}def get_system_status(self):”””获取系统状态”””return {

“metrics”: self.metrics,汇通网手机版app下载安装苹果“last_update”: datetime.now().isoformat()}扩展能力和未来发展多资产支持class MultiAssetTradingSystem(ProductionTradingSystem):

“””多资产交易系统”””def __init__(self, config):super().__init__(config)self.supported_assets = [“stocks”, “crypto”, “forex”, “commodities”]

async def analyze_portfolio(self, assets: list, trade_dat汇通网手机版app下载安装苹果e: str):”””分析整个投资组合”””results = {}# 并行分析多个资产

tasks = [self.run_with_retry(asset, trade_date) for asset in assets]portfolio_results = await asyncio.gather(*tasks, return_exceptions=True)

for asset, result in zip(assets, portfolio_results):if isinstance(result, Exception):results[asset] = {“error”: str(汇通网手机版app下载安装苹果result)}

else:results[asset] = result# 投资组合级别的风险管理portfolio_decision = self._make_portfolio_decision(results)return {

“individual_decisions”: results,”portfolio_decision”: portfolio_decision}def _make_portfolio_decision(self, individual_results):

“””基于个别资产决策做出投资组合级决策”””# 这里实现投资组合优化逻辑pass实时数据流集成import w汇通网手机版app下载安装苹果ebsocketimport threadingclass RealtimeDataIntegration:

“””实时数据流集成”””def __init__(self, config):self.config = configself.data_buffer = {}self.callbacks = []def start_realtime_feed(self, symbols):

“””启动实时数据流”””for symbol in symbols:thread = threading.Thread(target=self._connect_websocket,args=(symbol,)

)汇通网手机版app下载安装苹果thread.daemon = Truethread.start()def _connect_websocket(self, symbol):”””连接WebSocket数据流”””def on_message(ws, message):

data = json.loads(message)self._process_realtime_data(symbol, data)def on_error(ws, error):print(f”WebSocket错误 {symbol}: {error}”)

ws = websocket.WebSocketApp(f”wss://api.example.com汇通网手机版app下载安装苹果/stream/{symbol}”,on_message=on_message,on_error=on_error

)ws.run_forever()def _process_realtime_data(self, symbol, data):”””处理实时数据”””self.data_buffer[symbol] = data

# 触发注册的回调函数for callback in self.callbacks:callback(symbol, data)def register_callback(self, callback):

“””注册数据更新回调”””self.callbacks.appen汇通网手机版app下载安装苹果d(callback)机器学习增强from sklearn.ensemble import RandomForestClassifier

from sklearn.preprocessing import StandardScalerimport numpy as npclass MLEnhaniedTradingSystem:”””机器学习增强的交易系统”””

def __init__(self, config):self.config = configself.ml_model = RandomForestClassifier(n_estimators=100)

self.scaler = St汇通网手机版app下载安装苹果andardScaler()self.is_trained = Falsedef extract_features(self, state):”””从智能体状态中提取特征”””

features = []# 技术指标特征market_report = state.get(market_report, )features.extend(self._extract_technical_features(market_report))

# 情绪特征sentiment_report = state.get(sentiment_report, )features.extend(self._extract_s汇通网手机版app下载安装苹果entiment_features(sentiment_report))

# 基本面特征fundamentals_report = state.get(fundamentals_report, )features.extend(self._extract_fundamental_features(fundamentals_report))

return np.array(features).reshape(1, -1)def train_model(self, historical_data):”””训练机器学习模型”””X = []y = []

for record in historical_d汇通网手机版app下载安装苹果ata:features = self.extract_features(record[state])X.append(features.flatten())

y.append(record[actual_outcome]) # 1: profit, 0: lossX = np.array(X)y = np.array(y)# 标准化特征X_scaled = self.scaler.fit_transform(X)

# 训练模型self.ml_model.fit(X_scaled, y)self.is_trained = Truedef predict_outcome(self, state):”汇通网手机版app下载安装苹果“”预测交易结果”””

if not self.is_trained:return Nonefeatures = self.extract_features(state)features_scaled = self.scaler.transform(features)

# 预测概率probability = self.ml_model.predict_proba(features_scaled)[0][1]return {“success_probability”: probability,

“confidence”: “high” if probability > 0.7 else “medium汇通网手机版app下载安装苹果” if probability > 0.5 else “low”}def _extract_technical_features(self, report):

“””提取技术分析特征”””# 这里实现技术指标的数值提取return [0.5, 0.6, 0.7] # 示例def _extract_sentiment_features(self, report):

“””提取情绪分析特征”””# 这里实现情绪指标的数值提取return [0.8, 0.2] # 示例def _extract_fundamental_features(self, report):”””提取基本面特征”””

# 这里实现汇通网手机版app下载安装苹果基本面指标的数值提取return [1.2, 0.9, 1.5] # 示例总结我们构建的这个基于LangGraph的多智能体量化交易系统代表了AI在金融决策领域的一个重要进展该系统通过模拟真实投资机构的协作模式，实现了从数据收集、深度分析、策略制定到风险管理的完整闭环。

核心优势多视角决策：通过不同专业背景的智能体协作，避免了单一视角的局限性对抗性验证：多头空头辩论机制确保投资逻辑经过充分质疑和验证层次化风险管理：多层次的风险评估机制最大化降低决策风险持续学习能力：基于向量记忆的经验积累，使系统能够不断改进

可解释性：每个决策步骤都有明确的推理过程，便于审计和优化这个系统不仅是一个技术演示，更是一汇通网手机版app下载安装苹果个可实际部署的交易决策辅助工具它能够：为量化基金提供智能化的研究支持帮助个人投资者做出更理性的投资决策为金融机构提供风险管理的自动化工具。

作为金融AI研究的基础平台随着大语言模型能力的不断提升和多智能体框架的成熟，这样的系统将在不久的将来成为金融决策的重要工具，为投资者和金融机构创造更大的价值本文展示的完整实现代码已经为读者提供了一个可运行的起点。

作者：Fareed Khan