Blame for src/engine.ts

/**
 * 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();