Python CSV 金额汇总：用 Decimal 控制舍入并校验输入

CSV 里的单价看起来只有几位小数，用 float 汇总后却可能出现尾数；换成 Decimal 后，两份报表仍可能相差一分钱，因为舍入发生的时机不同。

本文面向初学者到中级读者，用 Python 标准库写一个逐条读取的金额汇总工具：从原始字符串构造 Decimal，逐条舍入后求和，遇到坏数据立即失败。 示例约定只有一种币种、保留两位小数，不处理税费、退款或汇率。

1. 方法：先固定输入和舍入规则

Python Decimal 官方文档说明，十进制字符串可精确表示；从 float 构造则会保留它已有的二进制近似。这里直接使用 CSV 的字符串字段，避免先调用 float()。

本例把以下规则作为数据接口：

表头必须按顺序为 unit_price,quantity，每条记录正好两列。空行和只有表头的文件均报错。
单价是非负十进制文本，整数部分 1–9 位，小数部分可省略，若有则为 1–4 位。拒绝空格、千分位、科学计数法、负数、NaN 和 Infinity。
数量为 1–1000000 的整数，不接受前导零；最多处理 1000000 条数据记录。
每条金额先计算 单价 × 数量，再用 ROUND_HALF_UP 舍入到 0.01，最后汇总。该规则在正数恰好居中时向上舍入；这是本例约定，实际规则应与自己的报表一致。

localcontext() 把运算精度局限在函数内。40 位有效数字足以覆盖上述数值及记录数上限；精度不是小数位数，两位小数由 quantize() 指定。相关接口见 quantize 文档。

如果只想先拦截增删列、错位和重复列名，可以先读 CSV 结构校验教程。本例在结构检查后继续校验数值。

2. 准备环境和测试数据

使用已有的 Python 3.9 或更新版本，无需安装第三方包。以下命令和脚本已在 Linux、Python 3.11.4 上测试。

在 Bash 或 zsh 中新建演示目录，后续命令保持在这个目录。重复实验时换一个目录名：

python3 --version
mkdir decimal-csv-demo
cd decimal-csv-demo
cat > orders.csv <<'CSV'
unit_price,quantity
0.10,3
0.005,1
0.005,1
CSV

将下一节代码保存成 total_csv.py，也可下载完整脚本。输入支持 UTF-8 及带 BOM 的 UTF-8；其他编码先参考 CSV 编码检查与转换，不要使用忽略解码错误的方式掩盖数据问题。

3. 实现：校验、逐条舍入和汇总

"""Sum per-row rounded amounts from a validated CSV. Python 3.9+."""
import argparse
import csv
from decimal import Decimal, DecimalException, ROUND_HALF_UP, localcontext
import json
import re
import sys


PRICE = re.compile(r"[0-9]{1,9}(?:\.[0-9]{1,4})?")
QUANTITY = re.compile(r"[1-9][0-9]{0,6}")
CENT = Decimal("0.01")
MAX_ROWS = 1_000_000


def summarize(path):
    with localcontext() as context:
        context.prec = 40
        total = Decimal("0.00")
        count = 0
        with open(path, encoding="utf-8-sig", newline="") as source:
            reader = csv.reader(source, strict=True)
            if next(reader, None) != ["unit_price", "quantity"]:
                raise ValueError("header must be unit_price,quantity")
            for record, row in enumerate(reader, start=1):
                if record > MAX_ROWS:
                    raise ValueError("more than 1000000 data records")
                location = f"record {record}, ending at line {reader.line_num}"
                if len(row) != 2:
                    raise ValueError(f"{location}: expected 2 fields")
                price, quantity = row
                if not PRICE.fullmatch(price):
                    raise ValueError(f"{location}: invalid unit_price")
                if not QUANTITY.fullmatch(quantity) or int(quantity) > 1_000_000:
                    raise ValueError(f"{location}: quantity must be 1..1000000")
                amount = (Decimal(price) * int(quantity)).quantize(
                    CENT, rounding=ROUND_HALF_UP
                )
                total += amount
                count += 1
        if not count:
            raise ValueError("at least one data record is required")
        return {"rows": count, "total": format(total, ".2f")}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("input", help="UTF-8 CSV with unit_price,quantity header")
    args = parser.parse_args()
    try:
        result = summarize(args.input)
    except (OSError, UnicodeError, csv.Error, ValueError, DecimalException) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

