均线指标计算节点代码

在量化交易和技术分析中，均线（Moving Average，简称 MA）是最基础也是最核心的指标之一。

它的核心目的只有一个：消除市场噪音，看清真实趋势。

日常的K线价格（尤其是加密货币或期货）总是上下剧烈跳动，很容易让人迷失方向。均线计算就是通过数学平均的方式，把价格的尖锐波动“平滑”掉，连成一条相对平稳的趋势线。

简单来说，均线主要分为两种常见类型：

简单移动平均 (SMA - Simple Moving Average)

计算方式：最朴素的算术平均法。比如“20周期SMA”，就是把过去20个交易日的收盘价加总，然后除以20。
特点：平稳，但反应稍慢。它对过去20天内的每一天都一视同仁（权重相同），因此当市场突然发生剧烈反转时，SMA 的掉头速度会比较迟钝（具有滞后性）。

指数移动平均 (EMA - Exponential Moving Average)

计算方式：加权平均法。它认为“越近发生的事情越重要”，因此给最近几天的价格赋予了更高的权重，而离现在越远的价格权重越低。
特点：敏感，反应迅速。EMA 能够更快地贴合当前价格的最新变动，在量化策略中常用于快速捕捉趋势反转的信号。

均线在量化回测中的 3 个核心作用：

判断大趋势：价格在均线上方运行，通常被视为多头（看涨）市场；在均线下方运行，则是空头（看跌）市场。
寻找支撑与阻力：在上涨趋势中，均线往往会像一张“蹦床”一样托住回调的价格（支撑位）；在下跌趋势中，均线又会变成压制价格反弹的“天花板”（阻力位）。
生成买卖信号：这是你在回测节点中最常用的功能。比如经典的“双均线策略”：当短期均线（如5周期）向上穿越长期均线（如20周期）时形成“金叉”，产生买入信号；反之形成“死叉”，产生卖出信号。

from typing import Any, Optional, Type
from panda_plugins.base import BaseWorkNode, work_node, ui
from pydantic import BaseModel, Field

# 1. 定义【输入模型】
@ui(
    window_size={
        "input_type": "number_input", 
        "min": 1, 
        "max": 200, 
        "placeholder": "请输入均线周期"
    },
    ma_type={
        "input_type": "select", 
        "options": [{"label": "简单移动平均 (SMA)", "value": "SMA"}, 
                    {"label": "指数移动平均 (EMA)", "value": "EMA"}],
    }
)
class MACalculatorInputModel(BaseModel):
    # 用于接收上游连线传过来的数据
    market_data: Any = Field(default=None, title="市场数据")
    
    # UI 面板配置项
    window_size: int = Field(default=20, title="计算周期")
    ma_type: str = Field(default="SMA", title="均线类型")

# 2. 定义【输出模型】
class MACalculatorOutputModel(BaseModel):
    processed_data: Any = Field(default=None, title="处理后的因子数据")
    message: str = Field(default="", title="执行信息")

# 3. 定义【节点核心逻辑】
@work_node(name="均线指标计算", group="自定义策略节点")
class MACalculatorNode(BaseWorkNode):
    
    # 必须使用 @classmethod，让系统在不运行节点时也能读到输入接口
    @classmethod
    def input_model(cls) -> Optional[Type[BaseModel]]:
        return MACalculatorInputModel
        
    # 必须使用 @classmethod，让系统在不运行节点时也能读到输出接口
    @classmethod
    def output_model(cls) -> Optional[Type[BaseModel]]:
        return MACalculatorOutputModel
        
    # 核心执行逻辑，注意这里的入参名最好用 input，保持与模板一致
    def run(self, input: MACalculatorInputModel) -> MACalculatorOutputModel:
        df = input.market_data
        window = input.window_size
        ma_type = input.ma_type
        
        # 兼容处理：防止刚拖拽节点没连线时报错
        if df is None:
            return MACalculatorOutputModel(
                processed_data=None,
                message="等待上游数据接入..."
            )

        # ---------------------------------------------------------
        # 此处为实际的指标计算逻辑 (伪代码，实际需配合 pandas 使用)
        if df is not None and not df.empty:
            if ma_type == "SMA":
                df[f'SMA_{window}'] = df['close'].rolling(window=window).mean()
            elif ma_type == "EMA":
                df[f'EMA_{window}'] = df['close'].ewm(span=window, adjust=False).mean()
        # ---------------------------------------------------------

        # 将结果封装进输出模型并返回
        return MACalculatorOutputModel(
            processed_data=df,
            message=f"成功计算 {window} 周期 {ma_type} 指标"
        )

