Self-healing broker clients, per-firm scheduler isolation, eager boot
Why the autotrader sometimes broke after a weekend until `pm2 restart`: `next start` only logs uncaught errors, so nothing ever crashed, and a few paths could leave a client dead for the life of the process. Tradovate: a closed socket scheduled exactly one re-login, with no catch — one failure during weekend maintenance (or a rate-limit penalty returned as a 200 without a token) and the client never came back. connectAndAuth is now a single-flight loop that never throws: jittered backoff capped at 5 min, penalty tickets waited out, captcha demands backed off. A 15s watchdog abandons silent, never-authorized or stuck-closing sockets and stale syncs; renewal falls back to a fresh login; orders fail fast (OrderNotPlacedError) on a dead socket and time out instead of hanging; a bad sync answer no longer wipes accounts and positions. TopstepX: hub start/stop/invoke and every REST call are bounded; a wedged hub is discarded and rebuilt; a stuck sync is superseded after 3 min; re-login no longer tears down the sync loop; a watchdog re-logs when syncs go stale; a no-fill timeout cancels the order. Scheduler: one unready client used to hold every firm forever. It now holds the tick for a 10-minute grace, then trades the other firms without it. Per-firm try/catch in batch collection, one batch at a time, accounts marked in flight, legs un-counted only when the entry never reached the broker, and the scheduler starts before the first batch fires. Boot and supervision: instrumentation.ts boots the process at server start (it used to wait for the first dashboard request); a supervisor rebuilds a client that has been unhealthy for 20 min (not a TopstepX one holding a position); broker HTTP calls default to a 60s timeout; GET /api/health shows each client's liveness and the scheduler's last decision. Shared state the eager boot would otherwise split across bundles lives on globalThis. Logs no longer carry access tokens or whole AxiosErrors (which include the encoded Tradovate password). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Brandon Li
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
6146031a38
commit
b932452e98
+573
-131
@@ -18,39 +18,302 @@ import {
|
||||
type DayPlan,
|
||||
} from './trading-logic';
|
||||
import { getSetting, setSetting } from './db';
|
||||
import { errorMessage, isClientReady, isOrderNotPlaced, type PlatformClient } from './platform-client';
|
||||
|
||||
// ── process-wide state ───────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Everything the batches share, held on globalThis like the scheduler state
|
||||
* further down. Next can bundle this module more than once — the scheduler
|
||||
* tick runs from whichever bundle booted the clients, POST /api/trade and
|
||||
* /api/copy-trade from their own route bundles — and plain module-level state
|
||||
* would split into one private copy per bundle: a leg recorded by the tick
|
||||
* would be invisible to a copy-trade, the trade mutex would lock nothing, and
|
||||
* the session-open jitter could be rolled twice in one day.
|
||||
*/
|
||||
interface TradeRuntime {
|
||||
/** In-memory runaway guard: legs fired per account per trade date, plus
|
||||
* the sum of their targets — realized P&L falling short of that sum is
|
||||
* how a stop-out is detected. Resets on restart, which is fine — day
|
||||
* completion is derived from realized PnL, not from this counter. */
|
||||
legCounts: Map<number, { date: string; count: number; sum: number }>;
|
||||
/** Today's session-open delay, keyed by Central calendar day (see openJitterMinutes). */
|
||||
sessionOpenJitter: Map<string, number>;
|
||||
/** Accounts whose orders are being placed right now. Their position only
|
||||
* shows once the fill comes back over the socket, so until then this is
|
||||
* what keeps them out of every other batch. */
|
||||
accountsInFlight: Set<number>;
|
||||
/** Firm id → when the tick first saw its client not ready (see UNREADY_GRACE_MS). */
|
||||
unreadySince: Map<string, number>;
|
||||
/** Firm id → when the tick last logged it as quarantined (rate limit). */
|
||||
quarantineLoggedAt: Map<string, number>;
|
||||
/** When the batch holding the trade mutex took it (0 = no batch in flight). */
|
||||
tradeInFlightSince: number;
|
||||
// What getSchedulerHealth reports.
|
||||
lastTickAt: number;
|
||||
lastTickOutcome: TickOutcome;
|
||||
lastFireAt: number;
|
||||
holding: string[];
|
||||
quarantined: string[];
|
||||
}
|
||||
|
||||
/** What the scheduler's last tick decided (getSchedulerHealth). */
|
||||
export type TickOutcome =
|
||||
| 'not-running' | 'holding-for-sync' | 'positions-open' | 'closed'
|
||||
| 'no-eligible' | 'fired' | 'skipped-in-flight' | `error: ${string}`;
|
||||
|
||||
const _tradeGlobal = globalThis as typeof globalThis & { __autoTradeRuntime?: TradeRuntime };
|
||||
|
||||
function runtime(): TradeRuntime {
|
||||
if (!_tradeGlobal.__autoTradeRuntime) {
|
||||
_tradeGlobal.__autoTradeRuntime = {
|
||||
legCounts: new Map(),
|
||||
sessionOpenJitter: new Map(),
|
||||
accountsInFlight: new Set(),
|
||||
unreadySince: new Map(),
|
||||
quarantineLoggedAt: new Map(),
|
||||
tradeInFlightSince: 0,
|
||||
lastTickAt: 0,
|
||||
lastTickOutcome: 'not-running',
|
||||
lastFireAt: 0,
|
||||
holding: [],
|
||||
quarantined: [],
|
||||
};
|
||||
}
|
||||
return _tradeGlobal.__autoTradeRuntime;
|
||||
}
|
||||
|
||||
/** How long the tick holds every firm for one whose client isn't ready. Long
|
||||
* enough for a restart or a reconnect to finish syncing; past it the client
|
||||
* is presumed dead and its firm is quarantined (see assessReadiness). */
|
||||
const UNREADY_GRACE_MS = 10 * 60 * 1000;
|
||||
/** How often a quarantined firm is re-announced in the log. */
|
||||
const QUARANTINE_LOG_EVERY_MS = 5 * 60 * 1000;
|
||||
/** A batch that has held the trade mutex this long is presumed wedged. */
|
||||
const TRADE_WEDGED_MS = 10 * 60 * 1000;
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
/** How many accounts one batch may fire, never below 1. The settings PATCH
|
||||
* stores whatever the form sends, and a cleared field arrives as '' —
|
||||
* parseInt('') is NaN, which slice() takes as 0: every batch would fire
|
||||
* nobody while the tick reported 'fired'. Non-numeric falls back to the
|
||||
* default 5, like a missing setting; the form's own floor is 1. */
|
||||
function maxConcurrentAccounts(): number {
|
||||
const configured = parseInt(getSetting('max_concurrent_accounts') ?? '5', 10);
|
||||
return Math.max(1, Number.isFinite(configured) ? configured : 5);
|
||||
}
|
||||
|
||||
function isAccountDead(amount: number, autoLiqThreshold: number): boolean {
|
||||
// Tradovate floors are positive balances; reconstructed TopstepX MLL
|
||||
// floors are negative (or +$0.01 once locked at breakeven). 0 = no floor.
|
||||
return autoLiqThreshold !== 0 && amount <= autoLiqThreshold;
|
||||
}
|
||||
|
||||
// In-memory runaway guard: legs fired per account per trade date, plus the sum
|
||||
// of their targets — realized P&L falling short of that sum is how a stop-out
|
||||
// is detected. Resets on restart, which is fine — day completion is derived
|
||||
// from realized PnL, not from this counter.
|
||||
const _legCounts = new Map<number, { date: string; count: number; sum: number }>();
|
||||
|
||||
function legsToday(accountId: number, tradeDate: string): number {
|
||||
const entry = _legCounts.get(accountId);
|
||||
const entry = runtime().legCounts.get(accountId);
|
||||
return entry && entry.date === tradeDate ? entry.count : 0;
|
||||
}
|
||||
|
||||
function legSumToday(accountId: number, tradeDate: string): number {
|
||||
const entry = _legCounts.get(accountId);
|
||||
const entry = runtime().legCounts.get(accountId);
|
||||
return entry && entry.date === tradeDate ? entry.sum : 0;
|
||||
}
|
||||
|
||||
function recordLeg(accountId: number, tradeDate: string, legAmount: number): void {
|
||||
_legCounts.set(accountId, {
|
||||
runtime().legCounts.set(accountId, {
|
||||
date: tradeDate,
|
||||
count: legsToday(accountId, tradeDate) + 1,
|
||||
sum: Math.round((legSumToday(accountId, tradeDate) + legAmount) * 100) / 100,
|
||||
});
|
||||
}
|
||||
|
||||
/** Take back a leg recordLeg counted for an entry that never reached the
|
||||
* broker. Left standing, it would bench a stop-type account for the day —
|
||||
* realized P&L "falls short" of a leg that never traded — and burn one of
|
||||
* the day's MAX_LEGS_PER_DAY. */
|
||||
function unrecordLeg(accountId: number, tradeDate: string, legAmount: number): void {
|
||||
const legCounts = runtime().legCounts;
|
||||
const entry = legCounts.get(accountId);
|
||||
if (!entry || entry.date !== tradeDate || entry.count <= 0) return;
|
||||
if (entry.count === 1) {
|
||||
legCounts.delete(accountId);
|
||||
return;
|
||||
}
|
||||
legCounts.set(accountId, {
|
||||
date: tradeDate,
|
||||
count: entry.count - 1,
|
||||
sum: Math.round((entry.sum - legAmount) * 100) / 100,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a leg's market entry. The leg was recorded before this — the runaway
|
||||
* guard has to see it while the batch is still in flight — so it is taken
|
||||
* back here only when the client KNOWS the order never reached the broker
|
||||
* (OrderNotPlacedError). Anything ambiguous, like an ack or fill timeout,
|
||||
* keeps the leg: the order may well have filled, and a leg counted that never
|
||||
* happened only costs a leg, while one forgotten that did can trade the
|
||||
* account twice.
|
||||
*/
|
||||
async function sendEntry(
|
||||
tag: string,
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
client: any,
|
||||
acc: { id: number; name: string },
|
||||
contractName: string,
|
||||
contracts: number,
|
||||
action: 'Buy' | 'Sell',
|
||||
tradeDate: string,
|
||||
legAmount: number,
|
||||
) {
|
||||
try {
|
||||
return await client.sendOrder(acc.id, contractName, contracts, action, 'Market');
|
||||
} catch (err) {
|
||||
if (isOrderNotPlaced(err)) {
|
||||
unrecordLeg(acc.id, tradeDate, legAmount);
|
||||
console.warn(`${tag} ${acc.name}: entry never reached the broker — leg un-counted`);
|
||||
} else {
|
||||
console.warn(`${tag} ${acc.name}: entry outcome unknown — leg kept (it may have filled)`);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/** Run one account's part of a batch with the account marked in flight, so
|
||||
* no other batch picks it while its orders are still being placed. */
|
||||
async function withAccountInFlight<T>(accountId: number, fire: () => Promise<T>): Promise<T> {
|
||||
const inFlight = runtime().accountsInFlight;
|
||||
inFlight.add(accountId);
|
||||
try {
|
||||
return await fire();
|
||||
} finally {
|
||||
inFlight.delete(accountId);
|
||||
}
|
||||
}
|
||||
|
||||
// ── readiness ────────────────────────────────────────────────────────────────
|
||||
|
||||
type TradableFirm = ReturnType<typeof getTradableFirms>[number];
|
||||
interface FirmClient { firm: TradableFirm; client: PlatformClient }
|
||||
|
||||
/**
|
||||
* Tradable firms that have a client, split by isClientReady. Everything that
|
||||
* trades or gates works from `ready` only: a client whose sync is missing or
|
||||
* stale holds a frozen snapshot, whose positions may be long gone and which
|
||||
* may be missing ones opened since — it must neither trade nor gate.
|
||||
*/
|
||||
function firmClients(now = Date.now()): { ready: FirmClient[]; unready: FirmClient[] } {
|
||||
const clients = getClients();
|
||||
const ready: FirmClient[] = [];
|
||||
const unready: FirmClient[] = [];
|
||||
for (const firm of getTradableFirms()) {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client) continue;
|
||||
(isClientReady(client, now) ? ready : unready).push({ firm, client });
|
||||
}
|
||||
return { ready, unready };
|
||||
}
|
||||
|
||||
function countOpenPositions(ready: FirmClient[]): number {
|
||||
return ready.reduce(
|
||||
(count, { client }) => count + client.accountList.filter((acc) => !!client.positions[acc.id]).length,
|
||||
0,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Which firms hold the tick, and which it trades without.
|
||||
*
|
||||
* A firm whose client isn't ready might hold positions nobody can see, so for
|
||||
* the first UNREADY_GRACE_MS the whole tick waits for it, as it always has —
|
||||
* right for a restart or a reconnect, which settle within a minute or two.
|
||||
* Past the grace period the client is presumed dead (broker maintenance, a
|
||||
* login that keeps failing) and the firm is quarantined: the other firms
|
||||
* trade without it, rather than one broken session stopping every firm on the
|
||||
* machine. A quarantined firm trades nothing — runTrade and copyTrade only
|
||||
* ever see ready clients — and rejoins on its first good sync.
|
||||
*
|
||||
* A client that has synced and found no accounts is ignored outright: it
|
||||
* cannot hold positions or trade, so there is nothing to wait for.
|
||||
*/
|
||||
function assessReadiness(
|
||||
ready: FirmClient[],
|
||||
unready: FirmClient[],
|
||||
now: number,
|
||||
): { holding: TradableFirm[]; quarantined: { firm: TradableFirm; unreadyMs: number }[] } {
|
||||
const rt = runtime();
|
||||
const holding: TradableFirm[] = [];
|
||||
const quarantined: { firm: TradableFirm; unreadyMs: number }[] = [];
|
||||
const tracked = new Set<string>();
|
||||
|
||||
for (const { firm, client } of unready) {
|
||||
if (client.lastSyncAt > 0 && client.accountList.length === 0) continue;
|
||||
tracked.add(firm.id);
|
||||
const since = rt.unreadySince.get(firm.id) ?? now;
|
||||
rt.unreadySince.set(firm.id, since);
|
||||
const unreadyMs = now - since;
|
||||
if (unreadyMs < UNREADY_GRACE_MS) holding.push(firm);
|
||||
else quarantined.push({ firm, unreadyMs });
|
||||
}
|
||||
|
||||
// Forget firms that are ready again (or gone), and say so for any that
|
||||
// had been traded around.
|
||||
for (const id of [...rt.unreadySince.keys()]) {
|
||||
if (tracked.has(id)) continue;
|
||||
const back = ready.find((r) => r.firm.id === id);
|
||||
if (back && rt.quarantineLoggedAt.has(id)) {
|
||||
console.log(`[scheduler] ${back.firm.name} synced again — trading it with the others`);
|
||||
}
|
||||
rt.unreadySince.delete(id);
|
||||
rt.quarantineLoggedAt.delete(id);
|
||||
}
|
||||
|
||||
return { holding, quarantined };
|
||||
}
|
||||
|
||||
// ── trade mutex ──────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* One batch at a time, instance-wide. The tick, a master "trade" command,
|
||||
* POST /api/trade and /api/copy-trade can all fire a batch, and two batches
|
||||
* collecting at once would each see the other's accounts as still flat — the
|
||||
* position gate only trips once a fill has come back over the socket.
|
||||
*
|
||||
* A batch that has held the mutex past TRADE_WEDGED_MS is presumed stuck on a
|
||||
* broker call that will never answer, and the next one goes ahead without it
|
||||
* rather than nothing trading for the rest of the process's life. The stuck
|
||||
* batch's accounts stay in accountsInFlight, so they are never traded twice,
|
||||
* and if the stuck batch does wake up it finds it no longer holds the mutex
|
||||
* and abandons its now-stale selection (holdsTradeMutex).
|
||||
*
|
||||
* Returns the lease to hand to releaseTradeMutex, or null to skip.
|
||||
*/
|
||||
function acquireTradeMutex(tag: string): number | null {
|
||||
const rt = runtime();
|
||||
const now = Date.now();
|
||||
if (rt.tradeInFlightSince > 0) {
|
||||
const heldMs = now - rt.tradeInFlightSince;
|
||||
if (heldMs < TRADE_WEDGED_MS) {
|
||||
console.log(`${tag} another batch in flight for ${Math.round(heldMs / 1000)}s — skipping`);
|
||||
return null;
|
||||
}
|
||||
console.warn(`${tag} batch in flight for ${Math.round(heldMs / 60_000)}m looks wedged — proceeding without it`);
|
||||
}
|
||||
rt.tradeInFlightSince = now;
|
||||
return now;
|
||||
}
|
||||
|
||||
function holdsTradeMutex(lease: number): boolean {
|
||||
return runtime().tradeInFlightSince === lease;
|
||||
}
|
||||
|
||||
/** Only the holder unlocks: a wedged batch that finally returns must not
|
||||
* release the batch that took over from it. */
|
||||
function releaseTradeMutex(lease: number): void {
|
||||
if (holdsTradeMutex(lease)) runtime().tradeInFlightSince = 0;
|
||||
}
|
||||
|
||||
/** Obfuscation stop for the current leg, in dollars. Infinity = no stop.
|
||||
* Day 1 of a cycle carries the $1,000 account stop (firstDayStop), min-day
|
||||
* (small) trades a 2×-target stop (minDayStop) — and the day after a losing
|
||||
@@ -134,6 +397,9 @@ function evaluateAccountForTrade(client: any, acc: { id: number; name: string; a
|
||||
if (isAccountDead(cash.amount, autoLiqThreshold)) return { ...base, eligible: false, reason: 'dead' };
|
||||
if (!acc.active) return { ...base, eligible: false, reason: 'inactive (DLL)' };
|
||||
if (client.positions[acc.id]) return { ...base, eligible: false, reason: 'already in trade' };
|
||||
// Another batch is placing this account's order right now; its position
|
||||
// won't show until the fill comes back, so the check above can't see it.
|
||||
if (runtime().accountsInFlight.has(acc.id)) return { ...base, eligible: false, reason: 'order in flight' };
|
||||
|
||||
// Obfuscation stops: a stop-out shows as realized P&L falling short of the
|
||||
// legs fired today — the day is booked; the make-back ride is tomorrow.
|
||||
@@ -210,9 +476,9 @@ function evaluateAccountForTrade(client: any, acc: { id: number; name: string; a
|
||||
// Random 1–20 minute delay added to the session open, picked once per Central
|
||||
// calendar day. Re-rolling on every check would move the open boundary between
|
||||
// polls, letting a trade slip through a moment the next check would reject.
|
||||
const sessionOpenJitter = new Map<string, number>();
|
||||
|
||||
// Held in runtime() so every bundle of this module agrees on the day's roll.
|
||||
function openJitterMinutes(centralDay: string): number {
|
||||
const sessionOpenJitter = runtime().sessionOpenJitter;
|
||||
let jitter = sessionOpenJitter.get(centralDay);
|
||||
if (jitter === undefined) {
|
||||
jitter = 1 + Math.floor(Math.random() * 20); // 1–20 inclusive
|
||||
@@ -290,23 +556,46 @@ function isInNoTradeWindow(): boolean {
|
||||
|
||||
// ── core trade logic ──────────────────────────────────────────────────────────
|
||||
|
||||
export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string) {
|
||||
/** One account's outcome in a batch: its fill summary, or { status: 'error', reason }. */
|
||||
type LegResult = { status: string; [key: string]: unknown };
|
||||
type FirmTradeResults = { firm: string; results: LegResult[] };
|
||||
type TradeOutcome = Extract<TickOutcome, 'closed' | 'positions-open' | 'no-eligible' | 'fired' | 'skipped-in-flight'>;
|
||||
|
||||
export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string): Promise<FirmTradeResults[]> {
|
||||
return (await runTradeBatch(action, symbol)).results;
|
||||
}
|
||||
|
||||
/** runTrade plus what it decided, which the tick records for getSchedulerHealth. */
|
||||
async function runTradeBatch(
|
||||
action: 'Buy' | 'Sell' | 'Auto',
|
||||
symbol: string,
|
||||
): Promise<{ outcome: TradeOutcome; results: FirmTradeResults[] }> {
|
||||
const lease = acquireTradeMutex('[auto-trade]');
|
||||
if (lease === null) return { outcome: 'skipped-in-flight', results: [] };
|
||||
try {
|
||||
return await runTradeLocked(action, symbol, lease);
|
||||
} finally {
|
||||
releaseTradeMutex(lease);
|
||||
}
|
||||
}
|
||||
|
||||
async function runTradeLocked(
|
||||
action: 'Buy' | 'Sell' | 'Auto',
|
||||
symbol: string,
|
||||
lease: number,
|
||||
): Promise<{ outcome: TradeOutcome; results: FirmTradeResults[] }> {
|
||||
if (isInNoTradeWindow()) {
|
||||
console.log('[auto-trade] CME market closed — skipping');
|
||||
return [];
|
||||
return { outcome: 'closed', results: [] };
|
||||
}
|
||||
|
||||
// Block if any account still has an open position (same gate as the scheduler tick)
|
||||
const allClients = getClients();
|
||||
const allFirms = getTradableFirms();
|
||||
const openPositions = allFirms.reduce((count, firm) => {
|
||||
const c = allClients.get(firm.id);
|
||||
if (!c) return count;
|
||||
return count + c.accountList.filter(acc => !!c.positions[acc.id]).length;
|
||||
}, 0);
|
||||
// Block if any account still has an open position (same gate as the scheduler tick).
|
||||
// Ready clients only, like everything below: a stale snapshot must neither trade nor gate.
|
||||
const { ready } = firmClients();
|
||||
const openPositions = countOpenPositions(ready);
|
||||
if (openPositions > 0) {
|
||||
console.log(`[auto-trade] ${openPositions} position(s) still open — skipping`);
|
||||
return [];
|
||||
return { outcome: 'positions-open', results: [] };
|
||||
}
|
||||
|
||||
// Resolve 'Auto' symbol once per batch so all accounts trade the same symbol
|
||||
@@ -324,9 +613,7 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
if (!pointValue) throw new Error(`Unknown symbol: ${resolvedSymbol}`);
|
||||
|
||||
const tradeDate = currentTradeDateCT();
|
||||
const maxConcurrent = parseInt(getSetting('max_concurrent_accounts') ?? '5', 10);
|
||||
const firms = getTradableFirms();
|
||||
const clients = getClients();
|
||||
const maxConcurrent = maxConcurrentAccounts();
|
||||
|
||||
// ── Phase 1: collect ALL eligible accounts across ALL firms in parallel ──
|
||||
type EligibleItem = {
|
||||
@@ -346,9 +633,8 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
|
||||
const allEligible: EligibleItem[] = [];
|
||||
|
||||
await Promise.all(firms.map(async (firm) => {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client || client.accountList.length === 0) return;
|
||||
await Promise.all(ready.map(async ({ firm, client }) => {
|
||||
if (client.accountList.length === 0) return;
|
||||
|
||||
// Skip this firm entirely if the symbol is banned for it
|
||||
if (firm.bannedSymbols.includes(resolvedSymbol)) {
|
||||
@@ -356,26 +642,42 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
return;
|
||||
}
|
||||
|
||||
// One firm's trouble (a contract lookup that throws, a malformed
|
||||
// snapshot) skips that firm, never the batch — a rejection here would
|
||||
// fail the Promise.all and every other firm would sit the tick out.
|
||||
try {
|
||||
const contract = await client.findFrontMonthContract(resolvedSymbol);
|
||||
if (!contract) return;
|
||||
|
||||
const contract = await client.findFrontMonthContract(resolvedSymbol);
|
||||
if (!contract) return;
|
||||
const found: EligibleItem[] = [];
|
||||
for (const acc of client.accountList) {
|
||||
const ev = evaluateAccountForTrade(client, acc, tradeDate);
|
||||
if (!ev || !ev.eligible || !ev.plan) continue;
|
||||
|
||||
for (const acc of client.accountList) {
|
||||
const ev = evaluateAccountForTrade(client, acc, tradeDate);
|
||||
if (!ev || !ev.eligible || !ev.plan) continue;
|
||||
|
||||
allEligible.push({
|
||||
firmName: firm.name, client, acc, contract,
|
||||
cash: ev.cash, dailyPnL: ev.dailyPnL, daysTraded: ev.daysTraded,
|
||||
plan: ev.plan, isExtraDay: ev.isExtraDay, drain: ev.drain,
|
||||
prefix: getAccountConfig(acc.id).accountType?.prefix ?? '',
|
||||
});
|
||||
found.push({
|
||||
firmName: firm.name, client, acc, contract,
|
||||
cash: ev.cash, dailyPnL: ev.dailyPnL, daysTraded: ev.daysTraded,
|
||||
plan: ev.plan, isExtraDay: ev.isExtraDay, drain: ev.drain,
|
||||
prefix: getAccountConfig(acc.id).accountType?.prefix ?? '',
|
||||
});
|
||||
}
|
||||
allEligible.push(...found);
|
||||
} catch (err) {
|
||||
console.error(`[auto-trade] ${firm.name}: skipped — ${errorMessage(err)}`);
|
||||
}
|
||||
}));
|
||||
|
||||
if (allEligible.length === 0) {
|
||||
console.log('[auto-trade] no eligible accounts found');
|
||||
return [];
|
||||
return { outcome: 'no-eligible', results: [] };
|
||||
}
|
||||
|
||||
// Collecting took so long that another batch presumed this one wedged and
|
||||
// took the mutex. That batch traded on fresher numbers than these, so
|
||||
// firing now could trade an account twice.
|
||||
if (!holdsTradeMutex(lease)) {
|
||||
console.warn('[auto-trade] superseded while collecting accounts — abandoning this batch');
|
||||
return { outcome: 'skipped-in-flight', results: [] };
|
||||
}
|
||||
|
||||
// Take only the first batch — all fired simultaneously, no rolling pool.
|
||||
@@ -395,9 +697,12 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
.sort((a, b) => b[1].length - a[1].length || a[0].localeCompare(b[0]))[0];
|
||||
const batch = batchGroup.slice(0, maxConcurrent);
|
||||
console.log(`[auto-trade] ${allEligible.length} eligible account(s) across ${groups.size} prefix group(s) — firing batch of ${batch.length} ${batchPrefix || '(no prefix)'}`);
|
||||
runtime().lastFireAt = Date.now();
|
||||
|
||||
// ── Phase 2: fire the batch simultaneously ──
|
||||
const tradeResults = await Promise.allSettled(batch.map(async (item) => {
|
||||
// Each account is marked in flight before anything is placed and released
|
||||
// once its orders are done, so no overlapping batch can pick it meanwhile.
|
||||
const tradeResults = await Promise.allSettled(batch.map((item) => withAccountInFlight(item.acc.id, async () => {
|
||||
const { client, acc, contract } = item;
|
||||
const cfg = getAccountConfig(acc.id).accountType!;
|
||||
|
||||
@@ -443,15 +748,19 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
minLegs - legsToday(acc.id, tradeDate),
|
||||
)
|
||||
: item.plan.remainingToday;
|
||||
// Counted before anything is sent so the runaway guard sees it while
|
||||
// the batch is in flight; taken back only if the entry never reached
|
||||
// the broker (here, or in sendEntry).
|
||||
recordLeg(acc.id, tradeDate, legAmount);
|
||||
|
||||
const inst = await legInstrument(client, item.drain, contract, pointValue);
|
||||
const inst = await legInstrument(client, item.drain, contract, pointValue)
|
||||
.catch((err: unknown) => { unrecordLeg(acc.id, tradeDate, legAmount); throw err; });
|
||||
const legContract = inst.contract;
|
||||
const pv = inst.pointValue;
|
||||
|
||||
const rawContracts = Math.max(1, Math.ceil(legAmount / 1000));
|
||||
const contracts = cfg.maxPositionSize > 0 ? Math.min(rawContracts, cfg.maxPositionSize) : rawContracts;
|
||||
const fill = await client.sendOrder(acc.id, legContract.name, contracts, resolvedAction, 'Market');
|
||||
const fill = await sendEntry('[auto-trade]', client, acc, legContract.name, contracts, resolvedAction, tradeDate, legAmount);
|
||||
|
||||
// Wait briefly for the cash balance WebSocket update to reflect entry commission
|
||||
await new Promise(r => setTimeout(r, 1000));
|
||||
@@ -481,9 +790,18 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
try {
|
||||
exitOrder = await client.placeOcoExit(acc.id, legContract.name, contracts, exitAction, exitPrice, stopPrice);
|
||||
} catch (err) {
|
||||
// The position is already open — never leave it exit-less. Fall
|
||||
// back to the classic lone limit and shout about it.
|
||||
console.error(`[auto-trade] ${acc.name}: placeoco REJECTED (${err instanceof Error ? err.message : err}) — falling back to lone limit exit, NO STOP`);
|
||||
// The position is already open — never leave it exit-less when
|
||||
// the pair never went in (refused, or not sent): fall back to
|
||||
// the classic lone limit and shout about it. An unknown outcome
|
||||
// (the ack timed out, the connection went after the send) is
|
||||
// different: the pair may be working, and a lone limit on top
|
||||
// doubles the exit — both filling flips the account into an
|
||||
// opposite position nothing manages. Place nothing more then.
|
||||
if (!isOrderNotPlaced(err)) {
|
||||
console.error(`[auto-trade] ${acc.name}: placeoco outcome UNKNOWN (${errorMessage(err)}) — NOT placing a second exit; check this position's exits at the broker`);
|
||||
throw new Error(`entry filled @ ${fill.price} but the exit's outcome is unknown — ${errorMessage(err)}`);
|
||||
}
|
||||
console.error(`[auto-trade] ${acc.name}: placeoco REJECTED (${errorMessage(err)}) — falling back to lone limit exit, NO STOP`);
|
||||
stopPrice = null;
|
||||
exitOrder = await client.placeOrderNoWait(acc.id, legContract.name, contracts, exitAction, 'Limit', exitPrice);
|
||||
}
|
||||
@@ -508,22 +826,28 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
exitPrice,
|
||||
commission: entryCommission,
|
||||
};
|
||||
}));
|
||||
})));
|
||||
|
||||
// Group results by firm for the response
|
||||
const firmResultsMap = new Map<string, unknown[]>();
|
||||
const firmResultsMap = new Map<string, LegResult[]>();
|
||||
for (let i = 0; i < batch.length; i++) {
|
||||
const firmName = batch[i].firmName;
|
||||
if (!firmResultsMap.has(firmName)) firmResultsMap.set(firmName, []);
|
||||
const r = tradeResults[i];
|
||||
if (r.status === 'rejected') {
|
||||
console.error(`[auto-trade] ${batch[i].acc.name} (${firmName}): failed — ${errorMessage(r.reason)}`);
|
||||
}
|
||||
firmResultsMap.get(firmName)!.push(
|
||||
r.status === 'fulfilled'
|
||||
? r.value
|
||||
: { status: 'error', reason: (r.reason as any)?.message ?? String(r.reason) }
|
||||
: { status: 'error', reason: errorMessage(r.reason) }
|
||||
);
|
||||
}
|
||||
|
||||
return Array.from(firmResultsMap.entries()).map(([firm, results]) => ({ firm, results }));
|
||||
return {
|
||||
outcome: 'fired',
|
||||
results: Array.from(firmResultsMap.entries()).map(([firm, results]) => ({ firm, results })),
|
||||
};
|
||||
}
|
||||
|
||||
// ── copy trade ───────────────────────────────────────────────────────────────
|
||||
@@ -535,14 +859,30 @@ export async function runTrade(action: 'Buy' | 'Sell' | 'Auto', symbol: string)
|
||||
*/
|
||||
export async function copyTrade() {
|
||||
console.log('[copy-trade] called');
|
||||
// Shares runTrade's mutex: a copy landing mid-batch would see the batch's
|
||||
// accounts as still flat and could fill past max_concurrent_accounts.
|
||||
const lease = acquireTradeMutex('[copy-trade]');
|
||||
if (lease === null) return [];
|
||||
try {
|
||||
return await copyTradeLocked(lease);
|
||||
} finally {
|
||||
releaseTradeMutex(lease);
|
||||
}
|
||||
}
|
||||
|
||||
async function copyTradeLocked(lease: number): Promise<unknown[]> {
|
||||
if (isInNoTradeWindow()) {
|
||||
console.log('[copy-trade] outside trading hours — skipping');
|
||||
return [];
|
||||
}
|
||||
|
||||
const firms = getTradableFirms();
|
||||
const clients = getClients();
|
||||
const maxConcurrent = Math.max(1, parseInt(getSetting('max_concurrent_accounts') ?? '5', 10));
|
||||
// Ready clients only, as in runTrade: a stale snapshot must neither be
|
||||
// copied from, counted against the slots, nor traded.
|
||||
const { ready, unready } = firmClients();
|
||||
if (unready.length > 0) {
|
||||
console.log(`[copy-trade] not synced — leaving out: ${unready.map((u) => u.firm.name).join(', ')}`);
|
||||
}
|
||||
const maxConcurrent = maxConcurrentAccounts();
|
||||
|
||||
// Find all accounts with open positions to determine direction + symbol
|
||||
let resolvedAction: 'Buy' | 'Sell' | null = null;
|
||||
@@ -554,9 +894,7 @@ export async function copyTrade() {
|
||||
let positionedClient: any = null;
|
||||
const positionedPrefixes = new Set<string>();
|
||||
|
||||
for (const firm of firms) {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client) continue;
|
||||
for (const { client } of ready) {
|
||||
for (const acc of client.accountList) {
|
||||
const pos = client.positions[acc.id];
|
||||
if (!pos) continue;
|
||||
@@ -585,10 +923,15 @@ export async function copyTrade() {
|
||||
|
||||
if (positionedClient) {
|
||||
for (const sym of enabledSymbols) {
|
||||
const contract = await positionedClient.findFrontMonthContract(sym);
|
||||
if (contract && contract.id === positionContractId) {
|
||||
resolvedSymbol = sym;
|
||||
break;
|
||||
// A lookup that throws rules out that one symbol, not the copy.
|
||||
try {
|
||||
const contract = await positionedClient.findFrontMonthContract(sym);
|
||||
if (contract && contract.id === positionContractId) {
|
||||
resolvedSymbol = sym;
|
||||
break;
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[copy-trade] ${sym}: contract lookup failed — ${errorMessage(err)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -629,10 +972,9 @@ export async function copyTrade() {
|
||||
};
|
||||
const eligible: CopyItem[] = [];
|
||||
|
||||
await Promise.all(firms.map(async (firm) => {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client || client.accountList.length === 0) {
|
||||
console.log(`[copy-trade] ${firm.name}: no client or empty account list — skip firm`);
|
||||
await Promise.all(ready.map(async ({ firm, client }) => {
|
||||
if (client.accountList.length === 0) {
|
||||
console.log(`[copy-trade] ${firm.name}: empty account list — skip firm`);
|
||||
return;
|
||||
}
|
||||
if (firm.bannedSymbols.includes(resolvedSymbol!)) {
|
||||
@@ -640,24 +982,31 @@ export async function copyTrade() {
|
||||
return;
|
||||
}
|
||||
|
||||
const contract = await client.findFrontMonthContract(resolvedSymbol!);
|
||||
if (!contract) {
|
||||
console.log(`[copy-trade] ${firm.name}: cannot resolve front-month contract for ${resolvedSymbol} — skip firm`);
|
||||
return;
|
||||
}
|
||||
// Same isolation as runTrade: one firm's throw skips that firm only.
|
||||
try {
|
||||
const contract = await client.findFrontMonthContract(resolvedSymbol!);
|
||||
if (!contract) {
|
||||
console.log(`[copy-trade] ${firm.name}: cannot resolve front-month contract for ${resolvedSymbol} — skip firm`);
|
||||
return;
|
||||
}
|
||||
|
||||
for (const acc of client.accountList) {
|
||||
const ev = evaluateAccountForTrade(client, acc, tradeDate);
|
||||
if (!ev) { console.log(`[copy-trade] ${acc.name}: no matching config prefix — skip`); continue; }
|
||||
if (!ev.eligible || !ev.plan) { console.log(`[copy-trade] ${acc.name}: ${ev.reason} — skip`); continue; }
|
||||
// copyTrade has no extra-day branch — those accounts wait for the scheduler
|
||||
if (ev.isExtraDay) { console.log(`[copy-trade] ${acc.name}: extra-day account — skip`); continue; }
|
||||
const found: CopyItem[] = [];
|
||||
for (const acc of client.accountList) {
|
||||
const ev = evaluateAccountForTrade(client, acc, tradeDate);
|
||||
if (!ev) { console.log(`[copy-trade] ${acc.name}: no matching config prefix — skip`); continue; }
|
||||
if (!ev.eligible || !ev.plan) { console.log(`[copy-trade] ${acc.name}: ${ev.reason} — skip`); continue; }
|
||||
// copyTrade has no extra-day branch — those accounts wait for the scheduler
|
||||
if (ev.isExtraDay) { console.log(`[copy-trade] ${acc.name}: extra-day account — skip`); continue; }
|
||||
|
||||
eligible.push({
|
||||
firmName: firm.name, client, acc, contract, cash: ev.cash, dailyPnL: ev.dailyPnL, plan: ev.plan,
|
||||
drain: ev.drain,
|
||||
prefix: getAccountConfig(acc.id).accountType?.prefix ?? '',
|
||||
});
|
||||
found.push({
|
||||
firmName: firm.name, client, acc, contract, cash: ev.cash, dailyPnL: ev.dailyPnL, plan: ev.plan,
|
||||
drain: ev.drain,
|
||||
prefix: getAccountConfig(acc.id).accountType?.prefix ?? '',
|
||||
});
|
||||
}
|
||||
eligible.push(...found);
|
||||
} catch (err) {
|
||||
console.error(`[copy-trade] ${firm.name}: skipped — ${errorMessage(err)}`);
|
||||
}
|
||||
}));
|
||||
|
||||
@@ -671,10 +1020,17 @@ export async function copyTrade() {
|
||||
return [];
|
||||
}
|
||||
|
||||
console.log(`[copy-trade] copying ${resolvedAction} ${resolvedSymbol} to ${batch.length} account(s)`);
|
||||
// Superseded while collecting (see runTrade) — this selection is stale.
|
||||
if (!holdsTradeMutex(lease)) {
|
||||
console.warn('[copy-trade] superseded while collecting accounts — abandoning this copy');
|
||||
return [];
|
||||
}
|
||||
|
||||
// Fire orders (same as Phase 2 of runTrade)
|
||||
const tradeResults = await Promise.allSettled(batch.map(async (item) => {
|
||||
console.log(`[copy-trade] copying ${resolvedAction} ${resolvedSymbol} to ${batch.length} account(s)`);
|
||||
runtime().lastFireAt = Date.now();
|
||||
|
||||
// Fire orders (same as Phase 2 of runTrade, accounts marked in flight the same way)
|
||||
const tradeResults = await Promise.allSettled(batch.map((item) => withAccountInFlight(item.acc.id, async () => {
|
||||
const { client, acc, contract } = item;
|
||||
const cfg = getAccountConfig(acc.id).accountType!;
|
||||
|
||||
@@ -686,15 +1042,17 @@ export async function copyTrade() {
|
||||
minLegs - legsToday(acc.id, tradeDate),
|
||||
)
|
||||
: item.plan.remainingToday;
|
||||
// Counted before sending, taken back only if the entry never reached the broker (see runTrade).
|
||||
recordLeg(acc.id, tradeDate, legAmount);
|
||||
|
||||
const inst = await legInstrument(client, item.drain, contract, pointValue);
|
||||
const inst = await legInstrument(client, item.drain, contract, pointValue)
|
||||
.catch((err: unknown) => { unrecordLeg(acc.id, tradeDate, legAmount); throw err; });
|
||||
const legContract = inst.contract;
|
||||
const pv = inst.pointValue;
|
||||
|
||||
const rawContracts = Math.max(1, Math.ceil(legAmount / 1000));
|
||||
const contracts = cfg.maxPositionSize > 0 ? Math.min(rawContracts, cfg.maxPositionSize) : rawContracts;
|
||||
const fill = await client.sendOrder(acc.id, legContract.name, contracts, resolvedAction!, 'Market');
|
||||
const fill = await sendEntry('[copy-trade]', client, acc, legContract.name, contracts, resolvedAction!, tradeDate, legAmount);
|
||||
|
||||
await new Promise(r => setTimeout(r, 1000));
|
||||
const updatedCash = client.accountCashBalances[acc.id] ?? { amount: 0, realizedPnL: 0 };
|
||||
@@ -720,7 +1078,12 @@ export async function copyTrade() {
|
||||
try {
|
||||
exitOrder = await client.placeOcoExit(acc.id, legContract.name, contracts, exitAction, exitPrice, stopPrice);
|
||||
} catch (err) {
|
||||
console.error(`[copy-trade] ${acc.name}: placeoco REJECTED (${err instanceof Error ? err.message : err}) — falling back to lone limit exit, NO STOP`);
|
||||
// Lone-limit fallback only for a pair that never went in (see runTrade).
|
||||
if (!isOrderNotPlaced(err)) {
|
||||
console.error(`[copy-trade] ${acc.name}: placeoco outcome UNKNOWN (${errorMessage(err)}) — NOT placing a second exit; check this position's exits at the broker`);
|
||||
throw new Error(`entry filled @ ${fill.price} but the exit's outcome is unknown — ${errorMessage(err)}`);
|
||||
}
|
||||
console.error(`[copy-trade] ${acc.name}: placeoco REJECTED (${errorMessage(err)}) — falling back to lone limit exit, NO STOP`);
|
||||
stopPrice = null;
|
||||
exitOrder = await client.placeOrderNoWait(acc.id, legContract.name, contracts, exitAction, 'Limit', exitPrice);
|
||||
}
|
||||
@@ -745,15 +1108,18 @@ export async function copyTrade() {
|
||||
exitPrice,
|
||||
commission: entryCommission,
|
||||
};
|
||||
}));
|
||||
})));
|
||||
|
||||
const results: unknown[] = [];
|
||||
for (let i = 0; i < batch.length; i++) {
|
||||
const r = tradeResults[i];
|
||||
if (r.status === 'rejected') {
|
||||
console.error(`[copy-trade] ${batch[i].acc.name} (${batch[i].firmName}): failed — ${errorMessage(r.reason)}`);
|
||||
}
|
||||
results.push(
|
||||
r.status === 'fulfilled'
|
||||
? r.value
|
||||
: { status: 'error', reason: (r.reason as any)?.message ?? String(r.reason) }
|
||||
: { status: 'error', reason: errorMessage(r.reason) }
|
||||
);
|
||||
}
|
||||
return results;
|
||||
@@ -763,13 +1129,31 @@ export async function copyTrade() {
|
||||
|
||||
/** Returns true if any configured account could still trade today (eligible for a next leg / extra-day, or has an open position). */
|
||||
function hasRemainingConfiguredAccounts(): boolean {
|
||||
const firms = getTradableFirms();
|
||||
const clients = getClients();
|
||||
const { ready, unready } = firmClients();
|
||||
const tradeDate = currentTradeDateCT();
|
||||
|
||||
for (const firm of firms) {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client || client.accountList.length === 0) continue;
|
||||
// A client that isn't ready can't say whether its accounts are done — its
|
||||
// snapshot is missing or stale — so while one might hold any of ours, the
|
||||
// scheduler stays up for it to rejoin. Only a client that never synced
|
||||
// and knows no accounts is an unknown in full; one whose (stale) list
|
||||
// shows none of ours can be ruled out.
|
||||
//
|
||||
// Only while the session is open, though. Outside it the unready firm
|
||||
// couldn't trade anyway, and after the close its day is over. Unbounded,
|
||||
// one firm that never comes back (a wrong password, a TopstepX key with
|
||||
// no username) would keep a stop-after-all scheduler up for good, trading
|
||||
// the ready firms every session after — exactly what the operator opted
|
||||
// out of. (Before an open with every ready account already done, this
|
||||
// stops rather than wait on the unready firm: the safe side.)
|
||||
if (!isInNoTradeWindow()) {
|
||||
for (const { client } of unready) {
|
||||
if (client.lastSyncAt === 0 && client.accountList.length === 0) return true;
|
||||
if (client.accountList.some((acc) => getAccountConfig(acc.id).accountType)) return true;
|
||||
}
|
||||
}
|
||||
|
||||
for (const { client } of ready) {
|
||||
if (client.accountList.length === 0) continue;
|
||||
|
||||
for (const acc of client.accountList) {
|
||||
const cfg = getAccountConfig(acc.id).accountType;
|
||||
@@ -777,6 +1161,8 @@ function hasRemainingConfiguredAccounts(): boolean {
|
||||
|
||||
// Account with open position = still in play
|
||||
if (client.positions[acc.id]) return true;
|
||||
// Order being placed right now = still in play (evaluation calls it ineligible)
|
||||
if (runtime().accountsInFlight.has(acc.id)) return true;
|
||||
|
||||
const ev = evaluateAccountForTrade(client, acc, tradeDate);
|
||||
if (ev?.eligible) return true;
|
||||
@@ -810,6 +1196,21 @@ function getState(): SchedulerState {
|
||||
export function startScheduler(action: 'Buy' | 'Sell' | 'Auto', symbol: string, stopAfterAll: boolean = false) {
|
||||
const state = getState();
|
||||
|
||||
// The settings reads and writes are the part that can throw, so they go
|
||||
// before anything in memory changes: a failure then leaves whatever
|
||||
// schedule was running as it was, never one marked running with its
|
||||
// interval cleared and nothing scheduled. A non-numeric setting falls back
|
||||
// to 60s — setInterval takes NaN as 1ms.
|
||||
const configuredSecs = parseInt(getSetting('tick_interval_seconds') ?? '60', 10);
|
||||
const intervalSecs = Math.max(5, Number.isFinite(configuredSecs) ? configuredSecs : 60);
|
||||
|
||||
// Mirror to the settings table so a restart can pick the schedule back up.
|
||||
// The interval itself is in-memory only; resumeSchedulerIfPersisted() recreates it.
|
||||
setSetting('scheduler_running', '1');
|
||||
setSetting('scheduler_action', action);
|
||||
setSetting('scheduler_symbol', symbol);
|
||||
setSetting('scheduler_stop_after_all', stopAfterAll ? '1' : '0');
|
||||
|
||||
// Clear any existing interval
|
||||
if (state.intervalId !== null) {
|
||||
clearInterval(state.intervalId);
|
||||
@@ -820,43 +1221,44 @@ export function startScheduler(action: 'Buy' | 'Sell' | 'Auto', symbol: string,
|
||||
state.running = true;
|
||||
state.stopAfterAll = stopAfterAll;
|
||||
|
||||
// Mirror to the settings table so a restart can pick the schedule back up.
|
||||
// The interval itself is in-memory only; resumeSchedulerIfPersisted() recreates it.
|
||||
setSetting('scheduler_running', '1');
|
||||
setSetting('scheduler_action', action);
|
||||
setSetting('scheduler_symbol', symbol);
|
||||
setSetting('scheduler_stop_after_all', stopAfterAll ? '1' : '0');
|
||||
|
||||
const tick = async () => {
|
||||
if (!state.running) return;
|
||||
state.lastRun = new Date();
|
||||
|
||||
// Skip this tick until every client has completed its initial sync (positions are populated)
|
||||
const clients = getClients();
|
||||
const firms = getTradableFirms();
|
||||
const notReady = firms.filter(f => {
|
||||
const c = clients.get(f.id);
|
||||
return c && !c.syncComplete;
|
||||
});
|
||||
if (notReady.length > 0) {
|
||||
console.log(`[scheduler] waiting for sync: ${notReady.map(f => f.name).join(', ')}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip this tick if any account still has an open position from the previous batch
|
||||
const openPositions = firms.reduce((count, firm) => {
|
||||
const client = clients.get(firm.id);
|
||||
if (!client) return count;
|
||||
return count + client.accountList.filter(acc => !!client.positions[acc.id]).length;
|
||||
}, 0);
|
||||
if (openPositions > 0) {
|
||||
console.log(`[scheduler] ${openPositions} position(s) still open — skipping tick`);
|
||||
return;
|
||||
}
|
||||
const rt = runtime();
|
||||
const now = Date.now();
|
||||
rt.lastTickAt = now;
|
||||
|
||||
try {
|
||||
const results = await runTrade(state.action, state.symbol);
|
||||
const filled = results.flatMap((r: any) => r.results ?? []).filter((r: any) => r.status === 'filled').length;
|
||||
// Hold the tick while any firm's client is still inside its sync
|
||||
// grace period (positions may not be populated yet); trade around
|
||||
// firms that have been out past it (see assessReadiness).
|
||||
const { ready, unready } = firmClients(now);
|
||||
const { holding, quarantined } = assessReadiness(ready, unready, now);
|
||||
rt.holding = holding.map((f) => f.name);
|
||||
rt.quarantined = quarantined.map((q) => q.firm.name);
|
||||
if (holding.length > 0) {
|
||||
rt.lastTickOutcome = 'holding-for-sync';
|
||||
console.log(`[scheduler] waiting for sync: ${rt.holding.join(', ')}`);
|
||||
return;
|
||||
}
|
||||
for (const { firm, unreadyMs } of quarantined) {
|
||||
if (now - (rt.quarantineLoggedAt.get(firm.id) ?? 0) < QUARANTINE_LOG_EVERY_MS) continue;
|
||||
rt.quarantineLoggedAt.set(firm.id, now);
|
||||
console.warn(`[scheduler] ${firm.name} not synced for ${Math.round(unreadyMs / 60_000)}m — trading the other firms without it`);
|
||||
}
|
||||
|
||||
// Skip this tick if any account still has an open position from the
|
||||
// previous batch — one batch fires only when every (ready) account is flat.
|
||||
const openPositions = countOpenPositions(ready);
|
||||
if (openPositions > 0) {
|
||||
rt.lastTickOutcome = 'positions-open';
|
||||
console.log(`[scheduler] ${openPositions} position(s) still open — skipping tick`);
|
||||
return;
|
||||
}
|
||||
|
||||
const { outcome, results } = await runTradeBatch(state.action, state.symbol);
|
||||
rt.lastTickOutcome = outcome;
|
||||
const filled = results.flatMap((r) => r.results).filter((r) => r.status === 'filled').length;
|
||||
if (filled > 0) console.log(`[scheduler] tick: ${filled} account(s) filled`);
|
||||
|
||||
// Auto-stop if user opted in and no configured accounts can trade anymore
|
||||
@@ -865,11 +1267,11 @@ export function startScheduler(action: 'Buy' | 'Sell' | 'Auto', symbol: string,
|
||||
stopScheduler();
|
||||
}
|
||||
} catch (err) {
|
||||
console.error('[scheduler] tick error:', err);
|
||||
rt.lastTickOutcome = `error: ${errorMessage(err)}`;
|
||||
console.error(`[scheduler] tick error: ${errorMessage(err)}`);
|
||||
}
|
||||
};
|
||||
|
||||
const intervalSecs = Math.max(5, parseInt(getSetting('tick_interval_seconds') ?? '60', 10));
|
||||
state.intervalId = setInterval(tick, intervalSecs * 1_000);
|
||||
console.log(`[scheduler] started — ${action} ${symbol} every ${intervalSecs}s`);
|
||||
}
|
||||
@@ -881,6 +1283,17 @@ export function stopScheduler() {
|
||||
state.intervalId = null;
|
||||
}
|
||||
state.running = false;
|
||||
|
||||
// Nothing watches readiness while stopped, so the grace clocks would be
|
||||
// stale by the next start — that start gives every firm a fresh grace.
|
||||
// Reset before the settings write, which can throw.
|
||||
const rt = runtime();
|
||||
rt.unreadySince.clear();
|
||||
rt.quarantineLoggedAt.clear();
|
||||
rt.holding = [];
|
||||
rt.quarantined = [];
|
||||
rt.lastTickOutcome = 'not-running';
|
||||
|
||||
setSetting('scheduler_running', '0');
|
||||
console.log('[scheduler] stopped');
|
||||
}
|
||||
@@ -888,9 +1301,10 @@ export function stopScheduler() {
|
||||
/**
|
||||
* Restart the schedule that was running before the process went down.
|
||||
*
|
||||
* No sync-wait here on purpose: tick() already skips while any client reports
|
||||
* !syncComplete, and again while any account holds an open position. So the
|
||||
* worst case is a few logged no-op ticks until the clients finish syncing.
|
||||
* No sync-wait here on purpose: tick() already holds while any client isn't
|
||||
* ready (for up to UNREADY_GRACE_MS each), and again while any account holds
|
||||
* an open position. So the worst case is a few logged no-op ticks until the
|
||||
* clients finish syncing.
|
||||
*/
|
||||
export function resumeSchedulerIfPersisted(): void {
|
||||
if (getSetting('scheduler_running') !== '1') return;
|
||||
@@ -915,3 +1329,31 @@ export function getSchedulerStatus() {
|
||||
sessionOpenDelayMin: todaysSessionOpenDelay(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Why the scheduler is or isn't trading, for health checks — separate from
|
||||
* getSchedulerStatus, whose shape is what the master receives. `holding` are
|
||||
* the firms the last tick waited on (inside their sync grace period);
|
||||
* `quarantined` the ones past it, traded around. Timestamps are epoch ms,
|
||||
* 0 = never; lastFireAt counts batches from any caller, not just the tick.
|
||||
* Until the first tick after a start, the outcome still reads 'not-running'.
|
||||
*/
|
||||
export function getSchedulerHealth(): {
|
||||
lastTickAt: number;
|
||||
lastTickOutcome: TickOutcome;
|
||||
lastFireAt: number;
|
||||
holding: string[];
|
||||
quarantined: string[];
|
||||
tradeInFlightSince: number;
|
||||
} {
|
||||
const running = getState().running;
|
||||
const rt = runtime();
|
||||
return {
|
||||
lastTickAt: rt.lastTickAt,
|
||||
lastTickOutcome: running ? rt.lastTickOutcome : 'not-running',
|
||||
lastFireAt: rt.lastFireAt,
|
||||
holding: running ? [...rt.holding] : [],
|
||||
quarantined: running ? [...rt.quarantined] : [],
|
||||
tradeInFlightSince: rt.tradeInFlightSince,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user