每一笔行情里的三个时钟：时间戳来源与时钟偏差

09:25，A 股集合竞价结束前五分钟，你的行情告警服务跑在一台刚迁移过宿主机的虚拟机上。迁移过程里宿主时钟比标准时间慢 2.7 秒，容器启动后 chrony 还在缓慢校准。你的去重逻辑按到达时间开窗：同一标的、三秒内、价格相同即判为重复。宿主机慢下来的那 2.7 秒，把 11:02 到 11:06 之间连续几笔同价成交压进了同一个窗口，真实成交被当成重复丢弃，告警没有触发。当天盘后对账，这几分钟的记录是空的，日志里没有任何错误——去重函数按设计正常工作，只是它拿到的时间基准错了。

这次故障和"行情断了"不是一回事。连接一直在，订阅一直确认，帧一直在到。出问题的是你对"这一帧是什么时候发生的"的判断，而这个判断依赖三个不同来源的时钟，你只控制了其中一个。

每一笔行情都带着别人的时钟到达，又在你进程里被打上第三个时钟。到达时间减事件时间同时包含网络传输和本机时钟偏差，一个等式两个未知数，单次采样无法拆开；只有把一段窗口内的低点当作下限来读，才能把两者分开。

三个问题，三件仪器

上面那次故障的根因，是同一个数字被要求回答三个不同的问题。拆开之后，仪表反而变少。

字段的来源要先分清，否则后面全是猜测：

有一个容易被忽略的事实：报价行里的 timestampMs 是上游给这个价格打的时间，不是平台把 HTTP 响应推出去的瞬间。对成交活跃、且处于交易时段的标的，两者足够接近；对成交稀疏的标的，或者任何已经收盘的标的，这个字段可以是几小时前的，而且这是设计如此，不是故障。所以年龄必须按标的算，且只在该标的的交易时段内算。跳过时段判断，你会得到每个周末都响一次的告警，和每周一早上报告六十小时偏差的时钟看板。

第二个推论：年龄衡量的是新鲜度，不是延迟。如果你的服务等级目标需要一个延迟数字，年龄给不了，因为它包含上游自己的缓存和上游自己的时钟。真正能归因到你自己链路上的，是围绕单次请求用单调钟测出的往返时间；即便如此，那也只是客户端视角的路径耗时，不是单向传输时间。

用下限估出偏差的上界

把等式写下来，估计量自然浮现：

年龄 = 到达墙钟 − 事件时间
     = 传输耗时 + (本机时钟 − 源时钟)
     = 传输耗时 + 偏差

传输耗时不可能为负，所以每个样本都满足「年龄 ≥ 偏差」，一段窗口内的年龄低点就是「偏差 + 传输下限」。这是上界，不是点估计。本机时钟准确、网络传输下限 8 毫秒时，低点读数是 8 毫秒，结论是「偏差不超过 8 毫秒」——正确且无害。本机快 3000 毫秒时，低点永远压不到 3000 毫秒以下，结论立刻有用：主机错了数秒，你算出的每个年龄都同向错了数秒。

平均值是这里错误的选择。平均年龄等于偏差加平均传输，而平均传输里混着重试、TLS 握手、调度停顿和 TCP 长尾；把噪声平均进一个"下限"的估计里，得到的数字跟着负载走，不跟着时钟走。低点是收敛到下限的次序统计量，这也是各类时钟同步算法都从低点或靠近低点的分位数起步的原因。

两种偏差要分开命名，处置方式完全不同。

漂移是连续的。未纳管的虚拟机时钟以几十到几百 ppm 的速度累积误差：100 ppm 等于每小时 360 毫秒，一天 8.6 秒。度量方式是每十分钟取一次低点、连成序列、算斜率，单位毫秒每小时。斜率接近零只说明稳定，不说明准确——恒定偏差的斜率也是零。

跳变是间断的。NTP 步进校正、快照恢复、手工改时间、容器迁移到另一台宿主机、闰秒处理不一致，都会让墙钟瞬间移位。开篇那次故障就是跳变，而且它可以用零成本的本地比较抓出来：

t0w, t0m = time.time(), time.monotonic()
# ... 一次请求 ...
t1w, t1m = time.time(), time.monotonic()
step_ms = (t1w - t0w - (t1m - t0m)) * 1000

