一专家模式：VWAP 计算节点测试

1.1 测试背景

在量化策略开发中，VWAP 是衡量市场平均价格的重要指标。本次测试旨在验证通过专家模式（Python 脚本）自定义开发的“5日VWAP节点”在逻辑计算、数据流转及因子生成方面的准确性与稳定性。

1.2 节点配置与逻辑验证

点击左侧导航栏-节点库右侧的专家模式;
$图片1.png$
算法实现（专家模式代码）;

根据代码截图（图片2、图片3），VWAP 的核心逻辑如下：
输入参数：接收过去 5 个周期（ $t$ 到 $t-4$ ）的 Close（收盘价）、High（最高价）、Low（最低价）以及 Volume（成交量）。
典型价格计算：$$Typical_Price = \frac{Close + High + Low}{3}$$
加权计算：$$VWAP = \frac{\sum (Typical_Price \times Volume)}{\sum Volume}$$
异常处理：代码中包含了对分母（累计成交量）是否为零的检查（if abs(vwap_den) <= 1e-6），有效防止了除零错误。

1.3 测试结果分析

在专家模式的调试窗口中，输入了模拟数据：

输入样本：包括 5 组价格与成交量数据。
输出结果：vwap: 9.743668262367675（见图片3底部打印）。
结论：计算逻辑运行正常，输出值符合加权平均逻辑。

1.4 节点拓扑结构

将自定义节点，拖拽至右侧画布，即可保存VWAP计算节点。
图片4.png 图片5.png

上游节点：Python 自定义因子节点（提供原始行情数据）。
当前节点：因子-5日VWAP_20日…（计算核心）。
下游节点：SVM 模型节点或因子分析节点，用于进一步的机器学习训练或有效性检验。

二期货仿真实盘跑通啦！

图片7.png
参考图片6的实盘运行状态：

当前合约：CU2604.SHF（沪铜）。
运行记录：日志显示策略持续接收行情并进行因子计算。
警告提示：日志中出现“名义价值超出开仓资金上限”的提示。这属于风控模块反馈，说明已成功触发交易指令，但受限于资金管理规则。

========================================================================================
补充：VWAP计算逻辑如下：

from typing import List, Optional, Type

from panda_plugins.base import BaseWorkNode, work_node
from pydantic import BaseModel, Field


class InputModel(BaseModel):
    """VWAP 因子节点输入模型（简洁版）

    只需要传入最近 5 天的收盘价、高价、低价、成交量 4 条时间序列：
    - close[i], high[i], low[i], volume[i] 代表第 i 天的数据
    - 按时间顺序排列：
        index=-5: 最早那天
        index=-1: 最新当天
    """

    close: List[float] = Field(..., description="最近 5 日收盘价序列，长度必须为 5，按时间先后排序")
    high: List[float] = Field(..., description="最近 5 日最高价序列，长度必须为 5")
    low: List[float] = Field(..., description="最近 5 日最低价序列，长度必须为 5")
    volume: List[float] = Field(..., description="最近 5 日成交量序列，长度必须为 5")


class OutputModel(BaseModel):
    """VWAP 因子输出模型"""

    vwap: float


@work_node(name="因子-5日VWAP", group="因子节点")
class VWAP5DayPlugin(BaseWorkNode):
    """5 日“历史成交量权重”的 VWAP 因子节点（序列输入版）

    公式与原定义等价：

        typical_price_t   = (close_t + high_t + low_t) / 3

        vwap = sum_{i=0..4}( typical_price_{t-i} * volume_{t-i} ) \\
               / max( sum_{i=0..4}( volume_{t-i} ), 1e-6 )

    这里通过 List[float] 序列输入代替逐日 lag 字段，让节点输入更简洁。
    """

    @classmethod
    def input_model(cls) -> Optional[Type[BaseModel]]:
        return InputModel

    @classmethod
    def output_model(cls) -> Optional[Type[BaseModel]]:
        return OutputModel

    def run(self, input: BaseModel) -> BaseModel:
        # 校验长度，确保都是 5 日窗口
        n_close = len(input.close)
        n_high = len(input.high)
        n_low = len(input.low)
        n_vol = len(input.volume)

        if not (n_close == n_high == n_low == n_vol == 5):
            raise ValueError(
                f"VWAP5DayPlugin: 输入序列长度必须全部为 5，当前长度分别为 "
                f"close={n_close}, high={n_high}, low={n_low}, volume={n_vol}"
            )

        # 逐日计算典型价格并加权求和
        vwap_num = 0.0
        vwap_den = 0.0
        for c, h, l, v in zip(input.close, input.high, input.low, input.volume):
            tp = (c + h + l) / 3.0
            vwap_num += tp * v
            vwap_den += v

        # 防除零
        if abs(vwap_den) <= 1e-6:
            vwap_den = 1e-6

        vwap_value = vwap_num / vwap_den
        return OutputModel(vwap=vwap_value)


if __name__ == "__main__":
    # 简单自测：构造 5 日窗口序列
    node = VWAP5DayPlugin()
    sample_input = InputModel(
        close=[9.6, 9.7, 9.5, 9.8, 10.0],
        high=[10.1, 10.0, 9.9, 10.2, 10.5],
        low=[9.2, 9.3, 9.1, 9.4, 9.5],
        volume=[600.0, 700.0, 800.0, 900.0, 1000.0],
    )
    print(node.run(sample_input))

一 专家模式：VWAP 计算节点测试