1. 项目概述用LangGraph重构投资组合分析工作流不是写脚本是搭“决策流水线”你有没有试过用Python做一次完整的投资组合分析从拉取股票行情、计算收益率、优化权重到生成可视化报告——整个过程像在一条单行道上开车前一步没跑完后一步就卡死。我做过不下二十个类似项目最常遇到的不是模型不准而是流程太脆数据源接口变了整个分析链路就断客户临时要加个风险归因模块就得把原来几百行代码翻来覆去改三遍更别说多人协作时A写的回测模块和B写的业绩归因压根不兼容最后靠Excel手工拼接结果。这次我彻底换了一种思路不写“分析脚本”而是用LangGraph搭建一条可插拔、可追溯、可中断重试的投资组合分析流水线。核心关键词就是LangGraph、Python、Portfolio Analysis——它不是替代pandas或cvxpy而是让这些工具在统一的有状态图谱里各司其职。适合三类人金融量化岗想摆脱“脚本工程师”标签的从业者、金融科技公司正在设计投研中台的技术负责人、以及MBA/金工专业学生想真正理解“分析流程”如何工程化落地。它解决的不是“怎么算”而是“怎么让整个分析过程变得像银行柜台业务一样——每个环节有编号、有日志、能复盘、出错不丢进度”。这不是AI炒股也不是黑箱预测而是一次对传统投研工作流的底层操作系统级重构。2. 整体架构设计为什么非得用LangGraph传统方案的三个致命短板2.1 传统方案的“单线程幻觉”与真实业务场景的撕裂感先说一个我踩过的典型坑去年帮一家私募做季度组合回顾原始方案是用Jupyter Notebook串起6个cell——从akshare拉数据、yfinance补海外标的、pandas清洗、empyrical算夏普、scipy.optimize求解最小方差权重、matplotlib画图最后用Jinja2渲染PDF报告。表面看很顺但实际交付后客户提了三个需求直接让我推倒重来第一要求所有中间结果比如每只股票的滚动波动率必须存入数据库供风控系统调用第二当某只股票停牌导致收益率计算失败时不能整条链路报错退出而要跳过该标的继续算其他部分并标记异常第三合规部门要求每份报告附带完整执行日志精确到每个函数调用的输入参数和耗时。这三个需求任何一个都让原有脚本变成“不可维护的债务”。原因很简单传统脚本本质是隐式状态流——状态全靠变量名和执行顺序维系没有显式定义“哪里是入口”“哪里可能失败”“失败后该去哪”。就像修水管你只能顺着水流方向一节节查没法在分叉口装压力表和旁通阀。2.2 LangGraph的“有状态图谱”如何精准切中痛点LangGraph不是另一个LLM框架它的核心价值在于提供可编程的状态机抽象。我把整个组合分析拆成7个原子节点Node每个节点只做一件事且职责清晰fetch_data拉取原始行情、clean_prices处理缺失值与复权、compute_returns计算日频/周频收益率、calculate_risk_metrics波动率、最大回撤等、run_optimization调用cvxpy求解、generate_visualizations产出图表、compile_report组装PDF。关键在于这些节点不是线性排列而是通过边Edge定义明确的流转逻辑正常路径fetch_data→clean_prices→compute_returns→calculate_risk_metrics→run_optimization→generate_visualizations→compile_report异常路径compute_returns节点内部若检测到某只股票数据不足30个交易日则触发skip_asset边将该标的ID传给calculate_risk_metrics节点的降级分支只计算已有效标的的风险人工干预路径在run_optimization完成后插入review_weights人工审核节点若审核不通过可触发adjust_constraints边动态修改风险预算参数后重新进入优化节点这种设计让“分析流程”本身成为一等公民。你可以用graph.get_state()随时查看当前执行到哪一步、中间变量存了什么、哪些分支被跳过。这比任何日志文件都直观——因为状态本身就是结构化的。2.3 为什么不用Airflow/Luigi它们解决的是调度不是语义建模有人会问既然要编排流程为啥不选Airflow这里必须划清界限Airflow解决的是“什么时候跑”LangGraph解决的是“怎么跑”。举个例子Airflow能确保每天9点跑组合分析但它无法告诉你“当波动率计算超阈值时应自动降低该资产在优化中的权重上限”。前者是时间维度的调度后者是业务逻辑维度的状态响应。Luigi同理它擅长处理数据依赖如“报告生成依赖图表图表依赖收益率”但无法表达“如果收益率序列存在结构性断点则切换至鲁棒估计方法”这类条件逻辑。LangGraph的ConditionalEdge正是为此而生——它允许你在图谱中嵌入任意Python函数作为路由判断器。我实测过在同一个LangGraph实例里既可以用if volatility 0.3: return high_risk_branch做简单判断也能调用训练好的LSTM模型对收益率序列做突变检测再根据预测结果路由。这种灵活性是传统工作流引擎无法提供的。2.4 架构分层LangGraph不是替代品而是粘合剂与指挥官最终落地的架构是三层嵌套底层工具层pandas数据处理、cvxpy优化求解、empyrical业绩归因、plotly交互图表、weasyprintPDF生成——这些保持原样不作任何封装改造中层LangGraph层定义7个节点函数、3类边正常流、异常流、人工流、1个状态Schema包含portfolio_data: dict,risk_metrics: dict,optimization_result: dict,execution_log: list等字段顶层应用层一个PortfolioAnalyzer类封装graph.invoke()调用逻辑提供run_full_analysis()、resume_from_node(calculate_risk_metrics)、export_state_to_json()等业务友好的API这种分层让团队协作变得清晰量化研究员专注写run_optimization节点里的cvxpy代码前端工程师只管generate_visualizations节点的plotly配置而架构师只需维护图谱拓扑。没有人需要读懂全部500行代码。3. 核心细节解析状态Schema设计、节点契约与异常路由机制3.1 状态Schema不是万能字典而是强约束的协议LangGraph要求你明确定义State类型这看似繁琐实则是稳定性的基石。我定义的PortfolioState不是简单的dict而是继承自TypedDict的结构化类型from typing import TypedDict, List, Dict, Optional, Any import pandas as pd class PortfolioState(TypedDict): # 原始输入 tickers: List[str] # [AAPL, MSFT, 000001.SZ] start_date: str # 2020-01-01 end_date: str # 2023-12-31 # 数据层 raw_prices: Optional[pd.DataFrame] # 列为ticker索引为datetime cleaned_prices: Optional[pd.DataFrame] returns: Optional[pd.DataFrame] # 日收益率矩阵 # 风险指标层 risk_metrics: Optional[Dict[str, float]] # {volatility: 0.18, max_drawdown: -0.22} asset_risk_contributions: Optional[pd.Series] # 每只股票对组合波动率的贡献 # 优化层 optimization_config: Dict[str, Any] # {method: min_variance, constraints: {...}} optimization_result: Optional[Dict[str, Any]] # {weights: {...}, sharpe_ratio: 1.2} # 输出层 visualizations: Optional[Dict[str, bytes]] # {risk_contribution_chart: b..., cumulative_return_plot: b...} report_pdf: Optional[bytes] # 运行时元数据 execution_log: List[Dict[str, Any]] # [{node: fetch_data, duration_sec: 12.3, status: success}] skipped_assets: List[str] # [000001.SZ] 记录被跳过的标的 current_step: str # calculate_risk_metrics这个Schema的关键在于显式声明可空性Optional和类型。比如raw_prices标注为Optional[pd.DataFrame]意味着在fetch_data节点执行前它必为空执行后必须为DataFrame若某个节点意外返回NoneLangGraph会在类型检查阶段直接报错而不是让错误潜伏到下游导致AttributeError。这比运行时assert isinstance(df, pd.DataFrame)更早暴露问题。我在测试中故意让clean_prices节点返回字符串LangGraph立刻抛出TypeError: Expected pd.DataFrame for raw_prices, got class str定位速度比调试日志快5倍。3.2 节点契约每个函数必须遵守的“宪法条款”LangGraph节点不是普通函数它必须遵循严格契约输入是完整State输出是State的增量更新Delta。以compute_returns节点为例def compute_returns(state: PortfolioState) - PortfolioState: 计算日收益率矩阵自动处理停牌与复权 # 1. 输入校验确保cleaned_prices存在且非空 if state[cleaned_prices] is None: raise ValueError(cleaned_prices not available. Run clean_prices node first.) # 2. 核心计算使用pandas pct_change但增加容错 prices_df state[cleaned_prices] returns_df prices_df.pct_change().dropna(howall) # 删除全NaN行 # 3. 智能跳过识别数据不足的标的 min_valid_days 30 valid_days returns_df.count() insufficient_assets valid_days[valid_days min_valid_days].index.tolist() # 4. 更新状态只返回需变更的字段Delta update { returns: returns_df, skipped_assets: state[skipped_assets] insufficient_assets, execution_log: state[execution_log] [{ node: compute_returns, duration_sec: time.time() - start_time, status: success, processed_tickers: returns_df.columns.tolist(), skipped_tickers: insufficient_assets }] } return update注意三点绝不修改原statestate[cleaned_prices]是只读的所有变更通过返回字典实现。这保证了状态不可变性immutability避免多线程下的竞态问题增量更新原则返回字典只包含returns、skipped_assets、execution_log三个键其他字段如raw_prices自动沿用上一状态值。这极大减少内存拷贝开销日志内嵌execution_log不是单独写文件而是作为state一部分流转确保日志与数据严格同步。导出报告时compile_report节点可直接读取完整执行轨迹3.3 异常路由用ConditionalEdge实现“业务逻辑驱动”的失败处理传统try-except只能捕获异常但无法表达“异常后该走哪条业务路径”。LangGraph的ConditionalEdge解决了这个问题。在compute_returns节点后我定义了如下路由from langgraph.graph import END, START from langgraph.graph import StateGraph def route_after_returns(state: PortfolioState) - str: 根据returns计算结果决定下一步 if len(state[skipped_assets]) 0: return calculate_risk_metrics # 全部有效走主路径 elif len(state[skipped_assets]) len(state[tickers]) * 0.3: return calculate_risk_metrics_degraded # 少量跳过走降级分支 else: return alert_insufficient_data # 大量跳过触发告警 # 在图谱构建中注册 workflow.add_conditional_edges( compute_returns, route_after_returns, { calculate_risk_metrics: calculate_risk_metrics, calculate_risk_metrics_degraded: calculate_risk_metrics_degraded, alert_insufficient_data: alert_insufficient_data } )calculate_risk_metrics_degraded节点会主动忽略skipped_assets中的标的仅对有效标的计算风险指标并在risk_metrics字典中添加degraded_mode: True标记。而alert_insufficient_data节点则调用企业微信机器人API发送告警并返回END终止流程。这种设计让“失败”不再是终点而是业务规则的一部分——就像银行系统不会因单笔交易失败而停摆而是启动反洗钱核查流程。3.4 人工干预节点让“专家经验”无缝融入自动化流水线量化分析最怕“全自动黑箱”。LangGraph支持在任意节点插入人工审核且不破坏状态连续性。review_weights节点实现如下def review_weights(state: PortfolioState) - PortfolioState: 人工审核优化权重支持批准/拒绝/修改 weights state[optimization_result][weights] # 生成审核界面简化版实际对接Streamlit print( 权重审核界面 ) print(f当前组合{list(weights.keys())}) print(f建议权重{weights}) print(请选择操作[A]批准 [R]拒绝 [M]手动修改) choice input(输入选择).strip().upper() if choice A: return {current_step: generate_visualizations} elif choice R: # 拒绝后触发约束调整 new_constraints state[optimization_config][constraints].copy() new_constraints[max_single_weight] * 0.8 # 降低单只股票上限 return { optimization_config: { **state[optimization_config], constraints: new_constraints }, current_step: run_optimization } elif choice M: # 手动输入新权重 new_weights {} for ticker in weights.keys(): w float(input(f输入{ticker}新权重)) new_weights[ticker] w return { optimization_result: {weights: new_weights}, current_step: generate_visualizations }关键点在于无论用户选择哪个分支返回的都是合法State增量。批准后进入图表生成拒绝后自动收紧约束并重跑优化修改后直接使用新权重。整个过程state始终在线无需保存/加载中间文件。4. 实操全流程从零搭建可运行的组合分析图谱含完整代码4.1 环境准备与依赖安装避开版本地狱的实操技巧LangGraph生态更新快版本冲突是头号拦路虎。我实测有效的环境配置如下基于Python 3.10# 创建干净虚拟环境强烈推荐避免污染全局 python -m venv portfolio_env source portfolio_env/bin/activate # Linux/Mac # portfolio_env\Scripts\activate # Windows # 安装核心依赖按此顺序避免pip自动降级 pip install --upgrade pip setuptools wheel pip install langgraph0.1.42 # 固定小版本避免API变动 pip install pandas2.2.2 numpy1.26.4 cvxpy1.4.3 pip install yfinance0.2.37 akshare1.10.82 # 数据源 pip install plotly5.23.1 weasyprint60.2 # 可视化与PDF pip install python-dotenv1.0.1 # 环境变量管理提示不要用pip install langgraph[all]它会强制安装langchain而langchain的向量库依赖可能与你的CUDA版本冲突。LangGraph本身不依赖LLM纯本地运行完全OK。环境变量.env文件内容# 数据源配置 YFINANCE_TIMEOUT30 AKSHARE_RETRIES3 # 优化配置 CVXPY_SOLVERECOS # 推荐免费且稳定避免SCS需编译或MOSEK商业授权 CVXPY_VERBOSEFalse # 报告配置 PDF_FONT_PATH/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf # Linux路径Windows需改为C:/Windows/Fonts/msyh.ttc4.2 定义状态Schema与基础工具函数创建portfolio_state.pyfrom typing import TypedDict, List, Dict, Optional, Any import pandas as pd import numpy as np class PortfolioState(TypedDict): tickers: List[str] start_date: str end_date: str raw_prices: Optional[pd.DataFrame] cleaned_prices: Optional[pd.DataFrame] returns: Optional[pd.DataFrame] risk_metrics: Optional[Dict[str, float]] asset_risk_contributions: Optional[pd.Series] optimization_config: Dict[str, Any] optimization_result: Optional[Dict[str, Any]] visualizations: Optional[Dict[str, bytes]] report_pdf: Optional[bytes] execution_log: List[Dict[str, Any]] skipped_assets: List[str] current_step: str # 工具函数安全的数据拉取带重试与超时 def safe_fetch_prices(tickers: List[str], start: str, end: str) - pd.DataFrame: 统一数据拉取入口兼容yfinance/akshare import yfinance as yf import akshare as ak all_data {} for ticker in tickers: try: # 优先yfinance美股/港股 if . in ticker and not ticker.endswith(.SZ): data yf.download(ticker, startstart, endend, progressFalse) if not data.empty: all_data[ticker] data[Close] continue # 次选akshareA股 if ticker.endswith(.SZ) or ticker.endswith(.SH): symbol ticker.split(.)[0] df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart.replace(-, ), end_dateend.replace(-, )) if not df.empty: df[date] pd.to_datetime(df[日期]) df.set_index(date, inplaceTrue) all_data[ticker] df[收盘] continue raise ValueError(fNo data source available for {ticker}) except Exception as e: print(fWarning: Failed to fetch {ticker}: {e}) continue if not all_data: raise ValueError(No price data fetched for any ticker) return pd.DataFrame(all_data).dropna(howall) # 工具函数稳健的收益率计算处理复权与停牌 def robust_returns(prices_df: pd.DataFrame) - pd.DataFrame: 计算日收益率自动处理复权因子缺失 # 若价格有明显跳跃50%尝试用pct_change而非diff returns prices_df.pct_change().fillna(0) # 检测并修正极端值如分红除权导致的-99% extreme_mask (returns -0.8) | (returns 0.8) if extreme_mask.sum().sum() 0: print(fDetected {extreme_mask.sum().sum()} extreme returns, applying median filter) # 用前后5日中位数替换 for col in returns.columns: series returns[col].copy() for idx in series[extreme_mask[col]].index: window series.loc[idx - pd.Timedelta(days5):idx pd.Timedelta(days5)] series.loc[idx] window.median() returns[col] series return returns.dropna(howall)4.3 构建核心节点函数7个原子操作创建portfolio_nodes.py逐个实现import time import pandas as pd import numpy as np from typing import Dict, Any, Optional, List from portfolio_state import PortfolioState, safe_fetch_prices, robust_returns # Node 1: 数据拉取 def fetch_data(state: PortfolioState) - PortfolioState: start_time time.time() try: prices_df safe_fetch_prices( state[tickers], state[start_date], state[end_date] ) update { raw_prices: prices_df, execution_log: state[execution_log] [{ node: fetch_data, duration_sec: time.time() - start_time, status: success, fetched_tickers: prices_df.columns.tolist(), data_points: len(prices_df) }] } return update except Exception as e: error_log { node: fetch_data, duration_sec: time.time() - start_time, status: error, error: str(e) } return { execution_log: state[execution_log] [error_log], current_step: error_handling } # Node 2: 数据清洗 def clean_prices(state: PortfolioState) - PortfolioState: start_time time.time() if state[raw_prices] is None: raise ValueError(raw_prices missing. Run fetch_data first.) df state[raw_prices].copy() # 前向填充插值处理停牌 df df.fillna(methodffill).interpolate(methodlinear) # 删除仍含NaN的列彻底无效标的 valid_cols df.columns[df.isna().sum() 0] df df[valid_cols] update { cleaned_prices: df, execution_log: state[execution_log] [{ node: clean_prices, duration_sec: time.time() - start_time, status: success, cleaned_tickers: df.columns.tolist(), nan_filled: int(state[raw_prices].isna().sum().sum() - df.isna().sum().sum()) }] } return update # Node 3: 收益率计算含智能跳过 def compute_returns(state: PortfolioState) - PortfolioState: start_time time.time() if state[cleaned_prices] is None: raise ValueError(cleaned_prices missing. Run clean_prices first.) returns_df robust_returns(state[cleaned_prices]) min_valid_days 30 valid_days returns_df.count() insufficient_assets valid_days[valid_days min_valid_days].index.tolist() update { returns: returns_df, skipped_assets: state[skipped_assets] insufficient_assets, execution_log: state[execution_log] [{ node: compute_returns, duration_sec: time.time() - start_time, status: success, processed_tickers: returns_df.columns.tolist(), skipped_tickers: insufficient_assets }] } return update # Node 4: 风险指标计算主分支 def calculate_risk_metrics(state: PortfolioState) - PortfolioState: start_time time.time() if state[returns] is None: raise ValueError(returns missing. Run compute_returns first.) returns_df state[returns] cov_matrix returns_df.cov() portfolio_vol np.sqrt(np.dot(np.array([1/len(returns_df.columns)]*len(returns_df.columns)), np.dot(cov_matrix, np.array([1/len(returns_df.columns)]*len(returns_df.columns))))) # 最大回撤向量化计算 cum_ret (1 returns_df).cumprod() running_max cum_ret.cummax() drawdown (cum_ret - running_max) / running_max max_drawdown drawdown.min().min() update { risk_metrics: { volatility: float(portfolio_vol), max_drawdown: float(max_drawdown), sharpe_ratio: 0.0 # 待优化后填充 }, execution_log: state[execution_log] [{ node: calculate_risk_metrics, duration_sec: time.time() - start_time, status: success }] } return update # Node 4降级分支处理跳过标的 def calculate_risk_metrics_degraded(state: PortfolioState) - PortfolioState: start_time time.time() if state[returns] is None: raise ValueError(returns missing.) # 过滤掉skipped_assets valid_returns state[returns].drop(columnsstate[skipped_assets], errorsignore) if valid_returns.empty: raise ValueError(No valid assets left after skipping) cov_matrix valid_returns.cov() portfolio_vol np.sqrt(np.dot(np.array([1/len(valid_returns.columns)]*len(valid_returns.columns)), np.dot(cov_matrix, np.array([1/len(valid_returns.columns)]*len(valid_returns.columns))))) cum_ret (1 valid_returns).cumprod() running_max cum_ret.cummax() drawdown (cum_ret - running_max) / running_max max_drawdown drawdown.min().min() update { risk_metrics: { volatility: float(portfolio_vol), max_drawdown: float(max_drawdown), sharpe_ratio: 0.0, degraded_mode: True, excluded_tickers: state[skipped_assets] }, execution_log: state[execution_log] [{ node: calculate_risk_metrics_degraded, duration_sec: time.time() - start_time, status: success, excluded_tickers: state[skipped_assets] }] } return update # Node 5: 优化求解最小方差 def run_optimization(state: PortfolioState) - PortfolioState: start_time time.time() import cvxpy as cp if state[returns] is None: raise ValueError(returns missing.) returns_df state[returns] n len(returns_df.columns) # 定义变量 weights cp.Variable(n) # 目标最小化组合方差 cov_matrix returns_df.cov().values objective cp.Minimize(cp.quad_form(weights, cov_matrix)) # 约束权重和为1非负多空限制可在此调整 constraints [ cp.sum(weights) 1, weights 0 ] # 添加动态约束来自state if max_single_weight in state[optimization_config].get(constraints, {}): constraints.append(weights state[optimization_config][constraints][max_single_weight]) prob cp.Problem(objective, constraints) prob.solve(solvercp.ECOS, verboseFalse) if prob.status ! cp.OPTIMAL: raise ValueError(fOptimization failed: {prob.status}) weights_dict {ticker: float(w) for ticker, w in zip(returns_df.columns, weights.value)} update { optimization_result: { weights: weights_dict, optimal_variance: float(prob.value), status: prob.status }, execution_log: state[execution_log] [{ node: run_optimization, duration_sec: time.time() - start_time, status: success, optimal_variance: float(prob.value) }] } return update # Node 6: 可视化生成 def generate_visualizations(state: PortfolioState) - PortfolioState: start_time time.time() import plotly.graph_objects as go from plotly.subplots import make_subplots if state[optimization_result] is None: raise ValueError(optimization_result missing.) weights state[optimization_result][weights] # 权重分布饼图 fig go.Figure(data[go.Pie(labelslist(weights.keys()), valueslist(weights.values()))]) fig.update_layout(titleOptimal Portfolio Weights) weights_bytes fig.to_image(formatpng, width800, height600) # 累计收益曲线假设等权基准 if state[returns] is not None: cum_ret (1 state[returns]).cumprod() equal_weight cum_ret.mean(axis1) optimal_cum_ret (cum_ret * pd.Series(weights)).sum(axis1).cumprod() fig2 make_subplots() fig2.add_trace(go.Scatter(xequal_weight.index, yequal_weight, nameEqual Weight)) fig2.add_trace(go.Scatter(xoptimal_cum_ret.index, yoptimal_cum_ret, nameOptimal)) fig2.update_layout(titleCumulative Return Comparison) perf_bytes fig2.to_image(formatpng, width1000, height600) update { visualizations: { weights_pie: weights_bytes, performance_curve: perf_bytes }, execution_log: state[execution_log] [{ node: generate_visualizations, duration_sec: time.time() - start_time, status: success }] } return update # Node 7: PDF报告生成 def compile_report(state: PortfolioState) - PortfolioState: start_time time.time() from weasyprint import HTML, CSS import base64 if state[visualizations] is None: raise ValueError(visualizations missing.) # 将图片转为base64嵌入HTML weights_b64 base64.b64encode(state[visualizations][weights_pie]).decode() perf_b64 base64.b64encode(state[visualizations][performance_curve]).decode() html_content f html headstyle body {{ font-family: sans-serif; margin: 40px; }} .chart {{ width: 100%; height: 500px; }} /style/head body h1Portfolio Analysis Report/h1 h2Optimal Weights/h2 img srcdata:image/png;base64,{weights_b64} classchart/ h2Performance Comparison/h2 img srcdata:image/png;base64,{perf_b64} classchart/ h2Risk Metrics/h2 ul liVolatility: {state[risk_metrics][volatility]:.2%}/li liMax Drawdown: {state[risk_metrics][max_drawdown]:.2%}/li /ul /body /html html HTML(stringhtml_content) pdf_bytes html.write_pdf() update { report_pdf: pdf_bytes, execution_log: state[execution_log] [{ node: compile_report, duration_sec: time.time() - start_time, status: success, pdf_size_kb: len(pdf_bytes) // 1024 }] } return update4.4 组装图谱与运行入口创建main.pyfrom langgraph.graph import StateGraph, END, START from portfolio_state import PortfolioState from portfolio_nodes import ( fetch_data, clean_prices, compute_returns, calculate_risk_metrics, calculate_risk_metrics_degraded, run_optimization, generate_visualizations, compile_report ) # 定义路由函数 def route_after_returns(state: PortfolioState) - str: if len(state[skipped_assets]) 0: return calculate_risk_metrics elif len(state[skipped_assets]) len(state[tickers]) * 0.3: return calculate_risk_metrics_degraded else: return alert_insufficient_data def route_after_optimization(state: PortfolioState) - str: # 简单路由优化后直接进可视化 return generate_visualizations # 构建图谱 workflow StateGraph(PortfolioState) # 添加节点 workflow.add_node(fetch_data, fetch_data) workflow.add_node(clean_prices, clean_prices) workflow.add_node(compute_returns, compute_returns) workflow.add_node(calculate_risk_metrics, calculate_risk_metrics) workflow.add_node(calculate_risk_metrics_degraded, calculate_risk_metrics_degraded) workflow.add_node(run_optimization, run_optimization) workflow.add_node(generate_visualizations, generate_visualizations) workflow.add_node(compile_report, compile_report) # 添加边 workflow.add_edge(START, fetch_data) workflow.add_edge(fetch_data, clean_prices) workflow.add_edge(clean_prices, compute_returns) workflow.add_conditional_edges( compute_returns, route_after_returns, { calculate_risk_metrics: calculate_risk_metrics, calculate_risk_metrics_degraded: calculate_risk_metrics_degraded, alert_insufficient_data: END # 终止 } ) workflow.add_edge(calculate_risk_metrics, run_optimization) workflow.add_edge(calculate_risk_metrics_degraded, run_optimization) workflow.add_conditional_edges( run_optimization, route_after_optimization, {generate_visualizations: generate_visualizations} ) workflow.add_edge(generate_visualizations, compile_report) workflow.add_edge(compile_report, END) # 编译图谱 app workflow.compile() # 运行示例 if __name__ __main__: # 初始化状态 initial_state: PortfolioState { tickers: [AAPL, MSFT, GOOGL, TSLA], start_date: 2022-01-01, end_date: 2023-12-31, raw_prices: None, cleaned_prices: None, returns: None, risk_metrics: None, asset_risk_contributions: None, optimization_config: { method: min_variance, constraints