适用场景与现象
账户流水按发生时间展示，每行需要显示“处理完这一笔后的累计变动”。测试数据中时间都不相同，结果一直正常；接入批量入账后，同一秒出现多笔流水，第一笔的余额竟然已经包含第二笔金额。最终合计正确，中间余额却不符合逐笔对账规则。
本文使用 SQLite 和 Python 标准库构造可重复现场。金额用整数分表示，累计值是查询范围内的净变动；若要展示真实余额，还要纳入期初余额。例子不代表真实生产事故。
最小复现：先看排序键是否重复
在 SQLite 客户端执行以下 SQL，或在 Python sqlite3 连接上执行：
CREATE TABLE ledger (
id INTEGER PRIMARY KEY,
account_id INTEGER NOT NULL,
occurred_at INTEGER NOT NULL,
amount_cents INTEGER NOT NULL
);
INSERT INTO ledger VALUES
(1, 10, 1000, 100),
(2, 10, 1000, 200),
(3, 10, 1001, -50);
SELECT id, amount_cents,
SUM(amount_cents) OVER (
PARTITION BY account_id ORDER BY occurred_at
) AS running_cents
FROM ledger
ORDER BY account_id, occurred_at, id;
返回的累计值是 300、300、250，逐笔预期却是 100、300、250。这不是金额重复写入：流水仍只有三行，最终金额也相同。
排查顺序建议是：先按主键检查是否重复入账，再检查 JOIN 是否扩大行数，最后检查窗口定义。验证时间是否重复可以执行：
SELECT account_id, occurred_at, COUNT(*) AS row_count
FROM ledger
GROUP BY account_id, occurred_at
HAVING COUNT(*) > 1;
本例返回账户 10、时间 1000、行数 2。只有唯一时间的数据不能覆盖这个故障。
原因：输出排序与窗口边界是两回事
省略框架时，此处使用从分区起点到当前排序值及其同值行的默认范围。两笔流水的 occurred_at 相同，因此第一笔也会读到第二笔。SQLite 官方把这些排序值相同的行称为 peers。
外层 ORDER BY ... id 只改变展示顺序，不能改变 OVER (...) 内的计算。于是页面看起来逐笔有序，累计金额却按时间组跳变。
这也可能是正确的业务需求：如果每条记录都要展示“截至这一秒所有流水的合计”，两行同为 300 完全合理。先明确口径，再决定是否修复。
修复：同时固定业务顺序和逐行框架
逐笔余额必须有稳定的全序。本例约定同一发生时间按唯一流水 ID 排序；如果业务要求以记账序号为准，应使用真正的记账序号，而不是默认认为自增 ID 等于业务先后。
SELECT id, amount_cents,
SUM(amount_cents) OVER (
PARTITION BY account_id
ORDER BY occurred_at, id
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS running_cents
FROM ledger
WHERE account_id = ?
ORDER BY occurred_at, id;
PARTITION BY 隔离账户；复合排序消除同时间歧义；ROWS 明确按行累计。参数 ? 交给驱动绑定，不能拼接账户输入。只改 ROWS 却不增加稳定的同时间排序键，仍没有定义同组流水谁先谁后。
可直接运行的回归验证
保存为 test_running_balance.py，运行 python -m unittest -v test_running_balance.py。它只操作内存数据库，不依赖线上账号或外部服务。
import sqlite3
import unittest
def read_balances(
connection: sqlite3.Connection, account_id: int
) -> list[tuple[int, int]]:
"""按明确的流水顺序计算账户的累计净变动。"""
return connection.execute(
"""
SELECT id, SUM(amount_cents) OVER (
PARTITION BY account_id
ORDER BY occurred_at, id
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS running_cents
FROM ledger WHERE account_id = ?
ORDER BY occurred_at, id
""",
(account_id,),
).fetchall()
class RunningBalanceTests(unittest.TestCase):
"""覆盖同时间流水、账户边界及金额约束。"""
def setUp(self) -> None:
self.connection = sqlite3.connect(":memory:")
self.addCleanup(self.connection.close)
self.connection.execute(
"""
CREATE TABLE ledger (
id INTEGER PRIMARY KEY,
account_id INTEGER NOT NULL,
occurred_at INTEGER NOT NULL,
amount_cents INTEGER NOT NULL
)
"""
)
# 故意打乱插入顺序，避免测试暗中依赖输入顺序。
self.connection.executemany(
"INSERT INTO ledger VALUES (?, ?, ?, ?)",
[(3, 10, 1001, -50), (2, 10, 1000, 200),
(4, 20, 1000, 900), (1, 10, 1000, 100)],
)
def test_same_time_and_refund(self) -> None:
self.assertEqual(
read_balances(self.connection, 10),
[(1, 100), (2, 300), (3, 250)],
)
def test_account_boundary(self) -> None:
self.assertEqual(read_balances(self.connection, 20), [(4, 900)])
def test_empty_account(self) -> None:
self.assertEqual(read_balances(self.connection, 30), [])
def test_zero_amount(self) -> None:
self.connection.execute("INSERT INTO ledger VALUES (5, 10, 1002, 0)")
self.assertEqual(read_balances(self.connection, 10)[-1], (5, 250))
def test_null_amount_rejected(self) -> None:
with self.assertRaises(sqlite3.IntegrityError):
self.connection.execute(
"INSERT INTO ledger VALUES (?, ?, ?, ?)", (5, 10, 1002, None)
)
self.assertEqual(len(read_balances(self.connection, 10)), 3)
if __name__ == "__main__":
unittest.main()
验证时还应把函数里的窗口改回 PARTITION BY account_id ORDER BY occurred_at，保留外层排序。第一项测试应失败，实际值回到 [(1, 300), (2, 300), (3, 250)]。这说明测试捕捉的是中间余额错误，而非仅检查最终合计。
上线前容易遗漏的边界
- 日期筛选会影响累计范围。
WHERE occurred_at >= ?在窗口计算前过滤历史流水；月度报表如果需要真实余额，必须补期初，或在子查询中计算完整累计后在外层截取展示范围。分页也需要先明确计算范围。 - 金额字段不应允许 NULL，本例通过约束拒绝未知金额。整数分也有范围限制，应给单笔金额和最大累计金额设业务上限，避免整数求和溢出。
- 回填旧流水或修改发生时间会改变后续累计值。这是动态查询的结果；审计账本若要求不可变余额，需要额外的记账规则和快照设计。
- 可评估
(account_id, occurred_at, id)索引，并用EXPLAIN QUERY PLAN观察实际计划。索引能否减少排序、是否值得增加写入开销，应在真实数据规模上测量，不以小样本耗时作为性能结论。
本次在 SQLite 3.50.4 上执行上述 5 项测试，全部通过；恢复原窗口定义后，同时间流水用例失败，其余 4 项通过，无执行错误。最小 SQL 现场也返回预期的错误累计值。Ruff 未安装，未执行格式和静态检查。本地验证覆盖 SQLite 内存查询的结果和约束，未验证生产 ORM、并发写入、磁盘负载、分页接口或其他数据库。迁移到其他数据库时应在目标版本重跑用例。
总结与参考
逐笔累计需要明确三件事：按什么账户分区、同时间流水如何排序、每行窗口读到哪里。对账检查既要核对最终合计，也要逐行核对业务主键及累计值。
参考 SQLite 窗口函数官方文档，重点查看框架定义、同值行和输出排序说明。
Discussion
评论