墙钟会跳，单调钟不会。两秒的区间里，这个差值超过约 1000 毫秒，说明动的是墙钟而不是网络。一次减法就能把 2.7 秒的迁移事故变成一条日志。两条注意事项让它保持诚实：Linux 上 time.monotonic() 读的是 CLOCK_MONOTONIC，机器挂起期间它停走，所以恢复后算出的差值很大，即使墙钟完全正确，这时要用 CLOCK_BOOTTIME 或开机标识区分"墙钟动了"和"单调钟停了"；另外 chrony 的默认行为是缓慢调整小偏差而不是步进，上限约 500 ppm，于是一次校正可以让偏差越过你设的 200 毫秒阈值，而跳变检测全程沉默。两件事都要做：给上界设阈值，同时盯差值。

响应头 Date 是第三件仪器，比它看起来的要弱。它只有秒级精度，量化误差就有正负 500 毫秒，再叠加传输。做法是拿它做交叉校验，容差给两秒，不要拿它当精确偏差。它有一个不可替代的场合：标的收盘、timestampMs 停在几小时前时，Date 仍在走，这是你在周末仅能拿到的服务端时钟信号。

反向边界：这些场景不该做这件事

这套测量要付出三样成本：一个判断分支、一个落库字段、一份采样预算。六种场景不值得付。

1. 日线与收盘后研究。 openTimeMs 是桶的标签，不是事件时刻。拿它算年龄，数字会在盘中一路上涨、在换桶瞬间归零。决策周期是天级时，决定对错的是交易日历与复权规则，时钟精度改不动结论。
2. 收盘标的与冷门标的。 年龄在设计上就没有上界。03:00 UTC 查一个美股标的，报价几小时前是正常的。做法是先判断交易时段，或直接读 delayed 与 unavailable 标记，把"年龄过大"告警限制在时段内生效。
3. 无法纳管的机器。 内网隔离环境里，不要去估一个你无法验证的偏差。做法是在到达时间旁边写一个 recv_clock_untrusted = true，所有判断改用事件时间。承认未知比给出错误上界便宜，而错误上界会在凌晨两点把你叫起来。
4. 界面上的"几秒前更新"。 用浏览器时钟算相对时间，等于把一台刚从睡眠恢复的笔记本变成故障源。相对时间放在服务端算，或者只显示事件时间戳，别的都不显示。
5. 去重键与排序键。 到达时间永远不能当键。一次跳变就会重排你的数据流：去重窗口要么收缩放进重复，要么放宽吞掉真实成交。键用 (标的, 事件时间, 价格)，到达时间单独存列。相关的失效形态见重连后的重复 tick与 K 线聚合对账两篇。
6. 十毫秒级的服务等级承诺。 客户端视角的年龄撑不起毫秒结论，单个观测点的传输下限就盖过了你要测的东西。那种精度需要接入侧埋点、PTP，或者两者都有。写清楚你的承诺覆盖你能观测的部分。

采样预算要单独说，因为它会悄无声息地让这套探针失效。免费套餐每月 9,000 次 REST 调用、每日上限 300 次（定价页，2026-10-20 访问）。每两秒探一次、跑满一天是 43,200 次调用，是每日上限的 144 倍。所以按窗口跑：五分钟间隔、十分钟一轮是 120 次调用，每周轮换一台机器，一周内仍能抓到一次跳变。需要连续监控时改走 WebSocket：握手只计一次额度，后续推送不按条计费；但流式接入需要 Growth 及以上套餐，具体额度与实时层级以定价页和你的控制台为准。

探针

下面这段用推送流做测量，和 REST 版本的区别在于样本来源是 tick 帧里的 at，每秒可以拿到多个样本，因此更适合观察传输下限的抖动。它同样把墙钟和单调钟在同一行取，把跳变量单独计数。

#!/usr/bin/env python3
"""
tick_clock_probe.py —— 用推送帧的 at 字段估计本机时钟偏差上界。

    TIQEX_API_KEY=... python tick_clock_probe.py

依赖：pip install websockets
权限：WebSocket 需 Growth 及以上套餐；免费套餐只含 REST 报价与 K 线。
计费：一次握手计一次额度，后续 tick 推送不计条数（定价页 2026-10-20）。
"""

