How to Test Five Live Trading Watchlist Events Without Flaky Timing

本文为原文前 6,000 字符的节选翻译，完整内容请查看原文。

Short answer: make event time a dependency, flush batches with an explicit watermark, and assert presence as a state transition rather than sleeping for an arbitrary number of milliseconds. That gives a stock-trading watchlist repeatable tests while still exercising the same batching rules used in production. A watchlist is a presence problem disguised as a delivery problem. The user cares whether AAPL is currently visible, whether a symbol disappeared after a disconnect, and whether the last quote is newer than the one already on screen. A batch that arrives 40 ms later is fine; a stale symbol that survives a reconnect is not.

简短的回答是：将事件时间作为依赖项，使用明确的水位线（watermark）刷新批次，并将存在性断言视为状态转换，而不是通过休眠任意毫秒数来等待。这使得股票交易观察列表的测试具有可重复性，同时还能执行生产环境中使用的相同批处理规则。观察列表本质上是一个伪装成交付问题的存在性问题。用户关心的是 AAPL 当前是否可见、符号在断开连接后是否消失，以及最新的报价是否比屏幕上已有的报价更新。批次晚到 40 毫秒没关系，但断开重连后残留的陈旧符号则是不可接受的。

I build RAG and agent features in Python, so I tend to start with a notebook containing a few events and then promote it into a small eval harness. The trap is carrying notebook timing assumptions into CI. I once saw a test pass locally, then fail with an assertion at event index 3 when the runner was busy. The test had a real sleep, a wall clock, and no definition of “present.” Three independent sources of noise.

我在 Python 中构建 RAG 和代理功能，因此我倾向于从包含少量事件的笔记本开始，然后将其提升为一个小型评估工具。陷阱在于将笔记本中的时间假设带入持续集成（CI）环境中。我曾经见过一个测试在本地通过，但在运行器繁忙时在事件索引 3 处因断言失败。该测试使用了真实的休眠、挂钟时间，且没有对“存在”的定义。这是三个独立的噪声源。

Why a realtime watchlist becomes flaky: Batching intentionally delays delivery. A server may wait for a short window, a maximum item count, or a transport watermark before sending a frame. WebRTC data channels add their own ordering and reliability choices; the W3C recommendation makes those properties explicit, but it does not decide what “current in the watchlist” means for your product. The usual anti-pattern is sleep(0.1) followed by “expect five messages.” It conflates transport progress with application progress. A slow CI host, a reconnect, or two symbols sharing a timestamp can change the result without changing the system’s correctness.

为什么实时观察列表会变得不稳定：批处理有意延迟了交付。服务器在发送帧之前可能会等待一个短窗口、最大项目数或传输水位线。WebRTC 数据通道增加了它们自己的排序和可靠性选择；W3C 建议明确了这些属性，但它并未决定“观察列表中的当前状态”对你的产品意味着什么。常见的反模式是使用 sleep(0.1) 后紧跟“期望收到五条消息”。这混淆了传输进度与应用程序进度。缓慢的 CI 主机、重连或两个符号共享同一个时间戳都可能改变结果，而无需改变系统的正确性。

Presence needs a small model. For each symbol, keep the latest sequence and a status such as present, removed, or unknown. A batch is then an input to a reducer. The reducer can be tested with an in-memory clock and a deterministic event stream; the network adapter can be tested separately for framing and reconnect behavior. How can Python test batched realtime stock watchlist events deterministically? Use three explicit boundaries: an event timestamp, a batch cutoff, and a presence assertion.

存在性需要一个小模型。对于每个符号，保留最新的序列号以及状态（如存在、已移除或未知）。批次随后作为归约器（reducer）的输入。归约器可以使用内存时钟和确定性事件流进行测试；网络适配器可以单独测试其分帧和重连行为。Python 如何确定性地测试批处理的实时股票观察列表事件？使用三个明确的边界：事件时间戳、批次截止时间和存在性断言。

Here is a complete, runnable example. It uses no broker or vendor SDK, which keeps the test focused on the contract your application owns.

这是一个完整且可运行的示例。它不使用任何代理或供应商 SDK，这使得测试能够专注于你应用程序所拥有的契约。

from dataclasses import dataclass
from typing import Dict, Iterable, List, Tuple

@dataclass(frozen=True)
class QuoteEvent:
    symbol: str
    sequence: int
    event_ms: int
    action: str # "upsert" or "remove"
    price: float | None = None

