场景与现象
离线对账脚本从接口分页读取订单，按客户汇总金额。数据少时结果正常，接口改成按创建时间返回后，同一客户的金额突然减少；没有异常，订单条数也没有变化。
下面只讨论有限批次、按客户汇总整数分金额的任务。订单已经由上游完成去重，退款使用负数；不处理币种换算。示例使用 Python 3.10 及以上标准库，不需要安装依赖。
用三条数据复现
保存为 reproduce.py，执行 python reproduce.py：
from itertools import groupby
rows = [("A", 100), ("B", 200), ("A", 300)]
totals = {
customer: sum(amount for _, amount in group)
for customer, group in groupby(rows, key=lambda row: row[0])
}
print(totals)
assert totals == {"A": 300, "B": 200}
正确业务结果应该是 A 为 400 分、B 为 200 分。实际 A 只有 300 分：输入中的 A 被切成两组，字典推导式第二次写入 A 时覆盖了第一组的 100 分。
排查时先记录输入条数、分组段数、不同客户数和汇总金额，不要直接打印原始订单或客户身份信息。若一个客户出现多段，优先检查输入排序合同。这个案例中，金额丢失发生在分组和构造字典之间，与浮点精度、数据库提交无关。
groupby 的合同是什么
Python itertools 官方文档说明，键值变化时便开始新的一组；它处理连续相同键，与 SQL 中聚合全部相同键的 GROUP BY 存在区别。
因此 A、B、A 是三段，而不是两个客户组。仅按时间排序、仅在每页内部按客户排序，都不能保证整个批次同一客户连续。
还有一个容易掩盖问题的细节：每个组也是迭代器，并与外层迭代器共享数据源。不能先把组迭代器全部存起来，等外层走完再计算。需要延后处理时，应在当前循环内转成列表；仅求和时直接在当前循环消费即可。
修复：先校验，再按同一个键排序并消费分组
有限批次可以先建立已校验列表，使用同一键排序和分组。下面限制批次最多 10000 条、客户键为 1—64 位 ASCII 标识、单笔金额绝对值不超过十亿分。这些是示例业务合同，接入实际系统时应调整为真实规则。金额不接受布尔值，避免 True 被当作整数 1。
把下面完整代码保存为 customer_totals.py，运行 python customer_totals.py 即可执行测试。
import re
import unittest
from collections.abc import Iterable
from itertools import groupby
from operator import itemgetter
MAX_ROWS = 10_000
MAX_AMOUNT_CENTS = 1_000_000_000
CUSTOMER_PATTERN = re.compile(r"[A-Za-z0-9_-]{1,64}")
def summarize(rows: Iterable[object]) -> dict[str, int]:
"""汇总有限批次，先拒绝非法行，再计算各客户的整数分金额。"""
validated: list[tuple[str, int]] = []
for index, row in enumerate(rows):
if index >= MAX_ROWS:
raise ValueError("批次订单数超过上限")
if not isinstance(row, (tuple, list)) or len(row) != 2:
raise ValueError(f"第 {index + 1} 行必须包含客户键和金额")
customer, amount = row
if (
not isinstance(customer, str)
or CUSTOMER_PATTERN.fullmatch(customer) is None
):
raise ValueError(f"第 {index + 1} 行客户键格式无效")
if type(amount) is not int or abs(amount) > MAX_AMOUNT_CENTS:
raise ValueError(f"第 {index + 1} 行金额必须是范围内的整数分")
validated.append((customer, amount))
ordered = sorted(validated, key=itemgetter(0))
return {
customer: sum(amount for _, amount in group)
for customer, group in groupby(ordered, key=itemgetter(0))
}
class SummaryTests(unittest.TestCase):
"""覆盖乱序、边界、输入合同与调用方数据保护。"""
def test_interleaved_customers(self) -> None:
self.assertEqual(
summarize([("A", 100), ("B", 200), ("A", 300)]),
{"A": 400, "B": 200},
)
def test_empty(self) -> None:
self.assertEqual(summarize([]), {})
def test_refund_and_zero(self) -> None:
self.assertEqual(summarize([("A", 100), ("A", -100), ("B", 0)]),
{"A": 0, "B": 0})
def test_generator(self) -> None:
self.assertEqual(summarize((row for row in [("B", 2), ("A", 1)])),
{"A": 1, "B": 2})
def test_input_unchanged(self) -> None:
rows = [("B", 2), ("A", 1)]
summarize(rows)
self.assertEqual(rows, [("B", 2), ("A", 1)])
def test_invalid_rows(self) -> None:
for row in [("A",), "A", ("", 1), ("客户", 1), ("A", True),
("A", 1.5), ("A", "100"), ("A", MAX_AMOUNT_CENTS + 1)]:
with self.subTest(row=row), self.assertRaises(ValueError):
summarize([row])
def test_batch_limit(self) -> None:
self.assertEqual(summarize([("A", 1)] * MAX_ROWS), {"A": MAX_ROWS})
with self.assertRaises(ValueError):
summarize([("A", 1)] * (MAX_ROWS + 1))
def test_amount_boundaries(self) -> None:
self.assertEqual(summarize([("A", MAX_AMOUNT_CENTS),
("A", -MAX_AMOUNT_CENTS)]), {"A": 0})
if __name__ == "__main__":
unittest.main()
sorted() 建立新列表，不改变调用方顺序；排序键与分组键均为第一个字段。组的求和发生在当前迭代内，不保存活的分组迭代器。函数在校验成功后才返回结果，调用方应在它返回后写入报表或事务，避免非法末行到达时前面客户已经写出。
这不等于给外部生成器增加回滚能力：如果生成器读取时会确认消息或修改外部状态，应在读取边界另外设计确认和重试。
验证修复是否针对根因
测试不能只有 A、A、B 这种已经排好序的数据。关键回归输入是 A、B、A。
将 ordered = sorted(validated, key=itemgetter(0)) 临时改为 ordered = validated 再运行测试，乱序客户用例必须失败；随后恢复排序重新运行。若移除修复后仍全部通过，测试并没有覆盖故障。
上线前还可校验输入金额总和与输出金额总和是否相等。但金额守恒仅是辅助检查：分错客户、正负金额抵消仍可能通过，必须同时核对客户维度结果。原示例中的乱序用例既检查总额，也检查具体客户。
数据量扩大后的选择
这里校验列表和排序列表都占用内存，排序时间为 O(n log n)，空间为 O(n)。不要把批次上限删除后直接用于无限消息流。
若只需汇总值，可在校验后用字典逐笔累加，时间通常为 O(n)、聚合状态为 O(客户数)，无须排序；仍应限制行数、客户数并明确异常时的写出行为。若还要逐客户处理全部明细，可由数据库按相同客户键排序后流式分组，但必须确认数据库排序规则与应用分组相等规则一致。
分页数据尤其要警惕：每页排序不代表跨页有序。跨页汇总需要统一聚合状态或全局排序；分批重试还需要订单去重规则。本文没有通过分组修复解决重复投递。
实测与适用边界
发布前在本地执行三行复现及上述 8 项 unittest，并执行移除排序的缺陷恢复检查。验证范围是内存数据与同步生成器；不包含真实接口分页、数据库游标、并发写入、崩溃恢复及大数据量性能。环境未提供 Ruff，未额外安装依赖。
遇到分组漏算，先确认键是否连续，再检查组迭代器消费位置和重复键覆盖。修复应落实到输入合同、排序或累加策略，以及能让原始缺陷重新失败的乱序测试。
Discussion
评论