| /** | |
| * CandleEngine — server-owned canonical bar store. | |
| * | |
| * Architecture: | |
| * | |
| * Tick ──► CandleEngine.onTick(symbol, price, volume, timestamp) | |
| * │ | |
| * ├──► applies session calendar (rejects off-hours ticks) | |
| * ├──► updates the 1m BarBucket (single source of truth) | |
| * ├──► broadcasts {t:'tick'} (conflated, 250ms) for header/watchlist | |
| * └──► on bar rollover OR each tick: | |
| * updates derived bars (5m, 15m, 1H, 4H, 1D, 1W, 1M) | |
| * emits {t:'bar', closed:false} to bar subscribers | |
| * on rollover, emits {t:'bar', closed:true} for the | |
| * just-closed bar, then opens the next. | |
| * | |
| * Single source of truth = 1m bars. All other timeframes are derived by | |
| * aggregating consecutive 1m bars (intraday) or by re-bucketing time | |
| * (daily/weekly/monthly). This eliminates the "alignment drift between | |
| * history and live" class of bugs entirely. | |
| * | |
| * Memory: ring buffers, BAR_BUCKET_CAPACITY per (sym, tf). Evicted after | |
| * BUCKET_IDLE_EVICTION_MS of no subscribers. | |
| * | |
| * Thread model: Node single-threaded; no locks needed. Every operation is | |
| * synchronous from the caller's perspective. | |
| */ | |
| import { EventEmitter } from 'node:events'; | |
| import { | |
| AssetClass, Bar, Tick, Timeframe, TIMEFRAMES, | |
| } from './types.js'; | |
| import { bucketStart, isTickAcceptable, isBucketAligned, alignToBucket } from './calendar.js'; | |
| import { getClass, getValidFromSec } from './symbols.js'; | |
| import { | |
| BUCKET_IDLE_EVICTION_MS, BAR_CONFLATION_INTERVAL_MS, | |
| MAX_SERVER_BAR_MEMORY_MB, BUCKET_IDLE_EVICTION_MIN, MAX_BARS_PER_BUCKET, | |
| } from './config.js'; | |
| import { createLogger } from './log.js'; | |
| import { recordFeedAge } from './load.js'; | |
| /** | |
| * Conservative per-bar memory footprint used for the server-wide LRU budget. | |
| * A Bar object holds 6 floats × 8 bytes = 48 bytes of payload; add ~16 bytes | |
| * of V8 object-header overhead = ~64 bytes. We pick 48 (payload-only) to | |
| * remain a conservative *lower* bound on real memory — the LRU is happy | |
| * underestimating since the budget bites earlier, not later, than the true | |
| * heap usage. Real overhead is absorbed by the 10 % headroom built into | |
| * `evictLRU` (drains to 90 % of budget). | |
| */ | |
| const BAR_BYTES = 48; | |
| const log = createLogger('engine'); | |
| /** | |
| * Defensive OHLC sanity check. Returns null if the bar passes, otherwise a | |
| * short reason string. Providers occasionally return bars with high<open, | |
| * NaN volume, or negative values — those must never enter a bucket. | |
| */ | |
| function validateBarOhlc(b: Bar): string | null { | |
| if (!isFinite(b.time) || b.time <= 0) return 'bad-time'; | |
| if (!isFinite(b.open) || !isFinite(b.high) || !isFinite(b.low) || !isFinite(b.close)) return 'nan-ohlc'; | |
| if (b.open <= 0 || b.high <= 0 || b.low <= 0 || b.close <= 0) return 'non-positive-ohlc'; | |
| if (!isFinite(b.volume) || b.volume < 0) return 'bad-volume'; | |
| if (b.high < Math.max(b.open, b.close)) return 'high<max(o,c)'; | |
| if (b.low > Math.min(b.open, b.close)) return 'low>min(o,c)'; | |
| if (b.high < b.low) return 'high<low'; | |
| return null; | |
| } | |
| /** Max ticks held in a bucket's seed-window queue. On overflow, the oldest is | |
| * dropped — at typical streaming rates (≤10 ticks/sec/sym) 500 covers a | |
| * ~50-second seed window, well beyond a normal Alpaca/Yahoo fetch (1–3 s). */ | |
| const PENDING_TICKS_MAX = 500; | |
| interface PendingTick { | |
| price: number; | |
| volume: number; | |
| timestamp: number; | |
| } | |
| // --------------------------------------------------------------------------- | |
| // BarBucket — ring buffer of bars for ONE (symbol, timeframe) | |
| // --------------------------------------------------------------------------- | |
| export class BarBucket { | |
| public readonly symbol: string; | |
| public readonly klass: AssetClass; | |
| public readonly tf: Timeframe; | |
| /** Bars in ascending order by .time. */ | |
| private bars: Bar[] = []; | |
| /** Wall-clock ms of the last activity (tick or subscribe). */ | |
| public lastActivityMs: number = Date.now(); | |
| /** | |
| * Wall-clock ms of the most recent read access (snapshot / applyTick / | |
| * acquire). Used by EngineRegistry.evictLRU to pick the coldest idle | |
| * bucket when bar-memory exceeds budget. | |
| */ | |
| public lastQueriedAt: number = Date.now(); | |
| /** Subscriber ref-count — when 0 the bucket is a candidate for eviction. */ | |
| public refCount: number = 0; | |
| /** | |
| * Per-bucket monotonic sequence number. Bumped on every emitted bar | |
| * event (including rollover's two events). Snapshots and catchups also | |
| * stamp the current seq so clients can drop out-of-order WS frames. | |
| * | |
| * Counts up to Number.MAX_SAFE_INTEGER before wrapping back to 1 (it never | |
| * returns 0). At 1000 events/sec that's a ~285,000-year lifetime, so the | |
| * wrap is purely theoretical — but keeping it strictly increasing-then- | |
| * reset (rather than the old 31-bit `| 0`, which wrapped every ~25 days) | |
| * means a long-lived bucket can never silently start handing out seq values | |
| * a client has already seen and would therefore drop. | |
| */ | |
| private seq: number = 0; | |
| /** | |
| * Instrument identity boundary (epoch sec) — bars whose bucket-aligned | |
| * start precedes it are rejected in seedHistory/prependHistory. This is | |
| * the AUTHORITATIVE enforcement point: the comparison runs after | |
| * alignToBucket, so it is immune to provider stamp conventions (the | |
| * history layer's raw-stamp filter is a bulk/efficiency pass only). | |
| * Undefined for continuous-identity assets. See SymbolMeta.validFromUtc. | |
| */ | |
| public readonly validFromSec?: number; | |
| constructor(symbol: string, klass: AssetClass, tf: Timeframe, validFromSec?: number) { | |
| this.symbol = symbol; | |
| this.klass = klass; | |
| this.tf = tf; | |
| this.validFromSec = validFromSec; | |
| } | |
| /** True once seedHistory has been called. Prevents ticks from creating | |
| * bars in an empty bucket that's still waiting for history to load. */ | |
| private seeded: boolean = false; | |
| /** True while a history fetch is in-flight for this bucket. Ticks arriving | |
| * during this window are queued (not dropped) and replayed after seed. */ | |
| private seeding: boolean = false; | |
| /** Ticks captured during the seed window. Replayed in order once seed | |
| * completes. Capped at PENDING_TICKS_MAX — oldest is dropped on overflow. */ | |
| private pendingTicks: PendingTick[] = []; | |
| /** | |
| * NEPSE 1D / 1W / 1M only — see CandleEngine.applySessionBar. | |
| * | |
| * `sessionTime` is the 1D start of the session currently folded into the | |
| * last bar; `sessionBase` is that last bar WITHOUT it (the earlier sessions | |
| * of the same week or month). Each update replaces the session's share | |
| * instead of adding to it, which is what keeps a weekly bar exact while its | |
| * current day is still trading. Always null for 1D: a day has one session. | |
| */ | |
| public sessionTime: number | null = null; | |
| public sessionBase: Bar | null = null; | |
| public setSessionOverlay(overlay: SessionOverlay | undefined): void { | |
| this.sessionTime = overlay?.sessionTime ?? null; | |
| this.sessionBase = this.tf === '1D' ? null : overlay?.base ?? null; | |
| } | |
| /** | |
| * Fold the latest figures of one session into this period bucket. | |
| * `session` is the exchange's whole-session bar, stamped at its 1D start. | |
| */ | |
| public applySession(session: Bar): TickResult { | |
| this.lastQueriedAt = Date.now(); | |
| const periodStart = bucketStart(this.klass, this.tf, session.time * 1000); | |
| const last = this.lastBar(); | |
| if (!last || periodStart > last.time) { | |
| // First session of a new day/week/month. | |
| const fresh: Bar = { ...session, time: periodStart }; | |
| this.bars.push(fresh); | |
| this.sessionTime = session.time; | |
| this.sessionBase = null; | |
| return last | |
| ? { kind: 'rollover', closed: { ...last }, fresh: { ...fresh } } | |
| : { kind: 'open', current: { ...fresh } }; | |
| } | |
| if (periodStart < last.time) return { kind: 'stale' }; | |
| if (this.sessionTime !== session.time) { | |
| // A later session inside the same week or month: everything already | |
| // in the bar is final, and becomes the base this session builds on. | |
| this.sessionBase = this.tf === '1D' ? null : { ...last }; | |
| this.sessionTime = session.time; | |
| } | |
| const base = this.sessionBase; | |
| last.open = base ? base.open : session.open; | |
| last.high = base ? Math.max(base.high, session.high) : session.high; | |
| last.low = base ? Math.min(base.low, session.low) : session.low; | |
| last.close = session.close; | |
| last.volume = (base ? base.volume : 0) + session.volume; | |
| return { kind: 'update', current: { ...last } }; | |
| } | |
| public bumpSeq(): number { | |
| // Wrap only at the float-safe-integer ceiling (never 0) instead of the | |
| // old 31-bit `| 0`, which silently wrapped every ~25 days of uptime. | |
| if (this.seq >= Number.MAX_SAFE_INTEGER) this.seq = 1; | |
| else this.seq += 1; | |
| return this.seq; | |
| } | |
| public currentSeq(): number { return this.seq; } | |
| /** Test seam: park the sequence at a chosen value to exercise wrap. */ | |
| public __setSeqForTest(v: number): void { this.seq = v; } | |
| /** | |
| * Replace the entire bar set (used at history-bootstrap time). | |
| * | |
| * Defensive: validates OHLC, drops invalid bars, and re-aligns any bar | |
| * whose `time` doesn't match `bucketStart()`. This is the safety net for | |
| * provider-vs-engine timestamp drift (e.g. OANDA H4 with custom | |
| * alignmentTimezone) that was the root cause of the Silver 4H phantom. | |
| */ | |
| public seedHistory(bars: Bar[]): void { | |
| const out: Bar[] = []; | |
| let invalid = 0; | |
| let realigned = 0; | |
| let preBoundary = 0; | |
| for (const raw of bars) { | |
| const reason = validateBarOhlc(raw); | |
| if (reason) { | |
| invalid++; | |
| continue; | |
| } | |
| let b: Bar = raw; | |
| if (!isBucketAligned(this.klass, this.tf, raw.time)) { | |
| const aligned = alignToBucket(this.klass, this.tf, raw.time); | |
| b = { ...raw, time: aligned }; | |
| realigned++; | |
| } | |
| // Instrument identity boundary — checked AFTER alignment so a | |
| // provider stamp above the boundary whose bucket starts below it | |
| // is still rejected (alignment only moves times backward). | |
| if (this.validFromSec !== undefined && b.time < this.validFromSec) { | |
| preBoundary++; | |
| continue; | |
| } | |
| out.push(b); | |
| } | |
| // Sort ascending + collapse duplicates (later overwrites earlier — bars | |
| // arrive in chronological order from providers, so "later" = "newer | |
| // close" which is what we want for the in-progress bar at the right edge). | |
| out.sort((a, b) => a.time - b.time); | |
| const deduped: Bar[] = []; | |
| for (const b of out) { | |
| const tail = deduped[deduped.length - 1]; | |
| if (tail && tail.time === b.time) { | |
| // Merge OHLC: open from earlier, h/l aggregated, close from later, volume summed. | |
| deduped[deduped.length - 1] = { | |
| time: tail.time, | |
| open: tail.open, | |
| high: Math.max(tail.high, b.high), | |
| low: Math.min(tail.low, b.low), | |
| close: b.close, | |
| volume: tail.volume + b.volume, | |
| }; | |
| } else { | |
| deduped.push(b); | |
| } | |
| } | |
| this.bars = deduped; | |
| this.seeded = true; | |
| this.seeding = false; | |
| this.lastActivityMs = Date.now(); | |
| this.lastQueriedAt = this.lastActivityMs; | |
| if (invalid > 0 || realigned > 0 || preBoundary > 0) { | |
| log.warn(`seedHistory ${this.symbol}/${this.tf}: ${invalid} invalid, ${realigned} realigned, ${preBoundary} pre-boundary, ${this.bars.length} kept`); | |
| } | |
| } | |
| public isSeeded(): boolean { return this.seeded; } | |
| /** | |
| * Approximate in-memory footprint of this bucket's bar array, in bytes. | |
| * Used by EngineRegistry.totalBarMemory / evictLRU. See BAR_BYTES above | |
| * for the conservative per-bar estimate. | |
| */ | |
| public get byteSize(): number { | |
| return this.bars.length * BAR_BYTES; | |
| } | |
| /** | |
| * Prepend older bars (backfill chunk). Bars MUST be strictly older than | |
| * any bar currently in the bucket — overlapping/duplicate timestamps are | |
| * silently dropped (the seeded range is authoritative for its window). | |
| * | |
| * Validates and realigns each input bar the same way `seedHistory` does, | |
| * preserving the per-bucket invariant: all bars sorted ascending, all | |
| * timestamps bucket-aligned, no duplicates. Returns the deduped, sorted, | |
| * strictly-older bars that actually landed in the bucket — `bars` is what | |
| * the WS hub should rebroadcast (never include filtered-out overlap). | |
| */ | |
| public prependHistory(olderBars: Bar[]): { added: number; bars: Bar[] } { | |
| if (!this.seeded) { | |
| // Prepend onto an unseeded bucket would be ambiguous (no boundary | |
| // for the overlap filter). Caller MUST seed first. Drop silently. | |
| return { added: 0, bars: [] }; | |
| } | |
| if (olderBars.length === 0) return { added: 0, bars: [] }; | |
| const oldestExisting = this.bars[0]?.time ?? Number.POSITIVE_INFINITY; | |
| const validated: Bar[] = []; | |
| let invalid = 0; | |
| let realigned = 0; | |
| let overlapped = 0; | |
| let preBoundary = 0; | |
| for (const raw of olderBars) { | |
| const reason = validateBarOhlc(raw); | |
| if (reason) { invalid++; continue; } | |
| let b: Bar = raw; | |
| if (!isBucketAligned(this.klass, this.tf, raw.time)) { | |
| const aligned = alignToBucket(this.klass, this.tf, raw.time); | |
| b = { ...raw, time: aligned }; | |
| realigned++; | |
| } | |
| // Instrument identity boundary — post-alignment, same as seedHistory. | |
| if (this.validFromSec !== undefined && b.time < this.validFromSec) { | |
| preBoundary++; | |
| continue; | |
| } | |
| // Strictly older than the seeded range — no merge, no overwrite. | |
| if (b.time >= oldestExisting) { overlapped++; continue; } | |
| validated.push(b); | |
| } | |
| if (validated.length === 0) { | |
| if (invalid > 0 || overlapped > 0 || preBoundary > 0) { | |
| log.debug(`prependHistory ${this.symbol}/${this.tf}: 0 added (${invalid} invalid, ${overlapped} overlap, ${preBoundary} pre-boundary)`); | |
| } | |
| return { added: 0, bars: [] }; | |
| } | |
| validated.sort((a, b) => a.time - b.time); | |
| // Dedup within the prepend itself (provider can return overlapping pages). | |
| const deduped: Bar[] = []; | |
| for (const b of validated) { | |
| const tail = deduped[deduped.length - 1]; | |
| if (tail && tail.time === b.time) { | |
| deduped[deduped.length - 1] = { | |
| time: tail.time, | |
| open: tail.open, | |
| high: Math.max(tail.high, b.high), | |
| low: Math.min(tail.low, b.low), | |
| close: b.close, | |
| volume: tail.volume + b.volume, | |
| }; | |
| } else { | |
| deduped.push(b); | |
| } | |
| } | |
| // Splice prepend in front; the seeded range stays untouched. | |
| this.bars = deduped.concat(this.bars); | |
| // Per-bucket safety cap: a bucket that stays subscribed through a deep | |
| // backfill is never an LRU eviction candidate, so bound its growth here | |
| // by trimming the OLDEST bars (keep the live right edge). Invisible to | |
| // clients — the prepend chunk was already returned for broadcast above; | |
| // only the server-retained copy is bounded. (Live append is the hot path | |
| // and is intentionally NOT capped here.) | |
| if (this.bars.length > MAX_BARS_PER_BUCKET) { | |
| const dropped = this.bars.length - MAX_BARS_PER_BUCKET; | |
| this.bars = this.bars.slice(dropped); | |
| log.warn(`prependHistory ${this.symbol}/${this.tf}: trimmed ${dropped} oldest bar(s) at MAX_BARS_PER_BUCKET=${MAX_BARS_PER_BUCKET}`); | |
| } | |
| this.lastActivityMs = Date.now(); | |
| this.lastQueriedAt = this.lastActivityMs; | |
| if (invalid > 0 || realigned > 0 || overlapped > 0 || preBoundary > 0) { | |
| log.debug(`prependHistory ${this.symbol}/${this.tf}: +${deduped.length} (${invalid} invalid, ${realigned} realigned, ${overlapped} overlap, ${preBoundary} pre-boundary)`); | |
| } | |
| return { added: deduped.length, bars: deduped }; | |
| } | |
| /** True between `beginSeed()` and the subsequent `seedHistory()` / | |
| * `markSeedEmpty()` call — ticks are queued, not applied. */ | |
| public isSeeding(): boolean { return this.seeding; } | |
| /** Mark this bucket as awaiting a history seed. Idempotent. Ticks that | |
| * arrive while seeding is true are pushed onto pendingTicks. */ | |
| public beginSeed(): void { | |
| this.seeding = true; | |
| this.lastActivityMs = Date.now(); | |
| } | |
| /** Called when the history fetch returned 0 bars (provider down / quota | |
| * exhausted / symbol off-hours with no recent data). Clears the seeding | |
| * flag so we stop queueing ticks, but leaves `seeded = false` so a | |
| * subsequent re-subscribe re-attempts the fetch. Drains the pending | |
| * queue (no baseline to apply them to). */ | |
| public markSeedEmpty(): void { | |
| this.seeding = false; | |
| this.pendingTicks = []; | |
| this.lastActivityMs = Date.now(); | |
| } | |
| /** Queue a tick that arrived during the seed window. Capped at | |
| * PENDING_TICKS_MAX — when full, the oldest tick is discarded. */ | |
| public enqueueTick(price: number, volume: number, timestamp: number): void { | |
| if (this.pendingTicks.length >= PENDING_TICKS_MAX) { | |
| this.pendingTicks.shift(); | |
| } | |
| this.pendingTicks.push({ price, volume, timestamp }); | |
| } | |
| /** Pop the pending tick queue. Returned in arrival order so the engine | |
| * can replay them via the normal applyTick path. */ | |
| public drainPendingTicks(): PendingTick[] { | |
| const drained = this.pendingTicks; | |
| this.pendingTicks = []; | |
| return drained; | |
| } | |
| /** Return a defensive copy of all bars (used for snapshot/catchup). */ | |
| public snapshot(): Bar[] { | |
| this.lastQueriedAt = Date.now(); | |
| return this.bars.map(b => ({ ...b })); | |
| } | |
| /** Bars with time >= since (catch-up after reconnect). */ | |
| public barsSince(since: number): Bar[] { | |
| // Binary-search lower bound | |
| let lo = 0, hi = this.bars.length; | |
| while (lo < hi) { | |
| const mid = (lo + hi) >>> 1; | |
| if (this.bars[mid].time < since) lo = mid + 1; else hi = mid; | |
| } | |
| return this.bars.slice(lo).map(b => ({ ...b })); | |
| } | |
| /** The most recent (possibly in-progress) bar, or null if empty. */ | |
| public lastBar(): Bar | null { | |
| return this.bars.length > 0 ? this.bars[this.bars.length - 1] : null; | |
| } | |
| public size(): number { return this.bars.length; } | |
| /** | |
| * Apply a tick to the bar bucket. Returns the result of the operation, | |
| * letting the engine know what to broadcast. | |
| */ | |
| public applyTick(price: number, volume: number, nowMs: number): TickResult { | |
| this.lastQueriedAt = Date.now(); | |
| const newStartSec = bucketStart(this.klass, this.tf, nowMs); | |
| const last = this.lastBar(); | |
| // Case 1: bucket is empty — open the first bar. | |
| if (!last) { | |
| const bar: Bar = { | |
| time: newStartSec, open: price, high: price, low: price, close: price, | |
| volume, | |
| }; | |
| this.bars.push(bar); | |
| return { kind: 'open', current: { ...bar } }; | |
| } | |
| // Case 2: still within the current bar — update H/L/C/V in place. | |
| if (newStartSec === last.time) { | |
| if (price > last.high) last.high = price; | |
| if (price < last.low) last.low = price; | |
| last.close = price; | |
| last.volume += volume; | |
| return { kind: 'update', current: { ...last } }; | |
| } | |
| // Case 3: new period started. | |
| if (newStartSec > last.time) { | |
| // GUARD: For long-period bars, do NOT create new bars from live | |
| // ticks. These should only come from history fetches. Live ticks | |
| // just update the last bar in-place. This prevents phantom | |
| // candles caused by calendar boundary misalignment between | |
| // history provider timestamps and our computed buckets. | |
| // | |
| // Always guarded: 1W, 1M (every asset class). | |
| // Also guarded for forex/commodity/index: 1D — the 17:00 ET | |
| // cutover doesn't always land where bucketStart computes for a | |
| // live tick arriving near the boundary, and the OANDA history | |
| // is the authoritative source for daily forex/metals bars. | |
| const dailyForexGuard = this.tf === '1D' | |
| && (this.klass === AssetClass.Forex | |
| || this.klass === AssetClass.Commodity | |
| || this.klass === AssetClass.Index); | |
| if (this.tf === '1W' || this.tf === '1M' || dailyForexGuard) { | |
| // Update the existing last bar instead of creating a new one | |
| if (price > last.high) last.high = price; | |
| if (price < last.low) last.low = price; | |
| last.close = price; | |
| last.volume += volume; | |
| return { kind: 'update', current: { ...last } }; | |
| } | |
| const closedCopy: Bar = { ...last }; | |
| const fresh: Bar = { | |
| time: newStartSec, | |
| open: price, high: price, low: price, close: price, | |
| volume, | |
| }; | |
| this.bars.push(fresh); | |
| return { kind: 'rollover', closed: closedCopy, fresh: { ...fresh } }; | |
| } | |
| // Case 4: backward timestamp (clock skew or out-of-order tick). Drop. | |
| // Log when the skew is significant — this surfaces "latest candle | |
| // not updating" bugs where every tick lands in an older bucket than | |
| // the most recent bar. | |
| const skewSec = last.time - newStartSec; | |
| if (skewSec > 60) { | |
| log.warn(`stale tick ${this.symbol}/${this.tf}: tick bucket=${newStartSec} last=${last.time} skew=${skewSec}s`); | |
| } | |
| return { kind: 'stale' }; | |
| } | |
| /** | |
| * Force-close the current bar without consuming a new tick. Useful for | |
| * the rollover-timer pathway (e.g. forex on quiet weekends) to ensure | |
| * bars roll over even when no trades arrive. | |
| */ | |
| public forceRollover(nowMs: number): TickResult { | |
| const last = this.lastBar(); | |
| if (!last) return { kind: 'stale' }; | |
| const newStartSec = bucketStart(this.klass, this.tf, nowMs); | |
| if (newStartSec === last.time) return { kind: 'stale' }; | |
| if (newStartSec < last.time) return { kind: 'stale' }; | |
| // Roll over by carrying the close price forward as the new bar's open. | |
| const closedCopy: Bar = { ...last }; | |
| const fresh: Bar = { | |
| time: newStartSec, | |
| open: last.close, high: last.close, low: last.close, close: last.close, | |
| volume: 0, | |
| }; | |
| this.bars.push(fresh); | |
| return { kind: 'rollover', closed: closedCopy, fresh: { ...fresh } }; | |
| } | |
| } | |
| type TickResult = | |
| | { kind: 'stale' } | |
| | { kind: 'open'; current: Bar } | |
| | { kind: 'update'; current: Bar } | |
| | { kind: 'rollover'; closed: Bar; fresh: Bar }; | |
| /** How a seeded NEPSE period bucket relates to the session still trading. See BarBucket.sessionBase. */ | |
| export interface SessionOverlay { | |
| sessionTime: number; | |
| base: Bar | null; | |
| } | |
| /** Timeframes whose NEPSE bars follow the exchange's session figures, not ticks. */ | |
| function isSessionTf(tf: Timeframe): boolean { | |
| return tf === '1D' || tf === '1W' || tf === '1M'; | |
| } | |
| // --------------------------------------------------------------------------- | |
| // Engine events | |
| // --------------------------------------------------------------------------- | |
| export interface BarEvent { | |
| symbol: string; | |
| tf: Timeframe; | |
| bar: Bar; | |
| closed: boolean; | |
| /** Per-bucket monotonic sequence — clients drop frames with seq <= lastSeq. */ | |
| seq: number; | |
| } | |
| export interface TickBroadcastEvent { | |
| symbol: string; | |
| price: number; | |
| timestamp: number; | |
| } | |
| interface EngineEvents { | |
| bar: (e: BarEvent) => void; | |
| tick: (e: TickBroadcastEvent) => void; | |
| } | |
| // --------------------------------------------------------------------------- | |
| // CandleEngine | |
| // --------------------------------------------------------------------------- | |
| export class CandleEngine extends EventEmitter { | |
| /** Map: symbol → Map<Timeframe, BarBucket> */ | |
| private readonly buckets = new Map<string, Map<Timeframe, BarBucket>>(); | |
| /** Last-broadcasted tick price per symbol (so we can dedupe redundant emits). */ | |
| private readonly lastTickPrice = new Map<string, number>(); | |
| /** Conflation throttle — wall time of last broadcast per symbol. */ | |
| private readonly lastTickEmitAt = new Map<string, number>(); | |
| /** Per-symbol latest price snapshot (used by clients subscribing later). */ | |
| private readonly latestPrice = new Map<string, number>(); | |
| /** | |
| * The trade's OWN time (Tick.timestamp) for the price in `latestPrice` — | |
| * written in the same statement, so the two always describe one trade. | |
| * Absent when the feed sent no usable time. See getLatestTradeAt. | |
| */ | |
| private readonly latestTradeAt = new Map<string, number>(); | |
| /** Rollover timer — sweeps every 5 s to close inactive bars. */ | |
| private rolloverTimer: NodeJS.Timeout | null = null; | |
| /** Eviction timer — sweeps every minute to drop idle buckets. */ | |
| private evictionTimer: NodeJS.Timeout | null = null; | |
| /** | |
| * Debounce flag for maybeEvictLRU. A market-open burst (or a 3-way backfill | |
| * wave of up to ~300 prepends) calls maybeEvictLRU many times in one | |
| * synchronous run; without this, each call does an O(buckets) totalBarMemory | |
| * scan. The flag collapses a burst to a single scheduled eviction. | |
| */ | |
| private evictPending = false; | |
| public on<E extends keyof EngineEvents>(event: E, listener: EngineEvents[E]): this { | |
| return super.on(event, listener); | |
| } | |
| public emit<E extends keyof EngineEvents>(event: E, ...args: Parameters<EngineEvents[E]>): boolean { | |
| return super.emit(event, ...args); | |
| } | |
| public start(): void { | |
| if (this.rolloverTimer || this.evictionTimer) return; | |
| // One-time migration: prior builds bucketed OANDA H4 on raw UTC clock | |
| // (00/04/08/12/16/20 UTC). The current build anchors H4 to 17:00 ET | |
| // (per OANDA's H4 "day-alignment" spec) so existing-process buckets | |
| // hold misaligned bars. Evict them so the first subscribe re-seeds | |
| // with correctly-anchored history. | |
| this.evictStaleOandaH4OnBoot(); | |
| // Roll bars over even in absence of ticks (e.g. quiet symbol over weekend). | |
| this.rolloverTimer = setInterval(() => this.sweepRollovers(), 5_000); | |
| this.evictionTimer = setInterval(() => this.sweepEvictions(), 60_000); | |
| log.info('CandleEngine started'); | |
| } | |
| /** | |
| * Boot-time eviction of OANDA-class 4H buckets. See `start()`. Safe to call | |
| * before any client subscribes — buckets at refCount===0 are dropped | |
| * outright; the next subscribe re-seeds them from HistoryFetcher. | |
| */ | |
| private evictStaleOandaH4OnBoot(): void { | |
| let evicted = 0; | |
| for (const [sym, perSym] of this.buckets) { | |
| const bucket = perSym.get('4H'); | |
| if (!bucket) continue; | |
| if (bucket.klass !== AssetClass.Forex | |
| && bucket.klass !== AssetClass.Commodity | |
| && bucket.klass !== AssetClass.Index) continue; | |
| perSym.delete('4H'); | |
| if (perSym.size === 0) this.buckets.delete(sym); | |
| evicted++; | |
| } | |
| if (evicted > 0) log.info(`evicted ${evicted} stale OANDA-class 4H bucket(s) on boot (re-anchor to 17:00 ET)`); | |
| } | |
| public stop(): void { | |
| if (this.rolloverTimer) clearInterval(this.rolloverTimer); | |
| if (this.evictionTimer) clearInterval(this.evictionTimer); | |
| this.rolloverTimer = null; | |
| this.evictionTimer = null; | |
| } | |
| // ----------------------------------------------------------------------- | |
| // Tick ingestion (the hot path) | |
| // ----------------------------------------------------------------------- | |
| /** | |
| * Ingest a normalized tick. The engine routes it to all BarBuckets for the | |
| * symbol and emits bar updates + a conflated tick broadcast. | |
| * | |
| * Off-hours ticks are FILTERED at the feed-source level (see oanda.ts / | |
| * alpaca.ts which both check calendar.isTickAcceptable before forwarding). | |
| * Here we just trust the caller — calling isTickAcceptable per tick would | |
| * recompute ET parts hundreds of times per second. | |
| */ | |
| public onTick(t: Tick): void { | |
| if (!isFinite(t.price) || t.price <= 0) return; | |
| const klass = getClass(t.symbol); | |
| if (!klass) return; // Unknown symbol — drop | |
| this.latestPrice.set(t.symbol, t.price); | |
| if (isFinite(t.timestamp) && t.timestamp > 0) this.latestTradeAt.set(t.symbol, t.timestamp); | |
| else this.latestTradeAt.delete(t.symbol); | |
| // Crypto trades around the clock with a real trade time, so its age says how | |
| // far behind the feed itself is running, whatever else the market is doing. | |
| if (klass === 'crypto') recordFeedAge(Date.now() - t.timestamp); | |
| // 1) Conflated tick broadcast for MarketHeader/WatchlistPanel | |
| this.maybeEmitTick(t.symbol, t.price, t.timestamp); | |
| // 2) Update bar buckets if any exist | |
| const tfBuckets = this.buckets.get(t.symbol); | |
| if (!tfBuckets || tfBuckets.size === 0) return; | |
| const nepse = klass === AssetClass.Nepse; | |
| for (const bucket of tfBuckets.values()) { | |
| // NEPSE daily/weekly/monthly bars take the exchange's own session | |
| // figures (applySessionBar), never tick samples. | |
| if (nepse && isSessionTf(bucket.tf)) continue; | |
| // Seed window: queue ticks instead of dropping. The engine replays | |
| // them in order via applyTick once the bucket is seeded, so the | |
| // right-edge live bar reflects activity that happened during the | |
| // (potentially 1–3s) Alpaca/Yahoo history fetch. | |
| // | |
| // NEPSE is the exception, and must be: its seed is read from the | |
| // feed's session book in the same turn the bucket is seeded, AFTER | |
| // the fetch — so the book already contains every tick that arrived | |
| // meanwhile. Replaying them too would count their volume twice. | |
| if (bucket.isSeeding()) { | |
| if (!nepse) bucket.enqueueTick(t.price, t.volume, t.timestamp); | |
| continue; | |
| } | |
| // Not seeding and not yet seeded → the bucket was created but no | |
| // seed was ever attempted (or the seed returned 0 bars and was | |
| // marked empty). Drop the tick — there's no baseline to apply it | |
| // to, and a fresh seed will run on the next client subscribe. | |
| if (!bucket.isSeeded()) continue; | |
| const result = bucket.applyTick(t.price, t.volume, t.timestamp); | |
| bucket.lastActivityMs = Date.now(); | |
| this.emitFromTickResult(bucket, result); | |
| } | |
| } | |
| /** How often a symbol's tick may be broadcast. The hub sets it by client count. */ | |
| private tickIntervalMs = BAR_CONFLATION_INTERVAL_MS; | |
| public setTickInterval(ms: number): void { this.tickIntervalMs = ms; } | |
| private maybeEmitTick(symbol: string, price: number, timestamp: number): void { | |
| const now = Date.now(); | |
| const last = this.lastTickEmitAt.get(symbol) ?? 0; | |
| const lastPrice = this.lastTickPrice.get(symbol); | |
| if (lastPrice === price && now - last < this.tickIntervalMs) return; | |
| if (now - last < this.tickIntervalMs) return; | |
| this.lastTickEmitAt.set(symbol, now); | |
| this.lastTickPrice.set(symbol, price); | |
| this.emit('tick', { symbol, price, timestamp }); | |
| } | |
| private emitFromTickResult(bucket: BarBucket, r: TickResult): void { | |
| switch (r.kind) { | |
| case 'stale': | |
| return; | |
| case 'open': | |
| case 'update': | |
| this.emit('bar', { | |
| symbol: bucket.symbol, tf: bucket.tf, | |
| bar: r.current, closed: false, seq: bucket.bumpSeq(), | |
| }); | |
| return; | |
| case 'rollover': | |
| // Emit final 'closed:true' for the just-finished bar, then open the new one. | |
| // Each gets its own seq so the client can distinguish them. | |
| this.emit('bar', { | |
| symbol: bucket.symbol, tf: bucket.tf, | |
| bar: r.closed, closed: true, seq: bucket.bumpSeq(), | |
| }); | |
| this.emit('bar', { | |
| symbol: bucket.symbol, tf: bucket.tf, | |
| bar: r.fresh, closed: false, seq: bucket.bumpSeq(), | |
| }); | |
| return; | |
| } | |
| } | |
| // ----------------------------------------------------------------------- | |
| // Bucket lifecycle (subscribe / unsubscribe / seed) | |
| // ----------------------------------------------------------------------- | |
| /** | |
| * Ensure a bucket exists for (symbol, tf) and return it. The caller is | |
| * responsible for incrementing refCount via `acquire()`. | |
| */ | |
| public getOrCreateBucket(symbol: string, tf: Timeframe): BarBucket | null { | |
| const klass = getClass(symbol); | |
| if (!klass) return null; | |
| let perSym = this.buckets.get(symbol); | |
| if (!perSym) { | |
| perSym = new Map(); | |
| this.buckets.set(symbol, perSym); | |
| } | |
| let bucket = perSym.get(tf); | |
| if (!bucket) { | |
| bucket = new BarBucket(symbol, klass, tf, getValidFromSec(symbol)); | |
| perSym.set(tf, bucket); | |
| } | |
| return bucket; | |
| } | |
| public getBucket(symbol: string, tf: Timeframe): BarBucket | undefined { | |
| return this.buckets.get(symbol)?.get(tf); | |
| } | |
| /** Increment refCount for a (sym, tf) subscription. */ | |
| public acquire(symbol: string, tf: Timeframe): BarBucket | null { | |
| const bucket = this.getOrCreateBucket(symbol, tf); | |
| if (!bucket) return null; | |
| bucket.refCount += 1; | |
| bucket.lastActivityMs = Date.now(); | |
| bucket.lastQueriedAt = bucket.lastActivityMs; | |
| return bucket; | |
| } | |
| /** Decrement refCount. When 0, the bucket becomes eligible for eviction. */ | |
| public release(symbol: string, tf: Timeframe): void { | |
| const bucket = this.getBucket(symbol, tf); | |
| if (!bucket) return; | |
| bucket.refCount = Math.max(0, bucket.refCount - 1); | |
| bucket.lastActivityMs = Date.now(); | |
| } | |
| /** | |
| * Mark a bucket as awaiting a history seed. Called by client-hub BEFORE | |
| * `await fetchHistory(...)` so that any ticks arriving during the fetch | |
| * window are queued (not dropped) and replayed once the seed lands. | |
| */ | |
| public beginSeed(symbol: string, tf: Timeframe): void { | |
| const bucket = this.getOrCreateBucket(symbol, tf); | |
| if (!bucket) return; | |
| bucket.beginSeed(); | |
| } | |
| /** | |
| * Seed a bucket's history (called once after the first subscribe, with | |
| * data from the HistoryFetcher). Clears the seeding flag and replays any | |
| * ticks captured during the fetch window in arrival order. | |
| */ | |
| public seedBucketHistory(symbol: string, tf: Timeframe, bars: Bar[], overlay?: SessionOverlay): void { | |
| const bucket = this.getOrCreateBucket(symbol, tf); | |
| if (!bucket) return; | |
| bucket.seedHistory(bars); | |
| bucket.setSessionOverlay(overlay); | |
| // Replay any ticks queued while the fetch was in flight. applyTick | |
| // handles backward-time ticks via its own stale-guard, so we don't | |
| // need to pre-filter — and emitting via the normal path lets bar | |
| // subscribers see the catch-up updates with monotonic seq numbers. | |
| const drained = bucket.drainPendingTicks(); | |
| for (const pt of drained) { | |
| if (!isFinite(pt.price) || pt.price <= 0) continue; | |
| const result = bucket.applyTick(pt.price, pt.volume, pt.timestamp); | |
| bucket.lastActivityMs = Date.now(); | |
| this.emitFromTickResult(bucket, result); | |
| } | |
| // Deep seeds (potentially thousands of bars × hundreds of subscriptions) | |
| // can push total bar memory over budget. Evict idle buckets if so. | |
| this.maybeEvictLRU(); | |
| } | |
| /** | |
| * Mark a bucket's seed attempt as having returned 0 bars (provider | |
| * unavailable or symbol has no recent data). Clears `seeding` so we stop | |
| * queueing ticks, but leaves `seeded === false` so the client's retry | |
| * triggers a fresh fetch on its next `sub_bars`. | |
| */ | |
| public markSeedEmpty(symbol: string, tf: Timeframe): void { | |
| const bucket = this.getBucket(symbol, tf); | |
| if (!bucket) return; | |
| bucket.markSeedEmpty(); | |
| } | |
| /** | |
| * Prepend a chunk of older bars (backfill page) to an already-seeded | |
| * bucket. The bucket's seeded range is authoritative — any input bar | |
| * whose timestamp falls inside it is dropped (no merge, no overwrite). | |
| * | |
| * Returns `{ added }` reporting how many bars actually landed after | |
| * validation + overlap filter, or `null` if the bucket doesn't exist | |
| * (e.g., evicted between the fetch and the prepend). | |
| * | |
| * Triggers server-wide LRU eviction post-write when total bar memory | |
| * exceeds MAX_SERVER_BAR_MEMORY_MB. | |
| */ | |
| public prependHistory(symbol: string, tf: Timeframe, olderBars: Bar[]): { added: number; bars: Bar[] } | null { | |
| const bucket = this.getBucket(symbol, tf); | |
| if (!bucket) return null; | |
| const result = bucket.prependHistory(olderBars); | |
| if (result.added > 0) this.maybeEvictLRU(); | |
| return result; | |
| } | |
| /** | |
| * Sum of `byteSize` across every bucket in the engine. Bytes; convert | |
| * to MB before comparing to MAX_SERVER_BAR_MEMORY_MB. | |
| */ | |
| public totalBarMemory(): number { | |
| let total = 0; | |
| for (const perSym of this.buckets.values()) { | |
| for (const bucket of perSym.values()) { | |
| total += bucket.byteSize; | |
| } | |
| } | |
| return total; | |
| } | |
| /** | |
| * Server-level LRU eviction. Called after every bar-array growth point | |
| * (seedBucketHistory, prependHistory) — if total bar memory > budget, | |
| * drop the oldest idle (refCount===0) buckets until total ≤ 90 % of | |
| * budget OR no idle candidates remain. | |
| * | |
| * Subscribed buckets are NEVER evicted regardless of how large they grow. | |
| * If all candidates are referenced and we're over budget, a single WARN | |
| * is logged per call — the process keeps running (real overhead is small | |
| * compared to V8 heap budget; this is a soft limit). | |
| */ | |
| /** Cumulative since boot. Exposed via diagnostics() — see the note there. */ | |
| private evictionsTotal = 0; | |
| private bytesFreedTotal = 0; | |
| private budgetPressureEvents = 0; | |
| public evictLRU(): { evicted: number; bytesFreed: number } { | |
| const budgetBytes = MAX_SERVER_BAR_MEMORY_MB * 1_048_576; | |
| let total = this.totalBarMemory(); | |
| if (total <= budgetBytes) return { evicted: 0, bytesFreed: 0 }; | |
| const targetBytes = Math.floor(budgetBytes * 0.9); | |
| const now = Date.now(); | |
| const minIdleMs = BUCKET_IDLE_EVICTION_MIN * 60_000; | |
| const candidates: Array<{ sym: string; tf: Timeframe; bucket: BarBucket }> = []; | |
| for (const [sym, perSym] of this.buckets) { | |
| for (const [tf, bucket] of perSym) { | |
| if (bucket.refCount > 0) continue; | |
| if (now - bucket.lastQueriedAt < minIdleMs) continue; | |
| candidates.push({ sym, tf, bucket }); | |
| } | |
| } | |
| candidates.sort((a, b) => a.bucket.lastQueriedAt - b.bucket.lastQueriedAt); | |
| let evicted = 0; | |
| let bytesFreed = 0; | |
| for (const { sym, tf, bucket } of candidates) { | |
| if (total <= targetBytes) break; | |
| const sz = bucket.byteSize; | |
| const perSym = this.buckets.get(sym); | |
| if (!perSym) continue; | |
| perSym.delete(tf); | |
| if (perSym.size === 0) this.buckets.delete(sym); | |
| total -= sz; | |
| bytesFreed += sz; | |
| evicted++; | |
| const idleMin = Math.round((now - bucket.lastQueriedAt) / 60_000); | |
| log.info(`evicted ${sym}:${tf} (${Math.round(sz / 1024)} KB, idle ${idleMin}m) — heap budget`); | |
| } | |
| this.evictionsTotal += evicted; | |
| this.bytesFreedTotal += bytesFreed; | |
| if (total > budgetBytes && candidates.length === 0) { | |
| this.budgetPressureEvents++; | |
| log.warn(`memory budget exceeded with no idle buckets — consider raising STRIX_MAX_BAR_MEMORY_MB or reducing concurrent deep subscriptions (total=${Math.round(total / 1_048_576)} MB / budget=${MAX_SERVER_BAR_MEMORY_MB} MB)`); | |
| } | |
| return { evicted, bytesFreed }; | |
| } | |
| /** | |
| * Internal: trigger LRU eviction iff memory exceeds budget. Debounced — a | |
| * synchronous burst of calls schedules exactly one eviction on the next | |
| * event-loop turn (setImmediate fires before the next I/O/tick), instead of | |
| * running an O(buckets) scan per call. evictLRU() re-checks the budget and | |
| * is a no-op if memory has since dropped under it, so the deferral is safe. | |
| * Direct evictLRU() callers (e.g. tests) are unaffected. | |
| */ | |
| private maybeEvictLRU(): void { | |
| if (this.evictPending) return; | |
| const budgetBytes = MAX_SERVER_BAR_MEMORY_MB * 1_048_576; | |
| if (this.totalBarMemory() <= budgetBytes) return; | |
| this.evictPending = true; | |
| setImmediate(() => { | |
| this.evictPending = false; | |
| this.evictLRU(); | |
| }); | |
| } | |
| /** | |
| * NEPSE: fold the exchange's latest whole-session figures into the 1D, 1W | |
| * and 1M buckets of `symbol` (whichever exist and are seeded), and emit the | |
| * change like any bar update. | |
| * | |
| * Why not ticks: the daily candle has an authoritative source — the | |
| * exchange's own open/high/low/last/volume for the session — while ticks | |
| * are samples taken every few seconds that can miss the trade that set the | |
| * high. The feed calls this after every poll that changed a security. | |
| */ | |
| public applySessionBar(symbol: string, session: Bar): void { | |
| if (validateBarOhlc(session)) return; | |
| const perSym = this.buckets.get(symbol); | |
| if (!perSym) return; | |
| for (const tf of ['1D', '1W', '1M'] as const) { | |
| const bucket = perSym.get(tf); | |
| if (!bucket || !bucket.isSeeded()) continue; | |
| this.emitFromTickResult(bucket, bucket.applySession(session)); | |
| bucket.lastActivityMs = Date.now(); | |
| } | |
| } | |
| public getLatestPrice(symbol: string): number { | |
| return this.latestPrice.get(symbol) ?? 0; | |
| } | |
| /** | |
| * Wall time of the most recent tick BROADCAST for a symbol, or 0 if none. | |
| * | |
| * This is the only per-symbol observation timestamp the engine keeps: | |
| * `latestPrice` is a bare number with no time attached. `lastTickEmitAt` is | |
| * written by `maybeEmitTick`, which is throttled to at most one broadcast per | |
| * BAR_CONFLATION_INTERVAL_MS — so while ticks flow it trails the true last | |
| * tick by under that interval, and when ticks stop it freezes. | |
| * | |
| * That makes it CONSERVATIVE for freshness: it over-states age by at most one | |
| * conflation interval and never under-states it. /quotes reports it verbatim | |
| * as `asOf` rather than synthesising `Date.now()`, which would present a | |
| * stored price as a live observation. | |
| */ | |
| public getLatestTickAt(symbol: string): number { | |
| return this.lastTickEmitAt.get(symbol) ?? 0; | |
| } | |
| /** | |
| * When the trade behind `getLatestPrice` happened, by the feed's own clock | |
| * (UNIX ms), or 0 when unknown. Unlike `getLatestTickAt` — the throttled | |
| * BROADCAST time, which can trail the price or, for a trade the feed | |
| * delivered late, post-date it — this always belongs to the price itself. | |
| * Use it to decide which session a price is from (quotes.ts). | |
| */ | |
| public getLatestTradeAt(symbol: string): number { | |
| return this.latestTradeAt.get(symbol) ?? 0; | |
| } | |
| public setLatestPrice(symbol: string, price: number): void { | |
| if (!(isFinite(price) && price > 0)) return; | |
| this.latestPrice.set(symbol, price); | |
| // A price set by hand has no trade time; never pair it with an old one. | |
| this.latestTradeAt.delete(symbol); | |
| } | |
| // ----------------------------------------------------------------------- | |
| // Background sweeps | |
| // ----------------------------------------------------------------------- | |
| private sweepRollovers(): void { | |
| const now = Date.now(); | |
| for (const perSym of this.buckets.values()) { | |
| for (const bucket of perSym.values()) { | |
| // CRITICAL: Do NOT force-roll bars during off-hours. | |
| // This prevents phantom empty candles from appearing when | |
| // the market is closed (e.g., stocks on weekends, forex on Sat). | |
| if (!isTickAcceptable(bucket.klass, now)) continue; | |
| // Never force-roll NEPSE. Its intraday bars exist only for | |
| // minutes that traded — the same rule its seed (the session | |
| // tape and floorsheet) follows — so a bucket that had lived | |
| // through a quiet stretch would otherwise disagree with a fresh | |
| // seed of the same stock. Its session bars follow the exchange. | |
| if (bucket.klass === AssetClass.Nepse) continue; | |
| // Never force-roll weekly/monthly — same guard as applyTick. | |
| // Also never force-roll daily for OANDA-class (forex/commodity/ | |
| // index) — those are aligned to 17:00 ET and we trust history. | |
| if (bucket.tf === '1W' || bucket.tf === '1M') continue; | |
| if (bucket.tf === '1D' | |
| && (bucket.klass === AssetClass.Forex | |
| || bucket.klass === AssetClass.Commodity | |
| || bucket.klass === AssetClass.Index)) continue; | |
| // Don't force-roll unseeded buckets | |
| if (!bucket.isSeeded()) continue; | |
| const last = bucket.lastBar(); | |
| if (!last) continue; | |
| const expectedStart = bucketStart(bucket.klass, bucket.tf, now); | |
| if (expectedStart > last.time) { | |
| // Time has moved past this bar's period — roll over. | |
| const result = bucket.forceRollover(now); | |
| this.emitFromTickResult(bucket, result); | |
| } | |
| } | |
| } | |
| } | |
| private sweepEvictions(): void { | |
| const now = Date.now(); | |
| for (const [sym, perSym] of this.buckets) { | |
| for (const [tf, bucket] of perSym) { | |
| if (bucket.refCount > 0) continue; | |
| if (now - bucket.lastActivityMs < BUCKET_IDLE_EVICTION_MS) continue; | |
| perSym.delete(tf); | |
| log.info(`evicted idle bucket ${sym}/${tf}`); | |
| } | |
| if (perSym.size === 0) this.buckets.delete(sym); | |
| } | |
| } | |
| // ----------------------------------------------------------------------- | |
| // Diagnostics | |
| // ----------------------------------------------------------------------- | |
| /** | |
| * Diagnostics, including whether the bar-memory budget is ACTUALLY BINDING. | |
| * | |
| * `barMemoryMb` vs `barBudgetMb` is the question that matters. The budget | |
| * (default 2048) is often described as "2 GB of a 16 GB box", which reads | |
| * like waste — it is not. The real ceiling is the V8 old-space set in the | |
| * Dockerfile (`--max-old-space-size=4096`), and the bar cache deliberately | |
| * sits at half of that so the other half remains for the Node baseline, WS | |
| * send buffers, parsers and GC headroom. The box's remaining ~12 GB is | |
| * unusable by this process without raising the V8 flag too. | |
| * | |
| * Raising both is possible but NOT automatically an improvement: a larger | |
| * old-space means longer major GC pauses, and on 2 vCPUs a GC pause is | |
| * event-loop lag — the exact signal that now governs client admission | |
| * (load.ts). More cached history could therefore cost concurrent users. | |
| * | |
| * So the numbers below exist to settle it with evidence. If, after real | |
| * traffic, `evictionsTotal` stays 0 and `barMemoryMb` never approaches | |
| * `barBudgetMb`, the budget is not the constraint and raising it would | |
| * achieve nothing. If `budgetPressureEvents` climbs, it is binding and | |
| * both values should be raised together. | |
| */ | |
| /** | |
| * Last traded price seen for a symbol, or undefined if none has arrived | |
| * since boot. Read-only view of the conflation map the tick path already | |
| * maintains — no new bookkeeping. | |
| * | |
| * Added for watch-live.ts, which needs the current price of a symbol that | |
| * did NOT produce the tick it is reacting to (a condition spanning two | |
| * symbols). Keeping a second price book there would be a copy of this one. | |
| */ | |
| public lastPrice(symbol: string): number | undefined { | |
| return this.lastTickPrice.get(symbol); | |
| } | |
| public diagnostics(): { | |
| symbols: number; | |
| buckets: number; | |
| totalBars: number; | |
| barMemoryMb: number; | |
| barBudgetMb: number; | |
| evictionsTotal: number; | |
| mbFreedTotal: number; | |
| budgetPressureEvents: number; | |
| } { | |
| let buckets = 0; | |
| let totalBars = 0; | |
| for (const perSym of this.buckets.values()) { | |
| for (const b of perSym.values()) { | |
| buckets++; | |
| totalBars += b.size(); | |
| } | |
| } | |
| return { | |
| symbols: this.buckets.size, | |
| buckets, | |
| totalBars, | |
| barMemoryMb: Math.round(this.totalBarMemory() / 1_048_576), | |
| barBudgetMb: MAX_SERVER_BAR_MEMORY_MB, | |
| evictionsTotal: this.evictionsTotal, | |
| mbFreedTotal: Math.round(this.bytesFreedTotal / 1_048_576), | |
| budgetPressureEvents: this.budgetPressureEvents, | |
| }; | |
| } | |
| /** Set of all timeframes (compile-time constant). */ | |
| public readonly allTimeframes: readonly Timeframe[] = TIMEFRAMES; | |
| } | |
| // --------------------------------------------------------------------------- | |
| // Singleton | |
| // --------------------------------------------------------------------------- | |
| export const engine = new CandleEngine(); | |