class Watchlist:
    def __init__(self) -> None:
        self._rows: Dict[str, QuoteEvent] = {}
        self._last_sequence: Dict[str, int] = {}

    def apply(self, event: QuoteEvent) -> None:
        previous = self._last_sequence.get(event.symbol, -1)
        if event.sequence <= previous:
            return # duplicate or late data cannot roll state backward
        self._last_sequence[event.symbol] = event.sequence
        if event.action == "remove":
            self._rows.pop(event.symbol, None)
        else:
            self._rows[event.symbol] = event

    def present(self) -> set[str]:
        return set(self._rows)

def make_batches(events: Iterable[QuoteEvent], window_ms: int) -> List[Tuple[int, List[QuoteEvent]]]:
    ordered = sorted(events, key=lambda item: (item.event_ms, item.sequence))
    batches: List[Tuple[int, List[QuoteEvent]]] = []
    current: List[QuoteEvent] = []
    cutoff = None
    for event in ordered:
        if cutoff is None:
            cutoff = event.event_ms + window_ms
        if event.event_ms >= cutoff and current:
            batches.append((cutoff, current))
            current = []
            cutoff = event.event_ms + window_ms
        current.append(event)
    if current and cutoff is not None:
        batches.append((cutoff, current))
    return batches

def test_presence_after_five_event_batch() -> None:
    events = [
        QuoteEvent("AAPL", 1, 1_000, "upsert", 190.10),
        QuoteEvent("MSFT", 1, 1_010, "upsert", 420.20),
        QuoteEvent("NVDA", 1, 1_020, "upsert", 900.30),
        QuoteEvent("TSLA", 1, 1_030, "upsert", 175.40),
        QuoteEvent("AMZN", 1, 1_040, "upsert", 180.50),
    ]
    watchlist = Watchlist()
    batches = make_batches(events, window_ms=100)
    assert len(batches) == 1
    cutoff, batch = batches[0]
    assert cutoff == 1_100
    for item in batch:
        watchlist.apply(item)
    assert watchlist.present() == {"AAPL", "MSFT", "NVDA", "TSLA", "AMZN"}

if __name__ == "__main__":
    test_presence_after_five_event_batch()
    print("presence contract passed")

The important assertion is the set of symbols, not the number of callback invocations. The cutoff is data, so the test never waits. If your production adapter uses a monotonic clock, inject the same clock interface and advance it in the test; do not patch time.sleep around an opaque coroutine.

重要的断言是符号集合，而不是回调调用的次数。截止时间是数据，因此测试永远不需要等待。如果你的生产适配器使用单调时钟，请注入相同的时钟接口并在测试中推进它；不要在不透明的协程周围修补 time.sleep。

Treat reconnects, duplicates, and late quotes as data: A watchlist test that only covers a clean connection is a demo. Add a second stream containing a duplicate sequence, a late quote, and a removal after reconnect. Feed both batches through the same reducer. The expected state should be explicit: a duplicate does nothing, a lower sequence cannot overwrite a newer price, and a removal wins only when its sequence is newer.

将重连、重复和延迟报价视为数据：仅涵盖干净连接的观察列表测试只是一个演示。添加第二个流，其中包含重复的序列号、延迟的报价以及重连后的移除操作。将两个批次通过同一个归约器处理。预期的状态应该是明确的：重复项不执行任何操作，较低的序列号不能覆盖较新的价格，且移除操作仅在其序列号更新时才生效。

def test_reconnect_does_not_resurrect_a_removed_symbol() -> None:
    watchlist = Watchlist()
    watchlist.apply(QuoteEvent("TSLA", 4, 2_000, "upsert", 175.40))
    watchlist.apply(QuoteEvent("TSLA", 5, 2_100, "remove"))
    # Replay from the reconnect includes an old snapshot and a new quote.
    watchlist.apply(QuoteEvent("TSLA", 4, 2_050, "upsert", 176.00))
    watchlist.apply(QuoteEvent("AAPL", 2, 2_060, "upsert", 191.00))
    assert "TSLA" not in watchlist.present()
    assert "AAPL" in watchlist.present()

That test is deliberately boring. Boring state transitions are easy to inspect in a failure report. For a real transport, record the batch id, first and last sequence, event-time range, and the watermark used to flush. Those fields let an eval harness explain a failure without reproducing network timing. Keep payload assertions separate from presence assertions; a quote can be present while its price is invalid, and those are different defects. Pick boundaries.

这个测试是故意设计得枯燥乏味的。枯燥的状态转换在失败报告中很容易检查。对于真实的传输，记录批次 ID、第一个和最后一个序列号、事件时间范围以及用于刷新的水位线。这些字段使评估工具能够在不重现网络时序的情况下解释失败原因。将有效载荷断言与存在性断言分开；报价可能存在但价格无效，这些是不同的缺陷。选择边界。