基于值而不是计数的 Pandas 滑动计算窗口

Question

我希望能找到一种像 pandas 的各种rolling_*函数那样的方法，但是我想要滚动计算的窗口由一系列值（比如DataFrame列的一系列值）定义，而不是由窗口中的行数定义。

举个例子，假设我有这些数据：

>>> print d
   RollBasis  ToRoll
0          1       1
1          1       4
2          1      -5
3          2       2
4          3      -4
5          5      -2
6          8       0
7         10     -13
8         12      -2
9         13      -5

如果我执行类似于rolling_sum(d, 5)的操作，我将得到一个滚动总和，其中每个窗口包含5行。但是我想要的是一个滚动总和，其中每个窗口包含RollBasis的某个值范围。也就是说，我希望能够执行类似于d.roll_by(sum,'RollBasis',5)的操作，并得到一个结果，其中第一个窗口包含所有RollBasis在1到5之间的行，然后第二个窗口包含所有RollBasis在2到6之间的行，然后第三个窗口包含所有RollBasis在3到7之间的行，以此类推。这些窗口的行数不会相等，但在每个窗口中选择的RollBasis值范围将是相同的。因此输出应该像这样：

>>> d.roll_by(sum, 'RollBasis', 5)
    1    -4    # sum of elements with 1 <= Rollbasis <= 5
    2    -4    # sum of elements with 2 <= Rollbasis <= 6
    3    -6    # sum of elements with 3 <= Rollbasis <= 7
    4    -2    # sum of elements with 4 <= Rollbasis <= 8
    # etc.

我不能使用groupby来完成这个任务，因为groupby总是产生不相交的组。我也无法使用滚动函数来完成它，因为它们的窗口总是按行数滚动，而不是按值滚动。那么我该怎么做呢？

- BrenBarn

Answer 1

我认为这样做可以满足你的需求：

In [1]: df
Out[1]:
   RollBasis  ToRoll
0          1       1
1          1       4
2          1      -5
3          2       2
4          3      -4
5          5      -2
6          8       0
7         10     -13
8         12      -2
9         13      -5

In [2]: def f(x):
   ...:     ser = df.ToRoll[(df.RollBasis >= x) & (df.RollBasis < x+5)]
   ...:     return ser.sum()

上述函数接受一个值，本例中为RollBasis，然后根据该值索引数据框列ToRoll。返回的序列由满足RollBasis + 5条件的ToRoll值组成。最后，该序列被加总并返回。

In [3]: df['Rolled'] = df.RollBasis.apply(f)

In [4]: df
Out[4]:
   RollBasis  ToRoll  Rolled
0          1       1      -4
1          1       4      -4
2          1      -5      -4
3          2       2      -4
4          3      -4      -6
5          5      -2      -2
6          8       0     -15
7         10     -13     -20
8         12      -2      -7
9         13      -5      -5

如果有人想尝试玩具示例数据框的代码，请使用以下代码：

In [1]: from pandas import *

In [2]: import io

In [3]: text = """\
   ...:    RollBasis  ToRoll
   ...: 0          1       1
   ...: 1          1       4
   ...: 2          1      -5
   ...: 3          2       2
   ...: 4          3      -4
   ...: 5          5      -2
   ...: 6          8       0
   ...: 7         10     -13
   ...: 8         12      -2
   ...: 9         13      -5
   ...: """

In [4]: df = read_csv(io.BytesIO(text), header=0, index_col=0, sep='\s+')

- Zelazny7

---

谢谢，看起来可以了。我添加了一个更通用的版本的答案，但我接受你的答案。 - BrenBarn

Answer 2

基于BrenBarns的答案，但通过使用基于标签的索引而不是基于布尔值的索引加快了速度:

def rollBy(what,basis,window,func,*args,**kwargs):
    #note that basis must be sorted in order for this to work properly     
    indexed_what = pd.Series(what.values,index=basis.values)
    def applyToWindow(val):
        # using slice_indexer rather that what.loc [val:val+window] allows
        # window limits that are not specifically in the index
        indexer = indexed_what.index.slice_indexer(val,val+window,1)
        chunk = indexed_what[indexer]
        return func(chunk,*args,**kwargs)
    rolled = basis.apply(applyToWindow)
    return rolled

使用索引列比不使用要快得多：

In [46]: df = pd.DataFrame({"RollBasis":np.random.uniform(0,1000000,100000), "ToRoll": np.random.uniform(0,10,100000)})

In [47]: df = df.sort("RollBasis")

In [48]: timeit("rollBy_Ian(df.ToRoll,df.RollBasis,10,sum)",setup="from __main__ import rollBy_Ian,df", number =3)
Out[48]: 67.6615059375763

In [49]: timeit("rollBy_Bren(df.ToRoll,df.RollBasis,10,sum)",setup="from __main__ import rollBy_Bren,df", number =3)
Out[49]: 515.0221037864685

值得注意的是，基于索引的解决方案是O(n)，而逻辑切片版本在平均情况下是O(n^2)（我想）。

我发现按照基础值的最小值到最大值来均匀分布窗口进行比较更有用，而不是在每个基础值上进行比较。这意味着需要修改函数：

def rollBy(what,basis,window,func,*args,**kwargs):
    #note that basis must be sorted in order for this to work properly
    windows_min = basis.min()
    windows_max = basis.max()
    window_starts = np.arange(windows_min, windows_max, window)
    window_starts = pd.Series(window_starts, index = window_starts)
    indexed_what = pd.Series(what.values,index=basis.values)
    def applyToWindow(val):
        # using slice_indexer rather that what.loc [val:val+window] allows
        # window limits that are not specifically in the index
        indexer = indexed_what.index.slice_indexer(val,val+window,1)
        chunk = indexed_what[indexer]
        return func(chunk,*args,**kwargs)
    rolled = window_starts.apply(applyToWindow)
    return rolled

