时间戳

首页 > 解决方案 > 在 Python 中分阶段迭代列表

python - 在 Python 中分阶段迭代列表

问题描述

我最近一直在使用 Python 中的时间序列数据，并且遇到了许多场景，我需要迭代数据值列表，并在迭代的不同点执行不同的操作，具体取决于数据领先到它。这些场景都可以通过单次迭代来解决，并且不难弄清楚如何编写执行它的代码，但每次感觉就像我过于复杂并且编写了不必要的代码。感觉应该有一种更简单、更 Pythonic 的方式。

一个简化的典型示例是这样的：

def find_above_threshold_for_n(readings, upper, n)
  segments = [] # A list-of-lists of contiguous readings above `upper`
  current = None
  for value in readings:
    if value < upper:
       # 1. Skip ahead to first threshold-crossing value.
       current = None
    else:
      if current is None:
        current = []
        count = 0
      current.append(value)
      count += 1
      # 2. Keep iterating until we find n above threshold
      if count >= n:
         # 3. If we find enough readings, keep this segment. We'll continue adding to it until we dip below threshold again.
         segments.append(current)
  return segments

这个概念是可行的，但感觉它应该可以用更少的代码行来实现，并且结构可以更好地映射我的心理模型。

正如你所看到的，当我们遍历列表时，有三个基本阶段——跳到第一个相关值，继续收集值直到 n，并继续收集值直到下一个不需要的值，然后冲洗并重复。每个阶段从前一个停止的地方继续。感觉应该有一个简单的习惯用法来实现这一点，因为上面的代码并没有真正让这些阶段变得明显。

它并不总是相同的阶段或阈值算法（例如，我正在处理的一些数据是位置，它是关于跟踪设备移动的距离），但通常有一个家庭相似的阶段移动列表。

我正在寻找类似的东西：

segments = []
while readings:
  while next(readings) < upper:
     continue
  segment = []
  reading = next(readings)
  while reading >= upper:
    segment.append(reading)
    reading = next(readings)
  if len(segment) >= n:
    segments.append(segment)

不幸的是，在 Python 中手动循环迭代器并不是那么漂亮，因为无论何时调用 next()，都必须有 3 行额外的 try/except 来检测 StopIteration。即使没有这个，上面的“改进”代码也不是很简单，尽管它确实更接近了我的心理模型。

我试图避免显式索引（即循环列表的长度），因为它们很容易出现一对一和意外溢出错误，并且它们很少导致更清晰的代码。

有没有一种更简洁的方法来分阶段迭代列表，在每个阶段从最后一个阶段停止的地方开始？或者，有没有更好的方法来解决这类问题？

标签： pythonwhile-loopiteration

解决方案

有了itertools.groupby它可以变得更短：

import itertools


def find_above_threshold_for_n(readings, upper, n):
    segments = []

    for valid, group in itertools.groupby(readings, lambda v: v >= upper):
        if not valid: continue

        group = list(group)
        if len(group) < n:
            continue

        segments.append(group)

    return segments


data = [1,7,9,11,10,9,8,6,7,8,9,1,2,5,8,3]
print(find_above_threshold_for_n(data, 7, 3))

更短但可能更不可读：

from itertools import groupby


def find_above_threshold_for_n(readings, upper, n):
    segments = [list(group) for valid, group in
            groupby(readings, lambda v: v >= upper) if valid]

    return list(filter(lambda g: len(g) >= n, segments))

推荐阅读