import asyncio
import json
import os
import statistics
import time
from datetime import datetime

import websockets

WS_URL = "wss://api.tiqex.io/api/v1/market/stream"
MARKET = "crypto"
SYMBOLS = ["BTCUSDT", "ETHUSDT"]
DURATION_S = 60
STEP_THRESHOLD_MS = 1000.0
IDLE_TIMEOUT_S = 5.0


def parse_at_ms(raw: str) -> float | None:
    """RFC3339 转毫秒。带 Z 的写法先换成本地可解析的偏移量。"""
    if not raw:
        return None
    try:
        return datetime.fromisoformat(raw.replace("Z", "+00:00")).timestamp() * 1000.0
    except ValueError:
        return None


async def probe(api_key: str) -> None:
    ages: dict[str, list[float]] = {}
    steps: list[float] = []
    frames = 0
    frames_without_ts = 0
    prev = None  # (墙钟, 单调钟)

    headers = {"X-API-Key": api_key}
    async with websockets.connect(WS_URL, additional_headers=headers, ping_interval=20) as ws:
        await ws.send(json.dumps({"action": "subscribe", "market": MARKET, "symbols": SYMBOLS}))

        loop = asyncio.get_running_loop()
        deadline = loop.time() + DURATION_S

        while loop.time() < deadline:
            try:
                raw = await asyncio.wait_for(ws.recv(), timeout=IDLE_TIMEOUT_S)
            except asyncio.TimeoutError:
                continue

            # 两个时钟在同一行取，保证描述的是同一瞬间
            recv_wall, recv_mono = time.time(), time.monotonic()
            try:
                frame = json.loads(raw)
            except json.JSONDecodeError:
                continue

            if not frame.get("succ"):
                print(f"错误帧：code={frame.get('code')} msg={frame.get('msg')}")
                break

            data = frame.get("data") or {}
            if data.get("type") != "tick":
                continue

            frames += 1
            if prev is not None:
                wall_delta = (recv_wall - prev[0]) * 1000.0
                mono_delta = (recv_mono - prev[1]) * 1000.0
                steps.append(wall_delta - mono_delta)
            prev = (recv_wall, recv_mono)

            event_ms = parse_at_ms(data.get("at"))
            symbol = data.get("symbol")
            if event_ms is None or not symbol:
                frames_without_ts += 1
                continue
            ages.setdefault(symbol, []).append(recv_wall * 1000.0 - event_ms)

    print(f"\n帧数：{frames}，缺失时间戳：{frames_without_ts}")
    if not ages:
        print("没有拿到可用样本：检查套餐是否含推送、标的代码是否正确")
        return

    print(f"{'标的':<12}{'样本':>6}{'低点':>10}{'中位':>10}{'p95':>10}{'上界':>10}")
    bounds = []
    spans = []
    for symbol, series in sorted(ages.items()):
        ordered = sorted(series)
        p95 = ordered[max(0, min(len(ordered) - 1, int(round(0.95 * len(ordered))) - 1))]
        bounds.append(ordered[0])
        spans.append(statistics.median(ordered) - ordered[0])
        print(f"{symbol:<12}{len(ordered):>6}{ordered[0]:>10.0f}"
              f"{statistics.median(ordered):>10.0f}{p95:>10.0f}{ordered[0]:>10.0f}")

    bound = max(bounds)          # 取差的标的，得到保守上界
    spread = max(spans)
    big_steps = [s for s in steps if abs(s) > STEP_THRESHOLD_MS]

    print(f"\n偏差上界：{bound:,.0f} 毫秒（含传输下限）")
    print(f"排队离散：{spread:,.0f} 毫秒（中位减低点）")
    print(f"疑似跳变：{len(big_steps)} 次" + (f"，幅度 {max(big_steps, key=abs):+,.0f} 毫秒"
                                          if big_steps else ""))

    if bound > 2000:
        print("\n结论：主机时钟错了数秒。先修 NTP，再谈年龄、去重窗口和告警顺序。")
    elif spread > 500:
        print("\n结论：时钟可用，离散来自传输与排队。抬新鲜度阈值，不要动超时。")
    else:
        print("\n结论：年龄小且稳定。保留交易时段判断，把上界记进值班手册。")


