场景与现象

日报按渠道统计访问次数与支付次数。原始计数正确,Python 接口返回的转化率却只有 0 和 1。把结果格式化成两位小数后,只是变成了 0.00 和 1.00。

本文把转化率定义为支付次数除以访问次数,返回 0 到 1 的比例;访问次数为零时返回 NULL,表示没有可计算的比例。示例使用 SQLite 与 Python 标准库,不需要安装数据库服务。它适合计数类报表,不适合需要精确十进制结算的金额运算。

先在数据库里复现

下面的 SQL 可直接在 SQLite 命令行中执行,或交给 Python 的 sqlite3 连接执行:

SELECT
    3 / 10 AS broken_rate,
    typeof(3 / 10) AS broken_type,
    CAST(3 / 10 AS REAL) AS late_cast,
    CAST(3 AS REAL) / 10 AS correct_rate;

结果依次是 0、integer、0.0、0.3。这一步把问题限定在 SQL 表达式内部:小数在数据库返回数据前就已经丢失,序列化和页面格式化无法补回来。

SQLite 在两个操作数都是整数且运算不溢出时执行整数算术,整数除法向零截断;参与计算的操作数含 REAL 时使用浮点算术。对这个计数场景,必须在除法发生前转换操作数。细节可查 SQLite 表达式文档。

CAST(paid / visits AS REAL) 转换的是已经截断的商;ROUND(paid / visits, 2) 同样不能恢复被截断的小数。100 * paid / visits 仍然可能进行整数除法,而且把口径改成了百分数,不能作为通用修复。

从实际表达式定位

不要只检查字段声明。先在测试库针对出问题的记录执行下面的参数化查询,保留应用传入的参数类型:

SELECT
    paid_count,
    visit_count,
    typeof(paid_count) AS paid_type,
    typeof(visit_count) AS visit_type,
    paid_count / visit_count AS raw_rate,
    typeof(paid_count / visit_count) AS rate_type
FROM channel_daily
WHERE channel = ?;

? 由驱动绑定渠道名,不能把外部输入拼入 SQL。重点看参与除法的值及其运行时类型;若真实查询先做 SUM 或条件计数,应把检查点放到聚合后的表达式上。数据库算出的值已经是 0,就先修 SQL;只有数据库输出正确而 API 错误时,再检查应用层转换。

表声明也不能代替表达式检查。SQLite 的 NUMERIC 亲和性可能把可无损表示的数值存成整数;给列起名 DECIMAL 不代表每次除法都会得到十进制小数。相关规则见 SQLite 数据类型文档。

修复查询并写清零分母语义

SELECT
    channel,
    CAST(paid_count AS REAL) / NULLIF(visit_count, 0) AS conversion_rate
FROM channel_daily
ORDER BY channel;

CAST 在除法前把分子转成 REAL。NULLIF(visit_count, 0) 把零分母明确转换为 NULL,使“无访问记录”的业务语义出现在查询里。SQLite 本身除零也会产生 NULL,这里使用 NULLIF 是为了让口径可读、可审查;它没有解决其他数据库的全部兼容性问题。

不要立即用 COALESCE 把 NULL 改成 0。零次支付、十次访问的比例是 0;零次访问没有比例,这两种状态应在接口和页面上区分。页面可把 NULL 展示成“暂无数据”,把 0.3 展示成 30.00%,展示时才乘以 100。

如果报表按多天聚合,应先求同一统计范围内的支付总次数和访问总次数,再计算比例。直接 AVG 每天的转化率会改变权重,属于另一个口径问题,不能靠 CAST 修复。

完整回归测试

保存为 test_conversion_rate.py,执行 python -m unittest -v test_conversion_rate。测试使用内存数据库,连接在清理阶段关闭。表约束限定计数为非负整数且支付次数不超过访问次数;如果业务按用户去重或允许一访多单,需要调整模型和指标定义。

import sqlite3
import unittest


RATE_SQL = """
SELECT CAST(paid_count AS REAL) / NULLIF(visit_count, 0)
FROM channel_daily WHERE channel = ?
"""


