Brandon LiandClaude Opus 5.5
5eb0ea7760
A firm's client used to stay logged in around the clock, renewing on a fixed timer overnight and at weekends. Now, once a firm has nothing open, nothing pending and nothing left to trade before its next session for a random 15–45 minutes, its client disconnects and the dashboard shows it Resting. It logs back in a random 20–50 minutes before the next session opens, at a check-in every 3–7 Central daytime hours, or at once when trading starts, a manual leg, copy trade, manual order or AutoBuyer run needs it. Token renewals fall at random points (Tradovate 15–45 min before expiry, TopstepX every 45–150 min). Settings → Trading → Rest idle broker sessions (session_rest) turns it off. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
848 lines
36 KiB
TypeScript
848 lines
36 KiB
TypeScript
import { TradovateClient } from './tradovate-class';
|
||
import { TopstepXClient } from './topstepx-class';
|
||
import { DXTradeClient } from './dxtrade-class';
|
||
import type { FirmPlatform } from './master-protocol';
|
||
import { errorMessage, isClientReady, type PlatformClient } from './platform-client';
|
||
import { getTradableFirms, startCatalogSync } from './firm-catalog';
|
||
import { firmInstruments, isFuturesFirm, prewarmMarket } from './instruments';
|
||
import { housekeepingRoot } from './instrument-specs';
|
||
import { resolveFirmAssignments } from './account-type-resolver';
|
||
import { resolveContracts } from './contract-resolver';
|
||
import { startReporter, sendReport } from './reporter';
|
||
import { startBackupSync } from './backup-sync';
|
||
import { applyBrokerHttpDefaults, applyBrokerProxy } from './outbound-proxy';
|
||
import axios from './http-client';
|
||
import { startMasterLink } from './master-link';
|
||
import { startUpdaterWatchdog } from './updater-watchdog';
|
||
import { getSchedulerStatus, nextSessionOpenFor, resumeSchedulerIfPersisted, sessionDemand, type SessionDemand } from './auto-trade';
|
||
import { superviseQuotes, SUPERVISE_INTERVAL_MS, type QuoteClient } from './live-quotes';
|
||
import { getSetting } from './db';
|
||
import { drawBetween, planRest, REST_LINGER_MAX_MS, REST_LINGER_MIN_MS } from './trading-logic';
|
||
|
||
// Use global to persist the client pool across HMR reloads in dev mode.
|
||
// Keyed by the master's firm id, which is a uuid string.
|
||
//
|
||
// It is also what keeps the pool single in production: instrumentation.ts
|
||
// boots it from its own server bundle, and the route handlers can each hold a
|
||
// separate instance of this module. Registry state therefore lives here, never
|
||
// in module-level variables.
|
||
const g = global as typeof globalThis & {
|
||
__tradovateClients?: Map<string, PlatformClient>;
|
||
__tradovateClientsInitialized?: boolean;
|
||
__tradovateClientCreds?: Map<string, string>;
|
||
__contractResolverTimer?: ReturnType<typeof setInterval>;
|
||
__contractResolveRetryTimer?: ReturnType<typeof setTimeout>;
|
||
__contractResolveRetries?: number;
|
||
__contractResolvedRoots?: Set<string>;
|
||
__clientSupervisorTimer?: ReturnType<typeof setInterval>;
|
||
__clientSupervision?: Map<string, SupervisionEntry>;
|
||
__quoteSupervisorTimer?: ReturnType<typeof setInterval>;
|
||
__sessionGovernorTimer?: ReturnType<typeof setInterval>;
|
||
__sessionRest?: Map<string, RestEntry>;
|
||
__sessionIdle?: Map<string, IdleEntry>;
|
||
__sessionWaking?: Map<string, WakeEntry>;
|
||
};
|
||
|
||
function ensureMap(): Map<string, PlatformClient> {
|
||
if (!g.__tradovateClients) {
|
||
g.__tradovateClients = new Map();
|
||
}
|
||
return g.__tradovateClients;
|
||
}
|
||
|
||
/** Tracks which credentials each live client was built with, so a password
|
||
* change can be noticed and the client rebuilt. */
|
||
function credsMap(): Map<string, string> {
|
||
if (!g.__tradovateClientCreds) g.__tradovateClientCreds = new Map();
|
||
return g.__tradovateClientCreds;
|
||
}
|
||
|
||
/** What a client was built from — a change to any of it rebuilds the client.
|
||
* DXTrade's server URL and login domain are the firm's (master), not the
|
||
* machine's, but they decide the login just as much. */
|
||
function credsSignature(firm: { platform: string; username: string; password: string; apiKey: string; serverUrl?: string; loginDomain?: string }): string {
|
||
return `${firm.platform} ${firm.username} ${firm.password} ${firm.apiKey} ${firm.serverUrl ?? ''} ${firm.loginDomain ?? ''}`;
|
||
}
|
||
|
||
export function initClient(
|
||
id: string,
|
||
firmName: string,
|
||
platform: FirmPlatform,
|
||
creds: { username: string; password: string; apiKey: string; serverUrl?: string; loginDomain?: string },
|
||
): PlatformClient {
|
||
const map = ensureMap();
|
||
const onSyncComplete = async () => {
|
||
const c = map.get(id);
|
||
if (!c) return;
|
||
console.log(`[${firmName}] sync complete — ${c.accountList.length} account(s)`);
|
||
// Every sync is a chance to bind accounts whose funding history has
|
||
// finally arrived. The first pass can already resolve untraded accounts
|
||
// from their exact cash balance; traded ones wait for the separate
|
||
// reports fetch that produces fund transactions.
|
||
try {
|
||
const counts = resolveFirmAssignments(id, c);
|
||
console.log(
|
||
`[resolver] ${firmName}: ${counts.assigned} assigned, `
|
||
+ `${counts.unassigned} unassigned, ${counts.manual} manual`,
|
||
);
|
||
} catch (err) {
|
||
console.error(`[resolver] ${firmName} failed`, err);
|
||
}
|
||
};
|
||
|
||
let client: PlatformClient;
|
||
switch (platform) {
|
||
case 'topstepx':
|
||
client = new TopstepXClient(creds.username, creds.apiKey, onSyncComplete);
|
||
break;
|
||
case 'dxtrade':
|
||
client = new DXTradeClient({
|
||
firmId: id,
|
||
firmName,
|
||
serverUrl: creds.serverUrl ?? '',
|
||
loginDomain: creds.loginDomain ?? '',
|
||
username: creds.username,
|
||
password: creds.password,
|
||
}, onSyncComplete);
|
||
break;
|
||
case 'tradovate':
|
||
client = new TradovateClient(creds.username, creds.password, onSyncComplete);
|
||
break;
|
||
default: {
|
||
const unknown: never = platform;
|
||
throw new Error(`no client for platform ${String(unknown)}`);
|
||
}
|
||
}
|
||
map.set(id, client);
|
||
credsMap().set(id, credsSignature({ platform, ...creds }));
|
||
return client;
|
||
}
|
||
|
||
export function removeClient(id: string): void {
|
||
const client = ensureMap().get(id);
|
||
// A throwing disconnect() must still take the client out of the pool:
|
||
// left in, half torn down, it would never reconnect — and the supervisor
|
||
// rebuilding it would fail the same way every time.
|
||
try {
|
||
client?.disconnect();
|
||
} catch (err) {
|
||
console.error(`[clients] disconnect failed for firm ${id}: ${errorMessage(err)}`);
|
||
}
|
||
ensureMap().delete(id);
|
||
credsMap().delete(id);
|
||
}
|
||
|
||
/** Disconnect all clients and clear the pool so they are recreated on next getClients() call. */
|
||
export function resetClients(): void {
|
||
const map = g.__tradovateClients;
|
||
if (map) {
|
||
for (const client of map.values()) {
|
||
try { client.disconnect(); } catch { /* ignore */ }
|
||
}
|
||
}
|
||
g.__tradovateClients = undefined;
|
||
g.__tradovateClientCreds = undefined;
|
||
g.__tradovateClientsInitialized = false;
|
||
}
|
||
|
||
/**
|
||
* Brings the client pool in line with the catalog and the stored credentials.
|
||
*
|
||
* Called at boot, whenever credentials are saved, and after the catalog
|
||
* changes — a firm can appear, disappear, or have its password corrected long
|
||
* after startup, none of which the old one-shot initialization could handle.
|
||
*/
|
||
export function reconcileClients(): void {
|
||
const map = ensureMap();
|
||
const firms = getTradableFirms();
|
||
const wanted = new Set(firms.map((f) => f.id));
|
||
|
||
// Firms removed on the master, or whose credentials were cleared.
|
||
for (const id of [...map.keys()]) {
|
||
if (!wanted.has(id)) {
|
||
console.log(`[clients] dropping client for firm ${id} — no longer tradable`);
|
||
removeClient(id);
|
||
}
|
||
}
|
||
|
||
for (const firm of firms) {
|
||
const signature = credsSignature(firm);
|
||
const existing = map.get(firm.id);
|
||
if (existing) {
|
||
if (credsMap().get(firm.id) === signature) continue;
|
||
console.log(`[clients] credentials changed for ${firm.name} — reconnecting`);
|
||
removeClient(firm.id);
|
||
}
|
||
console.log(`[clients] starting ${firm.platform} client for ${firm.name}`);
|
||
initClient(firm.id, firm.name, firm.platform, firm);
|
||
}
|
||
}
|
||
|
||
export function getClients(): Map<string, PlatformClient> {
|
||
const map = ensureMap();
|
||
|
||
// `next build` imports route modules to collect page data. Nothing there
|
||
// should call this, but a build that did must never dial the brokers or
|
||
// resume the scheduler — so hand back the empty pool and leave the flag
|
||
// unset, as if nothing had happened.
|
||
if (process.env.NEXT_PHASE === 'phase-production-build') return map;
|
||
|
||
if (!g.__tradovateClientsInitialized) {
|
||
g.__tradovateClientsInitialized = true;
|
||
try {
|
||
// Before anything dials out: a timeout on every broker HTTP call,
|
||
// and the operator's proxy (if any) under the broker clients.
|
||
// Master traffic stays direct by design. (http-client applies
|
||
// both when it loads, too; these are idempotent.)
|
||
applyBrokerHttpDefaults(axios.defaults);
|
||
applyBrokerProxy(axios.defaults);
|
||
|
||
// Pull the catalog before anything else — without it there are no
|
||
// firms to start clients for. Subsequent changes re-reconcile and
|
||
// re-resolve, so a new account type picks up accounts that were
|
||
// sitting Unassigned without waiting for a Tradovate sync.
|
||
startCatalogSync(() => {
|
||
reconcileClients();
|
||
// A newly listed root needs its front-month contract.
|
||
if (resolveRootsHere().some((r) => !g.__contractResolvedRoots?.has(r))) triggerContractResolve();
|
||
for (const [firmId, client] of ensureMap()) {
|
||
try {
|
||
resolveFirmAssignments(firmId, client);
|
||
} catch (err) {
|
||
console.error(`[resolver] re-resolve failed for ${firmId}`, err);
|
||
}
|
||
}
|
||
});
|
||
|
||
reconcileClients();
|
||
console.log(`[clients] Initialized ${map.size} platform client(s)`);
|
||
|
||
// Auto-resolve contracts after clients have time to authenticate,
|
||
// and measure every future's movement for the Auto draw.
|
||
setTimeout(() => triggerContractResolve(), 15_000);
|
||
prewarmMarket();
|
||
|
||
// Schedule daily resolve at midnight
|
||
scheduleDailyResolve();
|
||
|
||
// Rebuild any client that has been wedged for too long.
|
||
startClientSupervisor();
|
||
|
||
// Log idle sessions out, and back in before they're needed.
|
||
startSessionGovernor();
|
||
|
||
// Live quotes off one futures client, for the spread every entry
|
||
// pays and the Auto draw.
|
||
startQuoteSupervisor();
|
||
|
||
// Start master dashboard reporter
|
||
startReporter();
|
||
|
||
// Bring this machine's auto-updater back if its PM2 cron stops firing.
|
||
startUpdaterWatchdog();
|
||
|
||
// Push settings+credentials backups to the master, so a dead VPS
|
||
// can be replaced and this identity adopted.
|
||
startBackupSync();
|
||
|
||
// ...and the socket it prefers to report over, which is also how
|
||
// the master triggers trades and AutoBuyer runs here.
|
||
startMasterLink(() => { void sendReport(); });
|
||
|
||
// Pick the auto-trade schedule back up if it was running before restart
|
||
resumeSchedulerIfPersisted();
|
||
} catch (err) {
|
||
console.error('[clients] Failed to initialize clients', err);
|
||
}
|
||
}
|
||
return map;
|
||
}
|
||
|
||
// ── Contract auto-resolve ────────────────────────────────────────────────────
|
||
|
||
// With no authenticated Tradovate client yet — a cold catalog, a slow login,
|
||
// broker maintenance at boot — keep checking once a minute for half an hour
|
||
// instead of giving up until midnight. Missing it leaves the contract cache
|
||
// cold: the settings page shows nothing, TopstepX loses the resolver's
|
||
// front-month names, and the first trade pays for the lookups itself.
|
||
const CONTRACT_RESOLVE_RETRY_MS = 60_000;
|
||
const CONTRACT_RESOLVE_MAX_RETRIES = 30;
|
||
|
||
/** Every root a futures firm here lists, plus each one's drain micro. */
|
||
function resolveRootsHere(): string[] {
|
||
const roots = new Set<string>();
|
||
for (const firm of getTradableFirms().filter(isFuturesFirm)) {
|
||
const listed = firmInstruments(firm);
|
||
for (const r of listed) roots.add(r);
|
||
const micro = housekeepingRoot(listed);
|
||
if (micro) roots.add(micro);
|
||
}
|
||
return [...roots];
|
||
}
|
||
|
||
function triggerContractResolve(): void {
|
||
// A midnight run supersedes a pending retry; either way only one chain.
|
||
if (g.__contractResolveRetryTimer) clearTimeout(g.__contractResolveRetryTimer);
|
||
g.__contractResolveRetryTimer = undefined;
|
||
|
||
const map = ensureMap();
|
||
// Find first client with a valid access token (the Yahoo-driven resolver
|
||
// confirms contract ids against Tradovate; TopstepX clients resolve their
|
||
// own contracts but still reuse the resolver's front-month names).
|
||
// A token alone isn't enough: a client whose session died keeps its
|
||
// expired one, and resolving with that fails with nothing left to retry.
|
||
let accessToken: string | null = null;
|
||
for (const [, c] of map) {
|
||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||
const token = (c as any).accessInfo?.accessToken;
|
||
if (!token) continue;
|
||
let authenticated = true;
|
||
try { authenticated = c.health().authenticated; } catch { /* no health: trust the token */ }
|
||
if (authenticated) { accessToken = token; break; }
|
||
}
|
||
if (!accessToken && tradovateAllResting(map)) {
|
||
// Nothing to resolve with until one logs back in, and every wake
|
||
// resolves (wakeFirm) — no half hour of retries meanwhile.
|
||
console.log('[contract-resolver] Tradovate sessions are resting — resolving at the next login');
|
||
g.__contractResolveRetries = 0;
|
||
return;
|
||
}
|
||
if (!accessToken) {
|
||
const retries = (g.__contractResolveRetries ?? 0) + 1;
|
||
if (retries > CONTRACT_RESOLVE_MAX_RETRIES) {
|
||
console.log(
|
||
`[contract-resolver] No authenticated Tradovate client after ${CONTRACT_RESOLVE_MAX_RETRIES} retries`
|
||
+ ' — giving up until the midnight resolve',
|
||
);
|
||
g.__contractResolveRetries = 0;
|
||
return;
|
||
}
|
||
if (retries === 1) {
|
||
console.log('[contract-resolver] No authenticated Tradovate client yet — retrying every 60s');
|
||
}
|
||
g.__contractResolveRetries = retries;
|
||
g.__contractResolveRetryTimer = setTimeout(() => triggerContractResolve(), CONTRACT_RESOLVE_RETRY_MS);
|
||
return;
|
||
}
|
||
g.__contractResolveRetries = 0;
|
||
|
||
// Runs from timers: a throw here (the instruments read) would escape as
|
||
// an uncaught exception.
|
||
try {
|
||
const symbols = resolveRootsHere();
|
||
g.__contractResolvedRoots = new Set(symbols);
|
||
console.log(`[contract-resolver] Resolving ${symbols.length} instruments...`);
|
||
resolveContracts(symbols, accessToken)
|
||
.then((results) => {
|
||
const rolled = Object.entries(results).filter(([, v]) => v?.alternative);
|
||
console.log(`[contract-resolver] Done — ${rolled.length} contract(s) rolled via price match`);
|
||
})
|
||
.catch((err) => console.error(`[contract-resolver] Resolve failed: ${errorMessage(err)}`));
|
||
} catch (err) {
|
||
console.error(`[contract-resolver] Resolve failed: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
|
||
/** Every Tradovate client is a resting snapshot (and there is at least one). */
|
||
function tradovateAllResting(map: Map<string, PlatformClient>): boolean {
|
||
let resting = 0;
|
||
for (const [id, c] of map) {
|
||
let platform: string | undefined;
|
||
try { platform = c.health().platform; } catch { /* unknown: count it awake */ }
|
||
if (platform !== 'tradovate') continue;
|
||
if (!isResting(id)) return false;
|
||
resting++;
|
||
}
|
||
return resting > 0;
|
||
}
|
||
|
||
function scheduleDailyResolve(): void {
|
||
// Clear any existing timer (HMR safety)
|
||
if (g.__contractResolverTimer) clearInterval(g.__contractResolverTimer);
|
||
|
||
// Check every minute if it's midnight (00:00)
|
||
g.__contractResolverTimer = setInterval(() => {
|
||
const now = new Date();
|
||
if (now.getHours() === 0 && now.getMinutes() === 0) {
|
||
console.log('[contract-resolver] Midnight resolve triggered');
|
||
// Its own half hour of retries, not what's left of a chain it cuts off.
|
||
g.__contractResolveRetries = 0;
|
||
triggerContractResolve();
|
||
prewarmMarket();
|
||
}
|
||
}, 60_000);
|
||
}
|
||
|
||
// ── Client supervision ───────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Rebuilds a client that has been unable to trade for a long time.
|
||
*
|
||
* The clients reconnect themselves, but only along the paths their authors
|
||
* foresaw. A re-login that fails in an unexpected way, an await that never
|
||
* settles, a socket that dies without a close event — any of those used to
|
||
* leave a client wedged for the life of the process, and because the
|
||
* scheduler waits on every firm, one wedged firm stopped them all. After a
|
||
* weekend of broker maintenance that meant a pm2 restart. A fresh client is
|
||
* exactly what the restart gave it, so the registry hands one out itself:
|
||
* same firm, same credentials, new instance.
|
||
*
|
||
* "Unready" is isClientReady() saying no — the same rule the scheduler and
|
||
* /api/health use: not synced, or no good sync in the last 5 minutes.
|
||
*
|
||
* Deliberately slow — 20 minutes unready before a rebuild, and at most one
|
||
* rebuild per firm per 20 minutes. The clients' own retries get the first go,
|
||
* and a broker that is really down, or rate-limiting logins, must not be
|
||
* hammered with new sessions. Some clients are never replaced while unready
|
||
* (see rebuildHold): one whose own login retry is pending, and a TopstepX
|
||
* one holding a position.
|
||
*/
|
||
const SUPERVISOR_INTERVAL_MS = 60_000;
|
||
const REBUILD_AFTER_UNREADY_MS = 20 * 60_000;
|
||
const REBUILD_COOLDOWN_MS = 20 * 60_000;
|
||
|
||
/** Clients that can't log in until someone acts — a rebuild changes nothing,
|
||
* and fixing the credentials (or the firm's settings) rebuilds them through
|
||
* reconcileClients anyway. TopstepX with an API key but no username never
|
||
* tries to log in; DXTrade with no server URL can't, and after refused
|
||
* credentials it stops trying so as not to lock the account. */
|
||
function needsAttention(platform: FirmPlatform, client: PlatformClient): boolean {
|
||
if (platform === 'topstepx') return /username missing/i.test(client.lastFetchErrors?.['login'] ?? '');
|
||
if (platform === 'dxtrade') return (client as DXTradeClient).needsAttention === true;
|
||
return false;
|
||
}
|
||
|
||
interface SupervisionEntry {
|
||
/** The instance being judged. A different one (rebuilt here, or replaced
|
||
* by reconcileClients after a credentials change) starts a fresh clock. */
|
||
client: PlatformClient;
|
||
/** When it was first seen unready (0 = ready at the last check). */
|
||
unhealthySince: number;
|
||
lastRebuildAt: number;
|
||
/** Last time a due rebuild was held back and said so (log rate limit). */
|
||
holdLoggedAt: number;
|
||
}
|
||
|
||
function supervisionMap(): Map<string, SupervisionEntry> {
|
||
if (!g.__clientSupervision) g.__clientSupervision = new Map();
|
||
return g.__clientSupervision;
|
||
}
|
||
|
||
/** Keep the live quote feed on a futures client, subscribed to every root
|
||
* this machine may trade (lib/live-quotes.ts). */
|
||
function superviseLiveQuotes(): void {
|
||
// Runs from a timer: nothing may escape.
|
||
try {
|
||
const map = ensureMap();
|
||
const candidates = getTradableFirms().filter(isFuturesFirm).flatMap((firm) => {
|
||
const client = map.get(firm.id);
|
||
// A resting session holds no token; its feed (if it had one) drops.
|
||
return client && !isResting(firm.id) ? [{ firmId: firm.id, firmName: firm.name, platform: firm.platform, client: client as unknown as QuoteClient }] : [];
|
||
});
|
||
superviseQuotes(candidates, resolveRootsHere())
|
||
.catch((err) => console.error(`[quotes] supervision failed: ${errorMessage(err)}`));
|
||
} catch (err) {
|
||
console.error(`[quotes] supervision failed: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
|
||
function startQuoteSupervisor(): void {
|
||
if (g.__quoteSupervisorTimer) clearInterval(g.__quoteSupervisorTimer);
|
||
g.__quoteSupervisorTimer = setInterval(superviseLiveQuotes, SUPERVISE_INTERVAL_MS);
|
||
// The clients log in within seconds; the first look shouldn't wait 30.
|
||
setTimeout(superviseLiveQuotes, 10_000);
|
||
}
|
||
|
||
function startClientSupervisor(): void {
|
||
// HMR-safe, and safe when resetClients() re-runs the bootstrap.
|
||
if (g.__clientSupervisorTimer) clearInterval(g.__clientSupervisorTimer);
|
||
g.__clientSupervisorTimer = setInterval(superviseClients, SUPERVISOR_INTERVAL_MS);
|
||
}
|
||
|
||
/** Why a client isn't ready, short enough for one log line. Never the token. */
|
||
function describeUnready(client: PlatformClient, now: number): string {
|
||
const parts: string[] = [];
|
||
try {
|
||
const health = client.health();
|
||
parts.push(`transport ${health.transport}`);
|
||
if (health.lastLoginError) parts.push(`login error: ${health.lastLoginError.slice(0, 160)}`);
|
||
} catch {
|
||
parts.push('health unavailable');
|
||
}
|
||
parts.push(client.lastSyncAt > 0
|
||
? `last good sync ${Math.round((now - client.lastSyncAt) / 60_000)}m ago`
|
||
: 'never synced');
|
||
return parts.join(', ');
|
||
}
|
||
|
||
/**
|
||
* Why replacing this client now would do more harm than good, or null.
|
||
*
|
||
* - Its login keeps failing and its own retry is already scheduled. Nothing
|
||
* in the instance is broken — the broker is down, refusing the
|
||
* credentials, or rate-limiting — and a fresh instance would fail the same
|
||
* way while starting its backoff over at seconds: a burst of logins after
|
||
* every rebuild instead of one every few minutes, which is how a login
|
||
* rate-limit turns into a captcha lockout.
|
||
* - TopstepX with a position open. The gateway has no OCO, so the client
|
||
* itself cancels the surviving exit leg when the other fills; a fresh
|
||
* instance knows nothing of those pairs, and the survivor would be left
|
||
* working at the broker after the position closes — an order that can open
|
||
* a position nobody asked for. It waits until the account is flat.
|
||
*/
|
||
function rebuildHold(platform: string, client: PlatformClient, now: number): string | null {
|
||
try {
|
||
const health = client.health();
|
||
if (health.consecutiveLoginFailures > 0 && health.nextReconnectAt > now) {
|
||
return `its login retry is due in ${Math.max(1, Math.round((health.nextReconnectAt - now) / 60_000))}m`
|
||
+ ` (${health.consecutiveLoginFailures} failure(s) in a row)`;
|
||
}
|
||
} catch { /* no health: judge it on the rest */ }
|
||
if ((platform === 'topstepx' || platform === 'dxtrade') && Object.keys(client.positions ?? {}).length > 0) {
|
||
return 'it holds an open position (its exit pairs are tracked only inside the client)';
|
||
}
|
||
return null;
|
||
}
|
||
|
||
function superviseClients(): void {
|
||
// Runs inside setInterval: nothing may escape, or the process logs an
|
||
// uncaught exception and the supervisor keeps firing into the same fault.
|
||
try {
|
||
const map = ensureMap();
|
||
const book = supervisionMap();
|
||
const now = Date.now();
|
||
|
||
// A firm that left the registry takes its bookkeeping with it.
|
||
for (const id of [...book.keys()]) {
|
||
if (!map.has(id)) book.delete(id);
|
||
}
|
||
|
||
let firms: ReturnType<typeof getTradableFirms> | null = null;
|
||
for (const [id, client] of [...map]) {
|
||
try {
|
||
let entry = book.get(id);
|
||
if (!entry) {
|
||
entry = { client, unhealthySince: 0, lastRebuildAt: 0, holdLoggedAt: 0 };
|
||
book.set(id, entry);
|
||
} else if (entry.client !== client) {
|
||
entry.client = client;
|
||
entry.unhealthySince = 0;
|
||
}
|
||
|
||
// Logged out on purpose: rebuilding it would wake it.
|
||
if (isResting(id)) {
|
||
entry.unhealthySince = 0;
|
||
continue;
|
||
}
|
||
|
||
if (isClientReady(client, now)) {
|
||
entry.unhealthySince = 0;
|
||
continue;
|
||
}
|
||
if (!entry.unhealthySince) {
|
||
entry.unhealthySince = now;
|
||
continue;
|
||
}
|
||
if (now - entry.unhealthySince < REBUILD_AFTER_UNREADY_MS) continue;
|
||
if (now - entry.lastRebuildAt < REBUILD_COOLDOWN_MS) continue;
|
||
|
||
// Only read when a rebuild is actually due — it hits the DB.
|
||
firms ??= getTradableFirms();
|
||
const firm = firms.find((f) => f.id === id);
|
||
if (!firm) continue; // no longer tradable: reconcileClients drops it
|
||
if (needsAttention(firm.platform, client)) continue;
|
||
|
||
const minutes = Math.round((now - entry.unhealthySince) / 60_000);
|
||
const hold = rebuildHold(firm.platform, client, now);
|
||
if (hold) {
|
||
if (now - entry.holdLoggedAt >= REBUILD_COOLDOWN_MS) {
|
||
entry.holdLoggedAt = now;
|
||
console.log(`[clients] ${firm.name} unhealthy for ${minutes}m — not rebuilding while ${hold}`);
|
||
}
|
||
continue;
|
||
}
|
||
console.log(
|
||
`[clients] ${firm.name} unhealthy for ${minutes}m (${describeUnready(client, now)})`
|
||
+ ' — rebuilding client',
|
||
);
|
||
entry.lastRebuildAt = now;
|
||
removeClient(id);
|
||
entry.client = initClient(firm.id, firm.name, firm.platform, firm);
|
||
entry.unhealthySince = now;
|
||
} catch (err) {
|
||
console.error(`[clients] supervisor check failed for firm ${id}: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
} catch (err) {
|
||
console.error(`[clients] supervisor tick failed: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
|
||
// ── Session rest ─────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Logs a firm's broker session out while nothing needs it, and back in before
|
||
* something does.
|
||
*
|
||
* A session used to stay up from boot to shutdown: logged in all night and all
|
||
* weekend, renewing its token on a timer throughout. Now a firm that has been
|
||
* idle for a random 15–45 minutes — nothing held, nothing pending, nothing
|
||
* left to trade before its next session (sessionDemand in lib/auto-trade.ts)
|
||
* — has its client disconnected: the client keeps its last snapshot for the
|
||
* dashboards and the report, but holds no token and runs no timers. A fresh
|
||
* client logs in a random 20–50 minutes before the firm's next session opens
|
||
* (planRest), on the same path a restart takes. A long rest is broken by a
|
||
* check-in every few hours of the Central daytime, which refreshes balances
|
||
* and finds new accounts: one that can still trade keeps the session up. A
|
||
* command that needs the firm — trading started, a manual leg, a copy, an
|
||
* AutoBuyer run — wakes it at once.
|
||
*
|
||
* Trading never sees a resting firm (isSessionAway, in firmClients): it holds
|
||
* nothing, so it neither trades nor gates, and its lane doesn't wait for it.
|
||
* A woken one stays out of sight until its first good sync (WAKE_GRACE_MS at
|
||
* most), so a check-in never holds up the firms still trading.
|
||
*
|
||
* Setting session_rest = '0' keeps every session up, as before.
|
||
*/
|
||
const GOVERNOR_INTERVAL_MS = 60_000;
|
||
const WAKE_GRACE_MS = 15 * 60_000;
|
||
|
||
interface RestEntry {
|
||
/** The disconnected client, left in the pool as a snapshot. */
|
||
client: PlatformClient;
|
||
since: number;
|
||
/** When its next session wants it back (null: only a command or a check-in will). */
|
||
wakeAt: number | null;
|
||
checkInAt: number | null;
|
||
/** How long before its next session's open it logs back in (drawn per rest). */
|
||
leadMs: number;
|
||
/** The trade date whose accounts were all done, or null: the market was closed. */
|
||
afterTradeDate: string | null;
|
||
/** What wakeAt was planned from: the Central day (its open delay may be
|
||
* drawn later) and whether the scheduler was running. */
|
||
plannedDay: string;
|
||
plannedRunning: boolean;
|
||
}
|
||
|
||
interface IdleEntry {
|
||
client: PlatformClient;
|
||
since: number;
|
||
lingerMs: number;
|
||
}
|
||
|
||
interface WakeEntry {
|
||
client: PlatformClient;
|
||
since: number;
|
||
}
|
||
|
||
const CENTRAL_DAY = new Intl.DateTimeFormat('en-CA', { timeZone: 'America/Chicago' });
|
||
const CENTRAL_TIME = new Intl.DateTimeFormat('en-US', {
|
||
timeZone: 'America/Chicago', weekday: 'short', hour: 'numeric', minute: '2-digit',
|
||
});
|
||
|
||
/** "Mon 6:12 PM CT". */
|
||
export function fmtCentralTime(ms: number): string {
|
||
return `${CENTRAL_TIME.format(ms)} CT`;
|
||
}
|
||
|
||
function restMap(): Map<string, RestEntry> {
|
||
if (!g.__sessionRest) g.__sessionRest = new Map();
|
||
return g.__sessionRest;
|
||
}
|
||
|
||
function idleMap(): Map<string, IdleEntry> {
|
||
if (!g.__sessionIdle) g.__sessionIdle = new Map();
|
||
return g.__sessionIdle;
|
||
}
|
||
|
||
function wakingMap(): Map<string, WakeEntry> {
|
||
if (!g.__sessionWaking) g.__sessionWaking = new Map();
|
||
return g.__sessionWaking;
|
||
}
|
||
|
||
export function sessionRestEnabled(): boolean {
|
||
return getSetting('session_rest') !== '0';
|
||
}
|
||
|
||
/** The firm's session is resting: its client is a disconnected snapshot. */
|
||
export function isResting(firmId: string): boolean {
|
||
const entry = g.__sessionRest?.get(firmId);
|
||
return !!entry && g.__tradovateClients?.get(firmId) === entry.client;
|
||
}
|
||
|
||
/** Logging back in after a rest, and not synced yet. */
|
||
export function isWaking(firmId: string, now = Date.now()): boolean {
|
||
const entry = g.__sessionWaking?.get(firmId);
|
||
if (!entry) return false;
|
||
if (g.__tradovateClients?.get(firmId) !== entry.client
|
||
|| isClientReady(entry.client, now)
|
||
|| now - entry.since >= WAKE_GRACE_MS) {
|
||
g.__sessionWaking?.delete(firmId);
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
/** Out of trading's sight: resting, or not yet back from a rest. */
|
||
export function isSessionAway(firmId: string, now = Date.now()): boolean {
|
||
return isResting(firmId) || isWaking(firmId, now);
|
||
}
|
||
|
||
/** A resting firm's rest, for /api/health and the dashboards; null when awake. */
|
||
export function sessionRestInfo(firmId: string): { since: number; wakeAt: number | null; checkInAt: number | null } | null {
|
||
if (!isResting(firmId)) return null;
|
||
const entry = restMap().get(firmId)!;
|
||
return { since: entry.since, wakeAt: entry.wakeAt, checkInAt: entry.checkInAt };
|
||
}
|
||
|
||
/** When a resting firm logs back in by itself (its wake or its check-in,
|
||
* whichever comes first); null when awake or resting until a command. */
|
||
export function restingWakeAt(firmId: string): number | null {
|
||
const info = sessionRestInfo(firmId);
|
||
if (!info) return null;
|
||
const times = [info.wakeAt, info.checkInAt].filter((t): t is number => t !== null);
|
||
return times.length > 0 ? Math.min(...times) : null;
|
||
}
|
||
|
||
/** Log a resting firm back in now. True if it was resting. */
|
||
export function wakeFirm(firmId: string, why: string): boolean {
|
||
const resting = isResting(firmId);
|
||
restMap().delete(firmId);
|
||
if (!resting) return false;
|
||
idleMap().delete(firmId);
|
||
const firm = getTradableFirms().find((f) => f.id === firmId);
|
||
if (!firm) return false; // no longer tradable: reconcileClients drops it
|
||
console.log(`[sessions] ${firm.name}: logging back in — ${why}`);
|
||
removeClient(firmId);
|
||
const client = initClient(firm.id, firm.name, firm.platform, firm);
|
||
wakingMap().set(firmId, { client, since: Date.now() });
|
||
// A rest can span the midnight contract resolve; catch up once logged in.
|
||
if (firm.platform === 'tradovate') setTimeout(() => triggerContractResolve(), 15_000);
|
||
return true;
|
||
}
|
||
|
||
/** Log every resting firm back in now; how many there were. */
|
||
export function wakeAllFirms(why: string): number {
|
||
let woken = 0;
|
||
for (const id of [...restMap().keys()]) {
|
||
if (wakeFirm(id, why)) woken++;
|
||
}
|
||
return woken;
|
||
}
|
||
|
||
function startSessionGovernor(): void {
|
||
if (g.__sessionGovernorTimer) clearInterval(g.__sessionGovernorTimer);
|
||
g.__sessionGovernorTimer = setInterval(governSessions, GOVERNOR_INTERVAL_MS);
|
||
}
|
||
|
||
function governSessions(): void {
|
||
// Runs inside setInterval: nothing may escape.
|
||
try {
|
||
const map = ensureMap();
|
||
const now = Date.now();
|
||
// Clients replaced or dropped since (a credentials change, a firm gone).
|
||
for (const [id, entry] of [...restMap()]) if (map.get(id) !== entry.client) restMap().delete(id);
|
||
for (const [id, entry] of [...idleMap()]) if (map.get(id) !== entry.client) idleMap().delete(id);
|
||
|
||
if (!sessionRestEnabled()) {
|
||
idleMap().clear();
|
||
if (restMap().size > 0) wakeAllFirms('session rest switched off');
|
||
return;
|
||
}
|
||
|
||
for (const firm of getTradableFirms()) {
|
||
const client = map.get(firm.id);
|
||
if (!client) continue;
|
||
try {
|
||
const rest = restMap().get(firm.id);
|
||
if (rest) governResting(firm, rest, now);
|
||
else governAwake(firm, client, now);
|
||
} catch (err) {
|
||
console.error(`[sessions] ${firm.name}: check failed: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
} catch (err) {
|
||
console.error(`[sessions] governor tick failed: ${errorMessage(err)}`);
|
||
}
|
||
}
|
||
|
||
type GovernedFirm = ReturnType<typeof getTradableFirms>[number];
|
||
|
||
function governResting(firm: GovernedFirm, rest: RestEntry, now: number): void {
|
||
if (rest.wakeAt !== null && now >= rest.wakeAt) {
|
||
wakeFirm(firm.id, 'its next session opens soon');
|
||
return;
|
||
}
|
||
if (rest.checkInAt !== null && now >= rest.checkInAt) {
|
||
wakeFirm(firm.id, 'check-in');
|
||
return;
|
||
}
|
||
// Plan again once the Central day turns (its open delay is drawn then)
|
||
// or the scheduler starts or stops.
|
||
const running = getSchedulerStatus().running;
|
||
const day = CENTRAL_DAY.format(now);
|
||
if (day === rest.plannedDay && running === rest.plannedRunning) return;
|
||
rest.plannedDay = day;
|
||
rest.plannedRunning = running;
|
||
const opensAt = running ? nextSessionOpenFor(firm, now, rest.afterTradeDate) : null;
|
||
const wakeAt = opensAt === null ? null : Math.max(now, opensAt - rest.leadMs);
|
||
if (wakeAt === rest.wakeAt) return;
|
||
rest.wakeAt = wakeAt;
|
||
if (wakeAt !== null && rest.checkInAt !== null && rest.checkInAt >= wakeAt) rest.checkInAt = null;
|
||
console.log(`[sessions] ${firm.name}: next login ${wakeAt !== null ? `moved to ${fmtCentralTime(wakeAt)}` : 'when trading starts'}`);
|
||
}
|
||
|
||
function governAwake(firm: GovernedFirm, client: PlatformClient, now: number): void {
|
||
const demand = sessionDemand(firm, client, now);
|
||
if (demand.needed) {
|
||
idleMap().delete(firm.id);
|
||
return;
|
||
}
|
||
const idle = idleMap().get(firm.id);
|
||
if (!idle) {
|
||
idleMap().set(firm.id, {
|
||
client, since: now, lingerMs: Math.round(drawBetween(REST_LINGER_MIN_MS, REST_LINGER_MAX_MS)),
|
||
});
|
||
return;
|
||
}
|
||
if (now - idle.since < idle.lingerMs || demand.wait) return;
|
||
|
||
const opensAt = demand.scheduled ? nextSessionOpenFor(firm, now, demand.afterTradeDate) : null;
|
||
// Scheduled, with nothing opening in a week: something is off — stay up.
|
||
if (demand.scheduled && opensAt === null) return;
|
||
const plan = planRest(now, opensAt);
|
||
if (!plan) return; // its session opens too soon for a rest to be worth a login
|
||
idleMap().delete(firm.id);
|
||
restClient(firm, client, plan, demand, now);
|
||
}
|
||
|
||
function restClient(
|
||
firm: GovernedFirm,
|
||
client: PlatformClient,
|
||
plan: NonNullable<ReturnType<typeof planRest>>,
|
||
demand: SessionDemand,
|
||
now: number,
|
||
): void {
|
||
try {
|
||
client.disconnect();
|
||
} catch (err) {
|
||
// Half torn down is still not trading; the wake builds a fresh one.
|
||
console.error(`[sessions] ${firm.name}: disconnect failed: ${errorMessage(err)}`);
|
||
}
|
||
restMap().set(firm.id, {
|
||
client,
|
||
since: now,
|
||
wakeAt: plan.wakeAt,
|
||
checkInAt: plan.checkInAt,
|
||
leadMs: plan.leadMs,
|
||
afterTradeDate: demand.afterTradeDate,
|
||
plannedDay: CENTRAL_DAY.format(now),
|
||
plannedRunning: demand.scheduled,
|
||
});
|
||
const why = demand.afterTradeDate ? `every account is done for ${demand.afterTradeDate}`
|
||
: demand.scheduled ? 'its market is closed' : 'trading is stopped';
|
||
const back = plan.wakeAt !== null ? `back at ${fmtCentralTime(plan.wakeAt)}` : 'back when trading starts';
|
||
const checkIn = plan.checkInAt !== null ? `, check-in ${fmtCentralTime(plan.checkInAt)}` : '';
|
||
console.log(`[sessions] ${firm.name}: idle (${why}) — logged out; ${back}${checkIn}`);
|
||
}
|