Brandon LiandClaude Opus 5.5
1eb0736fbe
Fourth review round (52ed9c7):
- The one-and-done guard counts only entries the broker confirmed filled,
so an entry that timed out or was rejected unfilled no longer benches the
account for the day.
- TopstepX: an account is promoted only by a sync that actually checked its
funds; one with no account type stays out (not traded) without costing a
sync per pass, and once typed waits for that check. Wipe routes don't
count untyped accounts as failures.
- Tradovate: an unchecked balance-hold release needs Cash History refused
for a full hour straight (a 4xx other than 401/403/408/429, or an answer
without data); any other outcome resets it, so one stray reply during a
5xx outage can't release a payout.
- A wipe keeps the pre-wipe baseline and fund rows as its yardstick: the
rebuild must explain today's balance against them, so a payout the
rebuilt report hasn't booked yet stays held.
- The roll base of an unrolled record is kept per trade date in
account_meta, so a restart later the same trade date counts from the same
point instead of from a value that already includes the day's fills.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1751 lines
88 KiB
TypeScript
1751 lines
88 KiB
TypeScript
/**
|
||
* 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<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
|
||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||
const deadline = new Promise<never>((_, 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<string, string> = {};
|
||
public lastFetchRaw: Record<string, string> = {};
|
||
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<void>;
|
||
private hub: signalR.HubConnection | null = null;
|
||
private subscribedAccountIds = new Set<number>();
|
||
/** 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<typeof setInterval> | null = null;
|
||
private tokenRenewalInterval: ReturnType<typeof setInterval> | null = null;
|
||
private ocoPollInterval: ReturnType<typeof setInterval> | null = null;
|
||
private loginRetryTimer: ReturnType<typeof setTimeout> | null = null;
|
||
private watchdogInterval: ReturnType<typeof setInterval> | 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<number>();
|
||
/** 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<number>();
|
||
/** Settles when the sync in flight ends (requestSync hands it to a caller
|
||
* that arrives mid-pass). */
|
||
private syncRunning: Promise<void> = 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<number>();
|
||
|
||
/** 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<number, TsxTrade> } = {};
|
||
|
||
/** 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<string, number> } = {};
|
||
|
||
/** Front-month contract per product symbol, plus name→contract for order
|
||
* placement (sendOrder receives the contract NAME, e.g. "NQZ5"). */
|
||
private contractByProduct = new Map<string, { contract: TsxContract; resolvedAt: number }>();
|
||
private contractByName = new Map<string, TsxContract>();
|
||
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<void>) {
|
||
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<T>(path: string, body: unknown): Promise<T> {
|
||
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<void> {
|
||
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<number, string> = {
|
||
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<string> {
|
||
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<void> {
|
||
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<void> {
|
||
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<boolean> {
|
||
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<boolean> {
|
||
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<boolean> {
|
||
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<string, any> => {
|
||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||
const r = raw as Record<string, any>;
|
||
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<void> {
|
||
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<void>((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<number, TsxTrade>();
|
||
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 waits for a later pass — a position still open, or no
|
||
// account type yet (not traded until one is assigned, and then
|
||
// only after a check).
|
||
if (reconciled && 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<string, number>();
|
||
|
||
// One event per past date: booked P&L plus any fund transaction.
|
||
const events = new Map<string, number>();
|
||
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<string, number>();
|
||
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<string, number>();
|
||
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.)
|
||
// Only ours: an account with no type stays pending without costing
|
||
// a sync on every pass.
|
||
const waiting = () => this.accountList.some((a) => this.reconcilePendingIds.has(a.id)
|
||
&& (!onlyAccountIds || onlyAccountIds.includes(a.id))
|
||
&& Boolean(getAccountConfig(a.id).accountType));
|
||
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<PlatformContract | null> {
|
||
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<TsxContract> {
|
||
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<PlatformAccount[]> {
|
||
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<TsxContract> {
|
||
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<number> {
|
||
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<Record<string, unknown>> {
|
||
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<number | null> {
|
||
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<void> {
|
||
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<void> {
|
||
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<string, unknown>) => void): void {
|
||
this.fillWaiters.push({ orderId, callback: (trade) => callback(trade as unknown as Record<string, unknown>) });
|
||
}
|
||
|
||
// ── 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();
|
||
}
|
||
}
|