From 9b2d67766202eeb072f8ebf5b683dba5908b9b29 Mon Sep 17 00:00:00 2001 From: Brandon Li Date: Thu, 24 Sep 2026 17:55:32 -0500 Subject: [PATCH] History holds: only a booked fund transaction lifts them; pace, persist, and cover wipes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to 77d8c2e from its adversarial review: - Stop-after-all no longer shuts down before an open while histories are still loading (a regression in 77d8c2e): a pending account counts as in play everywhere but the weekday 2:55-5:00 PM CT stretch after the close. - Tradovate balance holds lift only when a Cash History books a fund transaction the account didn't have when the hold began; one that never does is released after 3 clean reports over 15 minutes, with a warning. The baseline comes from the balance when the report was requested, so a payout landing mid-report is held for the next one. Held accounts refetch every 5 minutes, not every sync. - The Cash History baseline is kept in account_meta, so a restart onto a failing report server still catches a payout from the DB cache. - A wipe holds the account until a Cash History rebuilds its fund transactions (the Fills report and the cache can't). - A report still queued after its polls, or answered without data, is a failure, not an empty history. - TopstepX: a newly fetched account (new, wiped, failed) waits for a sync that reconciles its funds, and the fetch starts that sync right away. - Accounts a fetch is already working through aren't fetched again by the targeted pass; the TopstepX REST fill search spans ±15 minutes. Co-Authored-By: Claude Opus 5.5 --- lib/auto-trade.ts | 21 +++- lib/topstepx-class.ts | 48 +++++++++- lib/tradovate-class.ts | 212 ++++++++++++++++++++++++++++++++++------- 3 files changed, 242 insertions(+), 39 deletions(-) diff --git a/lib/auto-trade.ts b/lib/auto-trade.ts index a8e91f1..c0b26e5 100644 --- a/lib/auto-trade.ts +++ b/lib/auto-trade.ts @@ -528,6 +528,19 @@ export function todaysSessionOpenDelay(): number { return openJitterMinutes(centralDay); } +/** Weekdays 2:55–5:00 PM CT: the trade date's session has closed and the 5 PM + * roll hasn't opened the next one. The rest of the no-trade window — the + * roll to the open, and weekends — is waiting on a session to come. */ +function isAfterSessionClose(now: Date = new Date()): boolean { + const parts = new Intl.DateTimeFormat('en-US', { + weekday: 'short', hour: 'numeric', minute: 'numeric', hour12: false, timeZone: 'America/Chicago', + }).formatToParts(now); + const get = (type: string) => parts.find((p) => p.type === type)?.value ?? ''; + if (get('weekday') === 'Sat' || get('weekday') === 'Sun') return false; + const minutes = (parseInt(get('hour'), 10) % 24) * 60 + parseInt(get('minute'), 10); + return minutes >= 14 * 60 + 55 && minutes < 17 * 60; +} + /** * Returns true when trading is not allowed based on the trading_hours setting. * @@ -1197,9 +1210,11 @@ function hasRemainingConfiguredAccounts(): boolean { const ev = evaluateAccountForTrade(client, acc, tradeDate); if (ev?.eligible) return true; - // Waiting on its history, not done: it may still owe today's trade. - // Only while the session is open, like the unready firms above. - if (ev?.reason === HISTORY_PENDING && !isInNoTradeWindow()) return true; + // Waiting on its history, not done: it may still owe today's trade — + // during the session and before it opens alike (a restart between + // the 5 PM roll and the open loads histories right then). Only + // once the session has closed is a still-loading account let go. + if (ev?.reason === HISTORY_PENDING && !isAfterSessionClose()) return true; } } diff --git a/lib/topstepx-class.ts b/lib/topstepx-class.ts index f7c0eee..1947dd9 100644 --- a/lib/topstepx-class.ts +++ b/lib/topstepx-class.ts @@ -236,6 +236,15 @@ export class TopstepXClient implements PlatformClient { * is the firm's first pass; this is per account, so one that appears * later, or is wiped, stays out of trading until its own fetch lands. */ private historyLoadedIds = new Set(); + /** Fetched, but not yet past a sync that reconciled its funds. TopstepX + * has no funds API: reconcileFunds, inside requestSync, is what books a + * withdrawal from the balance the trades don't explain, so a history + * fetched without it (after a wipe took the fund rows) would plan the + * account as stage 1 until the next sync. */ + private reconcilePendingIds = new Set(); + /** Accounts a fetchDaysTraded call is working through right now, so the + * targeted pass doesn't fetch them again (a wipe fetches outside it). */ + private fetchingIds = new Set(); /** Trades seen for the current CT trade date, per account — realizedPnL is * recomputed from these on every REST sync and every live trade event. */ @@ -877,6 +886,9 @@ export class TopstepXClient implements PlatformClient { } catch (err) { console.error(`[topstepx] fund reconcile failed for ${acc.name}: ${errorMessage(err)}`); } + // A freshly fetched history is tradable once a fresh pass has + // had its go at booking the funds behind the balance. + if (accountFresh && this.reconcilePendingIds.delete(acc.id)) this.historyLoadedIds.add(acc.id); try { this.updateMllFloor(acc); @@ -1106,7 +1118,11 @@ export class TopstepXClient implements PlatformClient { // 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)).map((a) => a.id) + ? 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; @@ -1147,6 +1163,18 @@ export class TopstepXClient implements PlatformClient { * 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; @@ -1216,7 +1244,10 @@ export class TopstepXClient implements PlatformClient { this.applyCycleSlice(account.id); saveDailyPnL(account.id, account.name, fresh); delete this.lastFetchErrors[account.name]; - this.historyLoadedIds.add(account.id); + // 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); @@ -1238,6 +1269,12 @@ export class TopstepXClient implements PlatformClient { } } 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. (The firm's first + // pass syncs from maybeFetchDaysTraded once readiness allows.) + if (this.reconcilePendingIds.size > 0 && this.historyLoaded && !this.intentionalDisconnect) { + void this.requestSync(); + } return { failedAccounts }; } @@ -1254,6 +1291,7 @@ export class TopstepXClient implements PlatformClient { * 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]; @@ -1421,7 +1459,9 @@ export class TopstepXClient implements PlatformClient { console.log(`[topstepx] order ${orderId} filled @ ${early} (before the place response)`); return { orderId, price: early, accountId, contractId: contract.id }; } - const since = new Date(Date.now() - 60_000).toISOString(); + // 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; @@ -1510,7 +1550,7 @@ export class TopstepXClient implements PlatformClient { /** The same, from the Gateway's trade history (REST). */ private async searchOrderFillPrice(accountId: number, orderId: number, since: string): Promise { const res = await this.post<{ trades?: TsxTrade[] }>('/api/Trade/search', { - accountId, startTimestamp: since, endTimestamp: new Date(Date.now() + 60_000).toISOString(), + accountId, startTimestamp: since, endTimestamp: new Date(Date.now() + 15 * 60_000).toISOString(), }); return firstFillPrice((res.trades ?? []).filter((t) => t.orderId === orderId)); } diff --git a/lib/tradovate-class.ts b/lib/tradovate-class.ts index 60a3f8d..6df213d 100644 --- a/lib/tradovate-class.ts +++ b/lib/tradovate-class.ts @@ -117,6 +117,27 @@ function reconnectBackoffMs(failures: number): number { return Math.min(BACKOFF_BASE_MS * 2 ** Math.max(0, failures - 1), BACKOFF_MAX_MS); } +interface CashBaseline { dayStart: number; realized: number } +interface HistoryHold { reason: 'balance' | 'wipe'; fundCount: number; since: number; checks: number; retryAt: number } + +/** A held account's refetch pace. */ +const HOLD_RETRY_MS = 5 * 60_000; +/** A balance hold no Cash History ever explains with a fund transaction — a + * fee, or an adjustment the report files as trading — is let go after this + * many clean reports spanning at least HOLD_RELEASE_MS, with a warning. */ +const HOLD_MAX_CHECKS = 3; +const HOLD_RELEASE_MS = 15 * 60_000; + +/** Whether cash is still what the baseline accounts for: the same start-of-day + * balance ('same'), or the daily reset having folded the day's realized P&L + * into the balance ('folded'). */ +function cashExplained(base: CashBaseline, cash: { amount: number; realizedPnL: number }): 'same' | 'folded' | null { + const dayStart = cash.amount - cash.realizedPnL; + if (Math.abs(dayStart - base.dayStart) <= FUND_DRIFT_TOLERANCE) return 'same'; + if (Math.abs(dayStart - (base.dayStart + base.realized)) <= FUND_DRIFT_TOLERANCE) return 'folded'; + return null; +} + /** A Tradovate cashBalance entity, as the sync snapshot and socket carry it. */ interface CashBalanceEntity { accountId: number; @@ -243,14 +264,20 @@ export class TradovateClient { * failed with a DB cache to fall back on. Until then the plan would read * a fresh stage-1 cycle, so historyReady() keeps them out of trading. */ private historyLoadedIds = new Set(); - /** Accounts whose balance moved in a way the loaded history can't explain - * (a payout posting between the hourly reports): held out of trading - * until a Cash History books it. */ - private historyRefreshIds = new Set(); + /** Accounts held out of trading until a Cash History books what their + * history is missing: 'balance' — a balance change it can't explain (a + * payout between the hourly reports); 'wipe' — a wipe that dropped their + * fund transactions, which only Cash History rebuilds. fundCount is the + * fund transactions held when the hold began; retryAt paces the refetch. */ + private historyHolds = new Map(); /** Per account, the start-of-day balance (amount − realizedPnL) the loaded * history was taken at, and the day's realized P&L as last seen — what the * daily reset folds into the balance. */ - private balanceBaseline: { [accountId: number]: { dayStart: number; realized: number } } = {}; + private balanceBaseline: { [accountId: number]: CashBaseline } = {}; + /** Accounts a fetchDaysTraded call is working through right now, so the + * targeted pass doesn't request the same reports again (a wipe fetches + * outside maybeFetchDaysTraded). */ + private fetchingIds = new Set(); /** Interval handles tracked so they can be cleared on reconnect */ private syncInterval: ReturnType | null = null; private heartbeatInterval: ReturnType | null = null; @@ -969,10 +996,16 @@ export class TradovateClient { // Accounts held out of trading for want of history — new since the // last pass, wiped, or flagged by a balance change — get their own // fetch now rather than wait out the hour. A failure falls back to - // the 5-minute retry like any other. + // the 5-minute retry like any other; a held account refetches on its + // own 5-minute pace; one already being fetched isn't asked for twice. const waiting = !initialFetchDue && !retryPending ? this.accountList - .filter((a) => !this.historyLoadedIds.has(a.id) || this.historyRefreshIds.has(a.id)) + .filter((a) => !this.fetchingIds.has(a.id)) + .filter((a) => { + if (!this.historyLoadedIds.has(a.id)) return true; + const hold = this.historyHolds.get(a.id); + return hold !== undefined && now >= hold.retryAt; + }) .map((a) => a.id) : []; const targeted = !retryDue && !hourlyRefreshDue && !initialFetchDue && waiting.length > 0; @@ -1061,6 +1094,18 @@ export class TradovateClient { /** Pass onlyAccountIds to refresh a subset (e.g. one wiped account) without * paying for a report round-trip on every other account of the firm. */ public async fetchDaysTraded(onlyAccountIds?: number[]): Promise<{ failedAccounts: number }> { + 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.accessInfo?.accessToken) return { failedAccounts: 0 }; this.fetchDaysComplete = false; let failedAccounts = 0; @@ -1125,7 +1170,15 @@ export class TradovateClient { )).data; pollAttempts++; } - return typeof reportData?.data === 'string' ? reportData.data : '[]'; + // Still queued after every poll, or an answer with no data + // (an errorText): a failure, not an empty history — read as + // '[]' it booked no rows and no fund transactions as fact. + if (reportData?.['p-ticket']) throw new Error(`${name} report still queued after ${pollAttempts} polls`); + if (typeof reportData?.data !== 'string') { + const why = typeof reportData?.errorText === 'string' ? `: ${reportData.errorText}` : ''; + throw new Error(`${name} report came back without data${why}`); + } + return reportData.data; } catch (err) { lastErr = err; if (!isTransient(err) || attempt === maxAttempts) throw err; @@ -1176,6 +1229,12 @@ export class TradovateClient { this.daysTraded[account.id] = active.filter((d) => d.pnl !== 0).length; } + // The balance as it stands before the report is asked for: the + // baseline this history is judged against, so a payout landing + // while the report is in flight isn't absorbed unexplained. + const cashNow = this.accountCashBalances[account.id]; + const cashBefore = cashNow ? { amount: cashNow.amount, realizedPnL: cashNow.realizedPnL } : null; + // --- Try Cash History first (true after-fee daily P&L) --- try { const raw = await requestReport('Cash History', account.name); @@ -1260,7 +1319,7 @@ export class TradovateClient { this.daysTraded[account.id] = active.filter((d) => d.pnl !== 0).length; saveDailyPnL(account.id, account.name, fresh); // upsert only fresh entries — preserves older cached rows delete this.lastFetchErrors[account.name]; - this.markHistoryLoaded(account.id, true); + this.markHistoryLoaded(account.id, true, cashBefore); continue; } catch { // fall through to Fills fallback @@ -1355,21 +1414,23 @@ export class TradovateClient { /** * Whether this account's history can be planned on — see PlatformClient. - * Loaded by this process, not wiped since, and no balance change the - * history can't explain. + * Loaded by this process, not wiped since, not on hold, and no balance + * change the history can't explain. */ public historyReady(accountId: number): boolean { if (!this.historyLoadedIds.has(accountId)) return false; if (this.dailyPnL[accountId] === undefined || this.fundTransactions[accountId] === undefined) return false; - if (this.historyRefreshIds.has(accountId)) return false; + if (this.historyHolds.has(accountId)) return false; return !this.balanceDrifted(accountId); } /** Drop the account's history from memory (a wipe): it reads unloaded, and - * stays out of trading, until the next fetch rebuilds it. */ + * stays out of trading, until a Cash History rebuilds it — the wipe took + * its fund transactions, which neither the Fills report nor the cache can + * bring back. */ public forgetHistory(accountId: number, accountName: string): void { this.historyLoadedIds.delete(accountId); - this.historyRefreshIds.delete(accountId); + this.historyHolds.set(accountId, { reason: 'wipe', fundCount: 0, since: Date.now(), checks: 0, retryAt: 0 }); delete this.balanceBaseline[accountId]; delete this.dailyPnL[accountId]; delete this.fullDailyPnL[accountId]; @@ -1383,20 +1444,101 @@ export class TradovateClient { /** * A fetch pass reached the account. Cash History is the only source that - * books fund transactions, so only it clears a balance-change hold; the - * Fills fallback and the DB cache still count as loaded otherwise. + * books fund transactions, so only it can lift a hold, and a balance hold + * only once it books a fund transaction the account didn't have when the + * hold began. The Fills fallback and the DB cache still count as loaded. + * + * cashBefore is the balance when the Cash History was requested: the + * baseline comes from it, not from the balance when the report lands, so + * a payout posting in between is still caught — and if the balance did + * move in between, this report may not have seen it, so the account is + * held for the next one. */ - private markHistoryLoaded(accountId: number, fromCashHistory: boolean): void { + private markHistoryLoaded( + accountId: number, + fromCashHistory: boolean, + cashBefore: { amount: number; realizedPnL: number } | null = null, + ): void { if (this.dailyPnL[accountId] === undefined || this.fundTransactions[accountId] === undefined) { this.historyLoadedIds.delete(accountId); return; } - if (this.historyRefreshIds.has(accountId) && !fromCashHistory) return; - this.historyRefreshIds.delete(accountId); this.historyLoadedIds.add(accountId); - const cash = this.accountCashBalances[accountId]; - if (cash) this.balanceBaseline[accountId] = { dayStart: cash.amount - cash.realizedPnL, realized: cash.realizedPnL }; - else delete this.balanceBaseline[accountId]; + const now = Date.now(); + const hold = this.historyHolds.get(accountId); + + if (!fromCashHistory) { + if (hold) { + hold.retryAt = now + HOLD_RETRY_MS; + return; + } + // Keep the baseline a Cash History set — this process's, or the + // last one on record (a restart onto a failing report server) — + // rather than take today's balance, which may hold an unbooked + // payout; the drift check then judges it. + if (!this.balanceBaseline[accountId]) { + const saved = loadAccountMeta(accountId, 'cash_baseline'); + try { + const parsed = saved ? JSON.parse(saved) as CashBaseline : null; + if (parsed && Number.isFinite(parsed.dayStart) && Number.isFinite(parsed.realized)) { + this.balanceBaseline[accountId] = parsed; + } + } catch { /* unreadable: judged from the live balance below */ } + } + this.balanceDrifted(accountId); + return; + } + + const cashAfter = this.accountCashBalances[accountId]; + const before = cashBefore ?? cashAfter ?? null; + const base: CashBaseline | null = before + ? { dayStart: before.amount - before.realizedPnL, realized: before.realizedPnL } + : null; + const movedDuringFetch = base !== null && cashAfter !== undefined && cashExplained(base, cashAfter) === null; + const name = this.accountList.find((a) => a.id === accountId)?.name ?? accountId; + + if (hold) { + hold.checks++; + const booked = (this.fundTransactions[accountId]?.length ?? 0) > hold.fundCount; + if (hold.reason === 'balance' && !booked) { + if (hold.checks < HOLD_MAX_CHECKS || now - hold.since < HOLD_RELEASE_MS) { + hold.retryAt = now + HOLD_RETRY_MS; + return; + } + console.warn(`${this.tag} ${name}: ${hold.checks} Cash History reports over` + + ` ${Math.round((now - hold.since) / 60_000)}m booked no fund transaction for its balance change —` + + ' releasing it (a fee, or an adjustment the report files as trading)'); + // Nothing will explain it: take the balance as it now stands. + this.historyHolds.delete(accountId); + this.setBaseline(accountId, cashAfter ?? null); + return; + } + this.historyHolds.delete(accountId); + } + + if (movedDuringFetch) { + this.historyHolds.set(accountId, { + reason: 'balance', + fundCount: this.fundTransactions[accountId]?.length ?? 0, + since: now, checks: 0, retryAt: now + 60_000, + }); + this.balanceBaseline[accountId] = base!; + console.warn(`${this.tag} ${name}: balance moved while its Cash History was in flight — held for the next report`); + return; + } + this.setBaseline(accountId, before); + } + + /** Set (and keep on record, for a restart that can't reach Cash History) + * the baseline a Cash History vouched for. */ + private setBaseline(accountId: number, cash: { amount: number; realizedPnL: number } | null): void { + if (!cash) { + delete this.balanceBaseline[accountId]; + return; + } + const base = { dayStart: cash.amount - cash.realizedPnL, realized: cash.realizedPnL }; + this.balanceBaseline[accountId] = base; + saveAccountMeta(accountId, 'cash_baseline', JSON.stringify(base)); } /** @@ -1412,25 +1554,29 @@ export class TradovateClient { private balanceDrifted(accountId: number): boolean { const cash = this.accountCashBalances[accountId]; if (!cash) return false; - const dayStart = cash.amount - cash.realizedPnL; const base = this.balanceBaseline[accountId]; if (!base) { - this.balanceBaseline[accountId] = { dayStart, realized: cash.realizedPnL }; + this.balanceBaseline[accountId] = { dayStart: cash.amount - cash.realizedPnL, realized: cash.realizedPnL }; return false; } - if (Math.abs(dayStart - base.dayStart) <= FUND_DRIFT_TOLERANCE) { + const how = cashExplained(base, cash); + if (how === 'same') { base.realized = cash.realizedPnL; return false; } - if (Math.abs(dayStart - (base.dayStart + base.realized)) <= FUND_DRIFT_TOLERANCE) { - this.balanceBaseline[accountId] = { dayStart, realized: cash.realizedPnL }; + if (how === 'folded') { + this.balanceBaseline[accountId] = { dayStart: cash.amount - cash.realizedPnL, realized: cash.realizedPnL }; return false; } - if (!this.historyRefreshIds.has(accountId)) { - this.historyRefreshIds.add(accountId); + if (!this.historyHolds.has(accountId)) { + this.historyHolds.set(accountId, { + reason: 'balance', + fundCount: this.fundTransactions[accountId]?.length ?? 0, + since: Date.now(), checks: 0, retryAt: 0, + }); const name = this.accountList.find((a) => a.id === accountId)?.name ?? accountId; - console.warn(`${this.tag} ${name}: balance moved $${Math.round((dayStart - base.dayStart) * 100) / 100}` - + ' that its history does not explain — held out of trading until Cash History refreshes'); + console.warn(`${this.tag} ${name}: balance moved $${Math.round((cash.amount - cash.realizedPnL - base.dayStart) * 100) / 100}` + + ' that its history does not explain — held out of trading until Cash History books it'); } return true; } @@ -1438,7 +1584,9 @@ export class TradovateClient { /** Run the drift check on every loaded account each sync, so the day's * last realized P&L is on record before the daily reset folds it in. */ private observeBalances(): void { - for (const accountId of this.historyLoadedIds) this.balanceDrifted(accountId); + for (const accountId of this.historyLoadedIds) { + if (!this.historyHolds.has(accountId)) this.balanceDrifted(accountId); + } } private async login(pTicket?: string): Promise {