# (可选) 用于在编辑器内本地测试是否能跑通
if __name__ == "__main__":
    node = MACalculatorNode()
    # 模拟一个空数据测试一下
    test_input = MACalculatorInputModel(window_size=20, ma_type="SMA")
    print(node.run(test_input))

代码结构解析

MACalculatorModel: 继承自 BaseModel，定义了这个节点在界面上需要用户输入的参数（如周期 window_size 和类型 ma_type）。

@ui 装饰器: 将后端的 Python 参数与前端编辑器的 UI 控件（如数字输入框、下拉选择菜单）绑定。

MACalculatorNode: 继承自 BaseWorkNode，包含 execute 方法，这是该节点接收上游数据、执行具体计算，并向下游输出数据的核心逻辑所在。

代码主要由三个核心部分构成：

MACalculatorInputModel (输入与界面模型)

作用：定义节点左侧的“接收口”和右侧属性面板的“配置项”。
market_data: 生成节点左侧的连线圆点，专门用来接收上游（比如行情源）传过来的 K 线数据表格。
@ui 装饰器及内部的 window_size / ma_type: 告诉平台前端，在这个节点的属性面板里画出一个数字输入框（填周期）和一个下拉菜单（选 SMA/EMA），让你可以在图形界面上直接调参，而不用每次都改代码。

MACalculatorOutputModel (输出模型)

作用：定义节点右侧的“发送口”。
processed_data: 生成节点右侧的连线圆点。等节点里的均线算完后，它负责把带有新均线列的数据表，顺着连线发送给下游的“期货回测”节点。
message: 方便你在运行日志里查看计算是否成功。

MACalculatorNode (节点大脑/核心逻辑)

作用：负责注册节点并执行真正的数学计算。
@work_node: 相当于给平台打个报告，让这个节点在左侧的“节点库”里挂牌上市。
input_model & output_model: 通过 @classmethod，让系统在不运行回测的情况下，就能提前读取模型并把左右两边的接口圆点画出来。
run 方法: 真正干活的车间。它把上游传进来的数据（df）和你在面板设置的参数（window）提取出来，利用 Pandas 进行移动平均线的计算，最后打包成 OutputModel 丢给下游。

均线指标计算节点应用步骤

在专家模式里写好代码，保存并重新加载窗口

注意：如果不重新加载会导致节点显示不正常（这个bug后续可能会修复）

从节点库中找到自定义节点，拖入画布
完成后续流程的设计

均线指标计算节点使用方法

连线：接入数据流

输入（左侧接源头）：找到你工作流里负责输出“历史行情数据”的节点（比如 CSV 文件读取、数据库加载，或者是你最开始的那个“Python代码输入”节点）。把它的数据输出点，连到我们这个均线节点的【市场数据】输入点上。
隐藏前提：上游传进来的数据必须是一张包含 ‘close’（收盘价）列的数据表（Pandas DataFrame）。
输出（右侧接下游）：把这个均线节点的【处理后的因子数据】输出点，连到右侧【期货回测】节点的对应输入点（比如“因子值”或“数据源”）上。

配置：在界面上调参

在画布上点击选中这个均线节点。
界面的侧边栏或属性区会自动弹出一个配置面板。这就是我们之前代码里 @ui 装饰器的功劳，你可以直接在图形界面上操作：
计算周期：输入框里填入你想要的数字（比如填 20 就是计算 20 周期均线，填 5 就是 5 周期）。
均线类型：在下拉菜单里点选 SMA（简单移动平均）或 EMA（指数移动平均）。
最大的好处：以后你想测试不同周期的策略，直接在界面上点一点、改改数字就行，再也不用去动底层的 Python 代码了！

运行：启动回测

点击整个工作流界面上的【启动工作流】。
底层发生了什么：节点收到原始行情数据后，会根据你的面板设置，自动在数据表的最右侧“无缝拼接”上一列新的数据（列名可能是 SMA_20）。然后，它把这张带着均线因子的新表，直接喂给后面的回测节点去计算收益率。