【问题标题】:Compute the cumulative sum of a list until a zero appears计算列表的累积和,直到出现零
【发布时间】:2018-07-26 01:54:46
【问题描述】:

我有一个(长)列表,其中随机出现零和一:

list_a = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]

我要获取列表_b

  • 列表的总和,直到出现 0 的位置
  • 出现0的地方,在列表中保留0

    list_b = [1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]
    

我可以这样实现:

list_b = []
for i, x in enumerate(list_a):
    if x == 0:
        list_b.append(x)
    else:
        sum_value = 0
        for j in list_a[i::-1]:
            if j != 0:
                sum_value += j
            else:
                break
        list_b.append(sum_value)
print(list_b)

但实际列表的长度很长。

所以,我想改进代码以提高速度。 (如果它不可读)

我把代码改成这样:

from itertools import takewhile
list_c = [sum(takewhile(lambda x: x != 0, list_a[i::-1])) for i, d in enumerate(list_a)]
print(list_c)

但这还不够快。我怎样才能更有效地做到这一点?

【问题讨论】:

  • 原来的列表中只有1s和0s吗?
  • 在列表长度上不可能有比线性更好的顺序解决方案。但是,您可以尝试通过从列表中的不同位置开始来并行化您的算法,搜索与零相邻的第一个非零条目并从那里继续。然后每个线程执行它的工作,直到它从相邻块的线程到达起点。奖励:您可以同时修改列表,因为线程将在不同的块上操作。
  • @Ev.库尼斯 是的。但是我将列表 [a, a, a, b, b, b, c, c .....] 更改为 list_a,如果更改值 0 s 而不是更改 1 s。
  • @HiroyukiTaniichi 如果你的实际数据真的是这样,那么试试:s.groupby(s.ne(s.shift()).cumsum()).cumcount()
  • @UniversE 这涉及到阅读列表至少比需要的时间多一次。您可以索引一部分并搜索下一个 0 并从那里开始,防止通读。然而,这是 python,所以你需要对其进行切片,然后 multi-process 操作,然后再次组合列表(python 线程在一个操作线程中进行时间切片。这 可能 比仅仅线性处理需要更长的时间)

标签: python performance list binary cumsum


【解决方案1】:

你想多了。

选项 1
您可以根据当前值是否为0 来迭代索引并相应地更新(计算累积总和)。

data = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]

for i in range(1, len(data)):
    if data[i]:  
        data[i] += data[i - 1] 

即如果当前元素不为零,则将当前索引处的元素更新为当前值与上一个索引处的值之和。

print(data)
[1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]

请注意,这会更新您的列表。如果您不想这样做,您可以提前创建一个副本 - new_data = data.copy() 并以相同的方式迭代 new_data


选项 2
如果您需要性能,可以使用 pandas API。根据0s 的位置查找分组,并使用groupby + cumsum 计算分组累积和,类似于上面:

import pandas as pd

s = pd.Series(data)    
data = s.groupby(s.eq(0).cumsum()).cumsum().tolist()

print(data)
[1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]

性能

首先,设置-

data = data * 100000
s = pd.Series(data)

接下来,

%%timeit
new_data = data.copy()
for i in range(1, len(data)):
    if new_data[i]:  
        new_data[i] += new_data[i - 1]

328 ms ± 4.09 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

并且,单独定时复制,

%timeit data.copy()
8.49 ms ± 17.2 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

所以,复制并不需要太多时间。最后,

%timeit s.groupby(s.eq(0).cumsum()).cumsum().tolist()
122 ms ± 1.69 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

pandas 方法在概念上是线性的(就像其他方法一样),但由于库的实现,速度更快。

【讨论】:

  • 我认为您应该将其作为另一个列表,而无需重写原始列表。可能还需要它。
  • @big_bang 是的,我已经提到 OP 可以创建一个副本:new_data = data.copy() 并迭代 new_data :)
  • 非常感谢您!我误解了列表理解类型比普通代码更快。
  • @HiroyukiTaniichi 一般来说,只要做得好,它就可以。但是,我不确定您的问题是否可以在不牺牲性能的情况下成功转换为列表理解。我在回答中添加了时间。
  • @HiroyukiTaniichi 不客气。 Pandas 是一个强大的 API,但需要一些时间来适应。如果您有任何问题,请不要害怕提出问题。祝你好运!
【解决方案2】:

您在发布的代码中过多地使用索引,而实际上您并不需要这样做。每次遇到0 时,您只需跟踪累积和并将其重置为0

list_a = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]

cum_sum = 0
list_b = []
for item in list_a:
    if not item:            # if our item is 0
        cum_sum = 0         # the cumulative sum is reset (set back to 0)
    else:
        cum_sum += item     # otherwise it sums further
    list_b.append(cum_sum)  # and no matter what it gets appended to the result
print(list_b)  # -> [1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]