class ConversionRateTests(unittest.TestCase):
    """验证比例精度、无数据语义和计数输入约束。"""

    def setUp(self) -> None:
        self.connection = sqlite3.connect(":memory:")
        self.addCleanup(self.connection.close)
        self.connection.execute("""
            CREATE TABLE channel_daily (
                channel TEXT PRIMARY KEY NOT NULL,
                paid_count INTEGER NOT NULL
                    CHECK(typeof(paid_count) = 'integer' AND paid_count >= 0),
                visit_count INTEGER NOT NULL
                    CHECK(typeof(visit_count) = 'integer' AND visit_count >= 0),
                CHECK(paid_count <= visit_count)
            )
        """)

    def rate(self, paid: int, visits: int) -> float | None:
        """通过生产同形查询读取一个渠道的比例。"""
        self.connection.execute(
            "INSERT INTO channel_daily VALUES (?, ?, ?)",
            ("search", paid, visits),
        )
        row = self.connection.execute(RATE_SQL, ("search",)).fetchone()
        if row is None:
            raise RuntimeError("未查询到已插入的渠道")
        return row[0]

    def test_fraction(self) -> None:
        self.assertAlmostEqual(self.rate(3, 10), 0.3)

    def test_small_fraction(self) -> None:
        self.assertAlmostEqual(self.rate(1, 1000), 0.001)

    def test_zero_payments(self) -> None:
        self.assertEqual(self.rate(0, 10), 0.0)

    def test_full_conversion(self) -> None:
        self.assertEqual(self.rate(10, 10), 1.0)

    def test_no_visits(self) -> None:
        self.assertIsNone(self.rate(0, 0))

    def test_invalid_counts(self) -> None:
        for paid, visits in [(-1, 10), (1, 0), (11, 10), (1.5, 10), (1, None)]:
            with self.subTest(paid=paid, visits=visits):
                with self.assertRaises(sqlite3.IntegrityError):
                    self.connection.execute(
                        "INSERT INTO channel_daily VALUES (?, ?, ?)",
                        ("invalid", paid, visits),
                    )

    def test_cast_after_division_is_too_late(self) -> None:
        row = self.connection.execute(
            "SELECT CAST(3 / 10 AS REAL), CAST(3 AS REAL) / 10"
        ).fetchone()
        self.assertEqual(row, (0.0, 0.3))


if __name__ == "__main__":
    unittest.main()

测试必须覆盖不能整除的数据。只测 0 / 10 和 10 / 10,错误查询也会通过。浮点比例使用近似断言,NULL 使用独立断言,不把缺失值和零混在一起。

确认测试能捕获原始缺陷的方法:临时把 RATE_SQL 中的 CAST(paid_count AS REAL) 换成 paid_count,再次运行测试。3 / 10 和 1 / 1000 两项应失败;恢复修复查询后全部通过。这个检查能发现“测试存在,但没有真正约束小数保留”的情况。

上线检查与边界

先在测试库对同一批报表输入比较旧、新结果:计数字段应完全相同,差异只出现在比例字段的小数部分。再检查接口是否把比例强制转换为 int,前端是否重复乘以 100,以及告警阈值使用的是 0.3 还是 30。保留原始计数,便于复算和审计。

本次修复只需要调整 SELECT,不需要迁移表结构。回滚应用查询即可回到旧行为,但旧查询会再次截断小数;若出现兼容问题,应优先修正消费者的类型假设,并保留无访问记录的 NULL 语义。

REAL 是二进制浮点数,不保证任意十进制值精确表示。超大计数可能失去整数精度,金额结算或精确阈值裁决应另行设计定点数或十进制运算方案。显式 CAST 也不是输入校验器,脏字符串应在写入边界拒绝,不能依赖转换后碰巧得到的数值。

本文验证范围是本地 Python sqlite3 内存数据库中的表达式、表约束和上述回归测试;未验证生产 ORM 生成 SQL、真实报表规模、并发写入及其他数据库引擎。修改聚合报表前,还要核对时间范围、去重方式和过滤条件是否一致。

总结

遇到比例只有 0 和 1,先查看数据库表达式及 typeof 结果。把转换放到除法操作数上,单独定义零分母语义,再用不能整除的数据做回归。数据库已经截断的小数,格式化层无法恢复。