适用场景与异常表现
离线任务跑了 26 小时，报表却显示只耗时 2 小时；开始、结束时间颠倒了 1 秒，监控却出现接近 24 小时的长任务；耗时 250 毫秒的接口被统计为 0 秒。这三种现象可能来自同一行代码：
duration_seconds = (finished_at - started_at).seconds
这里的 seconds 是时间差的一个分量。它并不代表整段时间差的总秒数。短任务测试只覆盖正数、整数秒且小于一天时，很容易漏掉这个问题。
本文使用 Python 3.11 及以上版本，只依赖标准库。示例时间是固定构造的 UTC 时间，不读取机器当前时间，不需要网络、数据库或第三方时区数据。
先保留时间差的三个分量
把下面代码保存为 reproduce_duration.py，执行 python reproduce_duration.py：
from datetime import timedelta
samples = {
"超过一天": timedelta(hours=26),
"负一秒": timedelta(seconds=-1),
"四分之一秒": timedelta(milliseconds=250),
}
for name, delta in samples.items():
print(name, delta.days, delta.seconds, delta.microseconds,
delta.total_seconds())
输出中后三列要连同 days 一起看：
| 样本 | days | seconds | microseconds | total_seconds() |
|---|---|---|---|---|
| 超过一天 | 1 | 7200 | 0 | 93600.0 |
| 负一秒 | -1 | 86399 | 0 | -1.0 |
| 四分之一秒 | 0 | 0 | 250000 | 0.25 |
timedelta 会将时间差规范化为天、秒、微秒三个分量。seconds 始终介于 0 和 86399 之间，负号可能体现在 days 上。总秒数对应 days * 86400 + seconds + microseconds / 1000000。Python 官方文档也直接提醒过误把 seconds 当总秒数的问题。timedelta 属性与方法
排查时先从同一条任务记录中取出原始起止时间、UTC 偏移、三个分量以及当前写入的耗时字段。不要先改监控阈值，否则只是让统计错误更难发现。日志应使用 started_at、finished_at、duration_ms 等稳定字段，固定消息使用中文，避免记录业务请求正文。
还要区分“跨午夜”和“超过 24 小时”：23:59 到次日 00:01 只有 120 秒，旧代码通常仍然正确；真正触发天数丢失的是时间差含有完整一天。
修复时同时明确输入合同
已有时间差需要浮点秒时，可以使用 delta.total_seconds()。要写入整数微秒字段，可以使用 delta // timedelta(microseconds=1)，避免先转浮点再乘回去。极大时间差的 total_seconds() 可能失去微秒精度，因此精确存储不应依赖该浮点往返。官方精度说明
仅替换 .seconds 还不够。负耗时可能暴露事件顺序颠倒、跨机器时钟偏差或错误字段映射；不带时区的时间也不能靠猜测补成 UTC。下面的入口约定：只接受带有效时区的 datetime，先转换到 UTC，再计算非负的精确微秒数。
将以下完整代码保存为 test_duration.py：
import unittest
from datetime import UTC, datetime, timedelta, timezone
def calculate_duration_us(started_at: datetime, finished_at: datetime) -> int:
"""计算事件间隔，拒绝缺少时区及结束早于开始的记录。"""
for field_name, value in (
("started_at", started_at),
("finished_at", finished_at),
):
if not isinstance(value, datetime):
raise TypeError(f"{field_name} 必须是日期时间对象")
if value.tzinfo is None or value.utcoffset() is None:
raise ValueError(f"{field_name} 必须包含有效时区")
delta = finished_at.astimezone(UTC) - started_at.astimezone(UTC)
if delta < timedelta(0):
raise ValueError("结束时间不能早于开始时间")
return delta // timedelta(microseconds=1)
class DurationTests(unittest.TestCase):
"""覆盖常见耗时、规范化边界和异常输入。"""
def setUp(self) -> None:
self.started_at = datetime(2026, 10, 21, tzinfo=UTC)
def test_normal_duration(self) -> None:
finished_at = self.started_at + timedelta(seconds=12)
self.assertEqual(calculate_duration_us(self.started_at, finished_at),
12_000_000)
def test_exact_day(self) -> None:
finished_at = self.started_at + timedelta(days=1)
self.assertEqual(calculate_duration_us(self.started_at, finished_at),
86_400_000_000)
def test_more_than_one_day(self) -> None:
finished_at = self.started_at + timedelta(hours=26)
self.assertEqual(calculate_duration_us(self.started_at, finished_at),
93_600_000_000)
def test_microsecond_precision(self) -> None:
finished_at = self.started_at + timedelta(microseconds=250_001)
self.assertEqual(calculate_duration_us(self.started_at, finished_at),
250_001)
def test_zero_duration(self) -> None:
self.assertEqual(calculate_duration_us(self.started_at, self.started_at), 0)
def test_negative_duration(self) -> None:
finished_at = self.started_at - timedelta(microseconds=1)
with self.assertRaisesRegex(ValueError, "结束时间"):
calculate_duration_us(self.started_at, finished_at)
def test_naive_input(self) -> None:
naive = self.started_at.replace(tzinfo=None)
for started_at, finished_at in (
(naive, self.started_at),
(self.started_at, naive),
):
with self.subTest(started_at=started_at, finished_at=finished_at):
with self.assertRaisesRegex(ValueError, "时区"):
calculate_duration_us(started_at, finished_at)
def test_wrong_input_type(self) -> None:
with self.assertRaises(TypeError):
calculate_duration_us("2026-10-21", self.started_at)
def test_different_offsets(self) -> None:
local_start = datetime(2026, 10, 21, 8,
tzinfo=timezone(timedelta(hours=8)))
utc_end = datetime(2026, 10, 21, 0, 1, tzinfo=UTC)
self.assertEqual(calculate_duration_us(local_start, utc_end), 60_000_000)
def test_long_interval_is_exact(self) -> None:
finished_at = self.started_at + timedelta(days=110_000, microseconds=1)
self.assertEqual(calculate_duration_us(self.started_at, finished_at),
9_504_000_000_000_001)
if __name__ == "__main__":
unittest.main(verbosity=2)
执行 python test_duration.py，应运行 10 项测试。跨天测试必须明确断言 26 小时对应 93600 秒；只断言结果大于零无法识别旧实现。负值测试必须验证报错，不能通过 abs() 或归零把异常记录伪装成有效样本。
代码里的字符串类型错误用例有意违反类型标注，用来证明运行时入口仍会校验。真实 JSON 接口还应在调用之前解析并限制字符串长度与格式，不能把这个函数当作完整的请求校验器。
UTC 转换用于统一时间点的比较语义，不能恢复丢失的偏移，也不能自动识别调用者构造出的不存在的本地时间。本文没有测试 IANA 地区时区或夏令时转换；此类入口需要单独定义歧义时间的选择与拒绝策略。
本进程耗时改用单调时钟
如果是在同一进程内测量一次调用耗时，可以直接用 time.monotonic_ns() 做差。单调时钟不受系统时钟更新影响，整数纳秒也避免了浮点转换，但它的起点没有业务日期含义。单调时钟说明
import time
from collections.abc import Callable
from typing import TypeVar
Result = TypeVar("Result")
def measure_call(
action: Callable[[], Result],
clock: Callable[[], int] = time.monotonic_ns,
) -> tuple[Result, int]:
"""返回调用结果与耗时；调用异常由业务边界统一处理。"""
started_ns = clock()
result = action()
duration_ns = clock() - started_ns
if duration_ns < 0:
raise ValueError("计时源返回了倒退的时间")
return result, duration_ns
ticks = iter([1_000_000_000, 1_250_000_000])
result, duration_ns = measure_call(lambda: "完成", lambda: next(ticks))
assert result == "完成"
assert duration_ns == 250_000_000
注入计时函数后，测试不用真的等待 250 毫秒。这里的包装器只在成功时返回耗时；失败调用的耗时应由业务监控边界通过 try/finally 记录，并保留原异常。不要把单调时钟读数作为持久化事件时间，也不要拿不同机器或重启前后的读数相减。
上线与历史数据修正
先保留旧字段，增加带单位的新字段，例如 duration_us_v2，用相同的原始起止时间双算，观察差异是否集中在超过一天、负值和亚秒任务。接口改字段类型或计量单位时同步更新消费者，避免把微秒误当毫秒。
历史记录若只留下了旧的 .seconds 结果，无法确定它曾经包含几天，也无法还原负值与微秒。只有起止时间完整、时区语义明确的记录才能重算；其余记录需要标记为历史统计不可恢复，不能直接加上猜测的 86400 秒。
新旧逻辑切换期间，监控耗时分布、非法顺序记录数和缺少时区记录数。出现输入合同不一致时，回退读路径或暂停回填，保留原始时间与新版结果，待修正源数据后继续。回退到旧字段会重新暴露原统计缺陷，只能作为暂时止损。
验证边界与结论
本文示例在 Python 3.14.6 上验证，包含 10 项事件耗时测试、三个分量复现及注入时钟示例；不涵盖真实数据库字段精度、跨主机时钟同步、IANA 时区和生产监控采集。运行环境没有 Ruff，本次未执行其格式化和静态检查，也未安装额外依赖。
排查这类问题的关键是先确认字段含义：seconds 是规范化后的分量，总耗时要使用完整时间差。精确整数存储使用整数微秒，单进程测量使用单调时钟；负耗时和时区缺失则作为数据质量问题单独处理。
Discussion
评论