如何用Pandas实现时间序列滚动5%分位缩尾后标准化？

我完全懂你的需求——先对时间序列做滚动窗口的分位缩尾（把超出指定分位的数值截断到对应分位值），再执行滚动标准化，而且之前尝试用rolling.apply没成功对吧？我来一步步帮你搞定这个问题。

第一步：自定义滚动缩尾函数 #

首先，我们需要写一个自定义函数，处理每个滚动窗口内的缩尾逻辑：对于每个窗口，先计算指定的上下分位值，然后把窗口的最后一个元素（也就是当前行的原始值）截断在这两个分位之间。

import numpy as np
import pandas as pd

def winsorize_single_value(window, lower_q=0.05, upper_q=0.95):
    # 计算窗口内的上下分位值（这里用线性插值，你可以根据需求调整interpolation参数）
    lower_bound = np.percentile(window, lower_q * 100, interpolation='linear')
    upper_bound = np.percentile(window, upper_q * 100, interpolation='linear')
    # 获取当前行的数值（窗口的最后一个元素）
    current_val = window[-1]
    # 执行缩尾：大于上界取上界，小于下界取下界，否则保留原值
    return max(min(current_val, upper_bound), lower_bound)

第二步：应用滚动缩尾到你的数据 #

接下来，用rolling.apply把这个函数应用到你的序列上。注意设置窗口大小（比如你需要的252天），以及min_periods参数（控制窗口最少需要多少个数据才计算，比如设为1的话，前面不足窗口大小的行也会计算）。

拿你给出的示例数据测试一下（窗口设为10，分位0.15和0.85）：

# 示例数据
df = pd.DataFrame({'series1':[78, 1, 3, 4, 5, 6, 7, 8, 99]})
# 应用滚动缩尾
df['winsorized'] = df['series1'].rolling(
    window=10, 
    min_periods=1
).apply(winsorize_single_value, args=(0.15, 0.85))

执行后得到的winsorized列就会符合你预期的缩尾逻辑——每个值都会根据其所在的滚动窗口分位被截断。

第三步：执行滚动标准化 #

在得到缩尾后的序列后，就可以按照你原来的逻辑执行滚动标准化了：

# 滚动窗口大小保持252
window_size = 252
df['std_series'] = (
    df['winsorized'] - df['winsorized'].rolling(window_size).mean()
) / df['winsorized'].rolling(window_size).std()

注意事项 #

• 性能优化：如果你的数据量非常大（比如几十万行），rolling.apply的Python层面循环会比较慢。这时候可以考虑用numba装饰自定义函数来加速，比如在函数上方加@numba.jit(nopython=True)（需要先安装numba）。
• 分位插值方式：np.percentile的interpolation参数可以调整，比如'nearest'、'lower'等，你可以根据自己的需求选择，这会影响分位值的计算结果。
• 窗口的包含逻辑：默认的rolling是包含当前行的，如果需要排除当前行（只用过去的252天数据），可以调整窗口的closed参数，比如closed='left'。

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