Live futures spreads off one client, and the master's max entry spread
One quote feed per machine: a TopstepX client's market hub, else a Tradovate client's market-data socket, subscribed to every listed root's front month and the drain micro. A supervisor keeps it on a working client, dropping one that fails or goes quiet while the market trades. The Auto draw weighs each instrument by its live spread (the typical tick when no fresh quote is at hand). No futures entry pays more than the master's max entry spread ($15 until set) to cross the book: the draw leaves such an instrument out, an explicit symbol sits the tick out, and fireLeg and the extra-day trade check it last before sending. The Settings page and the master's stats show the live spreads and the feed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Brandon Li
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
1782c16145
commit
db1d12327a
@@ -0,0 +1,392 @@
|
||||
/**
|
||||
* 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 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<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;
|
||||
}
|
||||
Reference in New Issue
Block a user