/** * TopstepX (ProjectX Gateway) client — the TopstepX twin of TradovateClient. * * REST: https://api.topstepx.com (auth via API key at POST /api/Auth/loginKey, * 24h session token, refreshed through POST /api/Auth/validate). * Realtime: SignalR user hub at https://rtc.topstepx.com/hubs/user, which * pushes account balances, orders, positions and trades — the analogue of * Tradovate's user-sync WebSocket. * * Differences from Tradovate that shape this file: * - There is no cash-history / fund-transaction report. Daily P&L comes from * POST /api/Trade/search (each half-turn carries fees; closing halves carry * profitAndLoss), grouped into Central-time trade dates with the same 5 PM * roll Tradovate uses. Deposits/withdrawals are invisible to the API, so * fund transactions only ever come from whatever the local DB already holds. * - There is no OCO order. placeOcoExit places the limit and stop separately * and cancels the sibling the moment one fills (order events, plus a safety * poll) — see the `ocoPairs` machinery. * - There is no auto-liquidation profile to read, so autoLiqThresholds stays * all zeros; a blown account shows up as canTrade=false → active=false. */ import axios from 'axios'; import * as signalR from '@microsoft/signalr'; import { currentTradeDateCT } from './trading-logic'; import { getCachedContract } from './contract-resolver'; import { signalrWebSocket } from './outbound-proxy'; import { saveDailyPnL, loadDailyPnL, loadAccountMeta, saveAccountMeta, loadFundTransactions, saveFundTransactions, saveEndedCycleStart } from './db'; import { getAccountConfig } from './account-type-resolver'; import type { PlatformClient, PlatformAccount, PlatformContract, ClientHealth } from './platform-client'; import { OrderNotPlacedError, isOrderNotPlaced, errorMessage } from './platform-client'; const API_BASE = 'https://api.topstepx.com'; const USER_HUB = 'https://rtc.topstepx.com/hubs/user'; /** * Self-healing timings. The weekend failure was a chain of awaits with no * deadline (hub.start, hub.invoke, a re-login) plus a re-login that tore down * the sync loop first — one wedged await left the firm "waiting for sync" * until PM2 restarted the process. Every hub await now has a deadline, sync * and hub setup detect their own wedges, and a watchdog keeps the client from * ever sitting with no session and nothing scheduled to get one. */ const HUB_START_TIMEOUT_MS = 30_000; const HUB_INVOKE_TIMEOUT_MS = 15_000; /** Automatic reconnect gives up by itself within ~45s (0/2/10/30s retries); * a hub still Connecting/Reconnecting after this is stuck, not retrying. */ const HUB_STALL_MS = 2 * 60_000; /** A hub setup or REST sync still running after this is wedged — start over. */ const HUB_SETUP_WEDGED_MS = 3 * 60_000; const SYNC_WEDGED_MS = 3 * 60_000; /** Failed logins back off 60s, 120s, … up to 10 min, ±20% so a fleet of * boxes coming back from the same outage doesn't log in in lockstep. */ const LOGIN_RETRY_BASE_MS = 60_000; const LOGIN_RETRY_MAX_MS = 10 * 60_000; const WATCHDOG_INTERVAL_MS = 60_000; /** Logged in longer than this with no good sync for as long → the session is * presumed dead in a way validate didn't catch; log in again. */ const SESSION_STALE_MS = 10 * 60_000; /** Auth calls get their own short deadlines: a login that never answers * would hold loginInFlight — and with it every retry — forever. */ const LOGIN_TIMEOUT_MS = 30_000; const VALIDATE_TIMEOUT_MS = 15_000; const TOKEN_LIFETIME_MS = 24 * 60 * 60 * 1000; /** Hard cap on every other REST call. axios's own timeout is a socket-idle * timer that only starts once the socket has connected, so a proxy CONNECT * that never answers hangs the call — and any single-flight flag it holds, * like the daily P&L fetch's — for good. Aborting (rather than racing) also * destroys a request that hasn't been written yet, so an order can't go out * minutes after its caller gave up on it. Longer than the idle timeout, so * it only bites where that one can't. */ const REST_DEADLINE_MS = 90_000; /** Rejects if `promise` hasn't settled within `ms`. The underlying operation * keeps running (callers discard whatever it belonged to); Promise.race * subscribes to it, so a late rejection never surfaces as unhandled. */ function withTimeout(promise: Promise, ms: number, label: string): Promise { let timer: ReturnType | undefined; const deadline = new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${label} timed out after ${Math.round(ms / 1000)}s`)), ms); }); return Promise.race([promise, deadline]).finally(() => clearTimeout(timer)); } /** * The gateway answered and said no (`success: false`) — as opposed to a * transport failure, where the request may or may not have been acted on. * placeOrder turns only this (and pre-handler HTTP refusals) into * OrderNotPlacedError. */ class GatewayRefusedError extends Error { readonly errorCode: number | undefined; /** Set when a refusal still names an order — that order exists. */ readonly orderId: number | undefined; constructor(message: string, errorCode: number | undefined, orderId: number | undefined) { super(message); this.name = 'GatewayRefusedError'; this.errorCode = errorCode; this.orderId = orderId; } } /** PlaceOrderErrorCode 6 (OrderPending) and 7 (UnknownError) don't say the * order wasn't created, so a refusal carrying them stays ambiguous. */ const AMBIGUOUS_PLACE_ERRORS = new Set([6, 7]); /** HTTP answers given before any order handler runs (by the gateway's * middleware or the Cloudflare edge in front of it): malformed request, * missing/expired token, forbidden, rate-limited. Nothing was placed. */ const REFUSED_BEFORE_HANDLER = new Set([400, 401, 403, 429]); interface TsxContract { id: string; name: string; description: string; tickSize: number; tickValue: number; activeContract: boolean; symbolId: string; } interface TsxTrade { id: number; accountId: number; contractId: string; creationTimestamp: string; price: number; profitAndLoss: number | null; // null = half-turn (entry only) fees: number; side: number; // 0 = Bid/Buy, 1 = Ask/Sell size: number; voided: boolean; orderId: number; } /** OrderStatus: 1 Open, 2 Filled, 3 Cancelled, 4 Expired, 5 Rejected, 6 Pending */ const ORDER_FILLED = 2; const ORDER_GONE = new Set([3, 4, 5]); /** An order's first fill price (by trade id), or null when it has none — * the price the hub's first trade event carries, so a whole tick. */ function firstFillPrice(trades: TsxTrade[]): number | null { let first: TsxTrade | null = null; for (const t of trades) { if (t.voided || typeof t.price !== 'number') continue; if (!first || t.id < first.id) first = t; } return first ? first.price : null; } const SIDE_BUY = 0; const SIDE_SELL = 1; const TYPE_LIMIT = 1; const TYPE_MARKET = 2; const TYPE_STOP = 4; export class TopstepXClient implements PlatformClient { private name: string; private apiKey: string; private token: string | null = null; public accountList: PlatformAccount[] = []; public accountCashBalances: { [accountId: number]: { amount: number; realizedPnL: number } } = {}; public positions: { [accountId: number]: { contractId: string; netPos: number; netPrice: number; timestamp: Date }; } = {}; public daysTraded: { [accountId: number]: number } = {}; public dailyPnL: { [accountId: number]: { date: string; pnl: number }[] } = {}; public fullDailyPnL: { [accountId: number]: { date: string; pnl: number }[] } = {}; public lastFundDates: { [accountId: number]: string | null } = {}; public fundTransactions: { [accountId: number]: { date: string; amount: number }[] } = {}; public priorProfit: { [accountId: number]: number } = {}; public autoLiqThresholds: { [accountId: number]: number } = {}; public fetchDaysComplete = false; public syncComplete = false; /** Epoch ms of the last sync that came back good (0 = never). */ public lastSyncAt = 0; public lastFetchErrors: Record = {}; public lastFetchRaw: Record = {}; public recentEntityEvents: { entityType: string; eventType: string; entity: unknown; ts: number }[] = []; /** Tradovate-only concept; empty keeps fee lookups a clean no-op. */ public products: { id: number; name: string }[] = []; private callbackOnSyncRequest: () => Promise; private hub: signalR.HubConnection | null = null; private subscribedAccountIds = new Set(); /** SubscribeAccounts done on the current hub connection. */ private accountsFeedSubscribed = false; /** ensureHub is single-flight; the generation lets a flight declared * wedged be superseded without its late finally clearing the new flag. */ private hubSetupInFlight = false; private hubSetupSince = 0; private hubSetupGeneration = 0; /** ensureHub was asked for while a flight ran (e.g. the first sync's * accounts landed mid-start) — run once more if the hub came up. */ private hubRecheckWanted = false; /** When the hub was first seen Connecting/Reconnecting/Disconnecting (0 = it wasn't). */ private hubStalledSince = 0; /** Last frame from the hub (event, invoke answer, handshake). */ private lastInboundAt = 0; private syncInterval: ReturnType | null = null; private tokenRenewalInterval: ReturnType | null = null; private ocoPollInterval: ReturnType | null = null; private loginRetryTimer: ReturnType | null = null; private watchdogInterval: ReturnType | null = null; private intentionalDisconnect = false; /** requestSync is single-flight too, with the same generation trick. */ private syncInFlight = false; private syncInFlightSince = 0; private syncGeneration = 0; private loginInFlight = false; private lastLoginAt = 0; private lastLoginError: string | null = null; private consecutiveLoginFailures = 0; private nextReconnectAt = 0; /** When the current token was issued or last renewed (24h lifetime). */ private tokenRefreshedAt = 0; /** Same hourly-with-retry throttle the Tradovate client uses for reports. */ private daysFetchInFlight = false; private daysFetchInFlightSince = 0; private lastDaysFetchStartedAt = 0; private lastDaysFetchSucceededAt = 0; private nextDaysFetchRetryAt = 0; /** The first full daily P&L pass has landed (cache fallbacks included). * No sync counts as good before it — see requestSync. */ private historyLoaded = false; /** Accounts whose history this client has loaded (a trade search came * back, or failed with a DB cache to fall back on). historyLoaded above * is the firm's first pass; this is per account, so one that appears * later, or is wiped, stays out of trading until its own fetch lands. */ private historyLoadedIds = new Set(); /** Fetched, but not yet past a sync that reconciled its funds. TopstepX * has no funds API: reconcileFunds, inside requestSync, is what books a * withdrawal from the balance the trades don't explain, so a history * fetched without it (after a wipe took the fund rows) would plan the * account as stage 1 until the next sync. */ private reconcilePendingIds = new Set(); /** Settles when the sync in flight ends (requestSync hands it to a caller * that arrives mid-pass). */ private syncRunning: Promise = Promise.resolve(); /** Accounts a fetchDaysTraded call is working through right now, so the * targeted pass doesn't fetch them again (a wipe fetches outside it). */ private fetchingIds = new Set(); /** Trades seen for the current CT trade date, per account — realizedPnL is * recomputed from these on every REST sync and every live trade event. */ private todayTrades: { [accountId: number]: Map } = {}; /** Reported fees per trade date, per account — weights for spreading the * off-book fee drift when reconstructing the MLL floor. */ private dailyFees: { [accountId: number]: Map } = {}; /** Front-month contract per product symbol, plus name→contract for order * placement (sendOrder receives the contract NAME, e.g. "NQZ5"). */ private contractByProduct = new Map(); private contractByName = new Map(); private static CONTRACT_TTL_MS = 30 * 60 * 1000; /** Fill waiters keyed by orderId — resolved by GatewayUserTrade events. */ private fillWaiters: { orderId: number; callback: (trade: TsxTrade) => void }[] = []; /** Emulated OCO exit pairs: when one leg fills, cancel the other. */ private ocoPairs: { accountId: number; limitId: number; stopId: number; done: boolean; placedAt: number }[] = []; constructor(name: string, apiKey: string, callbackOnSyncRequest: () => Promise) { this.name = name; this.apiKey = apiKey; this.callbackOnSyncRequest = callbackOnSyncRequest; this.watchdogInterval = setInterval(() => this.watchdog(), WATCHDOG_INTERVAL_MS); void this.connectAndAuth(); } // ── auth ───────────────────────────────────────────────────────────────── private authHeaders() { return { Authorization: `Bearer ${this.token}` }; } private async post(path: string, body: unknown): Promise { const res = await axios.post(`${API_BASE}${path}`, body, { headers: this.authHeaders(), signal: AbortSignal.timeout(REST_DEADLINE_MS), }).catch((err: unknown) => { // The deadline abort surfaces as a bare "canceled" — say what it // was. Still a plain Error: for an order it stays ambiguous. if (axios.isCancel(err)) throw new Error(`${path} timed out after ${REST_DEADLINE_MS / 1000}s`); throw err; }); const data = res.data as T & { success?: boolean; errorMessage?: string | null; errorCode?: number; orderId?: number | null }; // The gateway answers every call with a JSON object. Anything else — // an HTML maintenance or proxy page served with a 200, an empty body — // is a failed call, never an empty result: through the callers' // `?? []`, a positions answer would read as "flat", a trades answer // as "nothing traded today", and the sync would certify both. For an // order it stays a plain Error: it may have gone in. if (!data || typeof data !== 'object') { throw new Error(`${path} returned a non-JSON answer (HTTP ${res.status})`); } if (data.success === false) { throw new GatewayRefusedError( `${path} failed: ${data.errorMessage ?? `errorCode=${data.errorCode}`}`, typeof data.errorCode === 'number' ? data.errorCode : undefined, typeof data.orderId === 'number' && data.orderId > 0 ? data.orderId : undefined, ); } return data; } /** * Log in, single-flight. Safe to call from anywhere (constructor, retry * timer, token renewal, watchdog) — a second caller while one runs is a * no-op, and it never rejects. * * First login starts the hub, the sync loop and its 60s interval. A * RE-login only swaps the token and nudges a sync: it deliberately leaves * the sync interval and syncComplete alone. Tearing those down before the * new session was proven is what left a firm "waiting for sync" forever * when the re-login chain wedged. */ private async connectAndAuth(): Promise { if (this.intentionalDisconnect || this.loginInFlight) return; if (!this.name) { // loginKey requires the platform username the key was generated for. // The key alone is storable (per the credential form), but nothing // can be traded until the username arrives — say so loudly, once. this.lastFetchErrors['login'] = 'TopstepX username missing — the API key alone cannot log in. Add the username on the dashboard.'; this.lastLoginError = this.lastFetchErrors['login']; console.error(`[topstepx] ${this.lastFetchErrors['login']}`); return; } if (this.loginRetryTimer) { clearTimeout(this.loginRetryTimer); this.loginRetryTimer = null; } this.nextReconnectAt = 0; this.loginInFlight = true; try { // A hard deadline on top of axios's own: that one is a socket // inactivity timer, which never starts while a proxy CONNECT // hangs — and a login that never settles holds loginInFlight. const token = await withTimeout(this.login(), LOGIN_TIMEOUT_MS, 'loginKey'); if (this.intentionalDisconnect) return; this.token = token; this.lastLoginAt = this.tokenRefreshedAt = Date.now(); this.consecutiveLoginFailures = 0; this.lastLoginError = null; console.log(`[topstepx] logged in as ${this.name}`); delete this.lastFetchErrors['login']; // Session tokens live 24h; re-validate every hour to stay fresh. if (this.tokenRenewalInterval) clearInterval(this.tokenRenewalInterval); this.tokenRenewalInterval = setInterval(() => void this.renewToken(), 60 * 60 * 1000); if (!this.syncInterval) { // First login. The hub is never a hard dependency: REST polling // carries every feature (slower), and each sync tick calls // ensureHub again, so a hub that won't come up now gets retried. void this.ensureHub(); this.syncInterval = setInterval(() => void this.requestSync(), 60_000); void this.requestSync(); } else { // Re-login: the loop keeps running; just use the new session now. void this.requestSync(); void this.ensureHub(); } } catch (err) { if (this.intentionalDisconnect) return; this.consecutiveLoginFailures++; const msg = errorMessage(err); this.lastLoginError = msg; this.lastFetchErrors['login'] = msg; const backoff = Math.min( LOGIN_RETRY_BASE_MS * 2 ** Math.min(this.consecutiveLoginFailures - 1, 10), LOGIN_RETRY_MAX_MS, ); const delay = Math.round(backoff * (0.8 + Math.random() * 0.4)); this.nextReconnectAt = Date.now() + delay; console.error(`[topstepx] login failed for ${this.name}: ${msg} — attempt ${this.consecutiveLoginFailures}, retrying in ${Math.round(delay / 1000)}s`); this.loginRetryTimer = setTimeout(() => { this.loginRetryTimer = null; this.nextReconnectAt = 0; void this.connectAndAuth(); }, delay); } finally { this.loginInFlight = false; } } /** * Never-dead invariant, checked every minute. Every failure path above is * supposed to leave a retry scheduled; this catches the ones that don't * (and sessions that died without anything failing loudly): * - no token, no login running, no retry pending → log in now; * - logged in >10 min ago with no good sync for >10 min → the token may * have died in a way validate didn't catch → log in again. */ private watchdog(): void { if (!this.name || this.intentionalDisconnect) return; if (this.loginInFlight || this.loginRetryTimer) return; const now = Date.now(); if (!this.token) { console.warn(`[topstepx] watchdog: ${this.name} has no session and no login pending — logging in`); void this.connectAndAuth(); return; } if (now - this.lastLoginAt > SESSION_STALE_MS && now - this.lastSyncAt > SESSION_STALE_MS) { const since = this.lastSyncAt > 0 ? `in ${Math.round((now - this.lastSyncAt) / 60_000)} min` : 'since login'; console.warn(`[topstepx] watchdog: no good sync for ${this.name} ${since} — re-logging in`); void this.connectAndAuth(); } } /** LoginErrorCode names from the gateway swagger — the API itself sends * only the number with a null message. */ private static LOGIN_ERRORS: Record = { 1: 'UserNotFound', 2: 'PasswordVerificationFailed', 3: 'InvalidCredentials — check that the username is the TopstepX platform username the API key was generated for (not an email)', 4: 'AppNotFound', 5: 'AppVerificationFailed', 6: 'InvalidDevice', 7: 'AgreementsNotSigned — sign the API agreements in the TopstepX platform', 8: 'UnknownError', 9: 'ApiSubscriptionNotFound — the account has no API subscription', 10: 'ApiKeyAuthenticationDisabled', }; private async login(): Promise { const res = await axios.post(`${API_BASE}/api/Auth/loginKey`, { userName: this.name, apiKey: this.apiKey, }, { timeout: LOGIN_TIMEOUT_MS, // Just after connectAndAuth's own deadline: that one gives the // clean error; this one kills the request, so a stalled CONNECT // can't still log in minutes later. signal: AbortSignal.timeout(LOGIN_TIMEOUT_MS + 5_000), }); const data = res.data as { token?: string; success: boolean; errorCode: number; errorMessage: string | null }; if (!data.success || !data.token) { const detail = TopstepXClient.LOGIN_ERRORS[data.errorCode] ?? `errorCode=${data.errorCode}`; throw new Error(data.errorMessage ?? `loginKey failed: ${detail}`); } return data.token; } private async renewToken(): Promise { if (this.intentionalDisconnect || this.loginInFlight) return; if (!this.token) { void this.connectAndAuth(); return; } const sent = this.token; try { const res = await withTimeout(axios.post(`${API_BASE}/api/Auth/validate`, {}, { headers: this.authHeaders(), timeout: VALIDATE_TIMEOUT_MS, signal: AbortSignal.timeout(VALIDATE_TIMEOUT_MS + 5_000), }), VALIDATE_TIMEOUT_MS, 'token validate'); if (this.intentionalDisconnect) return; const data = res.data as { success: boolean; newToken?: string }; if (data.success && data.newToken) { // A login that landed meanwhile holds a newer session — keep it. if (this.token !== sent) return; this.token = data.newToken; this.tokenRefreshedAt = Date.now(); console.log('[topstepx] session token renewed'); } else if (!data.success) { console.warn('[topstepx] token validate failed — re-authenticating'); void this.connectAndAuth(); } } catch (err) { if (this.intentionalDisconnect) return; // Never the error object: its request config carries the Bearer token. console.error(`[topstepx] token renewal error — re-authenticating: ${errorMessage(err)}`); void this.connectAndAuth(); } } // ── realtime hub ───────────────────────────────────────────────────────── /** * Bring the user hub to connected-and-subscribed, as far as one pass can. * Called on login and every sync tick, and never awaited by anything that * matters: REST polling carries every feature on its own (slower), so a * wedged hub must never hold up a sync. * * - no hub → build a fresh one and start it; a failed start discards it * (a HubConnection whose first start failed never auto-reconnects), so * the next pass builds another; * - Disconnected (automatic reconnect gave up) → start it again, discard * on failure; * - Connecting/Reconnecting/Disconnecting for > 2 min → stuck; discard it * and build a fresh one; * - Connected → subscribe whatever isn't subscribed yet. * * Single-flight; a flight running past HUB_SETUP_WEDGED_MS is superseded * (every await in it has a deadline, so that is belt and braces). */ private async ensureHub(): Promise { if (this.intentionalDisconnect || !this.token) return; if (this.hubSetupInFlight) { const runningMs = Date.now() - this.hubSetupSince; if (runningMs < HUB_SETUP_WEDGED_MS) { this.hubRecheckWanted = true; return; } console.warn(`[topstepx] hub setup running for ${Math.round(runningMs / 1000)}s — looks wedged, starting over`); } const gen = ++this.hubSetupGeneration; this.hubSetupInFlight = true; this.hubSetupSince = Date.now(); this.hubRecheckWanted = false; let clean = false; try { clean = await this.ensureHubPass(gen); } catch (err) { console.error(`[topstepx] hub setup failed — running on REST polling only: ${errorMessage(err)}`); } finally { // A superseded flight must not clear its successor's flag. if (gen === this.hubSetupGeneration) { this.hubSetupInFlight = false; this.hubSetupSince = 0; // Asked for again mid-flight (typically: the first sync's // accounts landed while the hub was starting). Only worth a // rerun after a clean pass on a live hub — a failed start or // subscribe waits for the next tick rather than retrying hot. if (clean && this.hubRecheckWanted && !this.intentionalDisconnect && this.hub?.state === signalR.HubConnectionState.Connected) { this.hubRecheckWanted = false; void this.ensureHub(); } } } } /** One ensureHub pass. True when it ended connected with everything * subscribed; false when it stopped short (the next call carries on). */ private async ensureHubPass(gen: number): Promise { const State = signalR.HubConnectionState; let hub = this.hub; if (hub && (hub.state === State.Connecting || hub.state === State.Reconnecting || hub.state === State.Disconnecting)) { const now = Date.now(); if (!this.hubStalledSince) this.hubStalledSince = now; // Automatic reconnect is still working on it — let it. if (now - this.hubStalledSince < HUB_STALL_MS) return false; console.warn(`[topstepx] hub stuck ${hub.state} for ${Math.round((now - this.hubStalledSince) / 1000)}s — discarding it`); this.discardHub(); hub = null; } if (!hub) { hub = this.buildHub(); this.hub = hub; this.subscribedAccountIds.clear(); this.accountsFeedSubscribed = false; if (!(await this.startHub(hub, gen))) return false; console.log('[topstepx] user hub connected'); } else if (hub.state === State.Disconnected) { // Subscriptions died with the old connection. this.subscribedAccountIds.clear(); this.accountsFeedSubscribed = false; if (!(await this.startHub(hub, gen))) return false; console.log('[topstepx] user hub restarted'); } if (hub.state !== State.Connected) return false; this.hubStalledSince = 0; return this.subscribeAll(hub, gen); } /** Start `hub` against a deadline. False when it didn't come up (the hub * is discarded) or was discarded / superseded while starting. */ private async startHub(hub: signalR.HubConnection, gen: number): Promise { try { await withTimeout(hub.start(), HUB_START_TIMEOUT_MS, 'hub start'); } catch (err) { // REST polling still works without the hub; the next sync tick // builds a fresh one. console.error(`[topstepx] hub connect failed — falling back to REST polling: ${errorMessage(err)}`); if (this.hub === hub) this.discardHub(); return false; } if (this.hub !== hub) { // Discarded (disconnect, or a stall check) while it was starting. void hub.stop().catch(() => { /* discarded */ }); return false; } if (gen !== this.hubSetupGeneration) return false; this.hubStalledSince = 0; this.lastInboundAt = Date.now(); return true; } /** Subscribe everything the current connection isn't subscribed to yet. * A failed or timed-out invoke is left for the next pass. True when * nothing was left over. */ private async subscribeAll(hub: signalR.HubConnection, gen: number): Promise { const stillCurrent = () => this.hub === hub && gen === this.hubSetupGeneration && hub.state === signalR.HubConnectionState.Connected; const invoke = async (method: string, ...args: unknown[]) => { await withTimeout(hub.invoke(method, ...args), HUB_INVOKE_TIMEOUT_MS, method); this.lastInboundAt = Date.now(); }; try { if (!this.accountsFeedSubscribed) { await invoke('SubscribeAccounts'); if (!stillCurrent()) return false; this.accountsFeedSubscribed = true; } for (const acc of this.accountList) { if (this.subscribedAccountIds.has(acc.id)) continue; await invoke('SubscribeOrders', acc.id); await invoke('SubscribePositions', acc.id); await invoke('SubscribeTrades', acc.id); if (!stillCurrent()) return false; this.subscribedAccountIds.add(acc.id); } return true; } catch (err) { console.error(`[topstepx] hub subscribe failed — retrying on the next sync: ${errorMessage(err)}`); return false; } } /** Drop the current hub without waiting on it — stop() can hang as long * as the start it interrupts, so it is fire-and-forget. The hub's * handlers go quiet the moment this.hub stops pointing at it. */ private discardHub(): void { const hub = this.hub; this.hub = null; this.subscribedAccountIds.clear(); this.accountsFeedSubscribed = false; this.hubStalledSince = 0; if (hub) void hub.stop().catch(() => { /* discarded */ }); } /** A fresh HubConnection with every handler wired. Each handler first * checks its hub is still the current one: a discarded connection that * lingers (or reconnects on its own) must never write state. */ private buildHub(): signalR.HubConnection { // signalr's Node path does an unconditional dynamic require('ws') / // require('eventsource') when constructing the connection — bundled // builds throw "dynamic usage of require is not supported" on it, so // the package is listed in next.config serverExternalPackages and any // residual failure here is caught by ensureHub (REST polling covers // everything the hub does, just slower). // With skipNegotiation the hub is a single websocket to a broker // endpoint, so it follows the operator's proxy like the REST calls do. const proxiedWs = signalrWebSocket(); const hub = new signalR.HubConnectionBuilder() .withUrl(USER_HUB, { skipNegotiation: true, transport: signalR.HttpTransportType.WebSockets, accessTokenFactory: () => this.token ?? '', // eslint-disable-next-line @typescript-eslint/no-explicit-any ...(proxiedWs ? { WebSocket: proxiedWs as any } : {}), }) .withAutomaticReconnect() .configureLogging(signalR.LogLevel.Warning) .build(); // eslint-disable-next-line @typescript-eslint/no-explicit-any const unwrap = (raw: unknown): Record => { // eslint-disable-next-line @typescript-eslint/no-explicit-any const r = raw as Record; return (r && typeof r === 'object' && 'data' in r && typeof r.data === 'object') ? r.data : r; }; hub.on('GatewayUserAccount', (raw: unknown) => { if (this.hub !== hub) return; this.lastInboundAt = Date.now(); const acc = unwrap(raw); if (typeof acc?.id !== 'number') return; this.bufferEvent('account', acc); const existing = this.accountCashBalances[acc.id]; if (typeof acc.balance === 'number') { this.accountCashBalances[acc.id] = { amount: acc.balance, realizedPnL: existing?.realizedPnL ?? 0, }; } if (typeof acc.canTrade === 'boolean') { this.accountList = this.accountList.map((a) => a.id === acc.id ? { ...a, active: acc.canTrade } : a); } }); hub.on('GatewayUserPosition', (raw: unknown) => { if (this.hub !== hub) return; this.lastInboundAt = Date.now(); const pos = unwrap(raw); if (typeof pos?.accountId !== 'number') return; this.bufferEvent('position', pos); this.applyPosition(pos as { accountId: number; contractId: string; type: number; size: number; averagePrice: number; creationTimestamp?: string }); }); hub.on('GatewayUserTrade', (raw: unknown) => { if (this.hub !== hub) return; this.lastInboundAt = Date.now(); const trade = unwrap(raw) as unknown as TsxTrade; if (typeof trade?.accountId !== 'number') return; this.bufferEvent('trade', trade); this.recordTodayTrade(trade); // Wake anything waiting on this order's fill. const waiters = this.fillWaiters.filter((w) => w.orderId === trade.orderId); this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== trade.orderId); for (const w of waiters) { try { w.callback(trade); } catch (err) { console.error(`[topstepx] fill callback error: ${errorMessage(err)}`); } } }); hub.on('GatewayUserOrder', (raw: unknown) => { if (this.hub !== hub) return; this.lastInboundAt = Date.now(); const order = unwrap(raw); if (typeof order?.id !== 'number') return; this.bufferEvent('order', order); this.handleOcoOrderEvent(order.id, order.status); }); hub.onreconnecting(() => { if (this.hub !== hub) return; // The stall clock ensureHub judges "stuck Reconnecting" by. if (!this.hubStalledSince) this.hubStalledSince = Date.now(); }); hub.onreconnected(() => { if (this.hub !== hub) return; this.lastInboundAt = Date.now(); this.hubStalledSince = 0; console.log('[topstepx] hub reconnected — resubscribing'); this.subscribedAccountIds.clear(); this.accountsFeedSubscribed = false; void this.ensureHub(); void this.requestSync(); }); hub.onclose((err) => { if (this.hub !== hub || this.intentionalDisconnect) return; this.hubStalledSince = 0; console.warn(`[topstepx] hub closed — the next sync restarts it${err ? `: ${errorMessage(err)}` : ''}`); }); return hub; } private bufferEvent(entityType: string, entity: unknown): void { this.recentEntityEvents.push({ entityType, eventType: 'Updated', entity, ts: Date.now() }); if (this.recentEntityEvents.length > 50) this.recentEntityEvents.shift(); } private applyPosition(pos: { accountId: number; contractId: string; type: number; size: number; averagePrice: number; creationTimestamp?: string }): void { // PositionType: 1 Long, 2 Short. Size 0 (or type 0) means flat. if (pos.size > 0 && (pos.type === 1 || pos.type === 2)) { this.positions[pos.accountId] = { contractId: pos.contractId, netPos: pos.type === 1 ? pos.size : -pos.size, netPrice: pos.averagePrice, timestamp: pos.creationTimestamp ? new Date(pos.creationTimestamp) : new Date(), }; } else { delete this.positions[pos.accountId]; // Flat → any surviving emulated-OCO exit legs must go. void this.cancelOcoLegsForAccount(pos.accountId); } } private recordTodayTrade(trade: TsxTrade): void { if (trade.voided) return; const tradeDate = currentTradeDateCT(new Date(trade.creationTimestamp)); if (tradeDate !== currentTradeDateCT()) return; const map = (this.todayTrades[trade.accountId] ??= new Map()); map.set(trade.id, trade); this.recomputeRealized(trade.accountId); } private recomputeRealized(accountId: number): void { const today = currentTradeDateCT(); const map = this.todayTrades[accountId]; let realized = 0; if (map) { for (const t of map.values()) { if (t.voided) continue; if (currentTradeDateCT(new Date(t.creationTimestamp)) !== today) continue; realized += (t.profitAndLoss ?? 0) - (t.fees ?? 0); } } const existing = this.accountCashBalances[accountId]; this.accountCashBalances[accountId] = { amount: existing?.amount ?? 0, realizedPnL: Math.round(realized * 100) / 100, }; } // ── sync ───────────────────────────────────────────────────────────────── /** * One REST sync of accounts, positions and today's trades. Single-flight, * but a sync still running after SYNC_WEDGED_MS is presumed wedged and a * fresh one starts: nothing may hold syncInFlight forever, because a firm * that stops syncing stops trading. The superseded run checks its * generation after every await and bows out without touching state, and * its finally leaves the new run's flag alone. */ private async requestSync(): Promise { if (!this.token || this.intentionalDisconnect) return; const startedAt = Date.now(); if (this.syncInFlight) { const runningMs = startedAt - this.syncInFlightSince; // A caller that awaits gets the running pass's end. if (runningMs < SYNC_WEDGED_MS) return this.syncRunning; console.warn(`[topstepx] sync running for ${Math.round(runningMs / 1000)}s — looks wedged, starting a fresh one`); } const gen = ++this.syncGeneration; const superseded = () => gen !== this.syncGeneration || this.intentionalDisconnect; this.syncInFlight = true; this.syncInFlightSince = startedAt; let finished!: () => void; this.syncRunning = new Promise((resolve) => { finished = resolve; }); try { const accRes = await this.post<{ accounts: { id: number; name: string; balance: number; canTrade: boolean; isVisible: boolean }[] }>( '/api/Account/search', { onlyActiveAccounts: false }); if (superseded()) return; this.accountList = (accRes.accounts ?? []) .filter((a) => a.isVisible) .map((a) => ({ id: a.id, name: a.name, active: a.canTrade })); // Trades since yesterday cover the whole current CT trade date // (which opens at 5 PM the prior evening) with margin. const start = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000).toISOString(); const end = new Date(Date.now() + 60 * 60 * 1000).toISOString(); const today = currentTradeDateCT(); /** Accounts whose positions or trades didn't come back this pass. */ const staleAccounts: string[] = []; for (const acc of this.accountList) { const balance = (accRes.accounts ?? []).find((a) => a.id === acc.id)?.balance ?? 0; let accountFresh = true; try { const posRes = await this.post<{ positions: { accountId: number; contractId: string; type: number; size: number; averagePrice: number; creationTimestamp: string }[] }>( '/api/Position/searchOpen', { accountId: acc.id }); if (superseded()) return; const open = (posRes.positions ?? []).filter((p) => p.size > 0); if (open.length > 0) { this.applyPosition(open[0]); } else { delete this.positions[acc.id]; } } catch (err) { accountFresh = false; console.error(`[topstepx] position sync failed for ${acc.name}: ${errorMessage(err)}`); } if (superseded()) return; try { const tradeRes = await this.post<{ trades: TsxTrade[] }>( '/api/Trade/search', { accountId: acc.id, startTimestamp: start, endTimestamp: end }); if (superseded()) return; const map = new Map(); for (const t of tradeRes.trades ?? []) { if (t.voided) continue; if (currentTradeDateCT(new Date(t.creationTimestamp)) !== today) continue; map.set(t.id, t); } this.todayTrades[acc.id] = map; } catch (err) { accountFresh = false; console.error(`[topstepx] trade sync failed for ${acc.name}: ${errorMessage(err)}`); } if (superseded()) return; if (!accountFresh) staleAccounts.push(acc.name); this.accountCashBalances[acc.id] = { amount: balance, realizedPnL: 0 }; this.recomputeRealized(acc.id); this.autoLiqThresholds[acc.id] = 0; // Withdrawal/deposit detection — the books must explain the balance. // Never off a failed fetch: a fresh balance against stale trades // would book the unseen P&L as a (permanent) fund event. let reconciled = false; try { if (accountFresh) reconciled = this.reconcileFunds(acc); } catch (err) { console.error(`[topstepx] fund reconcile failed for ${acc.name}: ${errorMessage(err)}`); } // A freshly fetched history is tradable once a fresh pass has // actually checked the funds behind the balance (a skipped // check — a position still open — waits for a later pass). const ours = Boolean(getAccountConfig(acc.id).accountType); if ((reconciled || !ours) && this.reconcilePendingIds.delete(acc.id)) this.historyLoadedIds.add(acc.id); try { this.updateMllFloor(acc); } catch (err) { console.error(`[topstepx] MLL reconstruction failed for ${acc.name}: ${errorMessage(err)}`); } } console.log(`[topstepx] sync — ${this.accountList.length} account(s), ${Object.keys(this.positions).length} open position(s)`); // Fire-and-forget: the hub subscribes new accounts on its own // schedule and can never hold this sync up. void this.ensureHub(); this.maybeFetchDaysTraded(); this.syncComplete = true; // lastSyncAt is what the scheduler trades on (isClientReady), so it // certifies only a pass where every account's positions and trades // came back: a stale-flat position or a stale realized P&L walks // straight through the "already in trade" / "already traded today" // gates. Nor before the first daily P&L pass has landed — until // then the plan would run on no history at all. if (staleAccounts.length > 0) { console.warn(`[topstepx] sync incomplete — not refreshed: ${staleAccounts.join(', ')}; not counted as a good sync`); } else if (this.historyLoaded) { this.lastSyncAt = Date.now(); } // The resolver runs off this; its failure is not a sync failure. try { await this.callbackOnSyncRequest(); } catch (err) { console.error(`[topstepx] post-sync callback failed: ${errorMessage(err)}`); } } catch (err) { console.error(`[topstepx] sync failed: ${errorMessage(err)}`); } finally { if (gen === this.syncGeneration) { this.syncInFlight = false; this.syncInFlightSince = 0; } finished(); } } /** Re-slice the active cycle out of the full history using the last fund * transaction. A withdrawal's day belongs to the OLD cycle (its min-days * were booked before the payout) so it is excluded; a deposit's day is * included. */ private applyCycleSlice(accountId: number): void { const merged = this.fullDailyPnL[accountId] ?? []; const fundDate = this.lastFundDates[accountId]; const txns = this.fundTransactions[accountId] ?? []; const lastAmt = fundDate ? (txns.find((f) => f.date === fundDate)?.amount ?? null) : null; const isWithdrawal = lastAmt !== null && lastAmt < 0; const active = fundDate ? merged.filter((d) => (isWithdrawal ? d.date > fundDate : d.date >= fundDate)) : merged; const prior = fundDate ? merged.filter((d) => (isWithdrawal ? d.date <= fundDate : d.date < fundDate)) : []; this.priorProfit[accountId] = Math.round(prior.reduce((s, d) => s + d.pnl, 0) * 100) / 100; this.dailyPnL[accountId] = active; this.daysTraded[accountId] = active.filter((d) => d.pnl !== 0).length; } /** * Books invariant: for a $0-start account, balance must equal all trade * P&L plus all recorded fund transactions. TopstepX has no funds API, so a * divergence IS the fund event — a withdrawal (or deposit) the platform * never told us about. Recording it advances the withdrawal-stage counter * and starts a new cycle; the very first check also books any historical * net withdrawal in one lump (fine — the last stage repeats). * * $200 keeps fee noise and rounding out; the smallest real withdrawal in * the drain flow is well above it. Skipped while a position is open (fills * in flight) or while the trade history is known-stale. */ /** Books a withdrawal/deposit the trades can't explain. True once it has * actually checked the books against the balance (false when it had to * skip: a position open, no history or balance yet, no account type). */ private reconcileFunds(account: PlatformAccount): boolean { if (this.positions[account.id]) return false; if (this.lastFetchErrors[account.name]) return false; const history = this.fullDailyPnL[account.id]; if (!history) return false; // first daily-P&L fetch hasn't landed yet const cash = this.accountCashBalances[account.id]; if (!cash) return false; // The account's starting balance ($0 for Express, the funded size // otherwise) is part of the balance the books must explain — without // it a funded account books a phantom deposit for its whole size. Until // the account type is resolved that number is unknown, so hold off // rather than guess $0 and misfire on a funded account. const acctType = getAccountConfig(account.id).accountType; if (!acctType) return false; const startBal = acctType.accountSize; const today = currentTradeDateCT(); const tradePnL = history .filter((d) => d.date !== today) .reduce((s, d) => s + d.pnl, 0) + cash.realizedPnL; const txns = this.fundTransactions[account.id] ?? []; const funds = txns.reduce((s, f) => s + f.amount, 0); const delta = Math.round((cash.amount - startBal - tradePnL - funds) * 100) / 100; // The first check on this instance (or since a wipe) only has the // fetch window to explain the balance with: a lump it books is unseen // P&L and past payouts, not the end of a cycle this instance watched. const firstCheck = loadAccountMeta(account.id, 'funds_reconciled') === null; if (firstCheck) saveAccountMeta(account.id, 'funds_reconciled', today); if (Math.abs(delta) < 200) return true; console.log(`[topstepx] ${account.name}: balance ${cash.amount} vs books ${Math.round((tradePnL + funds) * 100) / 100} — recording ${delta < 0 ? 'withdrawal' : 'deposit'} of ${delta}`); // A new cycle starts today: note where the one it ends began, by // applyCycleSlice's rule for that date's own row, for the reporter's // catch-up. A second event today doesn't start another cycle. const prevFundDate = this.lastFundDates[account.id] ?? null; if (!firstCheck && prevFundDate !== today) { const prevAmt = prevFundDate ? (txns.find((f) => f.date === prevFundDate)?.amount ?? null) : null; saveEndedCycleStart(account.id, { fundDate: prevFundDate ?? '', includesDay: !(prevAmt !== null && prevAmt < 0) }); } // Same-day events merge into one row (the table keys on account+date), // so add today's delta onto whatever is already booked for today. const existingToday = txns.find((f) => f.date === today)?.amount ?? 0; saveFundTransactions(account.id, account.name, [ { date: today, amount: Math.round((existingToday + delta) * 100) / 100 }, ]); saveAccountMeta(account.id, 'last_fund_date', today); this.fundTransactions[account.id] = loadFundTransactions(account.id); this.lastFundDates[account.id] = today; this.applyCycleSlice(account.id); return true; } /** * Reverse-engineer the account's MLL floor — TopstepX never reports it. * * Topstep's Maximum Loss Limit sits `trailingDrawdown` below the END-OF-DAY * balance high, ratchets up only, and locks at the starting balance ($0 for * Express accounts) once the high has climbed the full drawdown: * * floor = min(0, eodBalanceHigh − trailingDrawdown) * * The EOD balance curve is rebuilt from daily P&L + recorded fund events. * Balance the books don't explain (sub-threshold off-book fees) is spread * across days proportionally to each day's reported fees — validated * against a platform-reported MLL to within $0.11. * * Stored in autoLiqThresholds (negative, or +$0.01 once locked at * breakeven — 0 must keep meaning "no floor"), which lights up the same * dead-account handling Tradovate's reported floors use. canTrade=false * accounts are skipped: the platform has already retired them. */ private updateMllFloor(account: PlatformAccount): void { // Deliberately computed for canTrade=false accounts too: when Topstep // liquidates a breach and disables the account, the floor must SURVIVE // so the dashboards show it as blown (☠), not merely inactive. Retired // accounts without an assigned type fall out below via dd = 0. const acctType = getAccountConfig(account.id).accountType; const dd = acctType?.trailingDrawdown ?? 0; // The account's real starting balance: $0 for Express, the funded size // otherwise. Everything below is in ABSOLUTE balance terms so the floor // is directly comparable to cash.amount. const startBal = acctType?.accountSize ?? 0; const history = this.fullDailyPnL[account.id]; const cash = this.accountCashBalances[account.id]; if (!(dd > 0) || !history || !cash) { this.autoLiqThresholds[account.id] = 0; return; } // Only move the floor on SETTLED books: never mid-trade (the fill // hasn't landed — the dead check must not fire off half-processed // state), and never while the balance disagrees with the books by // more than fee noise (a fill or fund event is still being absorbed; // reconcileFunds or the next sync settles it). The previous floor // stands until then, so `balance ≤ floor` can only come true off a // landed, booked fill. if (this.positions[account.id]) return; const today = currentTradeDateCT(); const fees = this.dailyFees[account.id] ?? new Map(); // One event per past date: booked P&L plus any fund transaction. const events = new Map(); const add = (date: string, amount: number) => events.set(date, (events.get(date) ?? 0) + amount); for (const d of history) if (d.date !== today) add(d.date, d.pnl); for (const f of this.fundTransactions[account.id] ?? []) if (f.date !== today) add(f.date, f.amount); const dates = [...events.keys()].sort(); if (dates.length === 0) { this.autoLiqThresholds[account.id] = 0; return; } // What the books don't explain (bounded by the reconcile threshold) is // off-book fee drift — attribute it in proportion to reported fees. // Today is handled live: the balance carries today's realized P&L and // any fund event detected today, neither of which is in `events`. const booked = dates.reduce((s, d) => s + events.get(d)!, 0); const todayFund = (this.fundTransactions[account.id] ?? []).find((f) => f.date === today)?.amount ?? 0; // Books invariant: cash.amount = startBal + Σ(booked P&L + funds) + // today's realized + today's fund. The residual is off-book fee drift. const residual = Math.round((cash.amount - startBal - cash.realizedPnL - todayFund - booked) * 100) / 100; // Books not settled (an unabsorbed fill or an unbooked fund event) — // keep the previous floor rather than skewing the curve with it. if (Math.abs(residual) >= 200) return; const totalFees = dates.reduce((s, d) => s + (fees.get(d) ?? 0), 0); let bal = startBal; let high = startBal; // the MLL trails EOD balance, starting a drawdown below startBal for (const d of dates) { const drift = totalFees > 0 ? residual * ((fees.get(d) ?? 0) / totalFees) : (d === dates[0] ? residual : 0); bal += events.get(d)! + drift; if (bal > high) high = bal; } // Absolute floor: a drawdown below the EOD high, locked at the starting // balance once the high has climbed the full drawdown above it. let floor = high >= startBal + dd ? startBal : Math.round((high - dd) * 100) / 100; // 0 is the "no floor" sentinel — a real $0 floor (Express, locked at // breakeven) takes a cent so it still reads as dead-eligible. if (floor === 0) floor = 0.01; this.autoLiqThresholds[account.id] = floor; } private maybeFetchDaysTraded(): void { if (this.daysFetchInFlight || this.intentionalDisconnect) 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, or wiped) get their own fetch now rather than wait out // the hour. A failure falls back to the 5-minute retry. const waiting = !initialFetchDue && !retryPending ? this.accountList .filter((a) => !this.historyLoadedIds.has(a.id) && !this.reconcilePendingIds.has(a.id) && !this.fetchingIds.has(a.id)) .map((a) => a.id) : []; const targeted = !retryDue && !hourlyRefreshDue && !initialFetchDue && waiting.length > 0; if (!initialFetchDue && !retryDue && !hourlyRefreshDue && !targeted) return; this.daysFetchInFlight = true; this.daysFetchInFlightSince = now; this.lastDaysFetchStartedAt = now; if (retryDue) this.nextDaysFetchRetryAt = 0; this.fetchDaysTraded(targeted ? waiting : undefined) .then(({ failedAccounts }) => { if (!this.historyLoaded && !this.intentionalDisconnect) { // Readiness waited on this; sync again now rather than // leaving the firm unready until the next 60s tick. this.historyLoaded = true; void this.requestSync(); } if (failedAccounts > 0) { this.nextDaysFetchRetryAt = Date.now() + 5 * 60 * 1_000; console.warn(`[topstepx] ${failedAccounts} account(s) failed daily P&L fetch — 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; console.error(`[topstepx] daily P&L fetch failed — retrying in 5 minutes: ${errorMessage(err)}`); }) .finally(() => { this.daysFetchInFlight = false; this.daysFetchInFlightSince = 0; }); } /** Daily P&L from the trades endpoint: net of fees per CT trade date, like * Tradovate's Cash History. Merged over the DB cache so history older than * the fetch window survives. */ public async fetchDaysTraded(onlyAccountIds?: number[]): Promise<{ failedAccounts: number }> { const ids = this.accountList .filter((a) => !onlyAccountIds || onlyAccountIds.includes(a.id)) .map((a) => a.id); for (const id of ids) this.fetchingIds.add(id); try { return await this.fetchDaysTradedPass(onlyAccountIds); } finally { for (const id of ids) this.fetchingIds.delete(id); } } private async fetchDaysTradedPass(onlyAccountIds?: number[]): Promise<{ failedAccounts: number }> { if (!this.token) return { failedAccounts: 0 }; this.fetchDaysComplete = false; let failedAccounts = 0; const start = new Date(Date.now() - 27 * 24 * 60 * 60 * 1000).toISOString(); const end = new Date(Date.now() + 60 * 60 * 1000).toISOString(); 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) { if (this.intentionalDisconnect) break; // a discarded client fetches (and writes) nothing more if (onlyAccountIds && !onlyAccountIds.includes(account.id)) continue; // Fund transactions never come from the TopstepX API — only what the // DB already holds (imports, manual rows). Usually empty. const storedFundDate = loadAccountMeta(account.id, 'last_fund_date'); this.lastFundDates[account.id] = storedFundDate; this.fundTransactions[account.id] = loadFundTransactions(account.id); const cached = loadDailyPnL(account.id); try { const res = await this.post<{ trades: TsxTrade[] }>( '/api/Trade/search', { accountId: account.id, startTimestamp: start, endTimestamp: end }); const trades = (res.trades ?? []).filter((t) => !t.voided); this.lastFetchRaw[account.name] = trades.length > 0 ? JSON.stringify(trades[0]) : '(no trades)'; const dailyMap: { [date: string]: { pnl: number; fees: number } } = {}; for (const t of trades) { const date = currentTradeDateCT(new Date(t.creationTimestamp)); const entry = (dailyMap[date] ??= { pnl: 0, fees: 0 }); entry.pnl += (t.profitAndLoss ?? 0) - (t.fees ?? 0); entry.fees += t.fees ?? 0; } const fresh = Object.entries(dailyMap) .map(([date, e]) => ({ date, pnl: Math.round(e.pnl * 100) / 100, fees: Math.round(e.fees * 10000) / 10000, })) .sort((a, b) => a.date.localeCompare(b.date)); const merged = mergePnL(cached, fresh); // Reported fees per day — the weights the MLL reconstruction // uses to place off-book fee drift. Fresh overwrites cached. const fees = new Map(); for (const e of cached) fees.set(e.date, e.fees ?? 0); for (const e of fresh) fees.set(e.date, e.fees); this.dailyFees[account.id] = fees; // The last fund transaction (usually one this client detected // itself — see reconcileFunds) splits cycles: withdrawals // exclude their day, deposits include it. this.fullDailyPnL[account.id] = merged; this.applyCycleSlice(account.id); saveDailyPnL(account.id, account.name, fresh); delete this.lastFetchErrors[account.name]; // One not loaded before (new, wiped, failed) is tradable after // the next sync reconciles its funds (requestSync); a refresh // of a loaded one keeps the funds it already has. if (!this.historyLoadedIds.has(account.id)) this.reconcilePendingIds.add(account.id); this.updateMllFloor(account); } catch (err) { const msg = errorMessage(err); console.error(`[topstepx] trade history failed for ${account.name}: ${msg}`); failedAccounts++; if (cached.length > 0) { this.fullDailyPnL[account.id] = cached; this.dailyFees[account.id] = new Map(cached.map((e) => [e.date, e.fees ?? 0])); this.applyCycleSlice(account.id); this.lastFetchErrors[account.name] = `${msg} (using ${cached.length} cached entries from database)`; // Not loaded before: no funds check has run on this cache, // and none will while the fetch error stands — so it waits // for a fetch that lands (reconcileFunds, requestSync). if (!this.historyLoadedIds.has(account.id)) this.reconcilePendingIds.add(account.id); } else { // No history at all is not an empty one: 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; } } } this.fetchDaysComplete = true; // The accounts just fetched wait on a reconciling sync: run one now // rather than leave them out for up to a minute, and before returning, // so a wipe route reads the result. (The firm's first pass syncs from // maybeFetchDaysTraded once readiness allows.) const waiting = () => this.accountList.some((a) => this.reconcilePendingIds.has(a.id) && (!onlyAccountIds || onlyAccountIds.includes(a.id))); if (waiting() && this.historyLoaded && !this.intentionalDisconnect) { await this.requestSync(); // a pass already running may have gone past them if (waiting()) await this.requestSync(); } return { failedAccounts }; } /** Whether this account's history can be planned on — see PlatformClient. * (Fund transactions come from reconcileFunds within a sync, so no * balance-drift hold is needed here.) */ public historyReady(accountId: number): boolean { return this.historyLoadedIds.has(accountId) && this.dailyPnL[accountId] !== undefined && this.fundTransactions[accountId] !== undefined; } /** 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.reconcilePendingIds.delete(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.dailyFees[accountId]; delete this.lastFetchErrors[accountName]; delete this.lastFetchRaw[accountName]; } // ── contracts ──────────────────────────────────────────────────────────── public async findFrontMonthContract(productName: string): Promise { if (!this.token) return null; const cached = this.contractByProduct.get(productName); if (cached && Date.now() - cached.resolvedAt < TopstepXClient.CONTRACT_TTL_MS) { const c = cached.contract; return { id: c.id, name: c.name, tickSize: c.tickSize }; } try { const res = await this.post<{ contracts: TsxContract[] }>( '/api/Contract/search', { searchText: productName, live: false }); const candidates = (res.contracts ?? []).filter((c) => c.name.startsWith(productName) && /^[A-Z]\d/.test(c.name.slice(productName.length))); if (candidates.length === 0) { console.log(`[topstepx] no contract found for ${productName}`); return null; } // Prefer the volume-resolved front month (shared with the Tradovate // path — names match across platforms, e.g. "NQZ5"), then TopstepX's // own activeContract flag. const resolvedName = getCachedContract(productName)?.name; const match = (resolvedName ? candidates.find((c) => c.name === resolvedName) : undefined) ?? candidates.find((c) => c.activeContract) ?? candidates[0]; this.contractByProduct.set(productName, { contract: match, resolvedAt: Date.now() }); this.contractByName.set(match.name, match); return { id: match.id, name: match.name, tickSize: match.tickSize }; } catch (err) { console.error(`[topstepx] contract search failed for ${productName}: ${errorMessage(err)}`); return null; } } /** Resolve a contract NAME (e.g. "NQZ5") to its TopstepX contract id. */ private async resolveContractId(contractSymbol: string): Promise { const known = this.contractByName.get(contractSymbol); if (known) return known; const res = await this.post<{ contracts: TsxContract[] }>( '/api/Contract/search', { searchText: contractSymbol, live: false }); const match = (res.contracts ?? []).find((c) => c.name === contractSymbol) ?? (res.contracts ?? []).find((c) => contractSymbol.startsWith(c.name)); if (!match) throw new Error(`TopstepX contract not found for symbol ${contractSymbol}`); this.contractByName.set(match.name, match); return match; } public async fetchInstrumentFees(): Promise<{ [symbol: string]: number }> { // No fee-parameter endpoint on the gateway; per-trade fees arrive with // each fill instead, which is what the commission delta reads. return {}; } public async getAccountList(): Promise { return this.accountList; } // ── orders ─────────────────────────────────────────────────────────────── /** * Everything an order needs before anything is sent: a session and the * contract. Failing here means nothing reached the broker, which is the * one case the scheduler may un-count a leg for — hence OrderNotPlacedError. */ private async contractForOrder(contractSymbol: string): Promise { if (!this.token) throw new OrderNotPlacedError('TopstepX not logged in — order not sent'); try { return await this.resolveContractId(contractSymbol); } catch (err) { throw new OrderNotPlacedError(`TopstepX contract lookup failed for ${contractSymbol} — order not sent: ${errorMessage(err)}`); } } /** * POST /api/Order/place. Only a failure that proves nothing was placed * becomes OrderNotPlacedError: no session, a gateway `success: false` * that names no order and no ambiguous code, or an HTTP refusal given * before any handler ran. Everything else (timeouts, resets, 5xx) may * have placed the order and stays a plain Error — rewrapped so the * AxiosError, whose config carries the Bearer token, never travels on * into someone's log. */ private async placeOrder( accountId: number, contractId: string, size: number, action: 'Buy' | 'Sell', type: number, limitPrice?: number, stopPrice?: number, ): Promise { if (!this.token) throw new OrderNotPlacedError('TopstepX not logged in — order not sent'); try { const res = await this.post<{ orderId: number }>('/api/Order/place', { accountId, contractId, type, side: action === 'Buy' ? SIDE_BUY : SIDE_SELL, size, limitPrice: limitPrice ?? null, stopPrice: stopPrice ?? null, trailPrice: null, customTag: null, }); return res.orderId; } catch (err) { if (err instanceof GatewayRefusedError) { const ambiguous = err.orderId !== undefined || (err.errorCode !== undefined && AMBIGUOUS_PLACE_ERRORS.has(err.errorCode)); if (!ambiguous) throw new OrderNotPlacedError(`TopstepX refused the order: ${err.message}`); throw new Error(`TopstepX order outcome unknown: ${err.message}`); } const response = (err as { response?: { status?: unknown; data?: { orderId?: unknown } } } | null)?.response; const status = response?.status; // Same rule as a refusal: an answer that names an order means it exists. const namesOrder = typeof response?.data?.orderId === 'number' && response.data.orderId > 0; if (typeof status === 'number' && REFUSED_BEFORE_HANDLER.has(status) && !namesOrder) { throw new OrderNotPlacedError(`TopstepX rejected the order request: ${errorMessage(err)}`); } throw new Error(`TopstepX order request failed (it may still have been placed): ${errorMessage(err)}`); } } /** Place an order and wait for its fill (trade event, with a REST poll as * backstop). Resolves with an object exposing the fill `price`. */ public async sendOrder( accountId: number, contractSymbol: string, quantity: number, action: 'Buy' | 'Sell', orderType: 'Market' | 'Limit', price?: number ): Promise> { const contract = await this.contractForOrder(contractSymbol); const orderId = await this.placeOrder( accountId, contract.id, quantity, action, orderType === 'Market' ? TYPE_MARKET : TYPE_LIMIT, orderType === 'Limit' ? price : undefined, ); console.log(`[topstepx] order ${orderId}: ${action} ${quantity}x${contractSymbol} ${orderType}${price ? ` @ ${price}` : ''}`); // A market fill can reach the hub before /api/Order/place answers, // when no waiter existed yet to hear it. The handler still recorded // it, so look there first. No await between here and the waiter // registered below, so nothing can slip in between. const early = this.bufferedFillPrice(accountId, orderId); if (early !== null) { console.log(`[topstepx] order ${orderId} filled @ ${early} (before the place response)`); return { orderId, price: early, accountId, contractId: contract.id }; } // Wide on purpose: the window is the local clock's, and a PC clock a // few minutes off would otherwise search past the fill. const since = new Date(Date.now() - 15 * 60_000).toISOString(); return new Promise((resolve, reject) => { let settled = false; const timeout = setTimeout(() => { if (settled) return; const late = this.bufferedFillPrice(accountId, orderId); if (late !== null) { this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); settle(late); return; } settled = true; clearInterval(poll); this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); // The caller gives up on this order and places no exits, so a // late fill would leave a position nothing manages — and one // open position blocks every firm through the open-position // gate. Cancel it, best effort. The rejection stays a plain // Error: it may have filled after all. console.warn(`[topstepx] order ${orderId} not filled within 30s — cancelling it`); void this.cancelOrder(accountId, orderId).catch((err) => console.error(`[topstepx] failed to cancel unfilled order ${orderId}: ${errorMessage(err)}`)); reject(new Error(`Order ${orderId} placed but no fill within 30s (market may be closed) — cancel requested`)); }, 30_000); const settle = (fillPrice: number) => { if (settled) return; settled = true; clearTimeout(timeout); clearInterval(poll); console.log(`[topstepx] order ${orderId} filled @ ${fillPrice}`); resolve({ orderId, price: fillPrice, accountId, contractId: contract.id }); }; this.fillWaiters.push({ orderId, callback: (trade) => settle(trade.price) }); // REST backstop for when the hub is down or the event is missed: // the order's trades (the same Trade/search fetchDaysTraded // reads), and every fifth poll whether the order died unfilled. // Both are documented Gateway endpoints; Order has no searchById. let polls = 0; const poll = setInterval(() => { polls++; void this.searchOrderFillPrice(accountId, orderId, since) .then((fillPrice) => { if (fillPrice === null) return; this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); settle(fillPrice); }) .catch(() => { /* transient — the next poll tries again */ }); if (polls % 5 !== 0) return; void this.post<{ orders?: { id: number; status: number }[] }>( '/api/Order/search', { accountId, startTimestamp: since }) .then(async (res) => { const order = res.orders?.find((o) => o.id === orderId); if (!order || !ORDER_GONE.has(order.status) || settled) return; // Cancelled after a partial fill still left a position. const fillPrice = this.bufferedFillPrice(accountId, orderId) ?? await this.searchOrderFillPrice(accountId, orderId, since); this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); if (fillPrice !== null) { settle(fillPrice); return; } if (settled) return; settled = true; clearTimeout(timeout); clearInterval(poll); reject(new Error(`Order ${orderId} ended without fill (status=${order.status})`)); }) .catch(() => { /* transient — the next poll tries again */ }); }, 2_000); }); } /** The price of an order's first fill already heard over the hub — the * price the fill waiter would have resolved with — or null. */ private bufferedFillPrice(accountId: number, orderId: number): number | null { const fills: TsxTrade[] = [...(this.todayTrades[accountId]?.values() ?? [])]; for (const event of this.recentEntityEvents) { if (event.entityType === 'trade') fills.push(event.entity as TsxTrade); } return firstFillPrice(fills.filter((t) => t?.accountId === accountId && t.orderId === orderId)); } /** The same, from the Gateway's trade history (REST). */ private async searchOrderFillPrice(accountId: number, orderId: number, since: string): Promise { const res = await this.post<{ trades?: TsxTrade[] }>('/api/Trade/search', { accountId, startTimestamp: since, endTimestamp: new Date(Date.now() + 15 * 60_000).toISOString(), }); return firstFillPrice((res.trades ?? []).filter((t) => t.orderId === orderId)); } /** Place an order and resolve on placement (does not wait for fill). */ public async placeOrderNoWait( accountId: number, contractSymbol: string, quantity: number, action: 'Buy' | 'Sell', orderType: 'Market' | 'Limit', price?: number ): Promise<{ orderId?: number }> { const contract = await this.contractForOrder(contractSymbol); const orderId = await this.placeOrder( accountId, contract.id, quantity, action, orderType === 'Market' ? TYPE_MARKET : TYPE_LIMIT, orderType === 'Limit' ? price : undefined, ); return { orderId }; } /** * Emulated OCO exit: the gateway has no linked-order type, so place the * take-profit limit and the stop separately and cancel the survivor when * either fills (order events; a poll and the flat-position hook back it * up). * * If the stop doesn't go in, what the caller may do next depends on what * is left at the broker — it falls back to a lone limit only on * OrderNotPlacedError, and places nothing more on a plain Error: * - stop refused, limit cancel confirmed → nothing of the pair is working: * OrderNotPlacedError, and the fallback starts from a clean book; * - stop refused, limit cancel failed → the limit may still be working: * plain Error (a fallback on top could double the exit); * - stop outcome unknown → it may be working, so the limit is kept (the * position keeps its take-profit, unpaired, as the fallback would have * left it) and the error is plain. */ public async placeOcoExit( accountId: number, contractSymbol: string, quantity: number, exitAction: 'Buy' | 'Sell', limitPrice: number, stopPrice: number, ): Promise<{ orderId?: number }> { const contract = await this.contractForOrder(contractSymbol); const limitId = await this.placeOrder(accountId, contract.id, quantity, exitAction, TYPE_LIMIT, limitPrice); let stopId: number; try { stopId = await this.placeOrder(accountId, contract.id, quantity, exitAction, TYPE_STOP, undefined, stopPrice); } catch (err) { if (!isOrderNotPlaced(err)) { throw new Error(`OCO stop leg outcome unknown — limit ${limitId} left working, unpaired: ${errorMessage(err)}`); } const limitCancelled = await this.cancelOrder(accountId, limitId).then(() => true, () => false); if (limitCancelled) { throw new OrderNotPlacedError(`OCO stop leg refused, limit ${limitId} cancelled — no exit working: ${errorMessage(err)}`); } throw new Error(`OCO stop leg refused and limit ${limitId} could not be cancelled — it may still be working: ${errorMessage(err)}`); } this.ocoPairs.push({ accountId, limitId, stopId, done: false, placedAt: Date.now() }); this.ensureOcoPoll(); console.log(`[topstepx] OCO exit pair placed — limit ${limitId} @ ${limitPrice}, stop ${stopId} @ ${stopPrice}`); return { orderId: limitId }; } private async cancelOrder(accountId: number, orderId: number): Promise { await this.post('/api/Order/cancel', { accountId, orderId }); } private handleOcoOrderEvent(orderId: number, status: number): void { const pair = this.ocoPairs.find((p) => !p.done && (p.limitId === orderId || p.stopId === orderId)); if (!pair) return; if (status === ORDER_FILLED) { pair.done = true; const sibling = pair.limitId === orderId ? pair.stopId : pair.limitId; console.log(`[topstepx] OCO leg ${orderId} filled — cancelling sibling ${sibling}`); void this.cancelOrder(pair.accountId, sibling).catch((err) => console.error(`[topstepx] failed to cancel OCO sibling ${sibling}: ${errorMessage(err)}`)); } } private async cancelOcoLegsForAccount(accountId: number): Promise { const pairs = this.ocoPairs.filter((p) => !p.done && p.accountId === accountId); for (const pair of pairs) { pair.done = true; for (const orderId of [pair.limitId, pair.stopId]) { await this.cancelOrder(accountId, orderId).catch(() => { /* already gone */ }); } } } /** Safety net under the event-driven OCO: while pairs are live, poll open * orders and reap legs whose sibling is gone or whose position is flat. */ private ensureOcoPoll(): void { if (this.ocoPollInterval) return; this.ocoPollInterval = setInterval(() => { void (async () => { const live = this.ocoPairs.filter((p) => !p.done); if (live.length === 0) { if (this.ocoPollInterval) { clearInterval(this.ocoPollInterval); this.ocoPollInterval = null; } this.ocoPairs = []; return; } for (const pair of live) { try { const res = await this.post<{ orders: { id: number }[] }>( '/api/Order/searchOpen', { accountId: pair.accountId }); const openIds = new Set((res.orders ?? []).map((o) => o.id)); const limitOpen = openIds.has(pair.limitId); const stopOpen = openIds.has(pair.stopId); if (!limitOpen && !stopOpen) { pair.done = true; } else if (limitOpen !== stopOpen && !this.positions[pair.accountId]) { // One leg gone, position flat — the survivor is stale. pair.done = true; const survivor = limitOpen ? pair.limitId : pair.stopId; console.log(`[topstepx] reaping stale OCO leg ${survivor} (position flat)`); await this.cancelOrder(pair.accountId, survivor).catch(() => { /* already gone */ }); } } catch { /* transient — retry next poll */ } } })(); }, 15_000); } public onFill(orderId: number, callback: (fill: Record) => void): void { this.fillWaiters.push({ orderId, callback: (trade) => callback(trade as unknown as Record) }); } // ── lifecycle ──────────────────────────────────────────────────────────── public health(): ClientHealth { return { platform: 'topstepx', authenticated: !!this.token, transport: this.hub ? String(this.hub.state) : 'none', syncComplete: this.syncComplete, lastSyncAt: this.lastSyncAt, lastLoginAt: this.lastLoginAt, lastLoginError: this.lastLoginError, consecutiveLoginFailures: this.consecutiveLoginFailures, nextReconnectAt: this.nextReconnectAt, // Validate hands out fresh 24h tokens; this is the latest one's expiry. tokenExpiresAt: this.token && this.tokenRefreshedAt ? this.tokenRefreshedAt + TOKEN_LIFETIME_MS : 0, lastInboundAt: this.lastInboundAt, pendingRequests: this.fillWaiters.length, syncInFlightSince: this.syncInFlight ? this.syncInFlightSince : 0, daysFetchInFlightSince: this.daysFetchInFlight ? this.daysFetchInFlightSince : 0, }; } /** Final: nothing this client scheduled fires afterwards, and every * in-flight login / sync / hub setup checks intentionalDisconnect (or * that its hub is still current) before touching state again. */ public disconnect(): void { this.intentionalDisconnect = true; if (this.syncInterval) { clearInterval(this.syncInterval); this.syncInterval = null; } if (this.tokenRenewalInterval) { clearInterval(this.tokenRenewalInterval); this.tokenRenewalInterval = null; } if (this.ocoPollInterval) { clearInterval(this.ocoPollInterval); this.ocoPollInterval = null; } if (this.loginRetryTimer) { clearTimeout(this.loginRetryTimer); this.loginRetryTimer = null; } if (this.watchdogInterval) { clearInterval(this.watchdogInterval); this.watchdogInterval = null; } this.nextReconnectAt = 0; this.daysFetchInFlight = false; this.daysFetchInFlightSince = 0; this.nextDaysFetchRetryAt = 0; this.discardHub(); } }