PandaAI工作流节点专家模式原理、步骤、自定义

步骤 1 ｜ 进入专家模式并理解节点框架

PandaAI QuantFlow 的专家模式允许用户直接查看和编辑工作流中每个功能节点对应的 Python 代码，是从"拖拽式策略搭建"进阶到"代码级策略定制"的关键入口。
如何进入：

点击「AI 工作流」，在已创建的工作流中点击「查看」。进入后，在左侧伸缩边框中点击「专家模式」，即可切换到代码视图。此时左侧文件面板会列出工作流中所有节点对应的 .py 文件，右侧画布仍保留节点连接的可视化视图。

节点框架的基本逻辑：

切换到专家模式后，你会看到每个功能节点背后的 Python 代码。

节点代码结构（以「Python代码输入」节点为例）：

专家模式下每个节点本质上是一个继承自 BaseWorkNode 的 Python 类，遵循统一的三层结构：

**输入模型（CodeInputModel）：**定义节点接受什么。这里是一个 code 字段，即用户在界面中输入的 Python 策略代码。模型内置了 @model_validator，会在输入时自动执行语法检查（check_syntax）和危险代码检测（danger_code_check），拦截诸如 os.system、文件读写等不安全操作。

**输出模型（CodeOutputModel）：**定义节点向下游传递什么。这个节点的逻辑很简单——校验通过后，原样透传代码字符串。

**节点注册（@work_node 装饰器）：**通过 name、group、type 等参数将这个类注册为可视化画布上的一个可拖拽节点，归属于「01-基础工具」分组。

简单说，专家模式让你看到：每个"拖拽方块"的背后，就是一个 Input → 处理逻辑 → Output 的 Python 类。 理解了这个结构，后续自建节点就有了基础。

再看一个进阶节点：「因子分析(期货)」

如果说「Python代码输入」是最简单的"透传型"节点，那「因子分析(期货)」就展示了节点框架真正的设计力量。

它的输入模型不再是一个简单的字符串，而是一组结构化参数：因子值 DataFrame、主力聚合标签（简单拼接/前复权/后复权）、调仓周期、分组数量、因子方向。每个参数通过 @ui 装饰器绑定了前端控件类型（下拉框 combobox），意味着代码里定义的字段结构，直接决定了画布上节点的交互界面——这是一套声明式的 UI 绑定机制。

另一个值得注意的设计是中文映射层：用户在界面上看到的是「后复权」，但 run() 方法内部通过 type_mapping 将其转换为底层数据标识 “889”。这层抽象把用户认知和系统实现解耦了，用户不需要记代码，开发者也不需要改底层接口。

对我们开发者而言，专家模式的价值在于：

你不是在一个黑盒平台上"配置策略"，而是能直接看到并复用整套节点开发范式——BaseModel 定义输入输出、@ui 控制前端渲染、@work_node 注册节点元信息、run() 承载核心逻辑。理解了这套约定，你就可以自己封装任何自定义节点：一个另类因子计算、一个风控过滤器、一个多策略调度器，都是同样的四步结构。专家模式本质上是把平台的扩展权交还给了用户。

最后，解读期货回测节点：

这个节点是工作流的"终端执行器"——前面所有的代码输入、因子构建，最终都汇聚到这里触发回测引擎。

输入模型结构：

FutureBacktestInputModel 定义了 8 个字段，分为两类：一类是上游节点传入的（策略代码 code、因子值 factors，通过 “input_type”: “None” 隐藏界面控件，只接受连线输入）；另一类是用户在画布上手动配置的（初始资金、佣金倍率、保证金倍率、回测频率、起止日期，分别绑定了 number_field、combobox、date_field 等前端控件）。

这里有个值得学习的校验设计：@model_validator 在所有字段解析完成后做了一次跨字段校验——检查结束日期必须晚于开始日期，且区间不超过 3 年。这种"先让每个字段各自合法，再检查字段间的逻辑关系"的分层校验思路，是 Pydantic v2 的推荐实践。

运行逻辑：

run() 方法的流程很清晰：先通过 get_backtest_id() 生成一个唯一回测 ID，然后调用 start() 函数把所有参数打包传给底层回测引擎。异常处理中做了一件实用的事——把报错信息压缩为"前两行 + 最后一行"输出，避免超长 traceback 淹没关键信息，同时通过 enrich_exception_with_source_code 把用户的策略代码附加到异常上下文中，方便定位是策略代码本身的问题还是引擎的问题。

输出模型：

只返回一个 backtest_id 字符串，下游的「策略回测结果」节点通过这个 ID 去查询和展示回测报告。这种异步设计意味着回测可能是后台排队执行的，节点本身不阻塞等待全部计算完成。

步骤 2 ｜ 创建自定义节点

理解了节点的代码结构后，下一步是动手建一个自己的节点。点击节点框下方的加号，即可用 Python 代码填充节点模板。新节点生成后拖拽到右侧画布，会自动存储在节点库的「用户自定义节点框」中，后续可复用和编辑。