【讨论】:

  • @cᴏʟᴅsᴘᴇᴇᴅ 感谢您安排时间!
  • 您可以通过避免使用append = list_b.append 查找append 方法来加快速度
  • @Leonhard 真的吗?没人这样做吗?您认为这会显着提高查找速度吗?对我来说似乎是命名空间污染
  • 我在 timeit 模块中进行了尝试,在这种情况下循环仅执行分支、添加和追加,它约为 15%。并不是说这可能是相关的,除了在这个线程中击败另一个答案。
【解决方案3】:

如果您想要一个紧凑的原生 Python 解决方案,它可能是内存效率最高但不是最快的(请参阅 cmets),您可以广泛借鉴 itertools

>>> from itertools import groupby, accumulate, chain
>>> list(chain.from_iterable(accumulate(g) for _, g in groupby(list_a, bool)))
[1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]

这里的步骤是:根据0的存在将列表分组为子列表(这是错误的),获取每个子列表中值的累积总和,展平子列表。

作为Stefan Pochmann cmets,如果您的列表是二进制内容(例如仅包含1s 和0s),那么您根本不需要将密钥传递给groupby()将依赖于身份功能。这比在这种情况下使用 bool 快约 30%:

>>> list(chain.from_iterable(accumulate(g) for _, g in groupby(list_a)))
[1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]

【讨论】:

  • :hmm.jpg: 如果 OP 是 looking for performance,这可能不是正确的解决方案。
  • @cᴏʟᴅsᴘᴇᴇᴅ 我赞成你的 Pandas 方法,但你对我的方法进行了基准测试吗?发生在我身上,groupby(list_a, int) 可能会更快?
  • 是的。我的设置在681 ms ± 4.26 ms per loop 打卡,我确保在评论之前验证了这一点:)
  • 你的回答还是不错的,你已经有我的投票了:)
  • 为什么不用groupby(list_a),不用钥匙?
【解决方案4】:

不必像问题中提出的那样复杂,一个非常简单的方法可以是这样。

list_a = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]
list_b = []
s = 0
for a in list_a:
    s = a+s if a !=0 else 0
    list_b.append(s)

print list_b