这里使用 csv.reader(..., strict=True)，不通过字符串切割解析逗号。按 CSV 官方文档要求，以 newline="" 打开文件；默认 reader 返回字符串字段，避免启用会将部分字段转成浮点数的 QUOTE_NONNUMERIC。

record 是数据记录序号，从 1 开始；reader.line_num 是该记录结束时累计读取的物理行号，跨行字段下二者不同。脚本只在整个文件检查成功后输出 JSON，金额以字符串返回，避免下游 JSON 解析器自动转成浮点数。它逐条读取，不保留全部记录；大文件处理方式可参考流式处理大型 CSV。

4. 验证：核对金额和舍入时机

查看帮助并运行：

python3 total_csv.py --help
python3 total_csv.py orders.csv

预期输出：

{"rows": 3, "total": "0.32"}

第一条金额为 0.30，后两条各从 0.005 舍入到 0.01。若先把所有未舍入金额相加再舍入，这个例子会得到 0.31，与本文规则不同。

用最小例子单独验证两种时机。运行 python3 后粘贴：

from decimal import Decimal, ROUND_HALF_UP

cent = Decimal("0.01")
items = [Decimal("0.005"), Decimal("0.005")]
print(sum(x.quantize(cent, rounding=ROUND_HALF_UP) for x in items))
print(sum(items).quantize(cent, rounding=ROUND_HALF_UP))

两行输出依次为 0.02、0.01。不要靠修改显示格式来解释这类差额；应先核对每条舍入还是总额舍入。

5. 验证：坏数据必须返回非零

在同一目录执行：

printf 'unit_price,quantity\nNaN,1\n' > bad-price.csv
printf 'unit_price,quantity\n1.20,0\n' > bad-quantity.csv
printf 'unit_price,quantity\n' > empty.csv
for file in bad-price.csv bad-quantity.csv empty.csv; do
  if python3 total_csv.py "$file"; then
    printf 'UNEXPECTED SUCCESS: %s\n' "$file"
  else
    code=$?
    printf 'exit=%s file=%s\n' "$code" "$file"
  fi
done

三个文件分别提示 invalid unit_price、quantity must be 1..1000000、at least one data record is required，均返回 1，且没有成功 JSON。正常任务返回 0；缺少输入路径等命令行用法错误由 argparse 返回 2。

6. 排障：先检查输入约定

现象	直接处理
`invalid unit_price`	检查原始 CSV 单价。按约定导出为 `12.3400` 这类纯十进制文本；不要先经浮点数转换，也不要随意删除逗号或货币符号。
`header must be unit_price,quantity`	将导出列名和顺序改成约定值。脚本已支持 UTF-8 BOM，不能据此自动接受分号分隔或不同列名。
`expected 2 fields`	根据记录号及结束行号检查缺列、多列和空行；使用 CSV 导出器正确引用字段。
总额与其他报表不同	用 `0.005` 两条记录复现，核对舍入时机、舍入模式及币种的小数位规则，再同步修改实现和预期结果。
`UnicodeDecodeError` 或 CSV 解析错误	先修正文件编码或引号结构，保留失败退出，不跳过坏记录。

本例只输出汇总，不写源文件。若要加入负数退款、多币种或更多小数位，必须先扩充数据规则、精度预算和边界测试。

7. 总结

从 CSV 原始字符串构造 Decimal，再明确每条舍入和汇总的顺序，可以让金额结果可复现。把格式、数量和记录数上限放进校验，成功后才输出固定两位小数的字符串。将非零退出码接入自动化流程，就能阻止坏数据继续进入下游报表。