如何在Snowflake中使用Snowpark计算指数加权移动平均（EWMA）

核心逻辑 #

指数加权移动平均（EWMA）的核心是递推计算，公式为：
EWMA_t = α * REVENUE_t + (1-α) * EWMA_{t-1}
分组内第一行的EWMA值直接取当期REVENUE作为初始值。

实现步骤 #

1. 导入依赖

from snowflake.snowpark.functions import col, lag, when, lit
from snowflake.snowpark.window import Window

2. 直接通过窗口函数组合实现

不需要单独封装函数的话，可直接编写计算逻辑：

alpha = 0.9
# 按资产分组、按月份排序的窗口
window_spec = Window.partition_by("ASSET_ID").order_by("MONTH")

# 新增EWMA计算列
rev_df_with_ewma = rev_df.with_column(
    "EWMA_REVENUE",
    # 分组首行无前置数据，直接取当前收入作为初始EWMA
    when(
        lag(col("REVENUE")).over(window_spec).is_null(),
        col("REVENUE")
    ).otherwise(
        # 非首行用递推公式计算EWMA
        lit(alpha) * col("REVENUE") + lit(1 - alpha) * lag(col("EWMA_REVENUE")).over(window_spec)
    )
)

3. 封装为可复用函数

如果需要将逻辑封装成独立的ewma函数，可按如下方式编写：

def ewma(col_name, alpha):
    window_spec = Window.partition_by("ASSET_ID").order_by("MONTH")
    return when(
        lag(col(col_name)).over(window_spec).is_null(),
        col(col_name)
    ).otherwise(
        lit(alpha) * col(col_name) + lit(1 - alpha) * lag(ewma(col_name, alpha)).over(window_spec)
    )

# 调用函数生成EWMA列
alpha = 0.9
rev_df_with_ewma = rev_df.with_column("EWMA_REVENUE", ewma("REVENUE", alpha))

注意事项 #

• 确保MONTH列是可排序的日期或数值类型，保证时间序列顺序正确。
• α取值范围为0~1，值越接近1，近期数据的权重越高。

内容的提问来源于stack exchange，提问作者user1742188