滚动或滑动窗口迭代器？

Question

我需要一个可在序列/迭代器/生成器上迭代的滚动窗口(也称为滑动窗口).默认的Python迭代可以被认为是一种特殊情况,窗口长度为1.我目前正在使用以下代码.有没有人有更多的Pythonic,更简洁,或更有效的方法来做到这一点？

def rolling_window(seq, window_size):
    it = iter(seq)
    win = [it.next() for cnt in xrange(window_size)] # First window
    yield win
    for e in it: # Subsequent windows
        win[:-1] = win[1:]
        win[-1] = e
        yield win

if __name__=="__main__":
    for w in rolling_window(xrange(6), 3):
        print w

"""Example output:

   [0, 1, 2]
   [1, 2, 3]
   [2, 3, 4]
   [3, 4, 5]
"""

Answer 1

旧版本的Python文档中有一个带有itertools示例:

from itertools import islice

def window(seq, n=2):
    "Returns a sliding window (of width n) over data from the iterable"
    "   s -> (s0,s1,...s[n-1]), (s1,s2,...,sn), ...                   "
    it = iter(seq)
    result = tuple(islice(it, n))
    if len(result) == n:
        yield result
    for elem in it:
        result = result[1:] + (elem,)
        yield result

来自文档的那个更简洁,并且用于itertools我想象的更大效果.

Answer 2

这看起来是量身定做的,collections.deque因为你基本上有一个FIFO(添加到一端,从另一端删除).但是,即使你使用了一个,list你也不应该切两次; 相反,你应该只是pop(0)从列表和append()新项目.

这是一个基于deque的优化实现,在您的原始版本之后进行图案化

from collections import deque

def window(seq, n=2):
    it = iter(seq)
    win = deque((next(it, None) for _ in xrange(n)), maxlen=n)
    yield win
    append = win.append
    for e in it:
        append(e)
        yield win

在我的测试中,它可以轻松地击败大部分时间在此处张贴的所有内容,尽管pillmuncher的tee版本比大型的可重复使用和小窗口更胜一筹.在较大的窗户上,deque以原始速度再次向前拉动.

访问单个项目deque可能比列表或元组更快或更慢.(如果使用负指数,则开头附近的项目会更快,或者靠近末尾的项目.)我sum(w)在循环体中放置了一个; 这对deque的强度起作用(从一个项目到另一个项目的迭代很快,所以这个循环比下一个最快的方法,即pillmuncher的速度快20%).当我将其更改为单独查找并在十个窗口中添加项目时,表格会转动,tee方法速度提高20%.我通过在添加中使用最后五个项的负索引来恢复一些速度,但tee仍然快一点.总的来说,我估计,对于大多数用途来说,任何一个都是快速的,如果你需要更多的性能,请选择最适合的那个.

Answer 3

我喜欢tee():

from itertools import tee, izip

def window(iterable, size):
    iters = tee(iterable, size)
    for i in xrange(1, size):
        for each in iters[i:]:
            next(each, None)
    return izip(*iters)

for each in window(xrange(6), 3):
    print list(each)

得到:

[0, 1, 2]
[1, 2, 3]
[2, 3, 4]
[3, 4, 5]

Answer 4

下面是添加了对泛化step,fillvalue参数:

from collections import deque
from itertools import islice

def sliding_window(iterable, size=2, step=1, fillvalue=None):
    if size < 0 or step < 1:
        raise ValueError
    it = iter(iterable)
    q = deque(islice(it, size), maxlen=size)
    if not q:
        return  # empty iterable or size == 0
    q.extend(fillvalue for _ in range(size - len(q)))  # pad to size
    while True:
        yield iter(q)  # iter() to avoid accidental outside modifications
        try:
            q.append(next(it))
        except StopIteration: # Python 3.5 pep 479 support
            return
        q.extend(next(it, fillvalue) for _ in range(step - 1))

如果需要,它一次产生块size项目,step每次迭代滚动位置,填充每个块fillvalue.示例size=4, step=3, fillvalue='*':

 [a b c d]e f g h i j k l m n o p q r s t u v w x y z
  a b c[d e f g]h i j k l m n o p q r s t u v w x y z
  a b c d e f[g h i j]k l m n o p q r s t u v w x y z
  a b c d e f g h i[j k l m]n o p q r s t u v w x y z
  a b c d e f g h i j k l[m n o p]q r s t u v w x y z
  a b c d e f g h i j k l m n o[p q r s]t u v w x y z
  a b c d e f g h i j k l m n o p q r[s t u v]w x y z
  a b c d e f g h i j k l m n o p q r s t u[v w x y]z
  a b c d e f g h i j k l m n o p q r s t u v w x[y z * *]

有关step参数用例的示例,请参阅有效地在python中处理大型.txt文件.

Answer 5

只是一个快速的贡献.

