/** * Live futures quotes: one feed per machine, on one futures client, and the * spread each entry pays read off it. * * Every firm trades the same CME book, so one feed serves them all. It rides * a TopstepX client's market hub when this machine has one, else a Tradovate * client's market-data socket (when its login carries market data). Each * listed root's front month is subscribed, plus the drain micro, and * resubscribed when the front month rolls. * * A quote older than QUOTE_FRESH_MS isn't live: the spread falls back to the * typical one tick (instrument-specs). A feed that fails, loses its client or * goes quiet while the market trades is dropped, and the next supervision * starts one on another client (the failed one rests RETRY_AFTER_MS). */ import * as signalR from '@microsoft/signalr'; import { brokerWebSocket, signalrWebSocket } from './outbound-proxy'; import { errorMessage } from './platform-client'; import { isNoTradeTime } from './trading-logic'; import { CME_SPECS, spreadDollars, tickValue } from './instrument-specs'; import { mergeTopstepxQuote, parseTradovateFrame, bookSpread, type BookTop } from './quote-helpers'; const MARKET_HUB = 'https://rtc.topstepx.com/hubs/market'; const TRADOVATE_MD_WS = 'wss://md.tradovateapi.com/v1/websocket'; /** A quote older than this is stale: the typical spread stands in. */ export const QUOTE_FRESH_MS = 2 * 60_000; const START_TIMEOUT_MS = 30_000; const INVOKE_TIMEOUT_MS = 15_000; /** Silent this long while the market trades: dead, or a login with no data. */ const SILENT_MS = 5 * 60_000; /** A client whose feed was dropped waits this long to be tried again. */ const RETRY_AFTER_MS = 3 * 60_000; export const SUPERVISE_INTERVAL_MS = 30_000; type Platform = 'topstepx' | 'tradovate'; type Contract = { id: number | string; name: string }; type QuoteSink = (root: string, contract: string, patch: BookTop) => void; /** What the feed needs of a futures client. */ export interface QuoteClient { quoteToken?: () => string | null; findFrontMonthContract(root: string): Promise; } interface Feed { readonly platform: Platform; readonly firmId: string; readonly firmName: string; readonly startedAt: number; lastQuoteAt: number; /** Why it gave up; null while it works. */ failed: string | null; /** The contract id `root` is subscribed under. */ contractOf(root: string): string | null; subscribe(root: string, contract: Contract): void; stop(): void; } interface LiveBook extends BookTop { at: number; contract: string; } interface State { books: Map; feed: Feed | null; droppedAt: Map; lastError: string | null; supervising: boolean; } // On globalThis: instrumentation.ts boots the feed from its own bundle, and // the route handlers (the Settings page, the draw) must read the same books. const g = globalThis as typeof globalThis & { __liveQuotes?: State }; function state(): State { return (g.__liveQuotes ??= { books: new Map(), feed: null, droppedAt: new Map(), lastError: null, supervising: false }); } function withTimeout(p: Promise, ms: number, what: string): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error(`${what} timed out after ${ms / 1000}s`)), ms); p.then((v) => { clearTimeout(timer); resolve(v); }, (err) => { clearTimeout(timer); reject(err); }); }); } const onQuote: QuoteSink = (root, contract, patch) => { const books = state().books; const prev = books.get(root); // A new front month starts a fresh book: the old one's sides don't carry. books.set(root, { ...(prev?.contract === contract ? prev : {}), ...patch, contract, at: Date.now() }); }; // ── TopstepX: the market hub ──────────────────────────────────────────────── class TopstepxFeed implements Feed { readonly platform = 'topstepx' as const; readonly startedAt = Date.now(); lastQuoteAt = 0; failed: string | null = null; private ready = false; private stopped = false; private readonly hub: signalR.HubConnection; private readonly byContract = new Map(); private readonly byRoot = new Map(); constructor(readonly firmId: string, readonly firmName: string, token: () => string | null) { // Like the user hub: one websocket, through the operator's proxy. const proxiedWs = signalrWebSocket(); this.hub = new signalR.HubConnectionBuilder() .withUrl(MARKET_HUB, { skipNegotiation: true, transport: signalR.HttpTransportType.WebSockets, accessTokenFactory: () => token() ?? '', // eslint-disable-next-line @typescript-eslint/no-explicit-any ...(proxiedWs ? { WebSocket: proxiedWs as any } : {}), }) .withAutomaticReconnect() .configureLogging(signalR.LogLevel.Warning) .build(); this.hub.on('GatewayQuote', (contractId: unknown, data: unknown) => { const sub = typeof contractId === 'string' ? this.byContract.get(contractId) : undefined; if (!sub || this.stopped) return; this.lastQuoteAt = Date.now(); onQuote(sub.root, sub.name, mergeTopstepxQuote(undefined, data)); }); this.hub.onreconnected(() => { if (!this.stopped) for (const id of this.byContract.keys()) void this.send(id); }); this.hub.onclose((err) => { if (!this.stopped) this.failed ??= `market hub closed${err ? `: ${errorMessage(err)}` : ''}`; }); withTimeout(this.hub.start(), START_TIMEOUT_MS, 'market hub start') .then(() => { if (this.stopped) return; this.ready = true; for (const id of this.byContract.keys()) void this.send(id); }) .catch((err) => { this.failed ??= `market hub: ${errorMessage(err)}`; }); } contractOf(root: string): string | null { return this.byRoot.get(root) ?? null; } subscribe(root: string, contract: Contract): void { const id = String(contract.id); const old = this.byRoot.get(root); if (old && old !== id) { this.byContract.delete(old); if (this.ready) void this.hub.invoke('UnsubscribeContractQuotes', old).catch(() => { /* rolled away anyway */ }); } this.byRoot.set(root, id); this.byContract.set(id, { root, name: contract.name }); if (this.ready) void this.send(id); } private async send(id: string): Promise { try { await withTimeout(this.hub.invoke('SubscribeContractQuotes', id), INVOKE_TIMEOUT_MS, `subscribing ${id}`); } catch (err) { if (!this.stopped) this.failed ??= errorMessage(err); } } stop(): void { this.stopped = true; void this.hub.stop().catch(() => { /* discarded */ }); } } // ── Tradovate: the market-data socket ─────────────────────────────────────── class TradovateFeed implements Feed { readonly platform = 'tradovate' as const; readonly startedAt = Date.now(); lastQuoteAt = 0; failed: string | null = null; private authorized = false; private stopped = false; private nextId = 2; private heartbeat: ReturnType | null = null; private readonly ws: WebSocket; private readonly byContract = new Map(); private readonly byRoot = new Map(); private readonly pending = new Map(); constructor(readonly firmId: string, readonly firmName: string, token: () => string | null) { const ws = brokerWebSocket(`${TRADOVATE_MD_WS}?r=${Math.random().toString(36).slice(2, 12)}`); this.ws = ws; ws.onopen = () => { const t = token(); if (!t) this.failed ??= 'no market-data token'; else this.send(`authorize\n1\n\n${t}`); }; ws.onmessage = (event) => this.onFrame(event.data); ws.onerror = () => { /* onclose follows */ }; ws.onclose = (event) => { if (this.heartbeat) clearInterval(this.heartbeat); this.heartbeat = null; if (!this.stopped) this.failed ??= `market-data socket closed (code ${event.code})`; }; } private send(frame: string): void { if (this.ws.readyState === 1) this.ws.send(frame); } private request(name: string): void { const id = this.nextId++; this.pending.set(id, name); this.send(`md/subscribeQuote\n${id}\n\n${JSON.stringify({ symbol: name })}`); } private onFrame(raw: unknown): void { if (this.stopped) return; const { responses, quotes } = parseTradovateFrame(raw); for (const r of responses) { if (r.id === 1) { if (r.status !== 200) { this.failed ??= `market data refused (status ${r.status})`; continue; } this.authorized = true; this.heartbeat = setInterval(() => this.send('[]'), 2500); for (const { name } of this.byRoot.values()) this.request(name); continue; } const name = this.pending.get(r.id); this.pending.delete(r.id); if (name && r.status !== 200) { const detail = typeof r.data === 'string' ? `: ${r.data.slice(0, 120)}` : ''; this.failed ??= `quotes for ${name} refused (status ${r.status}${detail})`; } } for (const q of quotes) { const sub = this.byContract.get(q.contractId); if (!sub) continue; this.lastQuoteAt = Date.now(); onQuote(sub.root, sub.name, { ...(q.bid !== undefined ? { bid: q.bid } : {}), ...(q.ask !== undefined ? { ask: q.ask } : {}) }); } } contractOf(root: string): string | null { const sub = this.byRoot.get(root); return sub ? String(sub.id) : null; } subscribe(root: string, contract: Contract): void { const id = Number(contract.id); if (!Number.isFinite(id)) return; const old = this.byRoot.get(root); if (old && old.id !== id) { this.byContract.delete(old.id); if (this.authorized) this.send(`md/unsubscribeQuote\n${this.nextId++}\n\n${JSON.stringify({ symbol: old.name })}`); } this.byRoot.set(root, { id, name: contract.name }); this.byContract.set(id, { root, name: contract.name }); if (this.authorized) this.request(contract.name); } stop(): void { this.stopped = true; if (this.heartbeat) clearInterval(this.heartbeat); this.heartbeat = null; try { this.ws.close(); } catch { /* already closed */ } } } // ── supervision ───────────────────────────────────────────────────────────── export interface QuoteCandidate { firmId: string; firmName: string; platform: string; client: QuoteClient; } /** * Keep one feed running and subscribed to `roots`: drop one that failed, * lost its client or went quiet while the market trades; start one on the * first client that can (TopstepX first, then Tradovate) and hasn't just * been dropped; subscribe each root's current front month. */ export async function superviseQuotes(candidates: QuoteCandidate[], roots: readonly string[], now = Date.now()): Promise { const st = state(); if (st.supervising) return; st.supervising = true; try { const marketOpen = !isNoTradeTime(new Date(now), { everyDay: false, mode: 'full_cme', jitterMin: 0 }); let feed = st.feed; if (feed) { const owner = candidates.find((c) => c.firmId === feed!.firmId); const silentMs = now - Math.max(feed.lastQuoteAt, feed.startedAt); const reason = feed.failed ?? (!owner ? 'its firm is gone' : !owner.client.quoteToken?.() ? 'its client is logged out' : marketOpen && silentMs > SILENT_MS ? `no quotes for ${Math.round(silentMs / 60_000)} min` : null); if (reason) { console.warn(`[quotes] dropping the ${feed.platform} feed on ${feed.firmName} — ${reason}`); feed.stop(); st.droppedAt.set(feed.firmId, now); st.lastError = `${feed.firmName}: ${reason}`; st.feed = feed = null; } } if (!feed) { const byPreference = [ ...candidates.filter((c) => c.platform === 'topstepx'), ...candidates.filter((c) => c.platform === 'tradovate'), ]; const pick = byPreference.find((c) => !!c.client.quoteToken?.() && now - (st.droppedAt.get(c.firmId) ?? -Infinity) >= RETRY_AFTER_MS); if (!pick) return; const token = () => pick.client.quoteToken?.() ?? null; feed = pick.platform === 'topstepx' ? new TopstepxFeed(pick.firmId, pick.firmName, token) : new TradovateFeed(pick.firmId, pick.firmName, token); st.feed = feed; console.log(`[quotes] live quotes from ${pick.firmName} (${pick.platform}) for ${roots.join(', ')}`); } const client = candidates.find((c) => c.firmId === feed!.firmId)?.client; if (!client) return; for (const root of roots) { if (!CME_SPECS[root]) continue; try { const contract = await client.findFrontMonthContract(root); if (contract && feed.contractOf(root) !== String(contract.id)) feed.subscribe(root, contract); } catch (err) { console.warn(`[quotes] ${root}: contract lookup failed — ${errorMessage(err)}`); } } } finally { st.supervising = false; } } // ── reading ───────────────────────────────────────────────────────────────── export interface LiveSpread { dollars: number; ticks: number; bid: number; ask: number; ageMs: number; contract: string; } /** `root`'s spread off a fresh quote; null when there is none. */ export function liveSpread(root: string, now = Date.now()): LiveSpread | null { const book = state().books.get(root); const spec = CME_SPECS[root]; if (!book || !spec || now - book.at > QUOTE_FRESH_MS) return null; const spread = bookSpread(book, spec.pointValue, spec.tickSize); if (!spread) return null; return { ...spread, bid: book.bid!, ask: book.ask!, ageMs: now - book.at, contract: book.contract }; } /** * The spread an entry on `root` pays, per contract: the live one when a * fresh quote is at hand, else the typical one tick. Never under a tick — * a locked book is momentary. */ export function entrySpread(root: string, now = Date.now()): { dollars: number; ticks: number; live: boolean } { const live = liveSpread(root, now); if (live) return { dollars: Math.max(live.dollars, tickValue(root)), ticks: Math.max(live.ticks, 1), live: true }; return { dollars: spreadDollars(root), ticks: CME_SPECS[root]?.spreadTicks ?? 1, live: false }; } /** The feed as the Settings page and /api/instruments show it. */ export function quoteStatus(now = Date.now()): { source: string | null; platform: Platform | null; since: string | null; lastQuoteAt: string | null; lastError: string | null; spreads: Record; } { const st = state(); const feed = st.feed; return { source: feed?.firmName ?? null, platform: feed?.platform ?? null, since: feed ? new Date(feed.startedAt).toISOString() : null, lastQuoteAt: feed?.lastQuoteAt ? new Date(feed.lastQuoteAt).toISOString() : null, lastError: feed?.failed ?? st.lastError, spreads: Object.fromEntries([...st.books.keys()].map((root) => [root, liveSpread(root, now)])), }; } /** The firm the feed runs on, for the master's stats; null with no feed. */ export function quoteSource(): string | null { const feed = state().feed; return feed ? `${feed.firmName} (${feed.platform})` : null; }