def main() -> None:
    api_key = os.environ.get("TIQEX_API_KEY")
    if not api_key:
        raise SystemExit("请先设置 TIQEX_API_KEY")
    asyncio.run(probe(api_key))


if __name__ == "__main__":
    main()

三段代码里有三个刻意的选择，比输出格式重要得多。

两个时钟在同一行取。把墙钟放在函数入口、单调钟放在函数返回，等于比较的是 DNS 解析前与 TLS 握手后的两个时刻，跳变检测会频繁误报，而你会以为是网络问题。

上界按标的算完之后取更差的那个，不取更好的。一个对 BTCUSDT 成立、对冷门标的失效的上界，不是对时钟的上界，是对运气的上界。聚合之后仍然是上界，只是更保守。

结论分三态，而不是两态。上界大，修时钟；上界小而离散大，说明年龄被时钟之外的东西主导，那是排队或缓存问题，该换一本值班手册；两者都小，仪器可用，下一步是自动化而不是继续测量。

决策清单

常见问题

能不能从本机测出精确的单向延迟？ 不能。一个等式两个未知数，单个观测点解不出来；声称给出单向传输时间的供应商，要么在接入侧埋了点，要么在估计。你能做的是给两者之和定上界——本文的做法——以及把往返时间归因到自己的路径，后者通常才是可行动的那一半。

年龄为什么会是负数？ 因为你的时钟比源时钟慢了超过传输耗时的量。这不构成数据损坏的证据。把它记成时钟事件、加一个计数器、把这一帧留下。

有了 delayed 标记，还需要这套测量吗？ 判断新鲜度，大体不需要：标记是供应商的权威陈述，不是从两个时钟推出来的结论，读取成本为零。但它不告诉你本机时钟是否错了，而后者仍然影响去重窗口、告警顺序和日志关联。

只信响应头的 Date 行不行？ 做精确偏差不行，秒级量化摆在前面。回答"本机是否错了十秒以上"够用；而且在收盘标的的时间戳故意不动时，它是此刻仍在走的服务端时钟。

容差设多少？ 从容差服务的决策倒推，不要从时钟倒推。告警需要区分相隔 50 毫秒的两笔成交，200 毫秒的偏差就已经让排序失去意义；看板五秒刷新一次，两秒的上界与结论无关，不值得为它花采样额度。

回测要不要做这件事？ 只有当你在分钟以下粒度上拼接实时与历史数据时才需要。否则主导因素是交易日历与复权规则，见跨市场时区与日历一篇。为了改善日线回测去测时钟偏差，是改不动结论的工作。

下一步

按顺序三步。第一步，在已经在调接口的进程里加那行跳变检测，把结果和请求标识一起落日志；成本是一次减法，一天之内就能知道你的机器会不会动。第二步，在每一台产出可信时间戳的机器上跑一次上面的探针，把上界写进值班手册，而不是写进聊天记录。第三步，把交易时段判断放到年龄告警前面，让周末安静下来，告警才保得住含义。

如果上界很大，那是平台侧的工作，不是应用侧：给主机指一个可达的 NTP 源，确认出网规则没有拦掉它，再检查容器运行时是否屏蔽了宿主时钟。如果上界很小而离散很宽，接着该读的是六种静默失效：时钟正常而离散偏大，说明这是投递问题穿着一件时钟问题的外衣。

口径与范围。 本文的字段名、错误码与额度取自 TIQEX API v1 开发者文档 tiqex.io/docs/auth、tiqex.io/docs/errors、tiqex.io/docs/market-coverage，以及定价页 tiqex.io/pricing，访问日期均为 2026-10-20（文档版本：API v1）。额度、单次报价的标的数上限（美股 20、跨市场 50、加密 100）、WebSocket 计费方式与服务等级约定属于文档默认值，按套餐而定；控制台中的套餐为准，延迟层级、历史深度与市场覆盖同样取决于你的权限，引用前请在定价页核对。代码面向 Python 3.11 及以上版本，仅作工程示范：没有重试预算、没有持久化、没有处理重定向。本文只示范工程实践，不构成投资建议；行情数据仅供参考。