pandas - 统计高于/低于当前行值的连续数量

Question

我正在寻找一种方法，可以将一个pandas Series转换为新的Series，表示每行Series值之前连续高于/低于该行值的数量：

a = pd.Series([30, 10, 20, 25, 35, 15])

您需要输出：

Value   Higher than streak  Lower than streak
30      0                   0
10      0                   1
20      1                   0
25      2                   0
35      4                   0
15      0                   3

这将使某人能够确定时间序列中每个“区域最大/最小”值的重要性。谢谢！

- Bruno Vieira

Answer 1

由于您需要查看先前的值以查看是否存在连续值，因此必须以某种方式与索引进行交互。这个解决方案首先查看当前索引之前的任何值，以查看它们是否小于或大于该值，然后设置在其后有False的任何值为False。它还避免了在DataFrame上创建迭代器，这可以加速处理更大数据集的操作。

import pandas as pd
from operator import gt, lt

a = pd.Series([30, 10, 20, 25, 35, 15])

def consecutive_run(op, ser, i):
    """
    Sum the uninterrupted consecutive runs at index i in the series where the previous data
    was true according to the operator.
    """
    thresh_all = op(ser[:i], ser[i])
    # find any data where the operator was not passing.  set the previous data to all falses
    non_passing = thresh_all[~thresh_all]
    start_idx = 0
    if not non_passing.empty:
        # if there was a failure, there was a break in the consecutive truth values,
        # so get the final False position. Starting index will be False, but it
        # will either be at the end of the series selection and will sum to zero
        # or will be followed by all successive True values afterwards
        start_idx = non_passing.index[-1]
    # count the consecutive runs by summing from the start index onwards
    return thresh_all[start_idx:].sum()


res = pd.concat([a, a.index.to_series().map(lambda i: consecutive_run(gt, a, i)),
                 a.index.to_series().map(lambda i: consecutive_run(lt, a, i))],
       axis=1)
res.columns = ['Value', 'Higher than streak', 'Lower than streak']
print(res)

结果：

   Value  Higher than streak  Lower than streak
0     30                   0                  0
1     10                   1                  0
2     20                   0                  1
3     25                   0                  2
4     35                   0                  4
5     15                   3                  0

- benjwadams

---

谢谢，我没想到我们会找到一个避免循环的解决方案。 - Bruno Vieira

通过仅获取相邻值并进行求和，更新为使用稍微更有效的求和算法。 - benjwadams

@benjwadams：在这里提出了对该方法的轻微调整（https://dev59.com/Ksb5oIgBc1ULPQZF-SLo#73694097）。不要使用`a.index.to_series()`，而是首先剪切肯定需要`0`的索引值。即使用`lt`，执行`idx = a[a.diff().ge(0)].index，然后a.iloc[idx].index.to_series()（对于gt，使用.le(0)）。所有其他索引值都需要零，因此我们可以通过使用Series.reindex并应用.fillna(0)来重新附加它们。例如，如果s = a.iloc[idx].index.to_series().map(...)，那么我们执行：res = s.reindex(a.index).fillna(0)`。 - ouroboros1

Answer 2

import pandas as pd
import numpy as np

value = pd.Series([30, 10, 20, 25, 35, 15])



Lower=[(value[x]<value[:x]).sum() for x in range(len(value))]
Higher=[(value[x]>value[:x]).sum() for x in range(len(value))]


df=pd.DataFrame({"value":value,"Higher":Higher,"Lower":Lower})

print(df)





      Lower  Higher  value
0       0      0     30
1       1      0     10
2       1      1     20
3       1      2     25
4       0      4     35
5       4      1     15

- 2Obe

---

谢谢您的回答。不幸的是，这个解决方案没有达到我期望的结果，因为每一行只应该与它之前的行进行比较。例如，在第二次观察时，10比30低 - 因此Lower列=1，Higher列=0。 - Bruno Vieira

也许根据您所假设的逻辑，您必须更改Higher和Lower的名称。 - 2Obe

假设这是一种更“Pythonic”的解决方案。 - 2Obe

Answer 3

编辑：更新以真正计算连续值。我无法想出一个可行的pandas解决方案，所以我们回到循环。

df = pd.Series(np.random.rand(10000))

