Test a market-data health check with eight small Python fixtures

A socket can be connected, a message can have arrived recently, and the price inside that message can still be old. If your application collapses those facts into a single connected = True, a test suite that only checks successful connection will miss the problem.

Here is a small classification function you can test without credentials, a running market or a WebSocket library. It uses four inputs: whether the market is open, when that status was checked, the source timestamp, and the local receipt timestamp.

The example is about observability, not trading decisions. All numeric timestamps below are synthetic Unix seconds. Provider timestamps in milliseconds must be divided by 1,000 before calling the function.

Make the health decision a pure function

Save this as feed_health.py:

from dataclasses import dataclass
import math

@dataclass(frozen=True)
class FeedState:
    market_open: bool | None
    status_checked_at: float | None
    source_time: float | None
    received_at: float | None


def health(state, now, max_price_age=120, max_status_age=600):
    # All times are Unix seconds; normalize provider units before calling.
    times = [now, state.status_checked_at, state.source_time, state.received_at]
    if any(t is not None and (isinstance(t, bool) or not isinstance(t, (int, float)) or not math.isfinite(t)) for t in times):
        return 'invalid_timestamp'
    if not isinstance(state.market_open, bool):
        return 'status_unknown'
    if state.status_checked_at is None or now - state.status_checked_at > max_status_age:
        return 'status_unknown'
    if any(t is not None and t > now + 2 for t in times[1:]):
        return 'clock_check'
    if not state.market_open:
        return 'market_closed'
    if state.source_time is None or state.received_at is None:
        return 'waiting_for_first_tick'
    if now - state.received_at > max_price_age:
        return 'consumer_silent'
    if now - state.source_time > max_price_age:
        return 'source_value_old'
    return 'current'

The function takes now explicitly. That keeps tests independent of the machine clock and lets you reproduce a decision later with the same inputs. Production code can pass time.time(); tests can pass a fixed number.

The order matters. A market that was closed fifteen minutes ago is not necessarily closed now. Check the age of the status decision before using it to suppress an alert. If that status cannot be trusted, return status_unknown rather than quietly treating the market as closed.

Next, detect clock differences. A source timestamp more than two seconds ahead of the evaluation clock returns clock_check. That tolerance is an example policy; choose it for your environment and synchronize host clocks.

Only after those checks should a freshly confirmed closed market suppress a price-age warning. During an open market, a missing timestamp becomes waiting_for_first_tick. It does not become a healthy state just because there has never been enough data to classify it.

Finally, distinguish the two kinds of age. No recent receipt becomes consumer_silent. A recent receipt carrying an old source time becomes source_value_old. Both can require investigation, but they send you toward different parts of the system.

Give each failure a fixture

Save this as test_feed_health.py in the same directory:

import unittest
from feed_health import FeedState, health

class FeedHealthTests(unittest.TestCase):
    def test_recent_frame_with_old_value(self):
        self.assertEqual(health(FeedState(True, 950, 700, 999), 1000), 'source_value_old')
    def test_silent_consumer(self):
        self.assertEqual(health(FeedState(True, 950, 700, 700), 1000), 'consumer_silent')
    def test_closed_market(self):
        self.assertEqual(health(FeedState(False, 950, 700, 700), 1000), 'market_closed')
    def test_failed_gate(self):
        self.assertEqual(health(FeedState(False, 300, 700, 700), 1000), 'status_unknown')
    def test_never_seen(self):
        self.assertEqual(health(FeedState(True, 950, None, None), 1000), 'waiting_for_first_tick')
    def test_current(self):
        self.assertEqual(health(FeedState(True, 950, 990, 999), 1000), 'current')
    def test_future_clock(self):
        self.assertEqual(health(FeedState(True, 950, 1010, 999), 1000), 'clock_check')
    def test_invalid(self):
        self.assertEqual(health(FeedState(True, 950, float('nan'), 999), 1000), 'invalid_timestamp')

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

Run the tests with the Python standard library:

python -m unittest -v test_feed_health.py

The old-source fixture has a receipt time of 999 and an evaluation time of 1,000. Transport traffic is only one second old, but the source value is 300 seconds old. That is exactly the case a connection-only test would miss.

The failed-gate fixture is more subtle. It says market_open=False, but the status was checked at 300. By 1,000 it is 700 seconds old, beyond the 600-second example budget. The output must be status_unknown, not market_closed.

These fixtures should all pass before you connect the function to an actual feed. Add boundary tests for your chosen age budgets, session policies and delayed-data entitlements. The sample does not infer an exchange session from a ticker, classify source price changes, or prove that a price is executable.

Connect the classification to observability

Keep the latest source and receipt timestamps in your consumer state, using bounded market/symbol identifiers from your configured watchlist. Set source time from the feed's documented timestamp field and receipt time when your code actually receives the frame. Preserve the newest valid source timestamp if older frames arrive out of order.

When the connection drops, keep the last known price timestamp available so its age can grow; record transport state separately. Recompute the classification on a timer, even when no frames arrive. A callback that only runs on incoming messages cannot notice that messages have stopped.

For Prometheus, export the timestamps themselves. Calculate age at query time with time() - timestamp_metric, rather than exporting a frozen “age at the last update.” This follows the official instrumentation guidance.

Track the freshness of market-status polls separately. A failure should not refresh the successful-status timestamp. That gives you a useful alert when the very gate that decides whether silence is normal stops working.

A healthy transport and current payload are separate assertions. Small fixtures make that distinction explicit and keep it from disappearing into a reassuring green icon.