/** * 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 } from './db'; import { getAccountConfig } from './account-type-resolver'; import type { PlatformClient, PlatformAccount, PlatformContract } from './platform-client'; const API_BASE = 'https://api.topstepx.com'; const USER_HUB = 'https://rtc.topstepx.com/hubs/user'; 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]); 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; 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(); private syncInterval: ReturnType | null = null; private tokenRenewalInterval: ReturnType | null = null; private ocoPollInterval: ReturnType | null = null; private loginRetryTimer: ReturnType | null = null; private intentionalDisconnect = false; private syncInFlight = false; /** Same hourly-with-retry throttle the Tradovate client uses for reports. */ private daysFetchInFlight = false; private lastDaysFetchStartedAt = 0; private lastDaysFetchSucceededAt = 0; private nextDaysFetchRetryAt = 0; /** 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.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() }); const data = res.data as T & { success?: boolean; errorMessage?: string | null; errorCode?: number }; if (data && data.success === false) { throw new Error(`${path} failed: ${data.errorMessage ?? `errorCode=${data.errorCode}`}`); } return data; } private connectAndAuth(): void { if (this.intentionalDisconnect) 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.'; console.error(`[topstepx] ${this.lastFetchErrors['login']}`); return; } this.login() .then(async (token) => { this.token = token; 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) { clearInterval(this.syncInterval); this.syncInterval = null; } this.syncComplete = false; // Hub failure must never read as a login failure: REST polling // carries every feature (slower), and subscribeAll retries the // hub on each sync tick. try { await this.startHub(); } catch (err) { console.error('[topstepx] hub setup failed — running on REST polling only', err instanceof Error ? err.message : err); } await this.requestSync(); this.syncInterval = setInterval(() => void this.requestSync(), 60_000); }) .catch((err) => { const msg = err instanceof Error ? err.message : String(err); this.lastFetchErrors['login'] = msg; console.error(`[topstepx] login failed for ${this.name}: ${msg} — retrying in 60s`); if (this.loginRetryTimer) clearTimeout(this.loginRetryTimer); this.loginRetryTimer = setTimeout(() => this.connectAndAuth(), 60_000); }); } /** 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, }); 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 { try { const res = await axios.post(`${API_BASE}/api/Auth/validate`, {}, { headers: this.authHeaders() }); const data = res.data as { success: boolean; newToken?: string }; if (data.success && data.newToken) { this.token = data.newToken; console.log('[topstepx] session token renewed'); } else if (!data.success) { console.warn('[topstepx] token validate failed — re-authenticating'); this.connectAndAuth(); } } catch (err) { console.error('[topstepx] token renewal error — re-authenticating', err); this.connectAndAuth(); } } // ── realtime hub ───────────────────────────────────────────────────────── private async startHub(): Promise { if (this.hub) { try { await this.hub.stop(); } catch { /* ignore */ } this.hub = null; } this.subscribedAccountIds.clear(); // 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 the caller (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) => { 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) => { 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) => { 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', err); } } }); hub.on('GatewayUserOrder', (raw: unknown) => { const order = unwrap(raw); if (typeof order?.id !== 'number') return; this.bufferEvent('order', order); this.handleOcoOrderEvent(order.id, order.status); }); hub.onreconnected(() => { console.log('[topstepx] hub reconnected — resubscribing'); this.subscribedAccountIds.clear(); void this.subscribeAll(); void this.requestSync(); }); hub.onclose((err) => { if (this.intentionalDisconnect) return; console.warn('[topstepx] hub closed', err?.message ?? ''); }); this.hub = hub; try { await hub.start(); await this.subscribeAll(); console.log('[topstepx] user hub connected'); } catch (err) { // REST polling still works without the hub; automatic reconnect // only runs after a successful start, so retry via the next sync. console.error('[topstepx] hub connect failed — falling back to REST polling', err); } } private async subscribeAll(): Promise { if (!this.hub) return; // A hub whose initial start failed never auto-reconnects — each sync // tick gets another chance to bring it up. if (this.hub.state === signalR.HubConnectionState.Disconnected) { try { await this.hub.start(); this.subscribedAccountIds.clear(); console.log('[topstepx] user hub connected'); } catch { return; // REST polling carries on; retry next sync } } if (this.hub.state !== signalR.HubConnectionState.Connected) return; try { await this.hub.invoke('SubscribeAccounts'); for (const acc of this.accountList) { if (this.subscribedAccountIds.has(acc.id)) continue; await this.hub.invoke('SubscribeOrders', acc.id); await this.hub.invoke('SubscribePositions', acc.id); await this.hub.invoke('SubscribeTrades', acc.id); this.subscribedAccountIds.add(acc.id); } } catch (err) { console.error('[topstepx] hub subscribe failed', err); } } 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 ───────────────────────────────────────────────────────────────── private async requestSync(): Promise { if (!this.token || this.syncInFlight) return; this.syncInFlight = true; try { const accRes = await this.post<{ accounts: { id: number; name: string; balance: number; canTrade: boolean; isVisible: boolean }[] }>( '/api/Account/search', { onlyActiveAccounts: false }); 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(); for (const acc of this.accountList) { const balance = (accRes.accounts ?? []).find((a) => a.id === acc.id)?.balance ?? 0; 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 }); 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) { console.error(`[topstepx] position sync failed for ${acc.name}`, err instanceof Error ? err.message : err); } try { const tradeRes = await this.post<{ trades: TsxTrade[] }>( '/api/Trade/search', { accountId: acc.id, startTimestamp: start, endTimestamp: end }); 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) { console.error(`[topstepx] trade sync failed for ${acc.name}`, err instanceof Error ? err.message : err); } 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. try { this.reconcileFunds(acc); } catch (err) { console.error(`[topstepx] fund reconcile failed for ${acc.name}`, err instanceof Error ? err.message : err); } try { this.updateMllFloor(acc); } catch (err) { console.error(`[topstepx] MLL reconstruction failed for ${acc.name}`, err instanceof Error ? err.message : err); } } console.log(`[topstepx] sync — ${this.accountList.length} account(s), ${Object.keys(this.positions).length} open position(s)`); await this.subscribeAll(); this.maybeFetchDaysTraded(); this.syncComplete = true; await this.callbackOnSyncRequest(); } catch (err) { console.error('[topstepx] sync failed', err instanceof Error ? err.message : err); } finally { this.syncInFlight = false; } } /** 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. */ private reconcileFunds(account: PlatformAccount): void { if (this.positions[account.id]) return; if (this.lastFetchErrors[account.name]) return; const history = this.fullDailyPnL[account.id]; if (!history) return; // first daily-P&L fetch hasn't landed yet const cash = this.accountCashBalances[account.id]; if (!cash) return; // 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; 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; if (Math.abs(delta) < 200) return; console.log(`[topstepx] ${account.name}: balance ${cash.amount} vs books ${Math.round((tradePnL + funds) * 100) / 100} — recording ${delta < 0 ? 'withdrawal' : 'deposit'} of ${delta}`); // 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); } /** * 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) 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; if (!initialFetchDue && !retryDue && !hourlyRefreshDue) return; this.daysFetchInFlight = true; this.lastDaysFetchStartedAt = now; if (retryDue) this.nextDaysFetchRetryAt = 0; this.fetchDaysTraded() .then(({ failedAccounts }) => { 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; } 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', err); }) .finally(() => { this.daysFetchInFlight = false; }); } /** 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 }> { 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 (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]; this.updateMllFloor(account); } catch (err) { const msg = err instanceof Error ? err.message : String(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)`; } else { this.dailyPnL[account.id] ??= []; this.daysTraded[account.id] ??= 0; this.lastFetchErrors[account.name] = msg; } } } this.fetchDaysComplete = true; return { failedAccounts }; } // ── 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}`, err instanceof Error ? err.message : 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 ─────────────────────────────────────────────────────────────── private async placeOrder( accountId: number, contractId: string, size: number, action: 'Buy' | 'Sell', type: number, limitPrice?: number, stopPrice?: number, ): Promise { 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; } /** 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.resolveContractId(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}` : ''}`); return new Promise((resolve, reject) => { let settled = false; const timeout = setTimeout(() => { if (settled) return; settled = true; clearInterval(poll); this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); reject(new Error(`Order ${orderId} placed but no fill within 30s (market may be closed)`)); }, 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. const poll = setInterval(() => { void this.post<{ order?: { status: number; filledPrice?: number } }>( '/api/Order/searchById', { accountId, orderId }) .then((res) => { const order = res.order; if (!order) return; if (order.status === ORDER_FILLED && typeof order.filledPrice === 'number') { this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); settle(order.filledPrice); } else if (ORDER_GONE.has(order.status) && !settled) { settled = true; clearTimeout(timeout); clearInterval(poll); this.fillWaiters = this.fillWaiters.filter((w) => w.orderId !== orderId); reject(new Error(`Order ${orderId} ended without fill (status=${order.status})`)); } }) .catch(() => { /* transient — the next poll tries again */ }); }, 2_000); }); } /** 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.resolveContractId(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 can't be placed the limit is cancelled and the error * rethrown, so the caller's lone-limit fallback starts from a clean book. */ public async placeOcoExit( accountId: number, contractSymbol: string, quantity: number, exitAction: 'Buy' | 'Sell', limitPrice: number, stopPrice: number, ): Promise<{ orderId?: number }> { const contract = await this.resolveContractId(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) { await this.cancelOrder(accountId, limitId).catch(() => { /* best effort */ }); throw 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}`, err instanceof Error ? err.message : 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 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; } this.daysFetchInFlight = false; this.nextDaysFetchRetryAt = 0; void this.hub?.stop().catch(() => { /* ignore */ }); this.hub = null; } }