Python itertools.accumulate 如何按分组重新开始累计

我在处理按批次到达的流水数据时，最容易误解的是：itertools.accumulate 能不能遇到分组标记就自动把累计值清零。答案是不能。它只会按照输入顺序，把当前累计值和下一个元素交给累计函数；要让每组重新开始，必须先识别边界，再为每组建立独立的累计过程。

Python 官方文档：https://docs.python.org/3/library/itertools.html#itertools.accumulate

最稳妥的做法是把“识别分组”和“组内累计”拆成两个阶段：先切分输入流，再对每个分组调用 accumulate。如果数据已经按键连续排列，则用 groupby 逐组套用同一个累计函数。

先把问题边界说清楚

假设输入是一条带有分组标记的记录流，标记出现时代表下一条数据开启新批次。例如 None 表示边界，普通整数表示本组增量：

from itertools import accumulate

# None 表示开启新组，数字表示当前组要累加的增量
stream = [1, 3, None, 2, 4, 1, None, 5]

# accumulate 只会得到一条连续累计序列，不会识别 None 是重置标记
print(list(accumulate(value for value in stream if value is not None)))

这个例子会把所有数字拼成一条连续序列，得到的不是“每组从 1 开始”的结果。initial=0 也只能设置整条迭代器的初始输出，不能在中途反复触发重置，所以不要试图用一个全局 initial 解决分组边界。

完整工作流：先切分，再累计

第一步是把输入流转换为若干个连续分组。下面的生成器遇到边界标记就交出当前组，最后再补交一次未结束的组；这样既保留了输入顺序，也不会为了计算结果复制整条数据。

from collections.abc import Iterable, Iterator
from itertools import accumulate

def split_on_marker(stream: Iterable[int | None]) -> Iterator[list[int]]:
    """按 None 切分输入；空组不会被交出。"""
    group: list[int] = []
    for value in stream:
        if value is None:
            # 连续边界不产生空组，避免输出虚假的累计结果
            if group:
                yield group
                group = []
            continue
        group.append(value)

    # 输入不以边界结束时，仍要交出最后一组
    if group:
        yield group

stream = [1, 3, None, 2, 4, 1, None, 5]
result = [list(accumulate(group)) for group in split_on_marker(stream)]
print(result)  # [[1, 4], [2, 6, 7], [5]]

这里的输出结构是“每组一个列表”。它适合需要明确边界、后续还要按组处理的任务；如果只想连续地产出带组号的结果，可以把内层列表改成生成器，并在外层逐项 yield。

大输入不要一次性保存所有分组

如果输入来自文件、消息队列或网络迭代器，通常不希望先构造完整的二维列表。可以让函数直接输出“组号、组内位置、累计值”，这样调用方消费一条就处理一条。

from collections.abc import Iterable, Iterator
from itertools import accumulate

def accumulated_by_marker(stream: Iterable[int | None]) -> Iterator[tuple[int, int]]:
    """惰性输出 (组号, 当前累计值)，None 会开始下一组。"""
    group_id = 0
    total = 0
    has_value = False

    for value in stream:
        if value is None:
            # 只有已经读到数据时才推进组号，避免连续标记制造空组
            if has_value:
                group_id += 1
                total = 0
                has_value = False
            continue

        total += value
        has_value = True
        yield group_id, total

stream = [1, 3, None, 2, 4, 1, None, 5]
print(list(accumulated_by_marker(stream)))
# [(0, 1), (0, 4), (1, 2), (1, 6), (1, 7), (2, 5)]

这个版本没有直接调用 accumulate，但它表达了同一个“每组独立状态”的机制。若累计逻辑只是加法，显式状态机更容易把边界、组号和输出格式放在一起；若要复用任意二元函数，可以把 total += value 换成 total = function(total, value)，并为第一项单独初始化。

数据已按连续键排列时使用 groupby

另一种常见输入是记录带有分组键，而且相同键已经连续出现。例如订单流按日期排序，同一天的金额记录挨在一起。这时可以用 itertools.groupby 得到每个连续分组，再把分组迭代器交给 accumulate。

from itertools import accumulate, groupby

records = [
    ("2026-10-01", 1),
    ("2026-10-01", 3),
    ("2026-10-02", 2),
    ("2026-10-02", 4),
]

result: list[tuple[str, int]] = []
for day, rows in groupby(records, key=lambda item: item[0]):
    # 每个 rows 只属于一个连续键，累计器自然从第一条记录重新开始
    totals = accumulate(amount for _, amount in rows)
    result.extend((day, total) for total in totals)

print(result)  # [('2026-10-01', 1), ('2026-10-01', 4), ('2026-10-02', 2), ('2026-10-02', 6)]

groupby 只把相邻且键相同的记录归为一组，不是 SQL 意义上“把全表同键记录搜集到一起”。如果同一个日期在输入中被拆成两段，它就会被视为两个分组；因此使用前要确认上游已经排序或已经按连续键组织好。

累计函数与边界条件怎么选

默认累计函数是加法，也可以传入接收两个参数的函数，例如运行最大值或自定义状态合并函数。无论采用哪种函数，都要先决定“新组的第一条记录”如何初始化：是直接作为初始累计值，还是从固定基线开始。

from itertools import accumulate
from operator import mul

# 乘法累计的初值来自每组第一项，不要额外插入 1，避免改变输出长度
group = [2, 3, 4]
print(list(accumulate(group, mul)))  # [2, 6, 24]

# 如果确实需要固定初值，initial 会额外产生一项输出
print(list(accumulate(group, mul, initial=10)))  # [10, 20, 60, 240]

实践中还要处理三个边界：开头就是分组标记时是否产生空组；连续标记是否只代表一个边界；输入以标记结束时是否丢弃最后的空组。本文的两个实现都选择“不输出空组”，如果业务需要保留空批次，应把这个规则写进输出协议，而不是依赖调用方猜测。

一张表选实现方案

小结

itertools.accumulate 解决的是“沿输入顺序不断合并”，不是“识别分组并重置”。遇到分组标记时，先用切分生成器明确边界；遇到已经按键连续排列的记录时，用 groupby 把每段交给独立的 accumulate。最后根据数据规模选择列表结果或惰性输出，并提前约定空组、末尾边界和初始值的语义。