使用Pandas滚动应用函数对整个窗口数据框进行操作

Question

我希望对滑动窗口应用一个函数。我在这里看到的所有答案都是关注于将函数应用于单个行/列，但我想要将我的函数应用于整个窗口。以下是一个简化的示例：

import pandas as pd
data = [ [1,2], [3,4], [3,4], [6,6], [9,1], [11,2] ]
df = pd.DataFrame(columns=list('AB'), data=data)

这是 df：

    A   B
0   1   2
1   3   4
2   3   4
3   6   6
4   9   1
5   11  2

取一个函数应用于整个窗口：

df.rolling(3).apply(lambda x: x.shape)

在这个例子中，我希望得到类似于以下内容：

    some_name   
0   NA  
1   NA  
2   (3,2)   
3   (3,2)   
4   (3,2)   
5   (3,2)   

当然，这个形状只是用作示例，展示了f将整个窗口作为计算对象，而不仅仅是一行/一列。我尝试使用 rolling 的 axis 关键字以及 apply 的 raw 关键字进行操作，但都没有成功。其他方法 (agg, transform) 似乎也不能实现。

当然，我可以使用列表推导式来做到这一点。只是想知道有没有更简单/更清晰的方法。

- Yair Daon

---

也许这个能帮到你：https://dev59.com/zGIj5IYBdhLWcg3whFUI - Shaido

以下答案是否回答了您的问题？使用pandas，我认为没有更简洁的方法。 - Ouyang Ze

Answer 1

如果使用pd.DataFrame.rolling，该函数会迭代地应用于列中，接收一系列浮点数/NaN，并逐个返回一系列浮点数/NaN。我认为你更好的选择是凭直觉来做。

def rolling_pipe(dataframe, window, fctn):
    return pd.Series([dataframe.iloc[i-window: i].pipe(fctn) 
                      if i >= window else None 
                      for i in range(1, len(dataframe)+1)],
                     index = dataframe.index) 

df.pipe(rolling_pipe, 3, lambda x: x.shape)

- Ouyang Ze

---

你能否简要解释一下这段代码的作用？谢谢！ - user6400946

当然 - pd.DataFrame.pipe 是一种非常有用的方法。它以一个函数作为参数，该函数的第一个输入是一个 pd.DataFrame。要充分利用 pipe，通常希望它返回一个 Series 或 DataFrame 对象，以便可以将这些管道连接在一起...但那是一个单独的话题。 - Ouyang Ze

本例中，我们知道想要对数据框的子集进行“滚动应用”函数。首先，我们将使用window参数来定义数据框的第一个“切片”，然后使用.iloc[..].pipe(fctn)在该切片上运行fctn函数并返回一个值。接着，我们使用列表推导式不断向下滚动数据框。在这种情况下，我们希望返回的明显对象是一个具有与输入数据框相同索引（index = dataframe.index）的pd.Series对象。 - Ouyang Ze

还有两点需要注意：1. 这里的“fctn”是一个函数，它期望输入一个“pd.DataFrame”，并假定输出为非可迭代对象，例如数字或字符串。这个函数有一个版本可以返回数据帧而不是序列，只是它的写法与上面有所不同。2. 自从发表这篇文章以后，我遇到了一个类似的函数叫做“pd.rolling_apply”，但是它的文档很缺乏，所以你必须自己测试一下它是否和“rolling_pipe”做着相同的事情。 - Ouyang Ze

Answer 2

您的apply函数所提供的参数是一个Series，其中包含一个索引属性，该属性包含start、stop和step属性。

RangeIndex(start=0, stop=2, step=1)

您可以使用此功能来查询您的数据框。

df = pd.DataFrame([('Sean', i) for i in range(1,11)], columns=['name', 'value'])

def func(series):
    view = df.iloc[series.index]
    # use view to do something...
    count = len(view[view.value.isin([1,2,8])])
    return count

df['count'] = df.value.rolling(2).apply(func)

可能有更有效的方法来做这件事，但我不确定如何实现。

- seanbehan

---

太棒了！感谢您的发布。 - David R

Answer 3

如果您需要在类似日期时间的索引上滚动应用程序，则其他答案不足够。您必须手动迭代Rolling对象，并根据需要将结果重构为Series或DataFrame：

from datetime import (
    datetime as DateTime,
    timedelta as TimeDelta,
)
import pandas as pd

now = DateTime.now(tz=TimeZone.utc)

df = pd.DataFrame([
    {'t': now + TimeDelta(days=1), 'x': 11, 'y': 21},
    {'t': now + TimeDelta(days=2), 'x': 12, 'y': 22},
    {'t': now + TimeDelta(days=3), 'x': 13, 'y': 23},
    {'t': now + TimeDelta(days=4), 'x': 14, 'y': 24},
]).set_index('t')

results = {}
for group in df.rolling('2D'):
    # Perform a silly calculation, in this case an aggregation
    result = group['y'].min() * group['x'].max()
    # Choose a value to use as the resulting index
    index = group.index.min()
    results[index] = result
results = pd.Series(results)
print(results)

2022-07-15 01:41:05.121823+00:00    252
2022-07-16 01:41:05.121823+00:00    286
2022-07-17 01:41:05.121823+00:00    322
dtype: int64

这类似于迭代GroupBy对象。但不幸的是，与GroupBy不同，迭代并没有产生用于滚动窗口的实际边界。我不知道如何手动获取这些边界。

我原以为可以使用DataFrame.rolling中新的method=参数来实现这一点，但我无法使其正常工作。如果我弄清楚了，我会发表另一个答案！

- shadowtalker