From 51e2cc4b8f1f207befeab7c9ceb9ccd905bdb80e Mon Sep 17 00:00:00 2001 From: Precious Akpan <49040483+precious-akpan@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:15:06 +0100 Subject: [PATCH 1/2] feat(trade): expose order book source health and freshness (OB-049) The order book feed had only a connection status, so a healthy but quiet market and a dead transport were indistinguishable, and nothing stopped stale depth from being treated as current. Track separately: transport state, the time of the last frame of any kind (heartbeats and acks count), the last valid book revision, and the provider timestamp of each update. Derive fresh / stale / reconnecting / unavailable with an explicit reason: silence past the heartbeat timeout, a run of malformed frames, or a provider delivering updates late are stale; a long gap with no book change while frames keep arriving is a quiet market and stays fresh. Provider lag is measured against a rolling baseline of delivery delays, so a constant clock offset in either direction is not lag. Duplicate or out-of-order revisions never make a book look fresher. quoteFromDepth() is the only sanctioned way to derive a suggested price from depth and returns null unless the feed is fresh with an applied snapshot. The shared subscription now records this from the frames it receives, and the order book panel shows the existing source health badge. Verification: 42 pure-logic tests, 11 stream tests through a fake WebSocket with an explicit MSW snapshot handler, 9 hook tests and 3 panel tests, all passing. These are mocked-transport tests; no live Binance verification. The repository gate (lint, typecheck, check:tokens, tests, build) does not pass on upstream main independent of this change: a syntax error in useOrderEventPolling.ts breaks typecheck and build, and pre-existing lint, token and test failures remain. The failing test set is identical with and without this change. --- .../orderbook/OrderBookPanel.test.tsx | 56 +++ .../components/orderbook/OrderBookPanel.tsx | 8 +- .../trade/hooks/useSourceFreshness.test.ts | 156 +++++++ .../trade/hooks/useSourceFreshness.ts | 59 +++ .../lib/market-data-stream.freshness.test.ts | 240 ++++++++++ .../features/trade/lib/market-data-stream.ts | 80 +++- .../trade/lib/source-freshness.test.ts | 441 ++++++++++++++++++ .../features/trade/lib/source-freshness.ts | 381 +++++++++++++++ 8 files changed, 1418 insertions(+), 3 deletions(-) create mode 100644 apps/web/src/features/trade/hooks/useSourceFreshness.test.ts create mode 100644 apps/web/src/features/trade/hooks/useSourceFreshness.ts create mode 100644 apps/web/src/features/trade/lib/market-data-stream.freshness.test.ts create mode 100644 apps/web/src/features/trade/lib/source-freshness.test.ts create mode 100644 apps/web/src/features/trade/lib/source-freshness.ts diff --git a/apps/web/src/features/trade/components/orderbook/OrderBookPanel.test.tsx b/apps/web/src/features/trade/components/orderbook/OrderBookPanel.test.tsx index 91f0bd2..b395cb1 100644 --- a/apps/web/src/features/trade/components/orderbook/OrderBookPanel.test.tsx +++ b/apps/web/src/features/trade/components/orderbook/OrderBookPanel.test.tsx @@ -22,6 +22,12 @@ vi.mock("../../hooks/useOrderBook", () => ({ }, })) +let mockSource: unknown = null + +vi.mock("../../hooks/useSourceFreshness", () => ({ + useSourceFreshness: () => mockSource, +})) + vi.mock("../../../settings/store/preferences-store", () => ({ usePreferencesStore: () => ({ slippageTolerance: 0.5 }), })) @@ -209,4 +215,54 @@ describe("OrderBookPanel", () => { expect(container).toBeInTheDocument() expect(screen.getByText("Reference Data")).toBeInTheDocument() }) + + describe("source health (OB-049)", () => { + beforeEach(() => { + mockSource = null + }) + + afterEach(() => { + mockSource = null + }) + + it("shows no health badge until a state is available", () => { + render() + expect(screen.getByText("Reference Data")).toBeInTheDocument() + expect(screen.queryByText("Live")).not.toBeInTheDocument() + expect(screen.queryByText("Stale")).not.toBeInTheDocument() + }) + + it("shows Live for a fresh feed", () => { + mockSource = { + health: { + status: "connected", + lastUpdateTime: 1, + staleDuration: null, + reconnectAttempt: 0, + isExecutable: true, + message: "Live market data", + }, + } + render() + expect(screen.getByText("Live")).toBeInTheDocument() + }) + + it("labels a silent feed Stale and explains why", () => { + mockSource = { + health: { + status: "stale", + lastUpdateTime: 1, + staleDuration: 30_000, + reconnectAttempt: 0, + isExecutable: false, + message: "No data received — connection may be dead", + }, + } + render() + const badge = screen.getByText("Stale") + expect(badge).toBeInTheDocument() + expect(badge.closest("[title]")?.getAttribute("title")).toMatch(/connection may be dead/) + expect(screen.getByText("(30s ago)")).toBeInTheDocument() + }) + }) }) diff --git a/apps/web/src/features/trade/components/orderbook/OrderBookPanel.tsx b/apps/web/src/features/trade/components/orderbook/OrderBookPanel.tsx index 0f2494b..05f2273 100644 --- a/apps/web/src/features/trade/components/orderbook/OrderBookPanel.tsx +++ b/apps/web/src/features/trade/components/orderbook/OrderBookPanel.tsx @@ -1,5 +1,7 @@ import { useOrderBook } from "../../hooks/useOrderBook" +import { useSourceFreshness } from "../../hooks/useSourceFreshness" import { useChartPreferencesStore } from "../../store/chart-preferences-store" +import { SourceHealthBadge } from "./SourceHealthBadge" import { RecentTradesTape } from "./RecentTradesTape" import { DepthChart } from "./DepthChart" @@ -11,6 +13,7 @@ export function OrderBookPanel({ symbol }: Props) { const orderbookView = useChartPreferencesStore((s) => s.orderbookView) const setOrderbookView = useChartPreferencesStore((s) => s.setOrderbookView) const { bids, asks, status, isLoading } = useOrderBook(symbol) + const source = useSourceFreshness(symbol) return (
@@ -57,7 +60,10 @@ export function OrderBookPanel({ symbol }: Props) { Depth Chart
- Reference Data + + Reference Data + {source ? : null} + {/* ── Tab Content ──────────────────────────────────────────────────── */} diff --git a/apps/web/src/features/trade/hooks/useSourceFreshness.test.ts b/apps/web/src/features/trade/hooks/useSourceFreshness.test.ts new file mode 100644 index 0000000..1cf4675 --- /dev/null +++ b/apps/web/src/features/trade/hooks/useSourceFreshness.test.ts @@ -0,0 +1,156 @@ +import { act, renderHook } from "@testing-library/react" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" +import { FRESHNESS_TICK_MS, useSourceFreshness } from "./useSourceFreshness" +import type { FreshnessResult } from "../lib/source-freshness" +import type { SourceHealth } from "./useSourceHealth" + +type View = { health: SourceHealth; freshness: FreshnessResult } + +const freshness = (over: Partial = {}): FreshnessResult => ({ + state: "fresh", + reason: "live", + canSeedQuote: true, + lastMessageAgeMs: 0, + lastValidUpdateAgeMs: 0, + providerLagMs: 0, + clockOffsetMs: 40, + ...over, +}) + +const health = (over: Partial = {}): SourceHealth => ({ + status: "connected", + lastUpdateTime: 1, + staleDuration: null, + reconnectAttempt: 0, + isExecutable: true, + message: "Live market data", + ...over, +}) + +let current: View +const unsubscribe = vi.fn() +let bookListener: (() => void) | null = null +const getOrCreate = vi.fn() + +vi.mock("../lib/market-data-stream", () => ({ + marketSubscriptionManager: { getOrCreate: (...args: Array) => getOrCreate(...args) }, +})) + +beforeEach(() => { + vi.useFakeTimers() + current = { health: health(), freshness: freshness() } + bookListener = null + unsubscribe.mockReset() + getOrCreate.mockReset() + getOrCreate.mockReturnValue({ + getSourceHealth: () => current, + subscribeBook: (cb: () => void) => { + bookListener = cb + return unsubscribe + }, + }) +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe("useSourceFreshness", () => { + it("returns null without a symbol and does not subscribe", () => { + const { result } = renderHook(() => useSourceFreshness(undefined)) + expect(result.current).toBeNull() + expect(getOrCreate).not.toHaveBeenCalled() + }) + + it("reports the current freshness on mount", () => { + const { result } = renderHook(() => useSourceFreshness("BTC")) + expect(result.current?.freshness.state).toBe("fresh") + expect(result.current?.health.status).toBe("connected") + }) + + it("notices silence, which raises no event, on its timer", () => { + const { result } = renderHook(() => useSourceFreshness("BTC")) + expect(result.current?.freshness.state).toBe("fresh") + + current = { + health: health({ status: "stale", staleDuration: 21_000, isExecutable: false, message: "No data received — connection may be dead" }), + freshness: freshness({ state: "stale", reason: "transport-silent", canSeedQuote: false }), + } + act(() => { + vi.advanceTimersByTime(FRESHNESS_TICK_MS) + }) + + expect(result.current?.freshness).toMatchObject({ state: "stale", reason: "transport-silent" }) + }) + + it("re-evaluates immediately when the book changes, not only on the timer", () => { + const { result } = renderHook(() => useSourceFreshness("BTC")) + current = { + health: health({ status: "reconnecting", isExecutable: false, message: "Reconnecting to market data…" }), + freshness: freshness({ state: "reconnecting", reason: "transport-closed", canSeedQuote: false }), + } + act(() => bookListener?.()) + expect(result.current?.freshness.state).toBe("reconnecting") + }) + + it("does not re-render while nothing visible changed", () => { + let renders = 0 + renderHook(() => { + renders++ + return useSourceFreshness("BTC") + }) + const afterMount = renders + + // The ages move every tick, but the state, reason and message are the same. + for (let i = 1; i <= 10; i++) { + current = { + health: health(), + freshness: freshness({ lastMessageAgeMs: i * 1_000, lastValidUpdateAgeMs: i * 1_000 }), + } + act(() => { + vi.advanceTimersByTime(FRESHNESS_TICK_MS) + }) + } + expect(renders).toBe(afterMount) + }) + + it("re-renders each whole second once the feed is stale, so the shown age stays right", () => { + current = { + health: health({ status: "stale", staleDuration: 21_000, isExecutable: false, message: "stale" }), + freshness: freshness({ state: "stale", reason: "transport-silent", canSeedQuote: false }), + } + const { result } = renderHook(() => useSourceFreshness("BTC")) + expect(result.current?.health.staleDuration).toBe(21_000) + + current = { ...current, health: { ...current.health, staleDuration: 22_000 } } + act(() => { + vi.advanceTimersByTime(FRESHNESS_TICK_MS) + }) + expect(result.current?.health.staleDuration).toBe(22_000) + }) + + it("stops its timer and unsubscribes on unmount", () => { + const { unmount } = renderHook(() => useSourceFreshness("BTC")) + unmount() + expect(unsubscribe).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + }) + + it("switches subscription when the symbol changes", () => { + const { rerender } = renderHook(({ symbol }) => useSourceFreshness(symbol), { + initialProps: { symbol: "BTC" }, + }) + rerender({ symbol: "ETH" }) + expect(unsubscribe).toHaveBeenCalledTimes(1) + expect(getOrCreate).toHaveBeenLastCalledWith("ETH") + }) + + it("clears the view when the symbol goes away", () => { + const { result, rerender } = renderHook(({ symbol }) => useSourceFreshness(symbol), { + initialProps: { symbol: "BTC" }, + }) + expect(result.current).not.toBeNull() + rerender({ symbol: undefined }) + expect(result.current).toBeNull() + }) +}) diff --git a/apps/web/src/features/trade/hooks/useSourceFreshness.ts b/apps/web/src/features/trade/hooks/useSourceFreshness.ts new file mode 100644 index 0000000..d444ebe --- /dev/null +++ b/apps/web/src/features/trade/hooks/useSourceFreshness.ts @@ -0,0 +1,59 @@ +import { useEffect, useState } from "react" +import { marketSubscriptionManager } from "../lib/market-data-stream" +import type { FreshnessResult } from "../lib/source-freshness" +import type { SourceHealth } from "./useSourceHealth" + +export type SourceFreshnessView = { + health: SourceHealth + freshness: FreshnessResult +} + +/** How often freshness is re-evaluated. Silence has no event, so it has to be polled. */ +export const FRESHNESS_TICK_MS = 1_000 + +/** + * OB-049: Live source health and freshness for a symbol's order book feed. + * + * Staleness is the ABSENCE of messages, which raises no event, so the state is + * re-evaluated on a timer as well as whenever the book changes. Re-renders are + * limited to visible changes: a new state, reason or message, or a change in + * the whole seconds shown once the feed is not fresh. + */ +export function useSourceFreshness(symbol: string | undefined): SourceFreshnessView | null { + const [view, setView] = useState(null) + + useEffect(() => { + if (!symbol) { + setView(null) + return + } + + const shared = marketSubscriptionManager.getOrCreate(symbol) + let lastKey = "" + + const evaluate = () => { + const next = shared.getSourceHealth(Date.now()) + const { freshness, health } = next + const shownSeconds = + freshness.state === "fresh" || health.staleDuration === null + ? "" + : Math.floor(health.staleDuration / 1000) + const key = `${freshness.state}|${freshness.reason}|${freshness.canSeedQuote}|${health.message}|${shownSeconds}` + if (key === lastKey) return + lastKey = key + setView(next) + } + + // Subscribing keeps the shared stream alive and re-evaluates on every book change. + const unsubscribe = shared.subscribeBook(evaluate) + const timer = setInterval(evaluate, FRESHNESS_TICK_MS) + evaluate() + + return () => { + clearInterval(timer) + unsubscribe() + } + }, [symbol]) + + return view +} diff --git a/apps/web/src/features/trade/lib/market-data-stream.freshness.test.ts b/apps/web/src/features/trade/lib/market-data-stream.freshness.test.ts new file mode 100644 index 0000000..fcccb43 --- /dev/null +++ b/apps/web/src/features/trade/lib/market-data-stream.freshness.test.ts @@ -0,0 +1,240 @@ +// OB-049: the shared market subscription must report health from what it +// actually receives. These tests drive the real subscription through a fake +// WebSocket, with the REST depth snapshot served by an explicit MSW handler, +// and control the clock by faking `Date` only (timers stay real, so MSW and +// the stream's own watchdogs are undisturbed). +// +// Mocked-transport tests: nothing here talks to Binance. They pin how the +// stream classifies quiet markets, dead transports, malformed frames and a +// lagging provider; they do not verify the live venue. + +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest" +import { HttpResponse, http } from "msw" +import { setupServer } from "msw/node" +import { marketSubscriptionManager } from "./market-data-stream" +import { quoteFromDepth } from "./source-freshness" + +const server = setupServer() +beforeAll(() => server.listen({ onUnhandledRequest: "error" })) +afterAll(() => server.close()) + +const T0 = new Date("2026-01-01T00:00:00.000Z").getTime() +const SYMBOL = "FRESHTEST" + +class FakeWebSocket { + static OPEN = 1 + static instances: Array = [] + readyState = 1 + onopen: (() => void) | null = null + onmessage: ((evt: { data: unknown }) => void) | null = null + onerror: (() => void) | null = null + onclose: (() => void) | null = null + sent: Array = [] + + constructor(public url: string) { + FakeWebSocket.instances.push(this) + } + send(data: string) { + this.sent.push(data) + } + close() { + this.readyState = 3 + } + // Test helpers + open() { + this.onopen?.() + } + frame(payload: unknown) { + this.onmessage?.({ data: typeof payload === "string" ? payload : JSON.stringify(payload) }) + } + drop() { + this.onclose?.() + } +} + +function serveSnapshot(lastUpdateId = 100) { + server.use( + http.get("https://api.binance.com/api/v3/depth", () => + HttpResponse.json({ + lastUpdateId, + bids: [["99.00", "1"], ["98.00", "2"]], + asks: [["101.00", "1"], ["102.00", "2"]], + }), + ), + ) +} + +const ack = (id = 1) => ({ result: null, id }) +const depth = (u: number, providerTime?: number) => ({ + e: "depthUpdate", + ...(providerTime === undefined ? {} : { E: providerTime }), + u, + b: [["99.00", "1.5"]], + a: [], +}) + +let unsubscribe: (() => void) | null = null + +/** Subscribe to the book, connect the fake socket and wait for the snapshot to apply. */ +async function connectedFeed() { + serveSnapshot() + const shared = marketSubscriptionManager.getOrCreate(SYMBOL) + unsubscribe = shared.subscribeBook(() => {}) + const ws = FakeWebSocket.instances.at(-1)! + ws.open() + await vi.waitFor(() => expect(shared.getBookState().bids.length).toBeGreaterThan(0)) + return { shared, ws } +} + +beforeEach(() => { + FakeWebSocket.instances = [] + vi.stubGlobal("WebSocket", FakeWebSocket) + vi.useFakeTimers({ toFake: ["Date"], now: T0 }) +}) + +afterEach(() => { + unsubscribe?.() + unsubscribe = null + marketSubscriptionManager.removeInstance(SYMBOL) + server.resetHandlers() + vi.useRealTimers() + vi.unstubAllGlobals() +}) + +const advance = (ms: number) => vi.setSystemTime(Date.now() + ms) + +describe("source freshness from the live stream", () => { + it("is unavailable until the snapshot has been applied", async () => { + serveSnapshot() + const shared = marketSubscriptionManager.getOrCreate(SYMBOL) + unsubscribe = shared.subscribeBook(() => {}) + expect(shared.getFreshness()).toMatchObject({ state: "unavailable", reason: "connecting" }) + expect(shared.getFreshness().canSeedQuote).toBe(false) + + // Let the in-flight snapshot request settle so it cannot outlive this test. + await vi.waitFor(() => expect(shared.getBookState().bids.length).toBeGreaterThan(0)) + }) + + it("is fresh once connected with a snapshot, and can seed a quote from the book", async () => { + const { shared } = await connectedFeed() + const freshness = shared.getFreshness() + expect(freshness).toMatchObject({ state: "fresh", canSeedQuote: true }) + expect(quoteFromDepth(shared.getBookState(), freshness)).toEqual({ bid: 99, ask: 101, mid: 100 }) + }) + + it("stays fresh through minutes with no book changes while acks keep arriving (no trades)", async () => { + const { shared, ws } = await connectedFeed() + for (let i = 0; i < 24; i++) { + advance(10_000) + ws.frame(ack(i + 2)) + } + const freshness = shared.getFreshness() + expect(freshness).toMatchObject({ state: "fresh", reason: "quiet-market", canSeedQuote: true }) + expect(freshness.lastValidUpdateAgeMs).toBeGreaterThanOrEqual(240_000) + }) + + it("goes stale when the transport says open but goes silent", async () => { + const { shared } = await connectedFeed() + advance(21_000) + const freshness = shared.getFreshness() + expect(freshness).toMatchObject({ state: "stale", reason: "transport-silent", canSeedQuote: false }) + expect(quoteFromDepth(shared.getBookState(), freshness)).toBeNull() + }) + + it("is reconnecting after the socket drops, and its stale depth cannot seed a quote", async () => { + const { shared, ws } = await connectedFeed() + ws.drop() + advance(500) + const freshness = shared.getFreshness() + expect(freshness).toMatchObject({ state: "reconnecting", canSeedQuote: false }) + expect(shared.getSourceHealth().health.status).toBe("reconnecting") + expect(quoteFromDepth(shared.getBookState(), freshness)).toBeNull() + }) + + it("counts malformed frames, keeps the last valid book, and reports degraded", async () => { + const { shared, ws } = await connectedFeed() + const before = shared.getBookState().bids.map((l) => l.price) + + ws.frame("not json at all") + ws.frame("null") + ws.frame({ e: "depthUpdate", u: 101, b: "oops", a: [] }) // wrong shape + advance(1_000) + + expect(shared.getFreshness()).toMatchObject({ + state: "stale", + reason: "malformed-updates", + canSeedQuote: false, + }) + // The bad frames never touched the book. + expect(shared.getBookState().bids.map((l) => l.price)).toEqual(before) + // And the transport is clearly alive: this is not the dead-transport case. + expect(shared.getFreshness().lastMessageAgeMs).toBeLessThan(2_000) + }) + + it("recovers from a malformed streak with the next valid update", async () => { + const { shared, ws } = await connectedFeed() + ws.frame("garbage") + ws.frame("garbage") + ws.frame("garbage") + expect(shared.getFreshness().reason).toBe("malformed-updates") + + ws.frame(depth(101, Date.now() - 30)) + expect(shared.getFreshness()).toMatchObject({ state: "fresh", canSeedQuote: true }) + }) + + it("flags a provider that delivers updates far later than their timestamps", async () => { + const { shared, ws } = await connectedFeed() + for (let i = 0; i < 4; i++) { + advance(100) + ws.frame(depth(101 + i, Date.now() - 30)) // normal ~30ms delivery + } + expect(shared.getFreshness().state).toBe("fresh") + + advance(100) + ws.frame(depth(200, Date.now() - 9_000)) // 9s old on arrival + expect(shared.getFreshness()).toMatchObject({ + state: "stale", + reason: "provider-lagging", + canSeedQuote: false, + }) + }) + + it("does not read a constant clock offset as lag", async () => { + const { shared, ws } = await connectedFeed() + const offset = 15 * 60_000 // provider clock 15 minutes behind ours + for (let i = 0; i < 6; i++) { + advance(100) + ws.frame(depth(101 + i, Date.now() - offset - 30)) + } + expect(shared.getFreshness()).toMatchObject({ state: "fresh", providerLagMs: 0 }) + }) + + it("ignores a replayed old revision for freshness", async () => { + const { shared, ws } = await connectedFeed() + ws.frame(depth(150, Date.now() - 30)) + advance(15_000) + ws.frame(depth(120, Date.now() - 30)) // older revision replayed + const freshness = shared.getFreshness() + // The replay proves the transport is alive but does not make the book newer. + expect(freshness.lastMessageAgeMs).toBe(0) + expect(freshness.lastValidUpdateAgeMs).toBe(15_000) + }) + + it("requires a fresh snapshot after a reconnect before depth may seed a quote again", async () => { + const { shared, ws } = await connectedFeed() + ws.drop() + // Reconnect: a new socket opens and the depth snapshot is refetched. + serveSnapshot(500) + vi.useRealTimers() + await vi.waitFor(() => expect(FakeWebSocket.instances.length).toBeGreaterThan(1), { timeout: 3_000 }) + vi.useFakeTimers({ toFake: ["Date"], now: T0 + 1_000 }) + + const next = FakeWebSocket.instances.at(-1)! + next.open() + // Between the socket opening and the snapshot landing, depth is unusable. + expect(shared.getFreshness()).toMatchObject({ state: "unavailable", reason: "awaiting-snapshot" }) + + await vi.waitFor(() => expect(shared.getFreshness().canSeedQuote).toBe(true)) + expect(shared.getFreshness().state).toBe("fresh") + }) +}) diff --git a/apps/web/src/features/trade/lib/market-data-stream.ts b/apps/web/src/features/trade/lib/market-data-stream.ts index a29ab9d..6a0fdae 100644 --- a/apps/web/src/features/trade/lib/market-data-stream.ts +++ b/apps/web/src/features/trade/lib/market-data-stream.ts @@ -1,7 +1,20 @@ import { BINANCE_PERIOD, BINANCE_SYMBOL, fetchOracleCandles, type OhlcBar } from "./oracle" +import { + DEFAULT_FRESHNESS_POLICY, + createTelemetry, + deriveFreshness, + recordMalformed, + recordMessage, + recordSnapshot, + recordValidUpdate, + resetSnapshot, + toSourceHealth, +} from "./source-freshness" +import type { FeedTelemetry, FeedTransport, FreshnessResult } from "./source-freshness" import type { OrderBookLevel, OrderBookState } from "../hooks/useOrderBook" import type { TradeItem, UseRecentTradesResult } from "../hooks/useRecentTrades" import { deduplicateAndSortTrades } from "../hooks/useRecentTrades" +import type { SourceHealth } from "../hooks/useSourceHealth" export type StreamStatus = "connecting" | "connected" | "polling" | "disconnected" | "error" @@ -27,6 +40,13 @@ type RawBook = { lastUpdateId: number } +/** A depth frame must carry both sides as arrays and a numeric revision to be applied. */ +export function isValidDepthMsg(data: unknown): boolean { + if (typeof data !== "object" || data === null) return false + const m = data as { b?: unknown; a?: unknown; u?: unknown } + return Array.isArray(m.b) && Array.isArray(m.a) && typeof m.u === "number" && Number.isFinite(m.u) +} + export function applyDelta(map: Map, entries: Array<[string, string]>) { for (const [price, size] of entries) { if (parseFloat(size) === 0) { @@ -57,6 +77,8 @@ export function buildLevels(map: Map, ascending: boolean): Array type BinanceDiffMsg = { e?: string + /** Provider event time (ms since epoch), when present. */ + E?: number u: number b: Array<[string, string]> a: Array<[string, string]> @@ -134,6 +156,10 @@ class SharedMarketSubscription { isLoading: true, } + // OB-049: connection health, last message, last valid revision and provider + // timestamps, tracked separately so a quiet market is never read as a dead feed. + private telemetry: FeedTelemetry = createTelemetry() + // Trades state private currentTradesResult: UseRecentTradesResult = { trades: [], @@ -171,6 +197,34 @@ class SharedMarketSubscription { return this.liveBars.get(period) ?? null } + /** OB-049: Transport state, derived from the stream status so there is one source of truth. */ + private feedTransport(): FeedTransport { + switch (this.status) { + case "connected": + case "polling": // polling refreshes `lastMessageAt` with every successful fetch + return "open" + case "disconnected": + return "closed" + case "error": + return "error" + case "connecting": + return "connecting" + } + } + + /** OB-049: Fresh / stale / reconnecting / unavailable for the book, with the reason. */ + public getFreshness(now: number = Date.now()): FreshnessResult { + this.telemetry.transport = this.feedTransport() + this.telemetry.reconnectAttempt = this.reconnectAttempt + return deriveFreshness(this.telemetry, now, DEFAULT_FRESHNESS_POLICY) + } + + /** OB-049: The freshness result in the shape the source health badge renders. */ + public getSourceHealth(now: number = Date.now()): { health: SourceHealth; freshness: FreshnessResult } { + const freshness = this.getFreshness(now) + return { health: toSourceHealth(freshness, this.telemetry, now), freshness } + } + /** OB-113: Exposed for testing — number of WS reconnect attempts made. */ public getReconnectAttempt(): number { return this.reconnectAttempt @@ -377,6 +431,7 @@ class SharedMarketSubscription { // snapshot's lastUpdateId is known. if (this.bookSubscribers.size > 0) { this.snapshotLoaded = false + resetSnapshot(this.telemetry) this.bookBuffer = [] void this.fetchDepthSnapshot() } @@ -386,6 +441,7 @@ class SharedMarketSubscription { if (this.isDestroyed) return // OB-113: Any arriving message proves the connection is alive. this.resetHeartbeat() + recordMessage(this.telemetry, Date.now()) this.handleWsMessage(evt.data) } @@ -440,10 +496,17 @@ class SharedMarketSubscription { private handleWsMessage(raw: unknown) { try { const data = typeof raw === "string" ? JSON.parse(raw) : raw - if (!data) return + if (!data || typeof data !== "object") { + recordMalformed(this.telemetry, Date.now()) + return + } // Depth event if (data.e === "depthUpdate" || (data.b && data.a && data.u)) { + if (!isValidDepthMsg(data)) { + recordMalformed(this.telemetry, Date.now()) + return + } this.handleDepthUpdate(data as BinanceDiffMsg) } // Trade event @@ -455,11 +518,18 @@ class SharedMarketSubscription { this.handleKlineUpdate(data as BinanceKlineMsg) } } catch { - // Ignore malformed payloads + // Malformed payloads never reach the book; they are counted so a feed + // that only sends garbage is reported as degraded. + recordMalformed(this.telemetry, Date.now()) } } private handleDepthUpdate(msg: BinanceDiffMsg) { + recordValidUpdate(this.telemetry, DEFAULT_FRESHNESS_POLICY, { + receivedAt: Date.now(), + revision: msg.u, + providerTimestamp: typeof msg.E === "number" ? msg.E : null, + }) if (this.snapshotLoaded) { applyDelta(this.book.bids, msg.b) applyDelta(this.book.asks, msg.a) @@ -585,8 +655,14 @@ class SharedMarketSubscription { applyDelta(this.book.bids, msg.b) applyDelta(this.book.asks, msg.a) } + // The book is at the newest revision seen, including buffered updates. + const newestRevision = this.bookBuffer.reduce( + (max, msg) => Math.max(max, msg.u), + this.book.lastUpdateId, + ) this.bookBuffer = [] this.snapshotLoaded = true + recordSnapshot(this.telemetry, Date.now(), newestRevision) this.publishBook() } catch { if (this.isDestroyed) return diff --git a/apps/web/src/features/trade/lib/source-freshness.test.ts b/apps/web/src/features/trade/lib/source-freshness.test.ts new file mode 100644 index 0000000..4af2f32 --- /dev/null +++ b/apps/web/src/features/trade/lib/source-freshness.test.ts @@ -0,0 +1,441 @@ +import { describe, expect, it } from "vitest" +import { + + DEFAULT_FRESHNESS_POLICY as POLICY, + createTelemetry, + deriveFreshness, + quoteFromDepth, + recordMalformed, + recordMessage, + recordSnapshot, + recordTransport, + recordValidUpdate, + resetSnapshot, + toSourceHealth +} from "./source-freshness" +import type {FeedTelemetry} from "./source-freshness"; + +const T0 = 1_000_000 + +/** A healthy open feed: snapshot applied at T0, then live updates. */ +function liveFeed(opts: { providerOffset?: number } = {}): FeedTelemetry { + const t = createTelemetry() + recordTransport(t, "open") + recordSnapshot(t, T0, 100) + recordValidUpdate(t, POLICY, { + receivedAt: T0 + 100, + revision: 101, + providerTimestamp: T0 + 100 - 40 - (opts.providerOffset ?? 0), + }) + return t +} + +const book = { + bids: [{ price: 99, size: 1 }], + asks: [{ price: 101, size: 1 }], +} + +describe("initial and unavailable states", () => { + it("is unavailable, not stale, before the first connection completes", () => { + const t = createTelemetry() + const f = deriveFreshness(t, T0) + expect(f).toMatchObject({ state: "unavailable", reason: "connecting", canSeedQuote: false }) + expect(f.lastMessageAgeMs).toBeNull() + expect(f.lastValidUpdateAgeMs).toBeNull() + }) + + it("is reconnecting when a first-ever connection is being retried", () => { + const t = createTelemetry() + recordTransport(t, "connecting", 2) + expect(deriveFreshness(t, T0)).toMatchObject({ state: "reconnecting", reason: "connecting" }) + }) + + it("is unavailable while an open transport has not delivered a snapshot", () => { + const t = createTelemetry() + recordTransport(t, "open") + recordMessage(t, T0) + expect(deriveFreshness(t, T0 + 1)).toMatchObject({ + state: "unavailable", + reason: "awaiting-snapshot", + canSeedQuote: false, + }) + }) + + it("delta updates before any snapshot do not create a usable book", () => { + const t = createTelemetry() + recordTransport(t, "open") + recordValidUpdate(t, POLICY, { receivedAt: T0, revision: 5, providerTimestamp: T0 - 10 }) + expect(deriveFreshness(t, T0 + 1).canSeedQuote).toBe(false) + }) +}) + +describe("a healthy feed", () => { + it("is fresh and may seed a quote", () => { + const f = deriveFreshness(liveFeed(), T0 + 500) + expect(f).toMatchObject({ state: "fresh", reason: "live", canSeedQuote: true }) + }) + + it("reports ages from local receipt times", () => { + const f = deriveFreshness(liveFeed(), T0 + 1_100) + expect(f.lastMessageAgeMs).toBe(1_000) + expect(f.lastValidUpdateAgeMs).toBe(1_000) + }) +}) + +describe("a market with no trades is not a dead connection", () => { + it("stays fresh when the book is unchanged for minutes but heartbeats keep arriving", () => { + const t = liveFeed() + const start = T0 + 100 + // Heartbeat/ack frames every 10s for 4 minutes, and not one book change. + for (let s = 10_000; s <= 240_000; s += 10_000) recordMessage(t, start + s) + + const f = deriveFreshness(t, start + 240_000) + expect(f).toMatchObject({ state: "fresh", reason: "quiet-market", canSeedQuote: true }) + expect(f.lastValidUpdateAgeMs).toBe(240_000) + }) + + it("still reports quiet-market without heartbeat frames as long as some frame is recent", () => { + const t = liveFeed() + recordMessage(t, T0 + 60_000) + expect(deriveFreshness(t, T0 + 65_000)).toMatchObject({ + state: "fresh", + reason: "quiet-market", + }) + }) +}) + +describe("dead transport", () => { + it("goes stale when an open transport goes silent past the heartbeat timeout", () => { + const t = liveFeed() + const f = deriveFreshness(t, T0 + 100 + POLICY.heartbeatTimeoutMs + 1) + expect(f).toMatchObject({ state: "stale", reason: "transport-silent", canSeedQuote: false }) + }) + + it("is not stale exactly at the timeout boundary", () => { + const t = liveFeed() + expect(deriveFreshness(t, T0 + 100 + POLICY.heartbeatTimeoutMs).state).toBe("fresh") + }) + + it("is reconnecting once the transport reports closed, keeping the last book visible", () => { + const t = liveFeed() + recordTransport(t, "closed", 1) + expect(deriveFreshness(t, T0 + 2_000)).toMatchObject({ + state: "reconnecting", + reason: "transport-closed", + canSeedQuote: false, + }) + }) + + it("is reconnecting after a transport error when a book exists, unavailable when none does", () => { + const withBook = liveFeed() + recordTransport(withBook, "error", 1) + expect(deriveFreshness(withBook, T0 + 1_000)).toMatchObject({ + state: "reconnecting", + reason: "transport-error", + }) + + const noBook = createTelemetry() + recordTransport(noBook, "error", 1) + expect(deriveFreshness(noBook, T0)).toMatchObject({ + state: "unavailable", + reason: "transport-error", + }) + }) + + it("expires the last book after a long outage", () => { + const t = liveFeed() + recordTransport(t, "closed", 6) + const f = deriveFreshness(t, T0 + 100 + POLICY.expireAfterMs + 1) + expect(f).toMatchObject({ state: "unavailable", reason: "data-expired", canSeedQuote: false }) + }) + + it("expires an open but silent-for-ever feed too", () => { + const t = liveFeed() + expect(deriveFreshness(t, T0 + 100 + POLICY.expireAfterMs + 1)).toMatchObject({ + state: "unavailable", + reason: "data-expired", + }) + }) + + it("recovers to fresh once a reconnect delivers a new snapshot", () => { + const t = liveFeed() + recordTransport(t, "closed", 1) + resetSnapshot(t) + recordTransport(t, "open", 0) + recordMessage(t, T0 + 5_000) + expect(deriveFreshness(t, T0 + 5_001).reason).toBe("awaiting-snapshot") + + recordSnapshot(t, T0 + 5_200, 500) + expect(deriveFreshness(t, T0 + 5_300)).toMatchObject({ state: "fresh", canSeedQuote: true }) + }) +}) + +describe("malformed updates", () => { + it("counts as stale after the threshold of consecutive bad frames", () => { + const t = liveFeed() + for (let i = 1; i < POLICY.malformedThreshold; i++) recordMalformed(t, T0 + 200 + i) + expect(deriveFreshness(t, T0 + 300).state).toBe("fresh") + + recordMalformed(t, T0 + 400) + expect(deriveFreshness(t, T0 + 500)).toMatchObject({ + state: "stale", + reason: "malformed-updates", + canSeedQuote: false, + }) + }) + + it("is not mistaken for a dead transport: frames are arriving", () => { + const t = liveFeed() + for (let i = 0; i < 5; i++) recordMalformed(t, T0 + 1_000 * (i + 1)) + const f = deriveFreshness(t, T0 + 5_500) + expect(f.reason).toBe("malformed-updates") + expect(f.lastMessageAgeMs).toBe(500) + }) + + it("clears as soon as a valid update arrives", () => { + const t = liveFeed() + for (let i = 0; i < 5; i++) recordMalformed(t, T0 + 200 + i) + recordValidUpdate(t, POLICY, { receivedAt: T0 + 300, revision: 102, providerTimestamp: T0 + 260 }) + expect(deriveFreshness(t, T0 + 301)).toMatchObject({ state: "fresh", canSeedQuote: true }) + }) + + it("keeps a running total for diagnostics", () => { + const t = liveFeed() + recordMalformed(t, T0 + 200) + recordValidUpdate(t, POLICY, { receivedAt: T0 + 300, revision: 102 }) + recordMalformed(t, T0 + 400) + expect(t.malformedTotal).toBe(2) + expect(t.consecutiveMalformed).toBe(1) + }) +}) + +describe("lagging provider", () => { + /** Establish a normal ~40ms delivery baseline, then deliver one update very late. */ + function feedThenLate(extraDelay: number): FeedTelemetry { + const t = liveFeed() + for (let i = 0; i < 5; i++) { + const received = T0 + 200 + i * 100 + recordValidUpdate(t, POLICY, { receivedAt: received, revision: 102 + i, providerTimestamp: received - 40 }) + } + const received = T0 + 1_000 + recordValidUpdate(t, POLICY, { + receivedAt: received, + revision: 200, + providerTimestamp: received - 40 - extraDelay, + }) + return t + } + + it("goes stale when the provider delivers updates far later than its own timestamps", () => { + const f = deriveFreshness(feedThenLate(POLICY.maxProviderLagMs + 1_000), T0 + 1_001) + expect(f).toMatchObject({ state: "stale", reason: "provider-lagging", canSeedQuote: false }) + expect(f.providerLagMs).toBe(POLICY.maxProviderLagMs + 1_000) + }) + + it("tolerates lag within the policy", () => { + const f = deriveFreshness(feedThenLate(1_000), T0 + 1_001) + expect(f.state).toBe("fresh") + expect(f.providerLagMs).toBe(1_000) + }) + + it("is not the same as a quiet market: lag needs delivered-late updates, not a long gap", () => { + const t = liveFeed() + recordMessage(t, T0 + 30_000) + const f = deriveFreshness(t, T0 + 31_000) + expect(f.reason).toBe("quiet-market") + expect(f.providerLagMs).toBe(0) + }) + + it("recovers when the provider catches up", () => { + const t = feedThenLate(POLICY.maxProviderLagMs + 1_000) + const received = T0 + 2_000 + recordValidUpdate(t, POLICY, { receivedAt: received, revision: 201, providerTimestamp: received - 40 }) + expect(deriveFreshness(t, received + 1).state).toBe("fresh") + }) + + it("has no lag figure when the provider sends no timestamps", () => { + const t = createTelemetry() + recordTransport(t, "open") + recordSnapshot(t, T0, 1) + recordValidUpdate(t, POLICY, { receivedAt: T0 + 100, revision: 2 }) + const f = deriveFreshness(t, T0 + 200) + expect(f).toMatchObject({ state: "fresh", providerLagMs: null, clockOffsetMs: null }) + }) +}) + +describe("clock skew between us and the provider", () => { + it.each([ + ["our clock 10 minutes ahead", 600_000], + ["our clock 10 minutes behind", -600_000], + ["in step", 0], + ])("a constant offset is not lag (%s)", (_label, skew) => { + const t = createTelemetry() + recordTransport(t, "open") + recordSnapshot(t, T0, 1) + for (let i = 0; i < 10; i++) { + const received = T0 + 100 + i * 100 + // Provider timestamps are `skew` ms away from our clock, plus ~40ms transit. + recordValidUpdate(t, POLICY, { receivedAt: received, revision: 2 + i, providerTimestamp: received - skew - 40 }) + } + const f = deriveFreshness(t, T0 + 1_200) + expect(f).toMatchObject({ state: "fresh", providerLagMs: 0, canSeedQuote: true }) + expect(f.clockOffsetMs).toBe(skew + 40) + }) + + it("still detects real lag under a large skew", () => { + const skew = 600_000 + const t = createTelemetry() + recordTransport(t, "open") + recordSnapshot(t, T0, 1) + for (let i = 0; i < 5; i++) { + const received = T0 + 100 + i * 100 + recordValidUpdate(t, POLICY, { receivedAt: received, revision: 2 + i, providerTimestamp: received - skew - 40 }) + } + const received = T0 + 1_000 + recordValidUpdate(t, POLICY, { + receivedAt: received, + revision: 50, + providerTimestamp: received - skew - 40 - 8_000, + }) + expect(deriveFreshness(t, received + 1)).toMatchObject({ state: "stale", reason: "provider-lagging" }) + }) + + it("re-baselines when the offset drifts, because only a bounded window is kept", () => { + const t = createTelemetry() + recordTransport(t, "open") + recordSnapshot(t, T0, 1) + for (let i = 0; i < POLICY.skewWindow; i++) { + recordValidUpdate(t, POLICY, { receivedAt: T0 + i, revision: 2 + i, providerTimestamp: T0 + i - 40 }) + } + // The provider clock steps 5 minutes: all later samples share the new offset. + const shift = 300_000 + for (let i = 0; i < POLICY.skewWindow; i++) { + const received = T0 + 1_000 + i + recordValidUpdate(t, POLICY, { receivedAt: received, revision: 1_000 + i, providerTimestamp: received - shift - 40 }) + } + const f = deriveFreshness(t, T0 + 1_000 + POLICY.skewWindow) + expect(t.deliveryDelays).toHaveLength(POLICY.skewWindow) + expect(f).toMatchObject({ state: "fresh", providerLagMs: 0 }) + }) + + it("ignores a non-finite provider timestamp", () => { + const t = liveFeed() + const before = t.deliveryDelays.length + recordValidUpdate(t, POLICY, { receivedAt: T0 + 200, revision: 200, providerTimestamp: Number.NaN }) + expect(t.deliveryDelays).toHaveLength(before) + expect(deriveFreshness(t, T0 + 300).state).toBe("fresh") + }) +}) + +describe("revision handling", () => { + it("rejects duplicate and out-of-order revisions without freshening the book", () => { + const t = liveFeed() + const revision = t.lastValidRevision + const at = t.lastValidUpdateAt + + expect(recordValidUpdate(t, POLICY, { receivedAt: T0 + 5_000, revision: 101 })).toBe(false) + expect(recordValidUpdate(t, POLICY, { receivedAt: T0 + 6_000, revision: 50 })).toBe(false) + + expect(t.lastValidRevision).toBe(revision) + expect(t.lastValidUpdateAt).toBe(at) + // The frames still prove the transport is alive. + expect(t.lastMessageAt).toBe(T0 + 6_000) + }) + + it("accepts a strictly newer revision", () => { + const t = liveFeed() + expect(recordValidUpdate(t, POLICY, { receivedAt: T0 + 500, revision: 102 })).toBe(true) + expect(t.lastValidRevision).toBe(102) + }) + + it("a replayed old revision does not clear a malformed streak", () => { + const t = liveFeed() + for (let i = 0; i < 3; i++) recordMalformed(t, T0 + 200 + i) + recordValidUpdate(t, POLICY, { receivedAt: T0 + 300, revision: 100 }) + expect(deriveFreshness(t, T0 + 301).reason).toBe("malformed-updates") + }) +}) + +describe("stale depth cannot seed an actionable quote", () => { + it("quotes from fresh depth", () => { + expect(quoteFromDepth(book, deriveFreshness(liveFeed(), T0 + 200))).toEqual({ + bid: 99, + ask: 101, + mid: 100, + }) + }) + + it.each([ + ["transport silent", (t: FeedTelemetry) => void t, T0 + 100 + POLICY.heartbeatTimeoutMs + 1], + ["malformed", (t: FeedTelemetry) => { for (let i = 0; i < 3; i++) recordMalformed(t, T0 + 200 + i) }, T0 + 300], + ["reconnecting", (t: FeedTelemetry) => recordTransport(t, "closed", 1), T0 + 300], + ])("refuses to quote when %s", (_label, degrade, at) => { + const t = liveFeed() + degrade(t) + expect(quoteFromDepth(book, deriveFreshness(t, at))).toBeNull() + }) + + it("refuses a crossed, empty or nonsensical book even from a fresh feed", () => { + const fresh = deriveFreshness(liveFeed(), T0 + 200) + expect(quoteFromDepth({ bids: [{ price: 101, size: 1 }], asks: [{ price: 99, size: 1 }] }, fresh)).toBeNull() + expect(quoteFromDepth({ bids: [], asks: book.asks }, fresh)).toBeNull() + expect(quoteFromDepth({ bids: book.bids, asks: [] }, fresh)).toBeNull() + expect(quoteFromDepth({ bids: [{ price: 0, size: 1 }], asks: book.asks }, fresh)).toBeNull() + expect(quoteFromDepth({ bids: [{ price: Number.NaN, size: 1 }], asks: book.asks }, fresh)).toBeNull() + }) + + it("refuses before a snapshot even if updates are flowing", () => { + const t = createTelemetry() + recordTransport(t, "open") + recordValidUpdate(t, POLICY, { receivedAt: T0, revision: 9, providerTimestamp: T0 - 10 }) + expect(quoteFromDepth(book, deriveFreshness(t, T0 + 1))).toBeNull() + }) +}) + +describe("toSourceHealth", () => { + it("maps each state onto the statuses the badge renders", () => { + const now = T0 + 300 + const fresh = liveFeed() + expect(toSourceHealth(deriveFreshness(fresh, now), fresh, now)).toMatchObject({ + status: "connected", + isExecutable: true, + }) + + const silent = liveFeed() + const later = T0 + 100 + POLICY.heartbeatTimeoutMs + 5_000 + expect(toSourceHealth(deriveFreshness(silent, later), silent, later)).toMatchObject({ + status: "stale", + isExecutable: false, + }) + + const closed = liveFeed() + recordTransport(closed, "closed", 2) + expect(toSourceHealth(deriveFreshness(closed, now), closed, now)).toMatchObject({ + status: "reconnecting", + reconnectAttempt: 2, + message: "Reconnecting (attempt 2)…", + }) + + const fresh0 = createTelemetry() + expect(toSourceHealth(deriveFreshness(fresh0, now), fresh0, now).status).toBe("initial-load") + + const expired = liveFeed() + recordTransport(expired, "closed", 6) + const gone = T0 + 100 + POLICY.expireAfterMs + 1 + expect(toSourceHealth(deriveFreshness(expired, gone), expired, gone).status).toBe("error") + }) + + it("explains why a feed is stale", () => { + const t = liveFeed() + for (let i = 0; i < 3; i++) recordMalformed(t, T0 + 200 + i) + expect(toSourceHealth(deriveFreshness(t, T0 + 300), t, T0 + 300).message).toMatch(/invalid updates/) + }) + + it("does not call a quiet market a problem", () => { + const t = liveFeed() + recordMessage(t, T0 + 40_000) + const h = toSourceHealth(deriveFreshness(t, T0 + 41_000), t, T0 + 41_000) + expect(h).toMatchObject({ status: "connected", isExecutable: true }) + expect(h.message).toMatch(/no recent book changes/) + }) +}) diff --git a/apps/web/src/features/trade/lib/source-freshness.ts b/apps/web/src/features/trade/lib/source-freshness.ts new file mode 100644 index 0000000..f77a87a --- /dev/null +++ b/apps/web/src/features/trade/lib/source-freshness.ts @@ -0,0 +1,381 @@ +/** + * OB-049: Market-data source health and freshness. + * + * A market feed has several independent clocks, and mixing them up is how a + * quiet market gets reported as a dead connection (or a dead connection as a + * quiet market). They are tracked separately here: + * + * - connection health: the transport state, and `lastMessageAt`, the local + * time ANY frame arrived (data, heartbeat or ack). Silence on the transport + * is what proves it dead. + * - last valid update: the local time and revision of the last book update + * that parsed and advanced the revision. A market with no trades still + * has a healthy transport, so this alone never marks a feed stale. + * - provider timestamps: the provider's own event time on those updates. + * Comparing them with local receipt time measures how far the provider + * is lagging, after removing the constant offset between the two clocks. + * + * This module is pure: every function takes `now` explicitly and mutates + * nothing but the telemetry it is given, so behaviour is fully deterministic + * under test. + */ + +import type { SourceHealth } from "../hooks/useSourceHealth" + +export type FeedTransport = "connecting" | "open" | "closed" | "error" + +export type FreshnessState = "fresh" | "stale" | "reconnecting" | "unavailable" + +/** Why the feed is in its current state. Distinct so the UI and tests can tell them apart. */ +export type FreshnessReason = + | "live" // updates are arriving + | "quiet-market" // transport healthy and heartbeating, but no book change lately + | "connecting" // first connection attempt, nothing to show yet + | "awaiting-snapshot" // transport open but no valid book yet + | "transport-silent" // transport claims to be open but nothing has arrived for too long + | "transport-closed" // transport dropped; a retry is under way + | "transport-error" // transport failed + | "malformed-updates" // frames arrive but keep failing validation + | "provider-lagging" // provider delivers updates far later than its own timestamps + | "data-expired" // last valid book is too old to show at all + +export type FreshnessPolicy = { + /** Longest silence (no frame of any kind) tolerated on an open transport. */ + heartbeatTimeoutMs: number + /** Extra delivery delay, beyond the fastest recently seen, that marks the provider as lagging. */ + maxProviderLagMs: number + /** Consecutive malformed frames, with no valid update since, that mark the feed degraded. */ + malformedThreshold: number + /** After this long without a valid update the last book is discarded as unusable. */ + expireAfterMs: number + /** Delivery-delay samples kept for the baseline (clock offset) estimate. */ + skewWindow: number +} + +export const DEFAULT_FRESHNESS_POLICY: FreshnessPolicy = { + heartbeatTimeoutMs: 20_000, // matches the stream's heartbeat watchdog + maxProviderLagMs: 5_000, + malformedThreshold: 3, + expireAfterMs: 5 * 60_000, + skewWindow: 32, +} + +export type FeedTelemetry = { + transport: FeedTransport + reconnectAttempt: number + /** A book snapshot has been applied, so depth exists to show. */ + hasSnapshot: boolean + /** Local time the last frame of ANY kind arrived. */ + lastMessageAt: number | null + /** Local time the last valid book update (or snapshot) was applied. */ + lastValidUpdateAt: number | null + /** Revision of the last valid update; only ever increases. */ + lastValidRevision: number | null + /** Provider event time of the last valid update. */ + lastProviderTimestamp: number | null + /** Recent `receivedAt - providerTimestamp` samples, used to remove the clock offset. */ + deliveryDelays: Array + /** Delivery delay of the most recent valid update. */ + lastDeliveryDelay: number | null + consecutiveMalformed: number + malformedTotal: number +} + +export function createTelemetry(): FeedTelemetry { + return { + transport: "connecting", + reconnectAttempt: 0, + hasSnapshot: false, + lastMessageAt: null, + lastValidUpdateAt: null, + lastValidRevision: null, + lastProviderTimestamp: null, + deliveryDelays: [], + lastDeliveryDelay: null, + consecutiveMalformed: 0, + malformedTotal: 0, + } +} + +export function recordTransport( + t: FeedTelemetry, + transport: FeedTransport, + reconnectAttempt = t.reconnectAttempt +): void { + t.transport = transport + t.reconnectAttempt = reconnectAttempt +} + +/** Any frame arrived: data, heartbeat, subscription ack. Proves the transport is alive. */ +export function recordMessage(t: FeedTelemetry, receivedAt: number): void { + t.lastMessageAt = receivedAt +} + +/** A frame arrived but could not be used. Also counts as proof of a live transport. */ +export function recordMalformed(t: FeedTelemetry, receivedAt: number): void { + t.lastMessageAt = receivedAt + t.consecutiveMalformed += 1 + t.malformedTotal += 1 +} + +type ValidUpdate = { + receivedAt: number + revision: number + /** Provider event time in ms. Omit when the update carries none. */ + providerTimestamp?: number | null +} + +/** + * A book update that parsed. Returns false, and leaves the book bookkeeping + * untouched, for a duplicate or out-of-order revision: replaying an old + * revision must not make a feed look fresher than it is. + */ +export function recordValidUpdate( + t: FeedTelemetry, + policy: FreshnessPolicy, + update: ValidUpdate +): boolean { + t.lastMessageAt = update.receivedAt + if (t.lastValidRevision !== null && update.revision <= t.lastValidRevision) { + return false + } + t.lastValidRevision = update.revision + t.lastValidUpdateAt = update.receivedAt + t.consecutiveMalformed = 0 + + const ts = update.providerTimestamp + if (typeof ts === "number" && Number.isFinite(ts)) { + const delay = update.receivedAt - ts + t.lastProviderTimestamp = ts + t.lastDeliveryDelay = delay + t.deliveryDelays.push(delay) + if (t.deliveryDelays.length > policy.skewWindow) t.deliveryDelays.shift() + } + return true +} + +/** A full snapshot was applied. It establishes the book and its revision. */ +export function recordSnapshot( + t: FeedTelemetry, + receivedAt: number, + revision: number +): void { + t.lastMessageAt = receivedAt + t.hasSnapshot = true + t.lastValidUpdateAt = receivedAt + t.lastValidRevision = revision + t.consecutiveMalformed = 0 +} + +/** The book was discarded (e.g. resync after reconnect); depth must be rebuilt before use. */ +export function resetSnapshot(t: FeedTelemetry): void { + t.hasSnapshot = false + t.lastValidRevision = null + t.lastValidUpdateAt = null +} + +export type FreshnessResult = { + state: FreshnessState + reason: FreshnessReason + /** + * Whether depth from this feed may seed an actionable quote (a price + * suggested into an order ticket). Only a fresh feed with an applied + * snapshot may; stale depth must never. + */ + canSeedQuote: boolean + /** ms since any frame arrived, or null if none has. */ + lastMessageAgeMs: number | null + /** ms since the last valid book update, or null if none has been applied. */ + lastValidUpdateAgeMs: number | null + /** + * How much later than usual the latest update reached us, in ms, with the + * fixed offset between our clock and the provider's removed. Null without + * provider timestamps. + */ + providerLagMs: number | null + /** Estimated `local clock - provider clock` in ms (includes minimum network latency), or null. */ + clockOffsetMs: number | null +} + +function result( + t: FeedTelemetry, + now: number, + state: FreshnessState, + reason: FreshnessReason, + extra: { providerLagMs: number | null; clockOffsetMs: number | null } +): FreshnessResult { + return { + state, + reason, + canSeedQuote: state === "fresh" && t.hasSnapshot && t.lastValidRevision !== null, + lastMessageAgeMs: t.lastMessageAt === null ? null : Math.max(0, now - t.lastMessageAt), + lastValidUpdateAgeMs: + t.lastValidUpdateAt === null ? null : Math.max(0, now - t.lastValidUpdateAt), + ...extra, + } +} + +/** + * Fresh / stale / reconnecting / unavailable, with the reason. + * + * Order matters and encodes the policy: + * 1. No usable book at all -> unavailable (or reconnecting if a retry is under way). + * 2. Transport down -> reconnecting, and after `expireAfterMs` unavailable. + * 3. Open transport, checked from most to least severe: silence beyond the + * heartbeat timeout, then a stream of malformed frames, then a lagging + * provider. Any of them is stale. + * 4. Otherwise fresh. A long gap since the last book change is NOT a reason + * to distrust the feed: `quiet-market` is reported so it is visible, but + * stays fresh. + */ +export function deriveFreshness( + t: FeedTelemetry, + now: number, + policy: FreshnessPolicy = DEFAULT_FRESHNESS_POLICY +): FreshnessResult { + const clockOffsetMs = t.deliveryDelays.length > 0 ? Math.min(...t.deliveryDelays) : null + const providerLagMs = + t.lastDeliveryDelay !== null && clockOffsetMs !== null + ? Math.max(0, t.lastDeliveryDelay - clockOffsetMs) + : null + const extra = { providerLagMs, clockOffsetMs } + + const validAge = t.lastValidUpdateAt === null ? null : Math.max(0, now - t.lastValidUpdateAt) + const hasBook = t.hasSnapshot && t.lastValidRevision !== null + + if (t.transport === "error" || t.transport === "closed") { + if (!hasBook) return result(t, now, "unavailable", t.transport === "error" ? "transport-error" : "transport-closed", extra) + if (validAge !== null && validAge > policy.expireAfterMs) { + return result(t, now, "unavailable", "data-expired", extra) + } + return result( + t, + now, + "reconnecting", + t.transport === "error" ? "transport-error" : "transport-closed", + extra + ) + } + + if (t.transport === "connecting") { + if (!hasBook) { + return result( + t, + now, + t.reconnectAttempt > 0 ? "reconnecting" : "unavailable", + "connecting", + extra + ) + } + if (validAge !== null && validAge > policy.expireAfterMs) { + return result(t, now, "unavailable", "data-expired", extra) + } + return result(t, now, "reconnecting", "transport-closed", extra) + } + + // transport === "open" + if (!hasBook) return result(t, now, "unavailable", "awaiting-snapshot", extra) + + if (validAge !== null && validAge > policy.expireAfterMs) { + return result(t, now, "unavailable", "data-expired", extra) + } + + const silence = t.lastMessageAt === null ? Infinity : now - t.lastMessageAt + if (silence > policy.heartbeatTimeoutMs) { + return result(t, now, "stale", "transport-silent", extra) + } + + if (t.consecutiveMalformed >= policy.malformedThreshold) { + return result(t, now, "stale", "malformed-updates", extra) + } + + if (providerLagMs !== null && providerLagMs > policy.maxProviderLagMs) { + return result(t, now, "stale", "provider-lagging", extra) + } + + // Healthy transport. Distinguish a quiet market from a busy one for display only. + const quiet = validAge !== null && validAge > policy.heartbeatTimeoutMs + return result(t, now, "fresh", quiet ? "quiet-market" : "live", extra) +} + +/** Maps a freshness result onto the health model the existing badge renders. */ +export function toSourceHealth( + freshness: FreshnessResult, + t: FeedTelemetry, + now: number +): SourceHealth { + const lastUpdateTime = t.lastValidUpdateAt + const staleDuration = freshness.lastValidUpdateAgeMs + const base = { lastUpdateTime, reconnectAttempt: t.reconnectAttempt } + + switch (freshness.state) { + case "fresh": + return { + ...base, + status: "connected", + staleDuration: null, + isExecutable: freshness.canSeedQuote, + message: freshness.reason === "quiet-market" ? "Live — no recent book changes" : "Live market data", + } + case "stale": + return { + ...base, + status: "stale", + staleDuration, + isExecutable: false, + message: STALE_MESSAGES[freshness.reason] ?? "Market data is stale", + } + case "reconnecting": + return { + ...base, + status: "reconnecting", + staleDuration, + isExecutable: false, + message: + t.reconnectAttempt > 0 + ? `Reconnecting (attempt ${t.reconnectAttempt})…` + : "Reconnecting to market data…", + } + case "unavailable": + return { + ...base, + status: freshness.reason === "connecting" || freshness.reason === "awaiting-snapshot" ? "initial-load" : "error", + staleDuration: now === 0 ? null : staleDuration, + isExecutable: false, + message: + freshness.reason === "data-expired" + ? "Market data expired — waiting for a fresh book" + : freshness.reason === "connecting" || freshness.reason === "awaiting-snapshot" + ? "Connecting to market data feed…" + : "Market data unavailable", + } + } +} + +const STALE_MESSAGES: Partial> = { + "transport-silent": "No data received — connection may be dead", + "malformed-updates": "Receiving invalid updates — showing last valid book", + "provider-lagging": "Provider is delivering updates late", +} + +type Level = { price: number; size: number } +type Book = { bids: ReadonlyArray; asks: ReadonlyArray } + +/** + * The only sanctioned way to turn feed depth into a suggested price for an + * order ticket. Returns null unless the feed is fresh with an applied + * snapshot, and unless the book is uncrossed, so stale, unsynced or broken + * depth can never seed an actionable quote. + */ +export function quoteFromDepth( + book: Book, + freshness: FreshnessResult +): { bid: number; ask: number; mid: number } | null { + if (!freshness.canSeedQuote) return null + const bid = book.bids.at(0)?.price + const ask = book.asks.at(0)?.price + if (bid === undefined || ask === undefined) return null + if (!Number.isFinite(bid) || !Number.isFinite(ask) || bid <= 0 || ask <= 0) return null + if (bid >= ask) return null + return { bid, ask, mid: (bid + ask) / 2 } +} From e732bf1943a000c4dfe9fca534ea19c601d5b22e Mon Sep 17 00:00:00 2001 From: Precious Akpan <49040483+precious-akpan@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:17:54 +0100 Subject: [PATCH 2/2] docs(changelog): note order book source health (OB-049) --- .changelog/unreleased/818-ob-049-source-freshness.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changelog/unreleased/818-ob-049-source-freshness.md diff --git a/.changelog/unreleased/818-ob-049-source-freshness.md b/.changelog/unreleased/818-ob-049-source-freshness.md new file mode 100644 index 0000000..95c92d0 --- /dev/null +++ b/.changelog/unreleased/818-ob-049-source-freshness.md @@ -0,0 +1,7 @@ +--- +type: added +area: trade +pr: 818 +breaking: false +--- +The order book now shows whether its market data is live, stale or reconnecting, and no longer treats a quiet market as a dead connection.