def count_bigger_consecutives(values):
  length = len(values)
  result = np.zeros(length)
  for i in range(length):
    for j in range(i):
      if(values[i]>values[j]):
        result[i] += 1
      else:
        break
  return result

%timeit count_bigger_consecutives(df.values)
1 loop, best of 3: 365 ms per loop

如果您关心性能，可以使用numba实现加速，它是用于Python代码的即时编译器。在这个例子中，您确实可以看到numba的优势：

from numba import jit 
@jit(nopython=True)
def numba_count_bigger_consecutives(values):
  length = len(values)
  result = np.zeros(length)
  for i in range(length):
    for j in range(i):
      if(values[i]>values[j]):
        result[i] += 1
      else:
        break
  return result

%timeit numba_count_bigger_consecutives(df.values)
The slowest run took 543.09 times longer than the fastest. This could mean that an intermediate result is being cached.
10000 loops, best of 3: 161 µs per loop

- P.Tillmann

---

谢谢。非常有趣，我之前不熟悉expanding()函数。然而，这并不完全是期望的行为。我需要知道时间序列中连续过去观测值的最大数量，使得当前行等于max()或min()。 - Bruno Vieira

哇，这速度快多了。感谢分享这个解决方案。不幸的是，结果是array([ 0., 0., 0., 0., 4., 0.])，而我期望的是0、0、1、2、4、0。由于看起来解决方案仍需要循环，您建议使用numba仍然非常有用。 - Bruno Vieira

Answer 4

这是我的解决方案-它有一个循环，但迭代次数只会是最大连续长度。它存储了每行连续段是否已计算的状态，并在完成时停止。它使用shift来测试前一行是高还是低，并不断增加shift直到所有连续段都被找到。

a = pd.Series([30, 10, 20, 25, 35, 15, 15])

a_not_done_greater = pd.Series(np.ones(len(a))).astype(bool)
a_not_done_less = pd.Series(np.ones(len(a))).astype(bool)

a_streak_greater = pd.Series(np.zeros(len(a))).astype(int)
a_streak_less = pd.Series(np.zeros(len(a))).astype(int)

s = 1
not_done_greater = True
not_done_less = True

while not_done_greater or not_done_less:
    if not_done_greater:
        a_greater_than_shift = (a > a.shift(s))
        a_streak_greater = a_streak_greater + (a_not_done_greater.astype(int) * a_greater_than_shift)
        a_not_done_greater = a_not_done_greater & a_greater_than_shift
        not_done_greater = a_not_done_greater.any()

    if not_done_less:
        a_less_than_shift = (a < a.shift(s))
        a_streak_less = a_streak_less + (a_not_done_less.astype(int) * a_less_than_shift)
        a_not_done_less = a_not_done_less & a_less_than_shift
        not_done_less = a_not_done_less.any()

    s = s + 1


res = pd.concat([a, a_streak_greater, a_streak_less], axis=1)
res.columns = ['value', 'greater_than_streak', 'less_than_streak']
print(res)

这将给出数据框：

   value  greater_than_streak  less_than_streak
0     30                    0                 0
1     10                    0                 1
2     20                    1                 0
3     25                    2                 0
4     35                    4                 0
5     15                    0                 3
6     15                    0                 0

- Ken Syme

Answer 5

这是一个同事想出来的解决方案（可能不是最有效的，但它能解决问题）：

输入数据

a = pd.Series([30, 10, 20, 25, 35, 15])

创建“higher”列

b = []

for idx, value in enumerate(a):
    count = 0
    for i in range(idx, 0, -1):
        if value < a.loc[i-1]:
            break
        count += 1
    b.append([value, count])

higher = pd.DataFrame(b, columns=['Value', 'Higher'])

创建'lower'列

c = []

for idx, value in enumerate(a):
    count = 0
    for i in range(idx, 0, -1):
        if value > a.loc[i-1]:
            break
        count += 1
    c.append([value, count])

lower = pd.DataFrame(c, columns=['Value', 'Lower'])

合并这两个新系列

print(pd.merge(higher, lower, on='Value'))

   Value  Higher  Lower
0     30       0      0
1     10       0      1
2     20       1      0
3     25       2      0
4     35       4      0
5     15       0      3

- Bruno Vieira