选择「填充节点模板代码」，系统会生成一个"两数求和"的最小示例。这个模板就是所有自定义节点的骨架，总共只有三个部分：

① InputModel — 你的节点需要什么
pythonclass InputModel(BaseModel):
number1: int
number2: int
用 Pydantic 的 BaseModel 声明输入字段。这里定义了什么，画布上节点就会出现对应的输入接口。字段类型（int、str、float、DataFrame…）决定了前端自动渲染的控件形式。

② OutputModel — 你的节点产出什么
pythonclass OutputModel(BaseModel):
result: int
同样用 BaseModel 声明。输出字段就是下游节点可以连线接收的数据。

③ 节点类本身 — 注册 + 逻辑
python@work_node(name=“示例-两数求和”, group=“测试节点”)
class ExamplePluginAddition(BaseWorkNode):
def input_model(cls): return InputModel
def output_model(cls): return OutputModel
def run(self, input):
return OutputModel(result=input.number1 + input.number2)
@work_node 装饰器的 name 决定画布上显示的节点名称，group 决定它在节点库中归属哪个分类。核心业务逻辑全部写在 run() 方法里——输入是 InputModel 实例，输出是 OutputModel 实例，中间你想做什么都可以。

平台只约定了"接口契约"（输入什么、输出什么、叫什么名字），run() 里面的世界完全是你的。你要做的只是把自己的策略逻辑填进这个三段式结构里，它就能变成画布上一个可拖拽、可连线、可复用的节点。

步骤 3 ｜ 基于模板创建自定义节点——择时增强期货回测

理解了节点的三段式结构后，我们来做一件有实际意义的事：

把默认的「期货回测」节点改造成一个带择时过滤的版本。

现有工作流的数据流是：Python代码输入 → 线性因子构建 → 期货回测。因子告诉你"买什么"，但没有回答"什么时候该买"。

我们要做的是在回测节点内部加一层择时逻辑：当大盘/品种处于不利环境时，自动缩减仓位或空仓，避免逆势硬扛。

改造思路：

在原有期货回测节点的 InputModel 上新增三个字段：择时方法（均线择时 / 动量择时）、择时回看周期、以及空仓信号下的仓位比例。run() 方法中，在调用底层回测引擎之前，先根据择时参数生成信号，将其注入策略代码的头部，让回测引擎带着择时逻辑一起跑。

整个改造只做了三件事，且完全没有碰底层回测引擎的代码：

第一，InputModel 加了三个字段——择时方法、回看周期、减仓比例。配合 @ui 装饰器，它们会自动在画布节点上渲染为下拉框，用户无需写代码就能配置择时参数。

第二，写了一个 build_timing_code_snippet 函数，根据用户选择的择时方法生成对应的 Python 代码片段。均线择时的逻辑是"收盘价在 N 日均线之上则满仓，之下则按比例减仓"；动量择时是"N 日收益率为正则满仓，为负则减仓"。

第三，在 run() 中把择时片段拼接到用户策略代码的头部，然后把拼接后的完整代码传给 start() 引擎。这意味着择时逻辑和原有策略代码是共存的，用户的策略代码里可以直接调用 calc_timing_signal() 函数来获取仓位系数。

这就是专家模式的扩展边界：你不需要 fork 整个平台，只需要在节点的三段式结构里做加法。

输入加字段、逻辑加处理、输出加信息——新功能就长出来了。

步骤 4 ｜ 连接节点并查看择时回测结果

节点创建好了，但它还孤零零地挂在画布上，需要接入数据流才能跑起来。

连线操作：

代码端口 ← 连接「Python代码输入」节点的代码输出（你的策略代码）
因子值端口 ← 连接「线性因子构建」节点的因子值输出（构建好的因子 DataFrame）

连接完成后，数据流变为：Python代码输入 → 择时期货回测 ← 线性因子构建，两路数据汇入回测节点。

查看择时信息：

我们的改造在 OutputModel 里新增了 timing_info 和 backtest_id 两个输出字段。回测跑完后，backtest_id 可以连接「策略回测结果」节点查看完整的回测报告。而如果你想把择时信息单独导出，可以利用节点库中「01-基础工具」分组下的「数据下载」节点——它能把上游传来的数据转成 CSV 并生成 OSS 下载链接。

参数配置建议（首次测试）：

先用保守参数跑通流程：择时方法选「均线择时」，回看周期选 20（日频下相当于月线），空仓信号仓位比例选 0（即信号触发时完全空仓）。回测区间不要太长，先用 3 个月验证节点能正常出结果，再逐步拉长。

运行回测，切到「均线择时 + 回看20 + 空仓0」跑择时版本

略作修改再次运行

运行成功 但网络可能出了点问题，请求失败~后端接口报错。
今晚先到这里。