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.
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.
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.
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. Here is a complete, runnable example. It uses no broker or vendor SDK, which keeps the test focused on the contract your application owns.
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.
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 that match the user promise
A 100 ms batch window is an implementation choice, not a product promise. If traders need a symbol to appear within 500 ms, test that service-level objective with a simulated watermark and a distribution of event times. Do not turn the window itself into a magic constant scattered through fixtures.
Measure the promise.
The catch is that deterministic batching is not suitable when your requirement is genuinely latency-first streaming, where every tick must be rendered immediately. In that case, use a smaller or disabled batch policy and test per-event acknowledgement and backpressure instead. Stick with a batched design when your interface can tolerate bounded delay and benefits from fewer renders.
Presence accuracy also has a trust boundary. The client should not infer removal from a missing batch; absence may mean the batch was delayed. Send an explicit remove event or a snapshot watermark, and expire unknown state only after a documented timeout. That rule makes reconnect behavior auditable and prevents a momentary network gap from looking like a market signal.
Start with a table of event fixtures: symbol, sequence, event time, action, and expected presence. Generate edge cases from that table, including equal timestamps and sequence gaps. Run the reducer tests on every change, then run a smaller transport suite that verifies framing, ordering, and reconnect handoff.
Measure batch age and queue depth in production, but keep those measurements out of correctness assertions. Alert on watermark lag and on symbols stuck in unknown; sample payloads with redaction because watchlists can reveal trading intent.
Finally, review the contract with whoever owns the UI. “Present” might mean received, rendered, or eligible for trading, and those meanings have different tests. I'm not sure any single timeout works for every market or mobile connection, so make the timeout configurable and validate it against observed event-time distributions.
A reliable test suite does not make realtime predictable. It makes the unpredictable parts visible, while keeping the presence decision deterministic.
Sources
- W3C, WebRTC Recommendation: https://www.w3.org/TR/webrtc/
- Python documentation,
dataclasses: https://docs.python.org/3/library/dataclasses.html - Python documentation,
time.monotonic: https://docs.python.org/3/library/time.html#time.monotonic " }
Top comments (0)