/** * The firm catalog — firms and their account types, owned by the master * dashboard and pulled from its public GET /api/firms/catalog. * * This machine holds a cache of the last complete, valid response. That cache * is the whole point: lib/auto-trade.ts needs every account's profit target and * size on each 60s tick, so if those numbers only ever came over the network, a * master reboot or a flaky link would stop trading with positions open. The * cache is written only by a successful sync and never edited locally, so it * cannot drift the way the old per-machine account_configs did. * * Stored as one JSON blob in `settings` rather than mirrored into tables: the * catalog is only ever read whole, or asked "give me firm X's account types". * * (Unrelated to the `FIRMS` array in lib/automations.ts, which describes * prop-firm websites the AutoBuyer clicks through. Different notion of "firm"; * the two are not linked yet.) */ import { isFirmPlatform } from './master-protocol'; import { sanitizeInstruments, sanitizeInstrumentLimits, DEFAULT_MAX_ENTRY_SPREAD } from './instrument-specs'; import { normalizeAccountType } from './kept-rules'; import type { CatalogAccountType, CatalogFirm, CatalogTradingSettings, FirmCatalogPayload, FirmPlatform, KeptAccountRules, SymbolMap, SymbolMapCandidate, } from './master-protocol'; import { getSetting, setSetting, listCredentials, getCredentials, rekeyCredentials, } from './db'; const CATALOG_PATH = '/api/firms/catalog'; const POLL_INTERVAL_MS = 60_000; const FETCH_TIMEOUT_MS = 10_000; // Until the very first successful sync there is nothing to trade from, so back // off quickly rather than sitting on the full poll interval. const COLD_RETRY_MIN_MS = 10_000; const COLD_RETRY_MAX_MS = 60_000; /** The version of validatePayload below, stamped on the payload it caches. * Bump it whenever the parser starts keeping a field an older one dropped: * a cache from before then lacks the field however often it is re-read, and * `?since=` would answer "unchanged" until the master next changes, so the * first sync on a newer parser fetches the whole catalog. * 2: kept rule sets (keptAccountRules) and the firm's maxConcurrentAccounts. * Stamped only on a payload from a master that sends keptAccountRules: one * from an older master can't vouch for kept sets it never sent. */ export const CATALOG_PARSER_VERSION = 2; /** A Date header further off than this is somebody else's clock, not the master's. */ const MAX_MASTER_OFFSET_MS = 365 * 24 * 60 * 60 * 1000; interface CatalogCache { payload: FirmCatalogPayload | null; /** Index over every firm's account types, for the id lookup. */ typesById: Map; /** Each live account type's kept rule sets, by closedAt (oldest window * first). Kept ids are deliberately not in typesById: nothing can be * pinned to one. */ keptByLiveId: Map; } const g = global as typeof globalThis & { __firmCatalog?: CatalogCache; __firmCatalogTimer?: ReturnType; __firmCatalogColdDelay?: number; /** The master's clock minus this one, from the last sync's Date header. */ __masterOffsetMs?: number; }; function indexPayload(payload: FirmCatalogPayload | null): CatalogCache { const typesById = new Map(); const keptByLiveId = new Map(); for (const firm of payload?.firms ?? []) { for (const type of firm.accountTypes) typesById.set(type.id, type); for (const kept of firm.keptAccountRules ?? []) { const list = keptByLiveId.get(kept.keptFrom); if (list) list.push(kept); else keptByLiveId.set(kept.keptFrom, [kept]); } } for (const list of keptByLiveId.values()) list.sort((a, b) => Date.parse(a.closedAt) - Date.parse(b.closedAt)); return { payload, typesById, keptByLiveId }; } function cache(): CatalogCache { if (g.__firmCatalog) return g.__firmCatalog; let payload: FirmCatalogPayload | null = null; const raw = getSetting('catalog_payload'); if (raw) { try { const stored = JSON.parse(raw); payload = validatePayload(stored); // Whatever parsed the blob when it was stored, not this parser: a // field an older build dropped on the way in can't come back from // re-reading the blob. const parser = (stored as { parser?: unknown }).parser; if (typeof parser === 'number') payload.parser = parser; else delete payload.parser; } catch { console.error('[catalog] cached payload is unreadable — ignoring it'); } } g.__firmCatalog = indexPayload(payload); return g.__firmCatalog; } export function getCachedCatalog(): CatalogFirm[] { return cache().payload?.firms ?? []; } export function getCachedFirm(firmId: string): CatalogFirm | undefined { return getCachedCatalog().find((f) => f.id === firmId); } export function getAccountTypeById(typeId: string): CatalogAccountType | undefined { return cache().typesById.get(typeId); } /** The kept rule sets of one live account type, oldest window first ([] when * it keeps none). lib/account-type-resolver.ts picks an account's by its * rules date. */ export function getKeptRulesFor(liveTypeId: string): KeptAccountRules[] { return cache().keptByLiveId.get(liveTypeId) ?? []; } /** The kept rule set ids loaded for a firm, for the report — null while the * cached catalog came from a parser older than CATALOG_PARSER_VERSION, which * can't be trusted to hold them (the master then reads this machine as not * applying kept rules yet). */ export function getKeptRuleIdsForFirm(firmId: string): string[] | null { const payload = cache().payload; if (!payload || (payload.parser ?? 0) < CATALOG_PARSER_VERSION) return null; return (payload.firms.find((f) => f.id === firmId)?.keptAccountRules ?? []).map((k) => k.id); } /** The master's clock minus this machine's, as of the last good sync (null * before one this process). Dates new accounts on the master's clock when * this one runs behind (lib/kept-rules.ts stampRulesDate). */ export function getMasterOffsetMs(): number | null { return g.__masterOffsetMs ?? null; } /** When the cached catalog last changed on the master (its payload's * updatedAt); null with no cache. No account first seen after a change can * be dated before it. */ export function getCatalogUpdatedAt(): string | null { return cache().payload?.updatedAt ?? null; } export function getCatalogMeta(): { version: number; updatedAt: string | null; lastSyncAt: string | null; lastError: string | null; firmCount: number; } { return { version: Number(getSetting('catalog_version') ?? 0), updatedAt: getSetting('catalog_updated_at'), lastSyncAt: getSetting('catalog_last_sync_at'), lastError: getSetting('catalog_last_error'), firmCount: getCachedCatalog().length, }; } /** Whether stored credentials are what the firm's platform logs in with — a * TopstepX API key, or a username + password (Tradovate, DXTrade). */ export function hasPlatformCredentials( platform: FirmPlatform, creds: { username: string; password: string; apiKey: string } | null | undefined, ): boolean { if (!creds) return false; switch (platform) { case 'topstepx': return !!creds.apiKey; case 'tradovate': case 'dxtrade': return !!creds.username && !!creds.password; } } /** Firms this machine can actually trade: in the catalog AND holding the * credentials their platform needs. Everything that starts a platform client * goes through here. */ export function getTradableFirms(): (CatalogFirm & { username: string; password: string; apiKey: string })[] { const result: (CatalogFirm & { username: string; password: string; apiKey: string })[] = []; for (const firm of getCachedCatalog()) { const creds = getCredentials(firm.id); if (creds && hasPlatformCredentials(firm.platform, creds)) result.push({ ...firm, ...creds }); } return result; } // ── Sync ───────────────────────────────────────────────────────────────────── const g2 = global as typeof globalThis & { __catalogUnknownPlatformWarned?: Set; __catalogKeptSkipWarned?: Set; }; /** Once per firm per process — the catalog is re-validated every 60s. */ function warnUnknownPlatform(id: string, name: string, platform: unknown): void { const warned = (g2.__catalogUnknownPlatformWarned ??= new Set()); const key = `${id}|${String(platform)}`; if (warned.has(key)) return; warned.add(key); console.warn(`[catalog] skipping "${name}": platform ${JSON.stringify(platform)} is unknown to this build — update this instance`); } function nonNegative(value: unknown): number { return typeof value === 'number' && Number.isFinite(value) && value > 0 ? value : 0; } /** Drops anything malformed rather than failing the sync: a bad symbol-map * entry only means that candidate isn't traded. */ function sanitizeSymbolMap(raw: unknown): SymbolMap { const map: SymbolMap = {}; if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) return map; for (const [root, list] of Object.entries(raw as Record)) { if (!Array.isArray(list)) continue; const candidates = list.flatMap((c): SymbolMapCandidate[] => { if (typeof c !== 'object' || c === null) return []; const cand = c as Record; if (typeof cand.symbol !== 'string' || !cand.symbol || cand.symbol.includes('|')) return []; return [{ symbol: cand.symbol }]; }); if (candidates.length > 0) map[root.toUpperCase()] = candidates; } return map; } function isDateString(value: unknown): value is string { return typeof value === 'string' && Number.isFinite(Date.parse(value)); } /** * A firm's kept rule sets, each as strict as an account type: a malformed * entry throws and the whole sync is rejected. One that names no live type * of this firm is skipped instead (warned once) — no account can map to it. */ function parseKeptRules( firm: { id: string; name: string }, raw: unknown[], accountTypes: CatalogAccountType[], tradesEveryDay: boolean, ): KeptAccountRules[] { const liveIds = new Set(accountTypes.map((t) => t.id)); const kept: KeptAccountRules[] = []; raw.forEach((k: unknown, j: number) => { const where = `kept rules ${j} of ${firm.name}`; const type = normalizeAccountType(k, { tradesEveryDay, where }); const entry = k as Record; if (typeof entry.keptFrom !== 'string') throw new Error(`${where} is missing keptFrom`); if (entry.openedAt !== null && !isDateString(entry.openedAt)) throw new Error(`${where} has bad openedAt`); if (!isDateString(entry.closedAt)) throw new Error(`${where} has bad closedAt`); if (!liveIds.has(entry.keptFrom) || type.firmId !== firm.id) { const warned = (g2.__catalogKeptSkipWarned ??= new Set()); if (!warned.has(type.id)) { warned.add(type.id); console.warn(`[catalog] skipping kept rules ${type.id} of "${firm.name}": it names no account type of this firm (keptFrom ${entry.keptFrom}, firmId ${type.firmId})`); } return; } kept.push({ ...type, keptFrom: entry.keptFrom, openedAt: entry.openedAt, closedAt: entry.closedAt }); }); return kept; } /** Rejects anything that isn't a complete catalog. A half-parsed payload would * silently unassign accounts, so it is better to keep the old cache. */ function validatePayload(raw: unknown): FirmCatalogPayload { if (typeof raw !== 'object' || raw === null) throw new Error('payload is not an object'); const body = raw as Record; if (typeof body.version !== 'number') throw new Error('missing version'); if (typeof body.updatedAt !== 'string') throw new Error('missing updatedAt'); if (!Array.isArray(body.firms)) throw new Error('missing firms'); const firms: CatalogFirm[] = []; body.firms.forEach((f: unknown, i: number) => { if (typeof f !== 'object' || f === null) throw new Error(`firm ${i} is not an object`); const firm = f as Record; if (typeof firm.id !== 'string' || typeof firm.name !== 'string') { throw new Error(`firm ${i} is missing id or name`); } // An old master sends no platform: that is Tradovate. A platform this // build doesn't know is NOT — coercing it would start the wrong client // with another platform's credentials. Skip the firm (before its // account types can fail the whole sync); it comes back once this // build knows the platform. const platform = firm.platform === undefined ? 'tradovate' : firm.platform; if (!isFirmPlatform(platform)) { warnUnknownPlatform(firm.id, firm.name, platform); return; } // Percent commission and every-day trading are DXtrade settings; // futures firms keep measured per-contract fees and the CME calendar. const dx = platform === 'dxtrade'; const tradesEveryDay = dx && firm.tradesEveryDay === true; if (!Array.isArray(firm.accountTypes)) throw new Error(`firm ${firm.name} is missing accountTypes`); const accountTypes: CatalogAccountType[] = firm.accountTypes.map((t: unknown, j: number) => normalizeAccountType(t, { tradesEveryDay, where: `account type ${j}` })); // Only when the master sent the list: absent = a master from before // kept rules, [] = every account on the current rules. const keptAccountRules = Array.isArray(firm.keptAccountRules) ? parseKeptRules({ id: firm.id, name: firm.name }, firm.keptAccountRules, accountTypes, tradesEveryDay) : undefined; firms.push({ id: firm.id, name: firm.name, platform, // The master's list when it sends one; absent from an older // master, whose bannedSymbols / symbolMap then apply. ...(Array.isArray(firm.instruments) ? { instruments: sanitizeInstruments(platform, firm.instruments) } : {}), ...(platform !== 'dxtrade' ? { instrumentLimits: sanitizeInstrumentLimits(firm.instrumentLimits) } : {}), bannedSymbols: Array.isArray(firm.bannedSymbols) ? (firm.bannedSymbols as string[]) : [], accountTypes, ...(keptAccountRules ? { keptAccountRules } : {}), serverUrl: typeof firm.serverUrl === 'string' ? firm.serverUrl : '', loginDomain: typeof firm.loginDomain === 'string' ? firm.loginDomain : '', commissionPct: dx ? nonNegative(firm.commissionPct) : 0, tradesEveryDay, symbolMap: sanitizeSymbolMap(firm.symbolMap), targetHoldHours: nonNegative(firm.targetHoldHours), financingPctPerDay: nonNegative(firm.financingPctPerDay), // Lenient: absent or malformed = no firm cap. maxConcurrentAccounts: Math.floor(nonNegative(firm.maxConcurrentAccounts)), updatedAt: typeof firm.updatedAt === 'string' ? firm.updatedAt : new Date(0).toISOString(), }); }); const trading = parseTradingSettings(body.trading); // A kept-aware master sends keptAccountRules on every firm (even []). A // payload without it is a master from before kept rules: cached without // a parser version, so the cache isn't trusted to hold kept sets — every // sync fetches in full (no `since`), and the first one after a kept-aware // master is back brings them, though its catalog version never moved. const keptAware = body.firms.every((f: unknown) => typeof f === 'object' && f !== null && Array.isArray((f as Record).keptAccountRules)); return { version: body.version, updatedAt: body.updatedAt, firms, ...(trading ? { trading } : {}), ...(keptAware ? { parser: CATALOG_PARSER_VERSION } : {}), }; } /** The master's trading settings; undefined when it sent none (or nothing * usable), so the defaults apply. */ function parseTradingSettings(raw: unknown): CatalogTradingSettings | undefined { if (typeof raw !== 'object' || raw === null) return undefined; const max = (raw as Record).maxEntrySpreadDollars; return typeof max === 'number' && Number.isFinite(max) && max >= 0 ? { maxEntrySpreadDollars: max } : undefined; } /** What the master's Settings page says, or the defaults from before it. */ export function getTradingSettings(): CatalogTradingSettings { return cache().payload?.trading ?? { maxEntrySpreadDollars: DEFAULT_MAX_ENTRY_SPREAD }; } function catalogUrl(): URL | null { const base = getSetting('master_dashboard_url'); if (!base) return null; try { // new URL, never string concatenation: the stored value routinely has a // trailing slash, which would produce //api/firms/catalog. return new URL(CATALOG_PATH, base); } catch { return null; } } function recordError(message: string): void { setSetting('catalog_last_error', message); setSetting('catalog_last_sync_at', new Date().toISOString()); } /** The master's clock off a response's Date header (whole seconds, read on * arrival — close enough to date accounts by). An unreadable header, or one * a year off, leaves the last offset standing. */ function recordMasterClock(header: string | null): void { const at = header ? Date.parse(header) : NaN; if (!Number.isFinite(at)) return; const offset = at - Date.now(); if (Math.abs(offset) > MAX_MASTER_OFFSET_MS) return; g.__masterOffsetMs = offset; } /** If the master's database was rebuilt, its firms come back with fresh ids and * every credential row orphans. Match the survivors up by name (once) rather * than making someone retype every password. */ function rekeyCredentialsByName(firms: CatalogFirm[]): void { const byName = new Map(firms.map((f) => [f.name.toLowerCase(), f.id])); const knownIds = new Set(firms.map((f) => f.id)); for (const row of listCredentials()) { if (knownIds.has(row.firm_id)) continue; const match = byName.get(row.firm_name.toLowerCase()); if (!match) continue; console.warn(`[catalog] re-keying credentials for "${row.firm_name}": ${row.firm_id} -> ${match}`); rekeyCredentials(row.firm_id, match); } } export async function syncCatalog(): Promise<{ ok: boolean; changed: boolean; version: number; error?: string }> { const version = Number(getSetting('catalog_version') ?? 0); const url = catalogUrl(); if (!url) { const error = 'master_dashboard_url is not set'; recordError(error); return { ok: false, changed: false, version, error }; } // Only the URL is required — deliberately not instance_name, unlike the // reporter. A freshly imaged machine has to be able to pull its catalog // before anyone has got round to naming it. // `since` only against the master the cache came from: two masters' // version numbers can collide, and an 'unchanged' from the wrong one // would keep the other's catalog. Nor for a cache an older parser wrote: // it lacks what that parser dropped, and 'unchanged' would keep it so. const hasCache = getCachedCatalog().length > 0; const source = url.origin; const cacheCurrent = (cache().payload?.parser ?? 0) >= CATALOG_PARSER_VERSION; if (hasCache && cacheCurrent && getSetting('catalog_source_url') === source) url.searchParams.set('since', String(version)); try { const res = await fetch(url, { signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) }); if (!res.ok) throw new Error(`master returned ${res.status}`); recordMasterClock(res.headers.get('date')); const body = await res.json(); if (body && typeof body === 'object' && (body as Record).unchanged === true) { setSetting('catalog_source_url', source); setSetting('catalog_last_sync_at', new Date().toISOString()); setSetting('catalog_last_error', ''); return { ok: true, changed: false, version }; } const payload = validatePayload(body); // A master pointed at the wrong FIRMS_DB, or restored from an empty // one, serves {"firms":[]} as a perfectly valid 200. Mirroring that // would stop trading on every machine at once, so keep what we have. if (payload.firms.length === 0 && hasCache) { const error = 'master returned an empty catalog — ignored'; console.error(`[catalog] ${error}`); recordError(error); return { ok: false, changed: false, version, error }; } // Same reasoning one level down: a firm losing every account type // unassigns all of its accounts. Keep the previous types and flag it, // with the kept rule sets that went with them (whose keptFrom named // types this payload lacks, so none of its own survived parsing). An // empty kept list on its own is never suspicious: it is every account // on the current rules. const previous = new Map(getCachedCatalog().map((f) => [f.id, f])); const suspicious: string[] = []; for (const firm of payload.firms) { const before = previous.get(firm.id); if (firm.accountTypes.length === 0 && before && before.accountTypes.length > 0) { firm.accountTypes = before.accountTypes; if (before.keptAccountRules) firm.keptAccountRules = before.keptAccountRules; else delete firm.keptAccountRules; suspicious.push(firm.name); } } setSetting('catalog_payload', JSON.stringify(payload)); setSetting('catalog_version', String(payload.version)); setSetting('catalog_source_url', source); setSetting('catalog_updated_at', payload.updatedAt); setSetting('catalog_last_sync_at', new Date().toISOString()); setSetting('catalog_last_error', suspicious.length > 0 ? `kept cached account types for ${suspicious.join(', ')} — master sent none` : ''); g.__firmCatalog = indexPayload(payload); rekeyCredentialsByName(payload.firms); console.log(`[catalog] synced v${payload.version} — ${payload.firms.length} firm(s)`); return { ok: true, changed: true, version: payload.version }; } catch (err) { const error = err instanceof Error ? err.message : String(err); console.error('[catalog] sync failed:', error); recordError(error); return { ok: false, changed: false, version, error }; } } export function startCatalogSync(onChange?: () => void): void { stopCatalogSync(); const tick = async () => { const result = await syncCatalog(); if (result.changed) { try { onChange?.(); } catch (err) { console.error('[catalog] onChange handler failed', err); } } let delay = POLL_INTERVAL_MS; if (getCachedCatalog().length === 0) { // Never synced: retry hard, with backoff, until something lands. const current = g.__firmCatalogColdDelay ?? COLD_RETRY_MIN_MS; delay = current; g.__firmCatalogColdDelay = Math.min(current * 2, COLD_RETRY_MAX_MS); } else { g.__firmCatalogColdDelay = COLD_RETRY_MIN_MS; // Jitter so a fleet restarting together doesn't poll in lockstep. delay = POLL_INTERVAL_MS + Math.floor(Math.random() * 10_000); } g.__firmCatalogTimer = setTimeout(() => void tick(), delay); }; void tick(); } export function stopCatalogSync(): void { if (g.__firmCatalogTimer) clearTimeout(g.__firmCatalogTimer); g.__firmCatalogTimer = undefined; }