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; __tradovateClientsInitialized?: boolean; __tradovateClientCreds?: Map; __contractResolverTimer?: ReturnType; __contractResolveRetryTimer?: ReturnType; __contractResolveRetries?: number; __contractResolvedRoots?: Set; __clientSupervisorTimer?: ReturnType; __clientSupervision?: Map; __quoteSupervisorTimer?: ReturnType; __sessionGovernorTimer?: ReturnType; __sessionRest?: Map; __sessionIdle?: Map; __sessionWaking?: Map; }; function ensureMap(): Map { 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 { 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 { 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(); 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): 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 { 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 | 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 { if (!g.__sessionRest) g.__sessionRest = new Map(); return g.__sessionRest; } function idleMap(): Map { if (!g.__sessionIdle) g.__sessionIdle = new Map(); return g.__sessionIdle; } function wakingMap(): Map { 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[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>, 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}`); }