// Create a class with a constructor that takes a username and password and sets the credentials import axios from 'axios'; import type { AccountItem, AuthLoginResponse, Contract } from './tradovate-helpers'; import { computeSec, randomUUIDV4 } from './tradovate-helpers'; import { POINT_VALUES, currentTradeDateCT } from './trading-logic'; import { getCachedContract, resolveContracts } from './contract-resolver'; import { saveDailyPnL, loadDailyPnL, saveAccountMeta, loadAccountMeta, saveFundTransactions, loadFundTransactions, saveEndedCycleStart } from './db'; import { brokerWebSocket } from './outbound-proxy'; import { FILLS_FALLBACK_NOTE, OrderNotPlacedError, SYNC_STALE_MS, errorMessage, type ClientHealth } from './platform-client'; /** * Entity types Tradovate sends over the user-sync socket. Keep this as the * single source of truth: the inline socket-payload type and the indirect * callback list previously declared it separately and fell out of sync, which * made the 'position' and 'cashBalance' handlers unreachable to the compiler. */ type TradovateEntityType = | 'order' | 'orderVersion' | 'auditUserAction' | 'command' | 'commandReport' | 'fill' | 'executionReport' | 'position' | 'cashBalance'; /** * What auth/accesstokenrequest can answer. HTTP 200 is not success on its * own: a rate-limited login comes back 200 with no token and a penalty ticket * instead (p-ticket, plus p-time in seconds to wait before presenting it, and * p-captcha when Tradovate wants a human), and a refused one with errorText. */ type LoginResponseBody = Partial & { 'p-ticket'?: string; 'p-time'?: number; 'p-captcha'?: boolean; errorText?: string; }; type LoginOutcome = | { ok: true; info: AuthLoginResponse } | { ok: false; error: string; retryInMs?: number }; /** A listener for pushed entity events. One-shot: it is removed once it fires. * `cancel` (sendOrder's fill wait) rejects the waiter when the client is * discarded, so no caller is left hanging on a dead instance. */ type IndirectCallback = { entityType: TradovateEntityType; eventType: 'Created' | 'Updated'; // Since the entity is not always the same, we need a validator to check if the response is the one we are looking for validator: (response: any) => boolean; callback: (response: any) => void; cancel?: (reason: string) => void; }; /* * Connection supervision. * * This client used to be `login().then(...)` with exactly one retry scheduled * from ws.onclose. One failed re-login — weekend maintenance, a network blip, * a rate-limit penalty answered as HTTP 200 without a token — left it dead for * the life of the process, with syncComplete frozen, and every firm on the * instance waited on it until someone ran a pm2 restart. A session could also * die without onclose ever firing (a socket stuck CLOSING, sync answers that * just stop), and an order sent on a closed socket waited forever. * * Now: every failure path ends in scheduleReconnect(); a 15s watchdog notices * sockets that went quiet, never authorized, or got stuck, and re-arms a * reconnect if nothing is pending; nothing waits on the broker without a * deadline. */ /** Reconnect after a socket drop on a still-good session. */ const SOCKET_DROP_RECONNECT_MS = 5_000; /** Reconnect when the session itself is suspect — straight to a fresh login. */ const FORCED_RELOGIN_DELAY_MS = 1_000; /** Login failures back off 5s, 10s, 20s, … capped at 5 min. */ const BACKOFF_BASE_MS = 5_000; const BACKOFF_MAX_MS = 5 * 60_000; /** A captcha demand means we've logged in too often — go quiet for a long while. */ const CAPTCHA_RETRY_MS = 15 * 60_000; const PENALTY_MAX_WAIT_MS = 5 * 60_000; const MAX_PENALTY_ROUNDS = 2; /** Deadline for login / token renewal / contract lookups over REST. */ const HTTP_TIMEOUT_MS = 15_000; /** Report calls answer with a p-ticket to poll when they're slow, so a single * HTTP call taking this long is a hung connection, not a big report. */ const REPORT_TIMEOUT_MS = 60_000; const LIVENESS_CHECK_MS = 15_000; const HANDSHAKE_TIMEOUT_MS = 30_000; const AUTHORIZE_TIMEOUT_MS = 30_000; /** Tradovate heartbeats every few seconds and we sync every 60s — this much * silence means the session is gone even if the socket says OPEN. */ const INBOUND_SILENCE_MS = 150_000; const STUCK_CLOSING_MS = 15_000; const TOKEN_RENEW_CHECK_MS = 2 * 60_000; const TOKEN_RENEW_WINDOW_MS = 15 * 60_000; /** A failed renewal this close to expiry logs in afresh instead of waiting. */ const TOKEN_RELOGIN_WINDOW_MS = 2 * 60_000; /** A reconnect reuses the current token only if it has at least this long left. */ const TOKEN_REUSE_MIN_MS = 5 * 60_000; /** An order ack later than this is treated as lost (ambiguous — it may have been placed). */ const ACK_TIMEOUT_MS = 15_000; const FILL_TIMEOUT_MS = 30_000; /** How far the start-of-day balance may move without a fill before the * account's history is treated as overtaken (balanceDrifted). Fills move * amount and realizedPnL together, so this only has to absorb rounding; * a payout is hundreds. */ const FUND_DRIFT_TOLERANCE = 25; const CONTRACT_FIND_TIMEOUT_MS = 15_000; /** On-demand front-month resolve (several Yahoo lookups, then Tradovate). */ const CONTRACT_RESOLVE_DEADLINE_MS = 30_000; const READY_STATE_NAMES = ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED']; function reconnectBackoffMs(failures: number): number { return Math.min(BACKOFF_BASE_MS * 2 ** Math.max(0, failures - 1), BACKOFF_MAX_MS); } /** A Tradovate cashBalance entity, as the sync snapshot and socket carry it. */ interface CashBalanceEntity { accountId: number; amount: number; realizedPnL: number; tradeDate?: { year: number; month: number; day: number } | null; archived?: boolean; } /** * An account's cash as the plan must read it. Tradovate rolls each account's * cash record to the new trade date on its own schedule — archiving the old * record and opening one with realizedPnL 0 — well after the 17:00 CT roll * (18 minutes on 2026-09-24), or not until the account first trades. Until * then the latest record still carries the ended day's realizedPnL, which the * plan would read as today's: legs sized off yesterday, one-and-done accounts * read as already traded, the entry-commission delta swallowing a whole day's * P&L. That day is over, and its P&L is already in amount: nothing is * realized today yet. */ function todaysCash(cb: CashBalanceEntity): { amount: number; realizedPnL: number; tradeDate: string | null; archived: boolean } { const t = cb.tradeDate; const tradeDate = t?.year ? `${t.year}-${String(t.month).padStart(2, '0')}-${String(t.day).padStart(2, '0')}` : null; const recorded = cb.realizedPnL; return { amount: cb.amount, tradeDate, archived: cb.archived === true, // Judged when read, not when stored: a record stored at 16:59 has // ended by 17:01, whether or not a sync has come in since. get realizedPnL(): number { return tradeDate !== null && tradeDate < currentTradeDateCT() ? 0 : recorded; }, }; } /** * Store a cash record unless the one already held is newer. At the roll the * snapshot and the socket deliver the ended day's record (archived) and the * new one together, in either order; the older must never win back. */ function storeCash( into: { [accountId: number]: { amount: number; realizedPnL: number; tradeDate?: string | null; archived?: boolean } }, cb: CashBalanceEntity, ): void { const next = todaysCash(cb); const prev = into[cb.accountId]; if (prev?.tradeDate && next.tradeDate) { if (next.tradeDate < prev.tradeDate) return; if (next.tradeDate === prev.tradeDate && next.archived && !prev.archived) return; } into[cb.accountId] = next; } /** ±20%, so several clients on one IP that failed together don't retry in lockstep. */ function jitter(ms: number): number { return Math.round(ms * (0.8 + Math.random() * 0.4)); } function fmtDelay(ms: number): string { return ms >= 60_000 ? `${(ms / 60_000).toFixed(1)}m` : `${(ms / 1000).toFixed(1)}s`; } /** Settles like `work`, or rejects once `ms` pass first. The work itself * carries on — only the wait for it is cut short. */ function withDeadline(work: Promise, ms: number, what: string): Promise { let timer: ReturnType | undefined; const deadline = new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${what} took longer than ${ms / 1000}s`)), ms); }); return Promise.race([work, deadline]).finally(() => clearTimeout(timer)); } export class TradovateClient { private name: string; private password: string; private accessInfo!: AuthLoginResponse; private deviceId = randomUUIDV4(); public accountList: AccountItem[] = []; public accountCashBalances: { [accountId: number]: { amount: number; /** Today's realized P&L — see todaysCash. */ realizedPnL: number; /** The trade date of the Tradovate record it came from (YYYY-MM-DD). */ tradeDate?: string | null; /** That record was superseded at a roll. */ archived?: boolean; }; } = {}; public positions: { [accountId: number]: { contractId: number; netPos: number; netPrice: number; timestamp: Date; }; } = {}; public daysTraded: { [accountId: number]: number } = {}; public dailyPnL: { [accountId: number]: { date: string; pnl: number }[] } = {}; /** Full P&L history (all cycles) — used for calendar display */ public fullDailyPnL: { [accountId: number]: { date: string; pnl: number }[] } = {}; /** Date of the last fund transaction per account — days traded are counted from this date onwards */ public lastFundDates: { [accountId: number]: string | null } = {}; /** All fund transactions (deposits/withdrawals) per account */ public fundTransactions: { [accountId: number]: { date: string; amount: number }[] } = {}; /** Sum of daily P&L from before the last fund transaction — used for "Target Same Equity" mode */ public priorProfit: { [accountId: number]: number } = {}; /** Balance floor set by the prop firm — account is blown when amount <= this value (0 = not set) */ public autoLiqThresholds: { [accountId: number]: number } = {}; /** True once fetchDaysTraded() has finished its last full run */ public fetchDaysComplete = false; /** Prevent overlapping report fetches and throttle them to hourly cadence with retry backoff. */ private daysFetchInFlight = false; private lastDaysFetchStartedAt = 0; private lastDaysFetchSucceededAt = 0; private nextDaysFetchRetryAt = 0; /** Accounts whose history this process has loaded: a report came back, or * failed with a DB cache to fall back on. Until then the plan would read * a fresh stage-1 cycle, so historyReady() keeps them out of trading. */ private historyLoadedIds = new Set(); /** Accounts whose balance moved in a way the loaded history can't explain * (a payout posting between the hourly reports): held out of trading * until a Cash History books it. */ private historyRefreshIds = new Set(); /** Per account, the start-of-day balance (amount − realizedPnL) the loaded * history was taken at, and the day's realized P&L as last seen — what the * daily reset folds into the balance. */ private balanceBaseline: { [accountId: number]: { dayStart: number; realized: number } } = {}; /** Interval handles tracked so they can be cleared on reconnect */ private syncInterval: ReturnType | null = null; private heartbeatInterval: ReturnType | null = null; private tokenRenewalInterval: ReturnType | null = null; /** Set to true by disconnect() to suppress reconnect on close */ private intentionalDisconnect = false; /** Last error per account name from fetchDaysTraded() */ public lastFetchErrors: Record = {}; /** Raw reports API response sample per account (first 100 chars) for debugging */ public lastFetchRaw: Record = {}; public products: { id: number; name: string }[] = []; /** Rolling buffer of the last 50 raw entity events — useful for debugging */ public recentEntityEvents: { entityType: string; eventType: string; entity: any; ts: number }[] = []; /** True once the first requestSync has completed and positions/balances are populated. */ public syncComplete = false; /** Epoch ms of the last sync that came back good (0 = never). */ public lastSyncAt = 0; // ── Connection supervisor state (see the comment above the constants) ── /** Single-flight guard: true while connectAndAuth() is logging in / opening. */ private connecting = false; private reconnectTimer: ReturnType | null = null; /** When reconnectTimer (or a login penalty wait) fires; 0 = nothing scheduled. */ private nextReconnectAt = 0; private consecutiveLoginFailures = 0; /** Sessions in a row that died before producing a good sync (never * opened, never authorized, sync refused) — they back off like login * failures instead of reconnecting (or re-logging in) every few seconds. */ private consecutiveSocketFailures = 0; /** Last successful login. */ private lastLoginAt = 0; private lastLoginError: string | null = null; /** The current token is suspect (refused, never authorized a socket, sync * answers stopped): the next connect logs in instead of reusing it. */ private needsLogin = true; /** Socket creation time; reset to the open time once it opens. */ private socketOpenedAt = 0; /** The current socket got as far as presenting the token. */ private authorizeSent = false; /** When the current socket's authorize was accepted (0 = not yet). */ private authorizedAt = 0; private lastInboundAt = 0; /** First time the watchdog saw the current socket CLOSING/CLOSED. */ private closingSeenAt = 0; /** When the oldest still-unanswered syncrequest was sent (0 = none). */ private syncRequestedAt = 0; private livenessInterval: ReturnType | null = null; private renewInFlight = false; /** The current token's expirationTime was (nearly) past on arrival: this * machine's clock is off, so the expiry is ignored (noteTokenExpiry()). */ private tokenExpiryUntrusted = false; /** A login rate-limit penalty wait in progress — disconnect() cuts it short. */ private penaltyWait: { timer: ReturnType; resolve: () => void } | null = null; private ws: WebSocket | null = null; private callbackOnSyncRequest: () => Promise; /** Incrementing ID for outgoing WebSocket messages — ensures concurrent orders don't clobber each other's callbacks. */ private nextMsgId = 100; private getMsgId(): number { return this.nextMsgId++; } // Events that we sent out, and tradovate gives us a response for the id we sent out. // One-shot: removed when the answer arrives, and all dropped with the socket. private directEventCallbacks: { [id: number]: (response: any, status?: number) => void; } = {}; /** Rejecters for awaited direct requests (acks, contract finds), so a socket * going away fails them instead of leaving the caller waiting forever. */ private pendingRejecters: { [id: number]: (reason: string) => void; } = {}; private indirectEventCallbacks: IndirectCallback[] = []; constructor( name: string, password: string, callbackOnSyncRequest: () => Promise ) { this.name = name; this.password = password; this.callbackOnSyncRequest = callbackOnSyncRequest; // One watchdog and one renewal check for the life of the client — // they span reconnects, and only disconnect() stops them. this.livenessInterval = setInterval(() => this.checkLiveness(), LIVENESS_CHECK_MS); this.tokenRenewalInterval = setInterval(() => this.maybeRenewToken(), TOKEN_RENEW_CHECK_MS); void this.connectAndAuth(); } /** Log prefix — several Tradovate firms can share one instance's log. */ private get tag(): string { return `[tradovate ${this.name}]`; } /** * Logs in (unless the current token is still good) and opens a fresh * socket. Single-flight, never throws, never leaves a rejection unhandled: * every way it can fail ends in scheduleReconnect(). */ private async connectAndAuth(): Promise { if (this.connecting || this.intentionalDisconnect) return; this.connecting = true; try { // A reconnect after a plain socket drop reuses a token with life // left in it — fewer logins, fewer rate-limit penalties while // Tradovate flaps during maintenance. Anything that makes the token // suspect sets needsLogin. if (this.needsLogin || !this.tokenValidFor(TOKEN_REUSE_MIN_MS)) { const outcome = await this.attemptLogin(); if (this.intentionalDisconnect) return; if (!outcome.ok) { this.consecutiveLoginFailures++; this.lastLoginError = outcome.error; this.scheduleReconnect( `login failed: ${outcome.error} (${this.consecutiveLoginFailures} in a row)`, outcome.retryInMs ?? reconnectBackoffMs(this.consecutiveLoginFailures), ); return; } this.accessInfo = outcome.info; this.noteTokenExpiry(); this.lastLoginAt = Date.now(); this.lastLoginError = null; this.consecutiveLoginFailures = 0; this.needsLogin = false; console.log(`${this.tag} logged in — token expires ${outcome.info.expirationTime}`); } this.openSocket(); } catch (err) { // Nothing above is expected to throw (brokerWebSocket can, on a bad // proxy setting) — but a throw must never end the reconnect loop. this.consecutiveSocketFailures++; this.scheduleReconnect(`connect failed: ${errorMessage(err)}`, reconnectBackoffMs(this.consecutiveSocketFailures)); } finally { this.connecting = false; } } /** * One login attempt, riding out up to MAX_PENALTY_ROUNDS rate-limit * penalties: Tradovate answers a throttled login with a p-ticket to present * again after p-time seconds. p-captcha means it wants a human — never try * to solve it, back off for a long while instead. */ private async attemptLogin(): Promise { let pTicket: string | undefined; for (let round = 0; ; round++) { let body: LoginResponseBody | undefined; try { body = await this.login(pTicket); } catch (err) { // axios rejects every non-2xx (and timeouts); a penalty ticket // could ride on one of those too. const data = (err as { response?: { data?: unknown } })?.response?.data as LoginResponseBody | undefined; if (data && typeof data === 'object' && data['p-ticket']) { body = data; } else { const text = data && typeof data === 'object' && typeof data.errorText === 'string' ? `: ${data.errorText}` : ''; return { ok: false, error: `${errorMessage(err)}${text}` }; } } if (this.intentionalDisconnect) return { ok: false, error: 'client disconnected' }; if (body?.accessToken) return { ok: true, info: body as AuthLoginResponse }; if (body?.['p-captcha']) { return { ok: false, error: 'Tradovate asked for a captcha (too many logins)', retryInMs: CAPTCHA_RETRY_MS }; } const ticket = body?.['p-ticket']; const pTime = Number(body?.['p-time']); if (ticket && pTime > 0) { if (round >= MAX_PENALTY_ROUNDS) { return { ok: false, error: `still rate-limited after ${round} penalty waits` }; } const waitMs = Math.min(pTime * 1000, PENALTY_MAX_WAIT_MS); console.warn(`${this.tag} login rate-limited — waiting ${fmtDelay(waitMs)} to retry with the penalty ticket`); pTicket = ticket; await this.penaltySleep(waitMs); if (this.intentionalDisconnect) return { ok: false, error: 'client disconnected' }; continue; } return { ok: false, error: body?.errorText ? String(body.errorText) : 'login answered without an access token' }; } } private penaltySleep(ms: number): Promise { return new Promise((resolve) => { this.nextReconnectAt = Date.now() + ms; const timer = setTimeout(() => { this.penaltyWait = null; this.nextReconnectAt = 0; resolve(); }, ms); this.penaltyWait = { timer, resolve }; }); } /** * Arms the one reconnect timer. Never stacks: if one is already pending it * stays as it is (a login backoff isn't cut short by a later socket event). */ private scheduleReconnect(reason: string, delayMs = SOCKET_DROP_RECONNECT_MS): void { if (this.intentionalDisconnect) return; if (this.reconnectTimer) { console.warn(`${this.tag} ${reason} — reconnect already due in ${fmtDelay(Math.max(0, this.nextReconnectAt - Date.now()))}`); return; } const wait = jitter(delayMs); this.nextReconnectAt = Date.now() + wait; console.warn(`${this.tag} ${reason} — reconnecting in ${fmtDelay(wait)}`); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.nextReconnectAt = 0; void this.connectAndAuth(); }, wait); } private openSocket(): void { // Only ever one live socket per client — a leftover would keep feeding // callbacks and heartbeats alongside the new one. if (this.ws) this.dropSocket('replaced by a new connection'); const randomnumber = Math.random().toString(36).substring(2, 15); // Broker endpoint — dialled through the operator's proxy when one // is configured (lib/outbound-proxy.ts). const ws = brokerWebSocket(`wss://demo.tradovateapi.com/v1/websocket?r=${randomnumber}`); this.ws = ws; this.socketOpenedAt = Date.now(); this.authorizeSent = false; this.authorizedAt = 0; this.closingSeenAt = 0; // Reset sync state so scheduler waits for the new sync to complete this.syncComplete = false; // Every handler first checks it still belongs to the current socket: // a replaced socket's late events must not touch the new session. ws.onopen = () => { if (this.ws !== ws) return; this.socketOpenedAt = Date.now(); // the authorize deadline runs from here console.log(`${this.tag} websocket connected — authorizing`); this.directEventCallbacks[2] = (_response: unknown, status?: number) => this.onAuthorized(ws, status); if (this.sendFrame('authorize\n2\n\n' + this.accessInfo.accessToken)) { this.authorizeSent = true; } else { this.abandonSocket('authorize could not be sent'); } }; ws.onmessage = (event) => { if (this.ws !== ws) return; // Any frame ('o', 'h', 'a', …) proves the session is alive. this.lastInboundAt = Date.now(); this.handleFrame(event.data); }; ws.onerror = (event) => { if (this.ws !== ws) return; // onclose follows and does the recovery; just note it (no event dump). const msg = 'message' in event ? (event as ErrorEvent).message : ''; console.warn(`${this.tag} websocket error${typeof msg === 'string' && msg ? `: ${msg}` : ''}`); }; ws.onclose = (event) => { if (this.ws !== ws) return; this.abandonSocket(`websocket closed (code=${event.code})`); }; } private onAuthorized(ws: WebSocket, status?: number): void { if (this.ws !== ws) return; if (typeof status === 'number' && status !== 200) { // The token was refused — another socket with it would be refused too. this.abandonSocket(`websocket authorize refused (status ${status})`, true); return; } this.authorizedAt = Date.now(); console.log(`${this.tag} websocket authorized — syncing`); this.clearSessionIntervals(); this.requestSync(); this.syncInterval = setInterval(() => { if (this.ws === ws) this.requestSync(); }, 60000); // Every 2.5 seconds send a heartbeat this.heartbeatInterval = setInterval(() => { if (this.ws === ws) this.sendFrame('[]'); }, 2500); } private handleFrame(raw: unknown): void { // 'o' open, 'h' heartbeat, 'c' close — nothing to do beyond liveness. if (typeof raw !== 'string' || raw[0] !== 'a') return; let data: ( | { i?: number; // Id s?: number; // Status d?: any; // Data } | { e?: string; d?: { entityType: TradovateEntityType; eventType: 'Created' | 'Updated'; entity: any; }; } )[]; try { data = JSON.parse(raw.slice(1)); } catch { console.warn(`${this.tag} unparseable frame (${raw.length} bytes) ignored`); return; } if (!Array.isArray(data)) return; for (const response of data) { if (!response || typeof response !== 'object') continue; if ( 'i' in response && response.i && this.directEventCallbacks[response.i] ) { // One-shot: removed before it runs, so a handler that re-registers // its id (the fixed ids 2/3/10/30 are per request) isn't undone. const id = response.i; const callback = this.directEventCallbacks[id]; delete this.directEventCallbacks[id]; delete this.pendingRejecters[id]; try { callback(response.d, response.s); } catch (err) { console.error(`${this.tag} handler for request ${id} failed: ${errorMessage(err)}`); } } if ('e' in response && response.e) { // Handle callbacks — all one-shot (fill waits), so take them out // of the list before running them. const matched = this.indirectEventCallbacks.filter((info) => response.d?.entityType === info.entityType && response.d?.eventType === info.eventType && info.validator(response.d.entity) ); if (matched.length > 0) { this.indirectEventCallbacks = this.indirectEventCallbacks.filter((info) => !matched.includes(info)); for (const info of matched) { try { info.callback(response.d?.entity); } catch (err) { console.error(`${this.tag} ${info.entityType} handler failed: ${errorMessage(err)}`); } } } // Buffer recent entity events (last 50) if (response.d?.entityType) { this.recentEntityEvents.push({ entityType: response.d.entityType, eventType: response.d.eventType, entity: response.d.entity, ts: Date.now() }); if (this.recentEntityEvents.length > 50) this.recentEntityEvents.shift(); } // Update positions from WebSocket position events if (response.d?.entityType === 'position' && response.d?.entity) { const pos = response.d.entity; if (pos.netPos !== 0) { this.positions[pos.accountId] = { contractId: pos.contractId, netPos: pos.netPos, netPrice: pos.netPrice, timestamp: new Date(pos.timestamp), }; } else { delete this.positions[pos.accountId]; } } // Update cash balances from WebSocket cashBalance events if (response.d?.entityType === 'cashBalance' && response.d?.entity) { const cb = response.d.entity as CashBalanceEntity; if (cb.accountId) storeCash(this.accountCashBalances, cb); } } } } /** * Gives up on the current socket (if any) and schedules the way back. * Used for every way a session ends — onclose, and the watchdog for the * ways that never fire onclose. `relogin` means the token itself is * suspect: log in afresh, after a short delay. */ private abandonSocket(reason: string, relogin = false): void { if (this.intentionalDisconnect) return; if (relogin) this.needsLogin = true; // A connect in flight opens a fresh socket anyway (and logs in first // if needsLogin was just set before its login) — don't race it. if (this.connecting) return; const hadSocket = this.ws !== null; const wasAuthorized = this.authorizedAt > 0; const tokenPresented = this.authorizeSent; // lastSyncAt only moves on a good sync, and one on this socket can't // predate its open. const synced = wasAuthorized && this.lastSyncAt >= this.socketOpenedAt; this.dropSocket(reason); let delay = relogin ? FORCED_RELOGIN_DELAY_MS : SOCKET_DROP_RECONNECT_MS; if (hadSocket && !synced) { // A session that died before it produced a single good sync backs // off if it keeps happening, rather than reconnecting every few // seconds — and, when it forces a login (a sync refused with 401 // on every fresh session), rather than logging in every second. // If it presented the token and got no yes, the token is suspect // too: log in afresh. A handshake that never completed says // nothing about the token — keep it, so a websocket-host outage // doesn't turn into a login every few seconds (and a rate-limit // penalty on top). if (tokenPresented && !wasAuthorized) this.needsLogin = true; this.consecutiveSocketFailures++; if (this.consecutiveSocketFailures > 1) delay = reconnectBackoffMs(this.consecutiveSocketFailures); } this.scheduleReconnect(reason, delay); } /** Detaches and closes the current socket and clears everything that * belonged to its session. Pending requests fail with `reason`. */ private dropSocket(reason: string): void { const ws = this.ws; this.ws = null; if (ws) { ws.onopen = null; ws.onmessage = null; ws.onerror = null; ws.onclose = null; try { ws.close(); } catch { /* already closing */ } } this.clearSessionIntervals(); this.syncComplete = false; this.authorizeSent = false; this.authorizedAt = 0; this.closingSeenAt = 0; this.syncRequestedAt = 0; this.failPendingRequests(reason); } private clearSessionIntervals(): void { if (this.syncInterval) { clearInterval(this.syncInterval); this.syncInterval = null; } if (this.heartbeatInterval) { clearInterval(this.heartbeatInterval); this.heartbeatInterval = null; } } /** Answers to requests sent on a socket can't arrive on the next one: * reject the awaited ones, drop the internal ones (authorize, sync, * product list — re-sent on the new socket). Fill watchers stay: fills * pushed after the reconnect still arrive. */ private failPendingRequests(reason: string): void { const rejecters = Object.values(this.pendingRejecters); this.pendingRejecters = {}; this.directEventCallbacks = {}; for (const reject of rejecters) reject(reason); } /** Every 15s: catch the ways a session dies without onclose, and make sure * something is always pending while the client is meant to be connected. */ private checkLiveness(): void { // A connect in flight is bounded (HTTP deadlines, capped penalty waits). if (this.intentionalDisconnect || this.connecting) return; try { const problem = this.livenessProblem(Date.now()); if (problem) { this.abandonSocket(problem.reason, problem.relogin); return; } if (!this.ws && !this.reconnectTimer) this.scheduleReconnect('nothing scheduled'); } catch (err) { console.error(`${this.tag} liveness check failed: ${errorMessage(err)}`); } } private livenessProblem(now: number): { reason: string; relogin?: boolean } | null { const ws = this.ws; if (!ws) return null; const state = ws.readyState; if (state === WebSocket.CLOSING || state === WebSocket.CLOSED) { // onclose normally takes it from here — unless it never fires. if (!this.closingSeenAt) this.closingSeenAt = now; else if (now - this.closingSeenAt >= STUCK_CLOSING_MS) return { reason: 'websocket stuck closing' }; return null; } this.closingSeenAt = 0; // Renewal normally keeps this from happening; a session on an expired // token is dead whatever the socket says. const expiresAt = this.tokenExpiresAt(); if (expiresAt > 0 && expiresAt <= now) return { reason: 'access token expired', relogin: true }; if (state === WebSocket.CONNECTING) { return now - this.socketOpenedAt > HANDSHAKE_TIMEOUT_MS ? { reason: 'websocket handshake timed out' } : null; } if (!this.authorizedAt) { return now - this.socketOpenedAt > AUTHORIZE_TIMEOUT_MS ? { reason: 'websocket authorize never answered', relogin: true } : null; } if (now - Math.max(this.lastInboundAt, this.authorizedAt) > INBOUND_SILENCE_MS) { return { reason: `no frames for ${INBOUND_SILENCE_MS / 1000}s` }; } if (now - this.authorizedAt > SYNC_STALE_MS && now - this.lastSyncAt > SYNC_STALE_MS) { return { reason: 'sync responses stopped', relogin: true }; } return null; } /** Sends a frame on the current socket if it is open. Never throws. */ private sendFrame(frame: string): boolean { const ws = this.ws; if (!ws || ws.readyState !== WebSocket.OPEN) return false; try { ws.send(frame); return true; } catch (err) { console.warn(`${this.tag} websocket send failed: ${errorMessage(err)}`); return false; } } /** Why `ws` can't carry a request right now, or null if it can. */ private socketProblem(ws: WebSocket | null): string | null { if (!ws) return 'Tradovate socket not connected'; if (ws.readyState !== WebSocket.OPEN) return `Tradovate socket ${READY_STATE_NAMES[ws.readyState] ?? 'not open'}`; if (!this.authorizedAt) return 'Tradovate socket not authorized yet'; return null; } /** * Sends one request on the socket and settles on its answer; it can't * hang. No answer within `timeoutMs`, or the socket going away first, * rejects with a plain Error — ambiguous, the broker may have acted on it. * Failing before anything was sent rejects with OrderNotPlacedError. * `onAnswer` runs synchronously inside the frame handler, so a fill * watcher it registers still sees events later in the same frame. */ private sendRequest( endpoint: string, rest: string, timeoutMs: number, onAnswer: (data: unknown, status?: number) => T | PromiseLike, ): Promise { const ws = this.ws; const problem = this.socketProblem(ws); if (!ws || problem) return Promise.reject(new OrderNotPlacedError(`${endpoint} not sent — ${problem}`)); const id = this.getMsgId(); return new Promise((resolve, reject) => { const timer = setTimeout(() => { delete this.directEventCallbacks[id]; delete this.pendingRejecters[id]; reject(new Error(`${endpoint}: no answer from Tradovate within ${timeoutMs / 1000}s — outcome unknown`)); }, timeoutMs); this.directEventCallbacks[id] = (data: unknown, status?: number) => { clearTimeout(timer); try { resolve(onAnswer(data, status)); } catch (err) { reject(err); } }; this.pendingRejecters[id] = (reason: string) => { clearTimeout(timer); reject(new Error(`${endpoint}: ${reason} before Tradovate answered — outcome unknown`)); }; try { ws.send(`${endpoint}\n${id}\n${rest}`); } catch (err) { clearTimeout(timer); delete this.directEventCallbacks[id]; delete this.pendingRejecters[id]; reject(new OrderNotPlacedError(`${endpoint} not sent — ${errorMessage(err)}`)); } }); } /** Runs the sync-complete handler without letting it reject unhandled. */ private notifySyncComplete(): void { try { this.callbackOnSyncRequest().catch((err) => { console.error(`${this.tag} sync-complete handler failed: ${errorMessage(err)}`); }); } catch (err) { console.error(`${this.tag} sync-complete handler failed: ${errorMessage(err)}`); } } private requestSync(): void { if (this.socketProblem(this.ws)) return; this.directEventCallbacks[3] = (response: any, status?: number) => { this.syncRequestedAt = 0; // A refused or empty answer must not wipe the accounts, positions // and balances we have — nor pass for a fresh sync. if (!response || typeof response === 'string' || (typeof status === 'number' && status !== 200)) { console.error(`${this.tag} sync request failed (status ${status ?? '?'}): ${typeof response === 'string' && response ? response : 'empty response'}`); // 401: the session is gone — no point waiting for the watchdog. if (status === 401) this.abandonSocket('sync refused (status 401)', true); return; } try { // liquidateOnly flag lives in accountRiskStatuses const riskStatusById: { [accountId: number]: { liquidateOnly?: string } } = ( response.accountRiskStatuses || [] ).reduce((acc: any, item: any) => { const key = item.accountId ?? item.id; acc[key] = item; return acc; }, {}); // Auto-liquidation balance floor lives in userAccountAutoLiqs. // item.id IS the account ID. The floor is: trailingMaxDrawdownLimit - trailingMaxDrawdown. // Tradovate uses 999999999 as a sentinel for "no limit" — skip those. for (const item of (response.userAccountAutoLiqs ?? [])) { const accountId: number = item.id; const limit: number = item.trailingMaxDrawdownLimit ?? 0; const drawdown: number = item.trailingMaxDrawdown ?? 0; const isSentinel = limit >= 999999999; const floor = (!isSentinel && limit > 0 && drawdown > 0) ? limit - drawdown : 0; this.autoLiqThresholds[accountId] = floor; } this.accountList = ((response.accounts ?? []) as AccountItem[]).map((account) => { if (riskStatusById[account.id]?.liquidateOnly) { return { ...account, active: false }; } return account; }); const balances: TradovateClient['accountCashBalances'] = {}; for (const item of (response.cashBalances ?? []) as CashBalanceEntity[]) storeCash(balances, item); this.accountCashBalances = balances; this.positions = (response.positions ?? []) .filter((item: any) => item.netPos !== 0) .reduce( ( acc: { [accountId: number]: { contractId: number; netPos: number; netPrice: number; timestamp: Date } }, item: { accountId: number; contractId: number; netPos: number; netPrice: number; timestamp: Date } ) => { acc[item.accountId] = { contractId: item.contractId, netPos: item.netPos, netPrice: item.netPrice, timestamp: new Date(item.timestamp), }; return acc; }, {} as { [accountId: number]: { contractId: number; netPos: number; netPrice: number; timestamp: Date } } ); this.lastSyncAt = Date.now(); // A session counts as working once it has synced (abandonSocket). this.consecutiveSocketFailures = 0; console.log(`${this.tag} sync:${this.accountList.length} account(s), ${Object.keys(this.accountCashBalances).length} balance(s)`); // Ahead of the fetch, so a balance change flagged here is // fetched on this same pass. this.observeBalances(); this.maybeFetchDaysTraded(); // The product list only feeds fetchInstrumentFees. Wait for it // on a session's first sync, but not for ever: still unanswered // at the next sync, it is lost — sync without it (a late answer // still loads it) rather than hold syncComplete false, and every // firm on the instance with it, while the syncs themselves are good. if (this.products.length > 0 || this.directEventCallbacks[30]) { this.syncComplete = true; this.notifySyncComplete(); return; } this.directEventCallbacks[30] = (products: any) => { if (Array.isArray(products) && products.length > 0) { this.products = products.map((p: any) => ({ id: p.id, name: p.name })); console.log(`${this.tag} loaded ${this.products.length} products`); } this.syncComplete = true; this.notifySyncComplete(); }; this.sendFrame('product/list\n30\n\n'); } catch (err) { console.error(`${this.tag} error processing sync response: ${errorMessage(err)}`); } }; if (this.sendFrame('user/syncrequest\n3\n\n{"splitResponses":false}') && !this.syncRequestedAt) { this.syncRequestedAt = Date.now(); } } private maybeFetchDaysTraded(): void { if (this.daysFetchInFlight) return; const now = Date.now(); const initialFetchDue = this.lastDaysFetchStartedAt === 0; const retryPending = this.nextDaysFetchRetryAt > 0; const retryDue = retryPending && now >= this.nextDaysFetchRetryAt; const hourlyRefreshDue = !retryPending && this.lastDaysFetchSucceededAt > 0 && now - this.lastDaysFetchSucceededAt >= 60 * 60 * 1_000; // Accounts held out of trading for want of history — new since the // last pass, wiped, or flagged by a balance change — get their own // fetch now rather than wait out the hour. A failure falls back to // the 5-minute retry like any other. const waiting = !initialFetchDue && !retryPending ? this.accountList .filter((a) => !this.historyLoadedIds.has(a.id) || this.historyRefreshIds.has(a.id)) .map((a) => a.id) : []; const targeted = !retryDue && !hourlyRefreshDue && !initialFetchDue && waiting.length > 0; if (!initialFetchDue && !retryDue && !hourlyRefreshDue && !targeted) return; const reason = retryDue ? 'retry after failure' : initialFetchDue ? 'initial sync' : hourlyRefreshDue ? 'hourly refresh' : `history for ${waiting.length} waiting account(s)`; this.daysFetchInFlight = true; this.lastDaysFetchStartedAt = now; if (retryDue) this.nextDaysFetchRetryAt = 0; this.fetchDaysTraded(targeted ? waiting : undefined) .then(({ failedAccounts }) => { if (failedAccounts > 0) { this.nextDaysFetchRetryAt = Date.now() + 5 * 60 * 1_000; console.warn(`${this.tag} fetchDaysTraded: ${failedAccounts} account(s) failed during ${reason} — retrying in 5 minutes`); return; } // A partial pass doesn't restart the hourly clock. if (!targeted) this.lastDaysFetchSucceededAt = Date.now(); }) .catch((err) => { this.nextDaysFetchRetryAt = Date.now() + 5 * 60 * 1_000; // errorMessage(): an AxiosError logged whole dumps the Bearer token. console.error(`${this.tag} fetchDaysTraded ${reason} failed — retrying in 5 minutes: ${errorMessage(err)}`); }) .finally(() => { this.daysFetchInFlight = false; }); } /** * Returns true when a withdrawal happened before the first trade of the day, * meaning that day's trades belong to the NEW cycle. * Timestamp format from Tradovate: "MM/DD/YYYY HH:MM:SS" in Central Time. * Falls back to 9:00 AM CT cutoff when firstTradeTimestamp is not available. */ private static isWithdrawalBeforeFirstTrade( fundTimestamp: string | null, firstTradeTimestamp: string | null ): boolean { if (!fundTimestamp) return false; if (firstTradeTimestamp) { // Direct comparison — withdrawal before the day's first trade return fundTimestamp < firstTradeTimestamp; } // Fallback heuristic: before 9:00 AM CT const match = fundTimestamp.match(/\d{2}\/\d{2}\/\d{4}\s+(\d{2}):\d{2}:\d{2}/); if (!match) return false; return parseInt(match[1], 10) < 9; } /** * Filters daily PnL entries for the current cycle based on fund date and withdrawal timing. * - Deposit: include the fund date (trading can start same day) * - Withdrawal before first trade: include the fund date (day's trades are new cycle) * - Withdrawal during/after first trade: exclude the fund date (day's trades are old cycle) */ private static filterActivePnL( entries: { date: string; pnl: number }[], fundDate: string | null, isWithdrawal: boolean, fundTimestamp: string | null, firstTradeTimestamp: string | null = null ): { active: { date: string; pnl: number }[]; prior: { date: string; pnl: number }[] } { if (!fundDate) return { active: entries, prior: [] }; // Withdrawal before first trade → day belongs to NEW cycle (use >=) // Withdrawal during/after first trade → day belongs to OLD cycle (use >) // Deposit → always include the day (use >=) const excludeFundDate = isWithdrawal && !TradovateClient.isWithdrawalBeforeFirstTrade(fundTimestamp, firstTradeTimestamp); const active = entries.filter((d) => excludeFundDate ? d.date > fundDate : d.date >= fundDate); const prior = entries.filter((d) => excludeFundDate ? d.date <= fundDate : d.date < fundDate); return { active, prior }; } /** Pass onlyAccountIds to refresh a subset (e.g. one wiped account) without * paying for a report round-trip on every other account of the firm. */ public async fetchDaysTraded(onlyAccountIds?: number[]): Promise<{ failedAccounts: number }> { if (!this.accessInfo?.accessToken) return { failedAccounts: 0 }; this.fetchDaysComplete = false; let failedAccounts = 0; const authHeaders = { Authorization: `Bearer ${this.accessInfo.accessToken}` }; const now = new Date(); now.setDate(now.getDate() + 1); // endDate must be tomorrow — report server excludes today's fills when endDate=today const start = new Date(); start.setDate(start.getDate() - 27); // keep total window ≤ 28 days const fmtDate = (d: Date) => { const m = String(d.getMonth() + 1).padStart(2, '0'); const day = String(d.getDate()).padStart(2, '0'); return `${m}/${day}/${d.getFullYear()}`; }; if (!this.lastFetchRaw) this.lastFetchRaw = {}; if (!this.lastFetchErrors) this.lastFetchErrors = {}; // Treat 5xx and network-level errors as retryable (Tradovate report server is flaky). const isTransient = (err: unknown): boolean => { // eslint-disable-next-line @typescript-eslint/no-explicit-any const status = (err as any)?.response?.status; // eslint-disable-next-line @typescript-eslint/no-explicit-any const code = (err as any)?.code; return (typeof status === 'number' && status >= 500 && status < 600) || code === 'ECONNRESET' || code === 'ETIMEDOUT' || code === 'ECONNABORTED'; }; const requestReport = async (name: string, accountName: string) => { const maxAttempts = 3; let lastErr: unknown = null; for (let attempt = 1; attempt <= maxAttempts; attempt++) { // A discarded client stops polling rather than keep a report // conversation going on a session nobody uses any more. if (this.intentionalDisconnect) throw new Error('client disconnected'); try { let reportData = (await axios.post( 'https://rpt-demo.tradovateapi.com/v1/reports/requestreport', { name, representationType: 'json', timezone: -300, params: [ { name: 'startDate', value: fmtDate(start) }, { name: 'endDate', value: fmtDate(now) }, { name: 'startTime', value: '00:00:00' }, { name: 'endTime', value: '00:00:00' }, { name: 'account', value: accountName }, ], }, { headers: authHeaders, timeout: REPORT_TIMEOUT_MS } )).data; let pollAttempts = 0; while (reportData?.['p-ticket'] && pollAttempts < 30) { const pTicket: string = reportData['p-ticket']; const pTime: number = Math.max(1, reportData['p-time'] ?? 1); await new Promise((r) => setTimeout(r, pTime * 1000)); if (this.intentionalDisconnect) throw new Error('client disconnected'); reportData = (await axios.get( 'https://rpt-demo.tradovateapi.com/v1/reports/getreport', { params: { 'p-ticket': pTicket }, headers: authHeaders, timeout: REPORT_TIMEOUT_MS } )).data; pollAttempts++; } return typeof reportData?.data === 'string' ? reportData.data : '[]'; } catch (err) { lastErr = err; if (!isTransient(err) || attempt === maxAttempts) throw err; const backoffMs = 500 * Math.pow(2, attempt - 1); // 500ms, 1s, 2s console.warn(`${this.tag} requestReport ${accountName} ${name}: transient error (${errorMessage(err)}, attempt ${attempt}/${maxAttempts}), retrying in ${backoffMs}ms`); await new Promise((r) => setTimeout(r, backoffMs)); } } throw lastErr; }; // Merge fresh API entries on top of cached historical entries (fresh takes precedence for overlapping dates) const mergePnL = ( cached: { date: string; pnl: number }[], fresh: { date: string; pnl: number }[] ): { date: string; pnl: number }[] => { const map = new Map(); for (const e of cached) map.set(e.date, e.pnl); for (const e of fresh) map.set(e.date, e.pnl); return Array.from(map.entries()) .map(([date, pnl]) => ({ date, pnl })) .sort((a, b) => a.date.localeCompare(b.date)); }; for (const account of this.accountList) { // A discarded client (credentials changed, firm removed) stops here // rather than keep writing rows next to its replacement's. if (this.intentionalDisconnect) break; if (onlyAccountIds && !onlyAccountIds.includes(account.id)) continue; // Load last known fund date and fund transactions from DB const storedFundDate = loadAccountMeta(account.id, 'last_fund_date'); this.lastFundDates[account.id] = storedFundDate; this.fundTransactions[account.id] = loadFundTransactions(account.id); // Load cache first — serves as both the startup baseline and the fallback if API fails const cached = loadDailyPnL(account.id); const storedFundTimestamp = loadAccountMeta(account.id, 'last_fund_timestamp'); if (cached.length > 0) { const storedFundTxns = this.fundTransactions[account.id] ?? []; const storedLastFundAmt = storedFundDate ? (storedFundTxns.find((f) => f.date === storedFundDate)?.amount ?? null) : null; const storedIsWithdrawal = storedLastFundAmt !== null && storedLastFundAmt < 0; const { active, prior: priorEntries } = TradovateClient.filterActivePnL( cached, storedFundDate, storedIsWithdrawal, storedFundTimestamp ); this.priorProfit[account.id] = Math.round(priorEntries.reduce((s, d) => s + d.pnl, 0) * 100) / 100; this.fullDailyPnL[account.id] = cached; this.dailyPnL[account.id] = active; this.daysTraded[account.id] = active.filter((d) => d.pnl !== 0).length; } // --- Try Cash History first (true after-fee daily P&L) --- try { const raw = await requestReport('Cash History', account.name); const fixed = raw.replace(/"Date":\s*(\d{4}-\d{2}-\d{2})/g, '"Date": "$1"'); const rows: { Date: string; Delta: string; 'Cash Change Type': string; [k: string]: any }[] = JSON.parse(fixed); this.lastFetchRaw[account.name] = rows.length > 0 ? JSON.stringify(rows[0]) : '(empty)'; const fundMap: { [date: string]: number } = {}; const fundTimestampMap: { [date: string]: string } = {}; const dailyMap: { [date: string]: number } = {}; const firstTradeTsMap: { [date: string]: string } = {}; for (const row of rows) { const changeType = (row['Cash Change Type'] ?? '').trim(); if (changeType === 'Fund Transaction' || changeType === 'Manual Adjustment' || changeType === 'Challenge Payout') { const delta = parseFloat((row['Delta'] ?? '0').replace(/,/g, '')); if (!isNaN(delta)) { fundMap[row['Date']] = (fundMap[row['Date']] ?? 0) + delta; if (row['Timestamp']) fundTimestampMap[row['Date']] = row['Timestamp']; } continue; } const delta = parseFloat((row['Delta'] ?? '0').replace(/,/g, '')); if (isNaN(delta)) continue; dailyMap[row['Date']] = (dailyMap[row['Date']] ?? 0) + delta; // Track earliest trade timestamp per date const ts = row['Timestamp']; if (ts && (!firstTradeTsMap[row['Date']] || ts < firstTradeTsMap[row['Date']])) { firstTradeTsMap[row['Date']] = ts; } } // Persist fund transactions so they survive beyond the 28-day window const freshFundTxns = Object.entries(fundMap) .map(([date, amount]) => ({ date, amount: Math.round(amount * 100) / 100 })) .sort((a, b) => a.date.localeCompare(b.date)); if (freshFundTxns.length > 0) { saveFundTransactions(account.id, account.name, freshFundTxns); this.fundTransactions[account.id] = loadFundTransactions(account.id); // reload merged full history } // Use the most recent fund transaction as the reset point — persist it so it survives beyond the 28-day window const lastFundDate = [...Object.keys(fundMap)].sort().pop() ?? null; if (lastFundDate) { // A new cycle starts: note where the one it ends began — // with filterActivePnL's call on that date's own row — for // the reporter's catch-up. Only from a stored start: the // first load after a wipe or on a new instance never saw // the cycle before it begin. if (storedFundDate && lastFundDate > storedFundDate) { const prevAmt = this.fundTransactions[account.id]?.find((f) => f.date === storedFundDate)?.amount ?? null; saveEndedCycleStart(account.id, { fundDate: storedFundDate, includesDay: !(prevAmt !== null && prevAmt < 0) || TradovateClient.isWithdrawalBeforeFirstTrade( fundTimestampMap[storedFundDate] ?? storedFundTimestamp, firstTradeTsMap[storedFundDate] ?? null), }); } saveAccountMeta(account.id, 'last_fund_date', lastFundDate); this.lastFundDates[account.id] = lastFundDate; if (fundTimestampMap[lastFundDate]) { saveAccountMeta(account.id, 'last_fund_timestamp', fundTimestampMap[lastFundDate]); } } const fundDate = this.lastFundDates[account.id]; const fresh = Object.entries(dailyMap) .map(([date, pnl]) => ({ date, pnl: Math.round(pnl * 100) / 100 })) .sort((a, b) => a.date.localeCompare(b.date)); const merged = mergePnL(cached, fresh); const lastFundAmt = fundDate ? (fundMap[fundDate] ?? null) : null; const isWithdrawal = lastFundAmt !== null && lastFundAmt < 0; const lastFundTs = fundDate ? (fundTimestampMap[fundDate] ?? loadAccountMeta(account.id, 'last_fund_timestamp')) : null; const firstTradeTs = fundDate ? (firstTradeTsMap[fundDate] ?? null) : null; const { active, prior: priorEntries } = TradovateClient.filterActivePnL( merged, fundDate, isWithdrawal, lastFundTs, firstTradeTs ); this.priorProfit[account.id] = Math.round(priorEntries.reduce((s, d) => s + d.pnl, 0) * 100) / 100; this.fullDailyPnL[account.id] = merged; this.dailyPnL[account.id] = active; this.daysTraded[account.id] = active.filter((d) => d.pnl !== 0).length; saveDailyPnL(account.id, account.name, fresh); // upsert only fresh entries — preserves older cached rows delete this.lastFetchErrors[account.name]; this.markHistoryLoaded(account.id, true); continue; } catch { // fall through to Fills fallback } // --- Fallback: Fills report + FIFO (slightly pre-fee, used for passed accounts) --- try { const raw = await requestReport('Fills', account.name); const fixed = raw.replace(/"_tradeDate":\s*(\d{4}-\d{2}-\d{2})/g, '"_tradeDate": "$1"'); type Fill = { _tradeDate: string; _timestamp: string; _action: number; _qty: number; _price: number; Product: string; commission: number; }; interface Lot { price: number; qty: number; commPerUnit: number; } const fills: Fill[] = JSON.parse(fixed); this.lastFetchRaw[account.name] = `[fills fallback] ${raw.slice(0, 100)}`; const sorted = [...fills].sort((a, b) => a._timestamp.localeCompare(b._timestamp)); const longBook: Lot[] = []; const shortBook: Lot[] = []; const dailyMap: { [date: string]: number } = {}; for (const fill of sorted) { const pointValue = POINT_VALUES[fill.Product] ?? 1; const isBuy = fill._action === 0; let remaining = fill._qty; const commPerUnit = fill._qty > 0 ? fill.commission / fill._qty : 0; if (isBuy) { while (remaining > 0 && shortBook.length > 0) { const lot = shortBook[0]; const closed = Math.min(lot.qty, remaining); dailyMap[fill._tradeDate] = (dailyMap[fill._tradeDate] ?? 0) + (lot.price - fill._price) * closed * pointValue - (commPerUnit + lot.commPerUnit) * closed; lot.qty -= closed; remaining -= closed; if (lot.qty === 0) shortBook.shift(); } if (remaining > 0) longBook.push({ price: fill._price, qty: remaining, commPerUnit }); } else { while (remaining > 0 && longBook.length > 0) { const lot = longBook[0]; const closed = Math.min(lot.qty, remaining); dailyMap[fill._tradeDate] = (dailyMap[fill._tradeDate] ?? 0) + (fill._price - lot.price) * closed * pointValue - (commPerUnit + lot.commPerUnit) * closed; lot.qty -= closed; remaining -= closed; if (lot.qty === 0) longBook.shift(); } if (remaining > 0) shortBook.push({ price: fill._price, qty: remaining, commPerUnit }); } } // Fills report has no fund transaction data — use stored fund date and stored fund transactions const fundDate = this.lastFundDates[account.id]; const storedFundTxns = this.fundTransactions[account.id] ?? []; const lastFundAmt = fundDate ? (storedFundTxns.find((f) => f.date === fundDate)?.amount ?? null) : null; const isWithdrawalFills = lastFundAmt !== null && lastFundAmt < 0; const fresh = Object.entries(dailyMap) .map(([date, pnl]) => ({ date, pnl: Math.round(pnl * 100) / 100 })) .sort((a, b) => a.date.localeCompare(b.date)); const merged = mergePnL(cached, fresh); const fillsFundTs = loadAccountMeta(account.id, 'last_fund_timestamp'); const { active, prior: priorEntries } = TradovateClient.filterActivePnL( merged, fundDate, isWithdrawalFills, fillsFundTs ); this.priorProfit[account.id] = Math.round(priorEntries.reduce((s, d) => s + d.pnl, 0) * 100) / 100; this.fullDailyPnL[account.id] = merged; this.dailyPnL[account.id] = active; this.daysTraded[account.id] = active.filter((d) => d.pnl !== 0).length; saveDailyPnL(account.id, account.name, fresh); // upsert only fresh entries — preserves older cached rows this.lastFetchErrors[account.name] = FILLS_FALLBACK_NOTE; } catch (err) { if (this.intentionalDisconnect) break; const msg = err instanceof Error ? err.message : String(err); console.error(`${this.tag} fetchDaysTraded: both reports failed for ${account.name}: ${msg}`); failedAccounts++; // Cache already loaded and filtered at top of loop — just log the error if (cached.length > 0) { this.lastFetchErrors[account.name] = `${msg} (using ${cached.length} cached entries from database)`; console.log(`${this.tag} fetchDaysTraded: using ${cached.length} cached entries from DB for ${account.name}`); } else { // No history at all is not an empty one: leave dailyPnL // unset so the account stays out of trading until a fetch // lands, instead of trading as a fresh stage-1 cycle. this.daysTraded[account.id] ??= 0; this.lastFetchErrors[account.name] = msg; } } // The Fills fallback, or the DB cache after both reports failed. this.markHistoryLoaded(account.id, false); } this.fetchDaysComplete = true; return { failedAccounts }; } /** * Whether this account's history can be planned on — see PlatformClient. * Loaded by this process, not wiped since, and no balance change the * history can't explain. */ public historyReady(accountId: number): boolean { if (!this.historyLoadedIds.has(accountId)) return false; if (this.dailyPnL[accountId] === undefined || this.fundTransactions[accountId] === undefined) return false; if (this.historyRefreshIds.has(accountId)) return false; return !this.balanceDrifted(accountId); } /** Drop the account's history from memory (a wipe): it reads unloaded, and * stays out of trading, until the next fetch rebuilds it. */ public forgetHistory(accountId: number, accountName: string): void { this.historyLoadedIds.delete(accountId); this.historyRefreshIds.delete(accountId); delete this.balanceBaseline[accountId]; delete this.dailyPnL[accountId]; delete this.fullDailyPnL[accountId]; delete this.daysTraded[accountId]; delete this.priorProfit[accountId]; delete this.lastFundDates[accountId]; delete this.fundTransactions[accountId]; delete this.lastFetchErrors[accountName]; delete this.lastFetchRaw[accountName]; } /** * A fetch pass reached the account. Cash History is the only source that * books fund transactions, so only it clears a balance-change hold; the * Fills fallback and the DB cache still count as loaded otherwise. */ private markHistoryLoaded(accountId: number, fromCashHistory: boolean): void { if (this.dailyPnL[accountId] === undefined || this.fundTransactions[accountId] === undefined) { this.historyLoadedIds.delete(accountId); return; } if (this.historyRefreshIds.has(accountId) && !fromCashHistory) return; this.historyRefreshIds.delete(accountId); this.historyLoadedIds.add(accountId); const cash = this.accountCashBalances[accountId]; if (cash) this.balanceBaseline[accountId] = { dayStart: cash.amount - cash.realizedPnL, realized: cash.realizedPnL }; else delete this.balanceBaseline[accountId]; } /** * True when the balance moved in a way the loaded history can't explain — * a payout or other fund transaction posting between the hourly Cash * History runs. Planned on as is, the ended cycle's rows against the * post-payout balance re-open a completed account. Today's trades move * amount and realizedPnL together, and the daily reset folds the day's * realized P&L into the balance; any other move past a few dollars holds * the account until a Cash History books it (maybeFetchDaysTraded fetches * it on the next sync). */ private balanceDrifted(accountId: number): boolean { const cash = this.accountCashBalances[accountId]; if (!cash) return false; const dayStart = cash.amount - cash.realizedPnL; const base = this.balanceBaseline[accountId]; if (!base) { this.balanceBaseline[accountId] = { dayStart, realized: cash.realizedPnL }; return false; } if (Math.abs(dayStart - base.dayStart) <= FUND_DRIFT_TOLERANCE) { base.realized = cash.realizedPnL; return false; } if (Math.abs(dayStart - (base.dayStart + base.realized)) <= FUND_DRIFT_TOLERANCE) { this.balanceBaseline[accountId] = { dayStart, realized: cash.realizedPnL }; return false; } if (!this.historyRefreshIds.has(accountId)) { this.historyRefreshIds.add(accountId); const name = this.accountList.find((a) => a.id === accountId)?.name ?? accountId; console.warn(`${this.tag} ${name}: balance moved $${Math.round((dayStart - base.dayStart) * 100) / 100}` + ' that its history does not explain — held out of trading until Cash History refreshes'); } return true; } /** Run the drift check on every loaded account each sync, so the day's * last realized P&L is on record before the daily reset folds it in. */ private observeBalances(): void { for (const accountId of this.historyLoadedIds) this.balanceDrifted(accountId); } private async login(pTicket?: string): Promise { const chl = '' + (Date.now() - 1581e9); const sec = computeSec(this.name, this.password, this.deviceId, chl); const generateEncodedPassword = (e) => { const { name: t, password: n } = e, o = t.length % n.length, r = (n.slice(o) + n.slice(0, o)).split('').reverse().join(''); return btoa(r); }; // Non-2xx and timeouts reject (callers log errorMessage(), never the // AxiosError itself — its config carries this body). The abort signal // backs the timeout up for a connect that never gets a socket. const res = await axios.post('https://live.tradovateapi.com/v1/auth/accesstokenrequest', { name: this.name, password: generateEncodedPassword({ name: this.name, password: this.password }), environment: 'demo', appId: 'tradovate_trader(web)', appVersion: '3.251205.0', deviceId: this.deviceId, cid: '1', chl: chl, sec: sec, enc: true, ...(pTicket ? { 'p-ticket': pTicket } : {}), }, { timeout: HTTP_TIMEOUT_MS, signal: AbortSignal.timeout(HTTP_TIMEOUT_MS + 5_000) }); return res.data ?? {}; } /** Access token expiry as epoch ms (0 = none, unparseable, or not * comparable with this machine's clock — see noteTokenExpiry()). */ private tokenExpiresAt(): number { if (this.tokenExpiryUntrusted) return 0; const raw = this.accessInfo?.expirationTime; if (!raw) return 0; const t = new Date(raw).getTime(); return Number.isFinite(t) ? t : 0; } /** * Called whenever a token arrives (login or renewal). Tradovate hands out * tokens with well over an hour to live, so one that is already expired — * or nearly — by our clock means this machine's clock is ahead of * Tradovate's, not that the token is dead. Trusting that expiry would have * the watchdog "expire" every fresh token and log in again every 15s (a * rate-limit penalty, then a captcha, for every client on this IP). * Treat the expiry as unknown instead: renew every 2 minutes and log in on * reconnect, as the code did before it read the expiry at all. */ private noteTokenExpiry(): void { const raw = this.accessInfo?.expirationTime; const t = raw ? new Date(raw).getTime() : NaN; const untrusted = Number.isFinite(t) && t - Date.now() < TOKEN_RENEW_WINDOW_MS; if (untrusted && !this.tokenExpiryUntrusted) { console.warn(`${this.tag} new token expires ${raw}, which is (nearly) past by this machine's clock — check the system clock; ignoring the expiry`); } this.tokenExpiryUntrusted = untrusted; } /** Holding a token known to stay valid for at least `ms` more. */ private tokenValidFor(ms: number): boolean { return !!this.accessInfo?.accessToken && this.tokenExpiresAt() > Date.now() + ms; } /** The 2-minute renewal check. Never rejects, never overlaps itself. */ private maybeRenewToken(): void { if (this.intentionalDisconnect || this.connecting || this.renewInFlight || !this.accessInfo?.accessToken) return; const now = Date.now(); const expiresAt = this.tokenExpiresAt(); // An unparseable expiry renews anyway — an extra renewal beats a silent expiry. if (expiresAt > 0 && expiresAt - now > TOKEN_RENEW_WINDOW_MS) return; // Already expired: renewal can't save it; the watchdog / next connect logs in. if (expiresAt > 0 && expiresAt <= now) return; this.renewInFlight = true; this.renewAccessToken() .catch((err) => console.error(`${this.tag} token renewal failed: ${errorMessage(err)}`)) .finally(() => { this.renewInFlight = false; }); } private async renewAccessToken(): Promise { const token = this.accessInfo?.accessToken; if (!token) return; try { const res = await axios.get('https://live.tradovateapi.com/v1/auth/renewAccessToken', { headers: { Authorization: `Bearer ${token}`, }, timeout: HTTP_TIMEOUT_MS, signal: AbortSignal.timeout(HTTP_TIMEOUT_MS + 5_000), }); if (this.intentionalDisconnect) return; // A re-login while this was in flight already replaced the token. if (this.accessInfo?.accessToken !== token) return; const body = res.data as LoginResponseBody | undefined; if (!body?.accessToken) { throw new Error(body?.errorText ? String(body.errorText) : 'renewal answered without an access token'); } // Merge, so a renewal answer without expirationTime can't leave it undefined. this.accessInfo = { ...this.accessInfo, ...body } as AuthLoginResponse; this.noteTokenExpiry(); console.log(`${this.tag} renewed access token — expires ${this.accessInfo.expirationTime}`); } catch (err) { if (this.intentionalDisconnect || this.accessInfo?.accessToken !== token) return; console.warn(`${this.tag} token renewal failed: ${errorMessage(err)}`); const status = (err as { response?: { status?: unknown } })?.response?.status; const expiresAt = this.tokenExpiresAt(); if (status === 401 || (expiresAt > 0 && expiresAt - Date.now() < TOKEN_RELOGIN_WINDOW_MS)) { this.abandonSocket('token renewal failed', true); } } } async getAccountList(): Promise { if (!this.accessInfo?.accessToken) { console.log(`${this.tag} not authenticated`); return []; } const res = await axios.get('https://demo.tradovateapi.com/v1/account/list', { headers: { Authorization: `Bearer ${this.accessInfo.accessToken}`, }, timeout: HTTP_TIMEOUT_MS, }); return res.data; } async findFrontMonthContract(productName: string): Promise<{ id: number; name: string; tickSize: number } | null> { if (!this.accessInfo?.accessToken) return null; // Prefer the volume-based resolver so the trading path matches the settings page. const cached = getCachedContract(productName); if (cached) return { id: cached.id, name: cached.name, tickSize: cached.tickSize }; // Lookups fail (a 401 on an expiring token, a network blip) — that must // come back as "no contract", not a throw that aborts every firm's trade. try { // If the cache is cold, resolve on demand with the same logic the settings page uses. // Under a deadline: its lookups (Yahoo, then Tradovate) must not // hold every firm's trade. A late answer still fills the cache. const resolved = await withDeadline( resolveContracts([productName], this.accessInfo.accessToken), CONTRACT_RESOLVE_DEADLINE_MS, 'contract resolve', ); const contract = resolved[productName]; if (contract) { return { id: contract.id, name: contract.name, tickSize: contract.tickSize }; } } catch (err) { console.warn(`${this.tag} contract resolve for ${productName} failed: ${errorMessage(err)} — trying contract/suggest`); } try { // Last resort: fall back to Tradovate's ordered suggestions. const res = await axios.get( `https://demo.tradovateapi.com/v1/contract/suggest?t=${encodeURIComponent(productName)}&l=20`, { headers: { Authorization: `Bearer ${this.accessInfo.accessToken}` }, timeout: HTTP_TIMEOUT_MS } ); const contracts: Array<{ id: number; name: string; status: string; providerTickSize: number }> = res.data ?? []; const match = contracts.find((c) => c.name.startsWith(productName)); if (!match) return null; return { id: match.id, name: match.name, tickSize: match.providerTickSize ?? 0.25 }; } catch (err) { console.error(`${this.tag} contract lookup for ${productName} failed: ${errorMessage(err)}`); return null; } } async fetchInstrumentFees(symbols: string[]): Promise<{ [symbol: string]: number }> { if (!this.accessInfo?.accessToken || this.products.length === 0) return {}; const productIds = symbols .map((sym) => this.products.find((p) => p.name === sym)?.id) .filter((id): id is number => id !== undefined); const res = await axios.post( 'https://demo.tradovateapi.com/v1/contract/getproductfeeparams', { productIds }, { headers: { Authorization: `Bearer ${this.accessInfo.accessToken}` }, timeout: HTTP_TIMEOUT_MS } ); const result: { [symbol: string]: number } = {}; for (const param of (res.data?.params ?? [])) { const product = this.products.find((p) => p.id === param.productId); if (product && symbols.includes(product.name)) { const raw = (param.clearingFee ?? 0) + (param.exchangeFee ?? 0) + (param.nfaFee ?? 0) + (param.brokerageFee ?? 0) + (param.ipFee ?? 0) + (param.commission ?? 0) + (param.orderRoutingFee ?? 0); result[product.name] = parseFloat(raw.toFixed(4)); console.log( `${this.tag} fees ${product.name}: clearing=${param.clearingFee ?? 0}` + ` exchange=${param.exchangeFee ?? 0}` + ` nfa=${param.nfaFee ?? 0}` + ` brokerage=${param.brokerageFee ?? 0}` + ` ip=${param.ipFee ?? 0}` + ` commission=${param.commission ?? 0}` + ` routing=${param.orderRoutingFee ?? 0}` + ` → total=${result[product.name]}` ); } } return result; } async requestContractsFromSocket(names: string[]): Promise<{ [name: string]: Contract; }> { // Callers expect an object — a dead socket or a lost answer is {} with // a warning, never a promise left pending. try { return await this.sendRequest('contract/finds', `names=${names.join(',')}`, CONTRACT_FIND_TIMEOUT_MS, (response) => { const items: Contract[] = Array.isArray(response) ? response : []; return items.reduce((acc: { [name: string]: Contract }, item: Contract) => { return { ...acc, [item.name]: item }; }, {} as { [name: string]: Contract }); }); } catch (err) { console.warn(`${this.tag} contract/finds failed: ${errorMessage(err)}`); return {}; } } /** A string or empty ack is the broker refusing the order — nothing was placed. */ private static acceptedAck(response: unknown, endpoint: string): { orderId?: number; id?: number; commandStatus?: string } { if (typeof response === 'string') throw new OrderNotPlacedError(response || `${endpoint} refused`); if (!response) throw new OrderNotPlacedError(`Empty response from ${endpoint}`); if (typeof response !== 'object') return {}; const ack = response as { orderId?: number; id?: number; commandStatus?: string; failureReason?: string; failureText?: string }; // PlaceOrderResult / PlaceOCOResult: a refusal (SessionClosed, // LiquidationOnly, …) carries failureReason and no order id — no order // exists. Silently taking it as placed left a position with no exit // (placeoco) or waited 30s for a fill that could never come. if (ack.orderId == null && ack.id == null && ((ack.failureReason && ack.failureReason !== 'Success') || ack.failureText)) { throw new OrderNotPlacedError(`${endpoint} refused: ${ack.failureText || ack.failureReason}`); } return ack; } /** Resolves with the fill event for `orderId` (it carries the real * execution price). Registered synchronously from the ack handler, so a * fill later in the same frame as the ack is not missed. */ private waitForFill(orderId: number | undefined): Promise> { // Fills are pushed on the event stream, independently of the ack: one // that got ahead of it (an earlier frame, or earlier in the same one) // is already in the event buffer. Missed, a filled entry came back // "outcome unknown" and the caller placed no exit on an open position. const early = orderId == null ? undefined : this.recentEntityEvents.find((ev) => ev.entityType === 'fill' && ev.eventType === 'Created' && ev.entity?.orderId === orderId); if (early) { console.log(`${this.tag} fill (ahead of its ack):`, JSON.stringify(early.entity)); return Promise.resolve(early.entity); } return new Promise((resolve, reject) => { const watcher: IndirectCallback = { entityType: 'fill', eventType: 'Created', validator: (item: any) => item?.orderId === orderId, callback: (fill: any) => { clearTimeout(timeout); console.log(`${this.tag} fill:`, JSON.stringify(fill)); resolve(fill); }, cancel: (reason: string) => { clearTimeout(timeout); reject(new Error(`Order ${orderId} acknowledged but ${reason} before its fill arrived — outcome unknown`)); }, }; const timeout = setTimeout(() => { this.indirectEventCallbacks = this.indirectEventCallbacks.filter((info) => info !== watcher); reject(new Error(`Order ${orderId} acknowledged but no fill within 30s (market may be closed)`)); }, FILL_TIMEOUT_MS); this.indirectEventCallbacks.push(watcher); }); } async sendOrder( accountId: number, contractSymbol: string, // e.g. "NQH6" — WebSocket placeorder requires "symbol" quantity: number, action: 'Buy' | 'Sell', orderType: 'Market' | 'Limit', price?: number ): Promise> { // Checked before the audit frame so a dead socket is a clean // "not placed" rather than a send into nothing. const problem = this.socketProblem(this.ws); if (problem) throw new OrderNotPlacedError(`order/placeorder not sent — ${problem}`); if (!this.sendFrame( `user/registeraudituseraction\n25\n\n${JSON.stringify({ accountId, actionType: action + orderType, details: `${action} ${quantity} ${contractSymbol} ${orderType}${price ? ` @ ${price}` : ''}, TIF Day`, })}` )) { throw new OrderNotPlacedError('order/placeorder not sent — Tradovate socket rejected the audit frame'); } return this.sendRequest( 'order/placeorder', `\n${JSON.stringify({ accountId, action, symbol: contractSymbol, orderQty: quantity, orderType, price, timeInForce: 'Day', text: 'DOM', })}`, ACK_TIMEOUT_MS, (response) => { console.log(`${this.tag} sendOrder raw ack:`, JSON.stringify(response)); const ack = TradovateClient.acceptedAck(response, 'order/placeorder'); // Tradovate returns the command object: { id, commandStatus, orderId, ... } const orderId = ack.orderId ?? ack.id; console.log(`${this.tag} sendOrder orderId=${orderId} status=${ack.commandStatus}`); // Wait for the fill event — it carries the real execution price return this.waitForFill(orderId); }, ); } /** Place an order and resolve as soon as the command is acknowledged (does not wait for fill). */ async placeOrderNoWait( accountId: number, contractSymbol: string, quantity: number, action: 'Buy' | 'Sell', orderType: 'Market' | 'Limit', price?: number ): Promise<{ orderId?: number }> { return this.sendRequest( 'order/placeorder', `\n${JSON.stringify({ accountId, action, symbol: contractSymbol, orderQty: quantity, orderType, price, timeInForce: 'Day', text: 'DOM', })}`, ACK_TIMEOUT_MS, (response) => { const ack = TradovateClient.acceptedAck(response, 'order/placeorder'); return { orderId: ack.orderId ?? ack.id }; }, ); } /** Place an OCO exit pair after an entry fill: a limit take-profit linked * to a stop, via order/placeoco — Tradovate cancels the sibling when one * fills. Resolves on command ack (does not wait for a fill), like * placeOrderNoWait. */ async placeOcoExit( accountId: number, contractSymbol: string, quantity: number, exitAction: 'Buy' | 'Sell', limitPrice: number, stopPrice: number, ): Promise<{ orderId?: number }> { return this.sendRequest( 'order/placeoco', `\n${JSON.stringify({ accountId, action: exitAction, symbol: contractSymbol, orderQty: quantity, orderType: 'Limit', price: limitPrice, timeInForce: 'Day', text: 'DOM', other: { action: exitAction, orderType: 'Stop', stopPrice, timeInForce: 'Day', text: 'DOM', }, })}`, ACK_TIMEOUT_MS, (response) => { const ack = TradovateClient.acceptedAck(response, 'order/placeoco'); return { orderId: ack.orderId ?? ack.id }; }, ); } /** Liveness snapshot for /api/health and the registry supervisor. No tokens. */ public health(): ClientHealth { const now = Date.now(); const expiresAt = this.tokenExpiresAt(); return { platform: 'tradovate', authenticated: !!this.accessInfo?.accessToken && (expiresAt === 0 || expiresAt > now), transport: this.ws ? (READY_STATE_NAMES[this.ws.readyState] ?? String(this.ws.readyState)) : 'none', syncComplete: this.syncComplete, lastSyncAt: this.lastSyncAt, lastLoginAt: this.lastLoginAt, lastLoginError: this.lastLoginError, consecutiveLoginFailures: this.consecutiveLoginFailures, nextReconnectAt: this.nextReconnectAt, tokenExpiresAt: expiresAt, lastInboundAt: this.lastInboundAt, pendingRequests: Object.keys(this.directEventCallbacks).length + this.indirectEventCallbacks.length, syncInFlightSince: this.syncRequestedAt, daysFetchInFlightSince: this.daysFetchInFlight ? this.lastDaysFetchStartedAt : 0, }; } /** Close the WebSocket connection and stop all timers. Call before discarding * the instance — nothing it scheduled fires afterwards. */ public disconnect(): void { this.intentionalDisconnect = true; if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } this.nextReconnectAt = 0; if (this.livenessInterval) { clearInterval(this.livenessInterval); this.livenessInterval = null; } if (this.tokenRenewalInterval) { clearInterval(this.tokenRenewalInterval); this.tokenRenewalInterval = null; } if (this.penaltyWait) { // Cut the wait short; the login loop sees intentionalDisconnect and stops. const { timer, resolve } = this.penaltyWait; this.penaltyWait = null; clearTimeout(timer); resolve(); } // Closes the socket with its handlers detached, clears sync/heartbeat, // rejects pending acks (ambiguous: they may have been placed). this.dropSocket('client disconnected'); const watchers = this.indirectEventCallbacks; this.indirectEventCallbacks = []; for (const info of watchers) info.cancel?.('client disconnected'); this.daysFetchInFlight = false; this.nextDaysFetchRetryAt = 0; } /** Register a one-time callback for when a fill arrives for a given orderId. */ onFill(orderId: number, callback: (fill: any) => void): void { this.indirectEventCallbacks.push({ entityType: 'fill', eventType: 'Created', validator: (item: any) => item?.orderId === orderId, callback, }); } }