- Ian Sudbery

---

为了使其正常工作（至少在pandas 0.14中），我认为您需要将chunk = indexed_what[indexer]替换为chunk = indexed_what.iloc[indexer]。否则，该切片将被视为索引范围，但它是位置范围。 - feilchenfeldt

这与Bren的答案不返回相同的结果，因为它们处理开/闭区间的方式不同（请参见示例中索引5的结果）。它也不允许what成为DF而不仅仅是一个系列。 - Luis

使用datetime/timestamp仍然有效，但需要进行一些调整。至于Luis评论中提到的结果不同，只需将窗口int减1，因为slice_indexer在开始和结束时都是包含的（>=，<= vs >=，<）。 - Ranald Fong

另一个让事情正常工作的解决方案是用 chunk = indexed_what.index[indexer] 替换 chunk = indexed_what[indexer]。我还没有测试这是否比 @Luis 的 iloc 解决方案更快。 - PrinsEdje80

Answer 3

基于Zelazny7的回答，我创建了这个更通用的解决方案:

def rollBy(what, basis, window, func):
    def applyToWindow(val):
        chunk = what[(val<=basis) & (basis<val+window)]
        return func(chunk)
    return basis.apply(applyToWindow)

>>> rollBy(d.ToRoll, d.RollBasis, 5, sum)
0    -4
1    -4
2    -4
3    -4
4    -6
5    -2
6   -15
7   -20
8    -7
9    -5
Name: RollBasis

与rolling_apply相比，它仍然很慢，但也许这是不可避免的。

- BrenBarn

---

如果不是过滤第二列的值，而是过滤索引值，这会快得多。Pandas索引目前支持非唯一条目，因此您可以通过将基础列设置为索引，然后在其上进行过滤来加速解决方案。 - Ian Sudbery

Answer 4

为了扩展@Ian Sudbury的回答，我已经将其扩展成一种方法，可以直接将其绑定到DataFrame类上使用（我希望在速度方面肯定会有一些改进，因为我不知道如何访问类的所有内部）。

我还添加了向后窗口和居中窗口的功能。它们只在远离边缘时完美地发挥作用。

import pandas as pd
import numpy as np

def roll_by(self, basis, window, func, forward=True, *args, **kwargs):
    the_indexed = pd.Index(self[basis])
    def apply_to_window(val):
        if forward == True:
            indexer = the_indexed.slice_indexer(val, val+window)
        elif forward == False:
            indexer = the_indexed.slice_indexer(val-window, val)
        elif forward == 'both':
            indexer = the_indexed.slice_indexer(val-window/2, val+window/2)
        else:
            raise RuntimeError('Invalid option for "forward". Can only be True, False, or "both".')
        chunck = self.iloc[indexer]
        return func(chunck, *args, **kwargs)
    rolled = self[basis].apply(apply_to_window)
    return rolled

pd.DataFrame.roll_by = roll_by

对于其他测试，我使用了以下定义：

def rollBy_Ian_iloc(what,basis,window,func,*args,**kwargs):
    #note that basis must be sorted in order for this to work properly   
    indexed_what = pd.Series(what.values,index=basis.values)
    def applyToWindow(val):
        # using slice_indexer rather that what.loc [val:val+window] allows
        # window limits that are not specifically in the index
        indexer = indexed_what.index.slice_indexer(val,val+window,1)
        chunk = indexed_what.iloc[indexer]
        return func(chunk,*args,**kwargs)
    rolled = basis.apply(applyToWindow)
    return rolled

def rollBy_Ian_index(what,basis,window,func,*args,**kwargs):
    #note that basis must be sorted in order for this to work properly   
    indexed_what = pd.Series(what.values,index=basis.values)
    def applyToWindow(val):
        # using slice_indexer rather that what.loc [val:val+window] allows
        # window limits that are not specifically in the index
        indexer = indexed_what.index.slice_indexer(val,val+window,1)
        chunk = indexed_what[indexed_what.index[indexer]]
        return func(chunk,*args,**kwargs)
    rolled = basis.apply(applyToWindow)
    return rolled

def rollBy_Bren(what, basis, window, func):
    def applyToWindow(val):
        chunk = what[(val<=basis) & (basis<val+window)]
        return func(chunk)
    return basis.apply(applyToWindow)

时间和测试：

df = pd.DataFrame({"RollBasis":np.random.uniform(0,100000,10000), "ToRoll": np.random.uniform(0,10,10000)}).sort_values("RollBasis")

In [14]: %timeit rollBy_Ian_iloc(df.ToRoll,df.RollBasis,10,sum)
    ...: %timeit rollBy_Ian_index(df.ToRoll,df.RollBasis,10,sum)
    ...: %timeit rollBy_Bren(df.ToRoll,df.RollBasis,10,sum)
    ...: %timeit df.roll_by('RollBasis', 10, lambda x: x['ToRoll'].sum())
    ...: 
484 ms ± 28.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
1.58 s ± 10.2 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
3.12 s ± 22.7 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
1.48 s ± 45.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

结论：绑定方法不像@Ian Sudbury的方法那么快，但也不像@BrenBarn的方法那么慢，但它确实允许更多灵活性，可以调用更多的功能。

- PrinsEdje80