由于当前的python文档在itertool示例中没有"窗口"(即http://docs.python.org/library/itertools.html的底部),这里有一个基于石斑鱼代码的片段,是给出的例子之一:

import itertools as it
def window(iterable, size):
    shiftedStarts = [it.islice(iterable, s, None) for s in xrange(size)]
    return it.izip(*shiftedStarts)

基本上,我们创建了一系列切片迭代器,每个迭代器的起点都向前一个点.然后,我们将这些拉链在一起.注意,此函数返回一个生成器(它不直接是一个生成器本身).

与上面的appending-element和advance-iterator版本非常相似,性能(即最好)随列表大小和窗口大小而变化.我喜欢这个,因为它是一个双线程(它可能是一个单行,但我更喜欢命名概念).

事实证明上面的代码是错误的.如果传递给iterable的参数是一个序列但是如果它是一个迭代器则不行.如果它是一个迭代器,那么在islice调用中共享相同的迭代器(但不是tee'd),这会严重破坏它.

这是一些固定的代码:

import itertools as it
def window(iterable, size):
    itrs = it.tee(iterable, size)
    shiftedStarts = [it.islice(anItr, s, None) for s, anItr in enumerate(itrs)]
    return it.izip(*shiftedStarts)

此外,还有一本书的版本.这个版本不是复制迭代器然后多次推进复制,而是在我们向前移动起始位置时制作每个迭代器的成对副本.因此,迭代器t既提供了"完整"迭代器,也提供了t处的起始点,也提供了创建迭代器t + 1的基础:

import itertools as it
def window4(iterable, size):
    complete_itr, incomplete_itr = it.tee(iterable, 2)
    iters = [complete_itr]
    for i in xrange(1, size):
        incomplete_itr.next()
        complete_itr, incomplete_itr = it.tee(incomplete_itr, 2)
        iters.append(complete_itr)
    return it.izip(*iters)

Answer 6

只是为了展示如何结合itertools的食谱,我延长了pairwise尽可能直接配方回window用的配方consume配方:

def consume(iterator, n):
    "Advance the iterator n-steps ahead. If n is none, consume entirely."
    # Use functions that consume iterators at C speed.
    if n is None:
        # feed the entire iterator into a zero-length deque
        collections.deque(iterator, maxlen=0)
    else:
        # advance to the empty slice starting at position n
        next(islice(iterator, n, n), None)

def window(iterable, n=2):
    "s -> (s0, ...,s(n-1)), (s1, ...,sn), (s2, ..., s(n+1)), ..."
    iters = tee(iterable, n)
    # Could use enumerate(islice(iters, 1, None), 1) to avoid consume(it, 0), but that's
    # slower for larger window sizes, while saving only small fixed "noop" cost
    for i, it in enumerate(iters):
        consume(it, i)
    return zip(*iters)

该window配方是一样的pairwise,它只是代替在第二的单个元素"消费" tee-ed迭代器上逐步增加消耗n - 1迭代器.使用consume而不是包裹每个迭代器islice是稍快(足够大的iterables),因为您只需支付islice在包装开销consume阶段,而不是在提取每个窗口-ED值(所以它是由边界的过程n,而不是在项目数量iterable).

性能方面,与其他一些解决方案相比,这是非常好的(并且比我测试的任何其他解决方案都要好).在Python 3.5.0,Linux x86-64上测试,使用ipython %timeit魔法.

kindall的所述deque溶液,通过使用调整了性能/正确性islice,而不是家庭轧制发生器表达和测试所得到的长度,因此不会产生结果时,可迭代比窗口短,以及使所述maxlen的deque位置上,而不是通过关键字(对较小的输入产生惊人的差异):

>>> %timeit -r5 deque(windowkindall(range(10), 3), 0)
100000 loops, best of 5: 1.87 ?s per loop
>>> %timeit -r5 deque(windowkindall(range(1000), 3), 0)
10000 loops, best of 5: 72.6 ?s per loop
>>> %timeit -r5 deque(windowkindall(range(1000), 30), 0)
1000 loops, best of 5: 71.6 ?s per loop

与之前的改进的kindall解决方案相同,但是每个都yield win改变了,yield tuple(win)因此存储生成器的结果工作,而没有所有存储的结果真正是最近结果的视图(所有其他合理的解决方案在这种情况下是安全的),并添加tuple=tuple到函数定义移动使用tuple从Bin LEGB到L:

>>> %timeit -r5 deque(windowkindalltupled(range(10), 3), 0)
100000 loops, best of 5: 3.05 ?s per loop
>>> %timeit -r5 deque(windowkindalltupled(range(1000), 3), 0)
10000 loops, best of 5: 207 ?s per loop
>>> %timeit -r5 deque(windowkindalltupled(range(1000), 30), 0)
1000 loops, best of 5: 348 ?s per loop

consume上面显示的解决方案:

>>> %timeit -r5 deque(windowconsume(range(10), 3), 0)
100000 loops, best of 5: 3.92 ?s per loop
>>> %timeit -r5 deque(windowconsume(range(1000), 3), 0)
10000 loops, best of 5: 42.8 ?s per loop
>>> %timeit -r5 deque(windowconsume(range(1000), 30), 0)
1000 loops, best of 5: 232 ?s per loop

同样consume,但内联的else情况是consume避免函数调用和n is None测试以减少运行时间,特别是对于小型输入,其中设置开销是工作的一个有意义的部分:

>>> %timeit -r5 deque(windowinlineconsume(range(10), 3), 0)
100000 loops, best of 5: 3.57 ?s per loop
>>> %timeit -r5 deque(windowinlineconsume(range(1000), 3), 0)
10000 loops, best of 5: 40.9 ?s per loop
>>> %timeit -r5 deque(windowinlineconsume(range(1000), 30), 0)
1000 loops, best of 5: 211 ?s per loop

(侧注:一个变量pairwise使用tee默认参数2反复进行嵌套tee对象,因此任何给定的迭代器只提前一次,而不是独立消耗的次数越来越多,类似于MrDrFenner的答案类似于非内联consume并且比consume所有测试中的内联慢,所以我为了简洁而省略了那些结果.

正如您所看到的,如果您不关心调用者需要存储结果的可能性,那么我的优化版本的kindall解决方案大部分时间都会获胜,除了"大型可迭代,小窗口大小的情况"(内联consume获胜) ); 随着可迭代大小的增加,它会迅速降级,而随着窗口大小的增加,它不会降级(对于可迭代的大小增加,每个其他解决方案的降级速度都会降低,但窗口大小的增加也会降低).它甚至可以通过包装来适应"需要元组"的情况map(tuple, ...),其运行速度比将函数中的元组稍微慢一点,但它很简单(需要1-5%的时间)并且让你保持更快的运行灵活性当你可以忍受反复返回相同的值.

如果您需要对存储的返回进行安全保护,consume那么除了最小的输入大小外,其他所有输入都会内联(非内联consume略慢但缩放类似).所述deque&几倍基于溶液获得仅针对最小输入,由于更小的设置成本,并且增益小; 随着迭代次数变长,它会严重降级.

为了记录在案,即kindall的解决方案的改编版本yield小号tuple的I所用的是:

def windowkindalltupled(iterable, n=2, tuple=tuple):
    it = iter(iterable)
    win = deque(islice(it, n), n)
    if len(win) < n:
        return
    append = win.append
    yield tuple(win)
    for e in it:
        append(e)
        yield tuple(win)

删除tuple函数定义行中的缓存以及tuple每个中的缓存yield以获得更快但不太安全的版本.

Answer 7

有一个库可以完全满足您的需求：

import more_itertools
list(more_itertools.windowed([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15],n=3, step=3))

Out: [(1, 2, 3), (4, 5, 6), (7, 8, 9), (10, 11, 12), (13, 14, 15)]

Answer 8

我使用以下代码作为一个简单的滑动窗口,使用生成器大大提高可读性.到目前为止,它的速度足以用于我的经验中的生物信息学序列分析.

我在这里包括它因为我没有看到这个方法已经使用了.同样,我对其比较的表现没有任何说法.

def slidingWindow(sequence,winSize,step=1):
"""Returns a generator that will iterate through
the defined chunks of input sequence. Input sequence
must be sliceable."""

    # Verify the inputs
    if not ((type(winSize) == type(0)) and (type(step) == type(0))):
        raise Exception("**ERROR** type(winSize) and type(step) must be int.")
    if step > winSize:
        raise Exception("**ERROR** step must not be larger than winSize.")
    if winSize > len(sequence):
        raise Exception("**ERROR** winSize must not be larger than sequence length.")

    # Pre-compute number of chunks to emit
    numOfChunks = ((len(sequence)-winSize)/step)+1

    # Do the work
    for i in range(0,numOfChunks*step,step):
        yield sequence[i:i+winSize]

Answer 9

def GetShiftingWindows(thelist, size):
    return [ thelist[x:x+size] for x in range( len(thelist) - size + 1 ) ]

>> a = [1, 2, 3, 4, 5]
>> GetShiftingWindows(a, 3)
[ [1, 2, 3], [2, 3, 4], [3, 4, 5] ]

Answer 10

deque窗口的略微修改版本,使其成为真正的滚动窗口.所以它开始只用一个元素填充,然后增长到它的最大窗口大小,然后收缩,因为它的左边缘接近结束:

from collections import deque
def window(seq, n=2):
    it = iter(seq)
    win = deque((next(it, None) for _ in xrange(1)), maxlen=n)
    yield win
    append = win.append
    for e in it:
        append(e)
        yield win
    for _ in xrange(len(win)-1):
        win.popleft()
        yield win

for wnd in window(range(5), n=3):
    print(list(wnd))

这给了

[0]
[0, 1]
[0, 1, 2]
[1, 2, 3]
[2, 3, 4]
[3, 4]
[4]