【讨论】:

    【解决方案5】:

    如果你想要性能,我会使用生成器(而且它也很简单)。

    def weird_cumulative_sum(seq):
        s = 0
        for n in seq:
            s = 0 if n == 0 else s + n
            yield s
    
    list_b = list(weird_cumulative_sum(list_a_))
    

    我认为您不会比这更好,无论如何您都必须至少迭代 list_a 一次。

    请注意,我在结果上调用了 list() 以获取类似于您的代码中的列表,但是如果使用 list_b 的代码仅使用 for 循环或其他方法对其进行迭代,则将结果转换为列表是没有用的,只需将它传递给生成器即可。

    【讨论】:

      【解决方案6】:

      我个人更喜欢这样的简单生成器:

      def gen(lst):
          cumulative = 0
          for item in lst:
              if item:
                  cumulative += item
              else:
                  cumulative = 0
              yield cumulative
      

      没什么神奇的(当您知道yield 的工作原理时),易于阅读并且应该相当快。

      如果您需要更高的性能,您甚至可以将其包装为 Cython 扩展类型(我在这里使用 IPython)。因此你失去了“易于理解”的部分,它需要“高度依赖”:

      %load_ext cython
      
      %%cython
      
      cdef class Cumulative(object):
          cdef object it
          cdef object cumulative
          def __init__(self, it):
              self.it = iter(it)
              self.cumulative = 0
      
          def __iter__(self):
              return self
      
          def __next__(self):
              cdef object nxt = next(self.it)
              if nxt:
                  self.cumulative += nxt
              else:
                  self.cumulative = 0
              return self.cumulative
      

      两者都需要消耗,例如使用list 来提供所需的输出:

      >>> list_a = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]
      >>> list(gen(list_a))
      [1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]
      >>> list(Cumulative(list_a))
      [1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]
      

      不过,既然你问到速度,我想分享一下我的计时结果:

      import pandas as pd
      import numpy as np
      import random
      
      import pandas as pd
      from itertools import takewhile
      from itertools import groupby, accumulate, chain
      
      def MSeifert(lst):
          return list(MSeifert_inner(lst))
      
      def MSeifert_inner(lst):
          cumulative = 0
          for item in lst:
              if item:
                  cumulative += item
              else:
                  cumulative = 0
              yield cumulative
      
      def MSeifert2(lst):
          return list(Cumulative(lst))
      
      def original1(list_a):
          list_b = []
          for i, x in enumerate(list_a):
              if x == 0:
                  list_b.append(x)
              else:
                  sum_value = 0
                  for j in list_a[i::-1]:
                      if j != 0:
                          sum_value += j
                      else:
                          break
                  list_b.append(sum_value)
      
      def original2(list_a):
          return [sum(takewhile(lambda x: x != 0, list_a[i::-1])) for i, d in enumerate(list_a)]
      
      def Coldspeed1(data):
          data = data.copy()
          for i in range(1, len(data)):
              if data[i]:  
                  data[i] += data[i - 1] 
          return data
      
      def Coldspeed2(data):
          s = pd.Series(data)    
          return s.groupby(s.eq(0).cumsum()).cumsum().tolist()
      
      def Chris_Rands(list_a):
          return list(chain.from_iterable(accumulate(g) for _, g in groupby(list_a, bool)))
      
      def EvKounis(list_a):
          cum_sum = 0
          list_b = []
          for item in list_a:
              if not item:            # if our item is 0
                  cum_sum = 0         # the cumulative sum is reset (set back to 0)
              else:
                  cum_sum += item     # otherwise it sums further
              list_b.append(cum_sum)  # and no matter what it gets appended to the result
      
      def schumich(list_a):
          list_b = []
          s = 0
          for a in list_a:
              s = a+s if a !=0 else 0
              list_b.append(s)
          return list_b
      
      def jbch(seq):
          return list(jbch_inner(seq))
      
      def jbch_inner(seq):
          s = 0
          for n in seq:
              s = 0 if n == 0 else s + n
              yield s
      
      
      # Timing setup
      timings = {MSeifert: [], 
                 MSeifert2: [],
                 original1: [], 
                 original2: [],
                 Coldspeed1: [],
                 Coldspeed2: [],
                 Chris_Rands: [],
                 EvKounis: [],
                 schumich: [],
                 jbch: []}
      sizes = [2**i for i in range(1, 20, 2)]
      
      # Timing
      for size in sizes:
          print(size)
          func_input = [int(random.random() < 0.75) for _ in range(size)]
          for func in timings:
              if size > 10000 and (func is original1 or func is original2):
                  continue
              res = %timeit -o func(func_input)   # if you use IPython, otherwise use the "timeit" module
              timings[func].append(res)
      
      
      %matplotlib notebook
      
      import matplotlib.pyplot as plt
      import numpy as np
      
      fig = plt.figure(1)
      ax = plt.subplot(111)
      
      baseline = MSeifert2 # choose one function as baseline
      for func in timings:
          ax.plot(sizes[:len(timings[func])], 
                  [time.best / ref.best for time, ref in zip(timings[func], timings[baseline])], 
                  label=func.__name__)  # you could also use "func.__name__" here instead
      ax.set_ylim(0.8, 1e4)
      ax.set_yscale('log')
      ax.set_xscale('log')
      ax.set_xlabel('size')
      ax.set_ylabel('time relative to {}'.format(baseline)) # you could also use "func.__name__" here instead
      ax.grid(which='both')
      ax.legend()
      plt.tight_layout()
      

      如果您对确切的结果感兴趣,我将它们放入 this gist

      这是一个对数图,相对于 Cython 的答案。简而言之:越低越快,两个主要刻度之间的范围代表一个数量级。

      因此,除了您拥有的解决方案之外,所有解决方案都倾向于在一个数量级内(至少在列表很大时)。奇怪的是,与纯 Python 方法相比,pandas 解决方案相当慢。然而,Cython 解决方案比所有其他方法高出 2 倍。

      【讨论】:

      • 问题是我没有将列表转换为系列的时间,并认为这是给定的。转换占那里时间的 3/4。顺便说一句,答案很好。
      • @cᴏʟᴅsᴘᴇᴇᴅ 感谢您的详细说明。我怀疑这可能是转换,但没有调查过。尤其是pd.Series 转换可能会特别慢(可能就像np.array 这已经在列表中进行了多次传递)。
      • 我会使用cumulative = nxt 而不是cumulative = 0,因为这是您在问题的一般情况下所做的(即:计算累积总和,直到找到值x 和然后重新启动)。我怀疑性能的变化会很大。
      • @GiacomoAlzetta 如果你喜欢这样,但是考虑到if item 检查它只会进入else 分支以防它为零。因此,您使用哪一个并不重要。并且更加概括它会对OP主要关注的性能产生负面影响:)
      • 太好了,如果所有的基准测试都这么好,那将是一个更好的地方!还提醒我需要学习 Cython!
      【解决方案7】:

      Python 3.8 开始,并引入assignment expressions (PEP 572):= 运算符),我们可以在列表解析中使用和递增变量:

      # items = [1, 1, 1, 0, 1, 1, 0, 1, 0, 1, 1, 1]
      total = 0
      [total := (total + x if x else x) for x in items]
      # [1, 2, 3, 0, 1, 2, 0, 1, 0, 1, 2, 3]
      

      这个:

      • 将变量 total 初始化为 0,表示运行总和
      • 对于每个项目,这两个:
        • 通过赋值表达式total与当前循环项目(total := total + x)相加,或者如果项目为0,则将其设置回0
        • 同时,将x 映射到total 的新值

      【讨论】:

      • [total := x and total + x for x in items]。提到它有多快也很好(我认为在 MSeifert 的基准测试中,它会排在第二位,仅被 Cython 解决方案击败)。
      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 2020-07-18
      • 2020-12-17
      • 2016-07-23
      • 2019-10-07
      • 1970-01-01
      • 1970-01-01
      • 2017-10-05
      相关资源
      最近更新 更多