Pandas按组对多列应用自定义扩展函数（指数加权均值与标准差）

问题原因 #

你遇到的ValueError: Data must be 1-dimensional是因为expanding(method='table')默认仅支持处理单列一维数据，但你的自定义函数需要同时访问VAR_TIME（时间衰减）和VAR_X（目标变量）两列数据，直接调用无法满足多列输入的需求。

解决方案 #

通过groupby().apply()在每个分组内完整处理DataFrame，这样可以同时访问VAR_TIME和VAR_X来计算自定义权重的指数加权统计量，且该方案可无缝适配expanding（扩展窗口）和rolling（滚动窗口）两种场景。

步骤1：定义自定义加权函数 #

先实现基于时间衰减的加权均值和标准差函数，这里假设权重为指数衰减形式（你可根据需求调整衰减逻辑）：

import pandas as pd
import numpy as np

def calc_weighted_mean(window_df, tau=1.0):
    """计算指数加权均值，tau为衰减系数"""
    times = window_df['VAR_TIME'].values
    x_vals = window_df['VAR_X'].values
    # 以窗口内最新时间为基准计算权重，时间越近权重越高
    weights = np.exp(-(times[-1] - times) / tau)
    return np.sum(weights * x_vals) / np.sum(weights)

def calc_weighted_std(window_df, tau=1.0):
    """计算指数加权标准差"""
    mean_val = calc_weighted_mean(window_df, tau)
    times = window_df['VAR_TIME'].values
    x_vals = window_df['VAR_X'].values
    weights = np.exp(-(times[-1] - times) / tau)
    weighted_var = np.sum(weights * (x_vals - mean_val)**2) / np.sum(weights)
    return np.sqrt(weighted_var)

步骤2：按组计算扩展窗口统计量 #

对每个分组先按时间排序，再遍历扩展窗口计算结果：

def apply_expanding_stats(group):
    # 确保分组内数据按时间升序排列
    group_sorted = group.sort_values('VAR_TIME').reset_index(drop=True)
    # 遍历每个扩展窗口（从第1行到当前行）
    group_sorted['EWM_EXPAND'] = [
        calc_weighted_mean(group_sorted.iloc[:i+1]) 
        for i in range(len(group_sorted))
    ]
    group_sorted['EWSTD_EXPAND'] = [
        calc_weighted_std(group_sorted.iloc[:i+1]) 
        for i in range(len(group_sorted))
    ]
    return group_sorted

# 示例数据
df = pd.DataFrame({
    'VAR_GROUP': ['A', 'A', 'A', 'B', 'B', 'B'],
    'VAR_TIME': [1, 2, 3, 1, 2, 3],
    'VAR_X': [10, 20, 30, 5, 15, 25]
})

# 应用扩展窗口计算
result_expanding = df.groupby('VAR_GROUP').apply(apply_expanding_stats).reset_index(drop=True)
print(result_expanding)

步骤3：适配滚动窗口场景 #

只需修改分组内的窗口逻辑，指定滚动窗口大小即可：

def apply_rolling_stats(group, window_size=2, tau=1.0):
    group_sorted = group.sort_values('VAR_TIME').reset_index(drop=True)
    rolling_results = []
    for i in range(len(group_sorted)):
        # 确定滚动窗口的起始索引（避免越界）
        start_idx = max(0, i - window_size + 1)
        window_df = group_sorted.iloc[start_idx:i+1]
        rolling_results.append({
            'EWM_ROLLING': calc_weighted_mean(window_df, tau),
            'EWSTD_ROLLING': calc_weighted_std(window_df, tau)
        })
    # 合并结果到原分组数据
    rolling_df = pd.DataFrame(rolling_results, index=group_sorted.index)
    return pd.concat([group_sorted, rolling_df], axis=1)

# 应用滚动窗口计算（窗口大小为2）
result_rolling = df.groupby('VAR_GROUP').apply(apply_rolling_stats, window_size=2).reset_index(drop=True)
print(result_rolling)

关键说明 #

• 必须确保每个分组内的数据按VAR_TIME排序，否则时间衰减逻辑会失效；
• 自定义函数中的衰减系数tau可根据业务需求调整，tau越小衰减越快，历史数据权重越低；
• 该方案灵活度高，可根据需要修改权重计算逻辑（比如改为线性衰减、自定义时间差权重等）。

内容的提问来源于stack exchange，提问作者Roman velez jimenez