392 lines
16 KiB
TypeScript
392 lines
16 KiB
TypeScript
/**
|
|
* 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<Contract | null>;
|
|
}
|
|
|
|
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<string, LiveBook>;
|
|
feed: Feed | null;
|
|
droppedAt: Map<string, number>;
|
|
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<T>(p: Promise<T>, ms: number, what: string): Promise<T> {
|
|
return new Promise<T>((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<string, { root: string; name: string }>();
|
|
private readonly byRoot = new Map<string, string>();
|
|
|
|
constructor(readonly firmId: string, readonly firmName: string, token: () => string | null) {
|
|
// Like the user hub: one websocket, through the operator's proxy.
|
|
const configuredWs = signalrWebSocket();
|
|
this.hub = new signalR.HubConnectionBuilder()
|
|
.withUrl(MARKET_HUB, {
|
|
skipNegotiation: true,
|
|
transport: signalR.HttpTransportType.WebSockets,
|
|
accessTokenFactory: () => token() ?? '',
|
|
...{ WebSocket: configuredWs },
|
|
})
|
|
.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<void> {
|
|
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<typeof setInterval> | null = null;
|
|
private readonly ws: WebSocket;
|
|
private readonly byContract = new Map<number, { root: string; name: string }>();
|
|
private readonly byRoot = new Map<string, { id: number; name: string }>();
|
|
private readonly pending = new Map<number, string>();
|
|
|
|
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<void> {
|
|
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<string, LiveSpread | null>;
|
|
} {
|
|
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;
|
|
}
|