spb/market-atlas

1import type { CanonicalQuote, EventType, MarketEvent, Severity } from "@market-atlas/market-model";2import { makeEventId } from "@market-atlas/market-model";3import { shortHash } from "@market-atlas/connector-sdk";4import type { ProposedEvent } from "@market-atlas/connector-sdk";5import { pool } from "../db/pool.js";6import { logger } from "../logger.js";7import { bus } from "./bus.js";8import { telemetry } from "./telemetry.js";910interface InstrumentStats {11  lastPrice: number | null;12  lastEventPrice: number | null;13  sessionHigh: number;14  sessionLow: number;15  /** EWMA of absolute 1-tick returns for volatility baselines. */16  ewmaAbsRet: number;17  ewmaVar: number;18  ticks: number;19  lastEventAt: Record<string, number>;20  lastSourceDivergenceAt: number;21  recentReturns: number[];22}2324/**25 * Event engine: turns canonical quotes into meaningful, deduplicated events and canonicalizes26 * connector-proposed events (filings, halts…). Thresholds are relative to per-instrument baselines,27 * so a 0.5% move in a treasury yield is not treated like 0.5% on a crypto pair.28 */29export class EventEngine {30  private stats = new Map<string, InstrumentStats>();31  private recentFingerprints = new Map<string, number>();32  private pending: MarketEvent[] = [];33  private timer: NodeJS_Timeout | null = null;3435  private stat(id: string): InstrumentStats {36    let s = this.stats.get(id);37    if (!s) {38      s = { lastPrice: null, lastEventPrice: null, sessionHigh: -Infinity, sessionLow: Infinity, ewmaAbsRet: 0, ewmaVar: 0, ticks: 0, lastEventAt: {}, lastSourceDivergenceAt: 0, recentReturns: [] };39      this.stats.set(id, s);40    }41    return s;42  }4344  resetSession(instrumentId: string) {45    const s = this.stat(instrumentId);46    s.sessionHigh = -Infinity;47    s.sessionLow = Infinity;48  }4950  onQuote(q: CanonicalQuote, name: string): void {51    if (q.price == null || q.realtimeStatus === "STALE") return;52    const s = this.stat(q.instrumentId);53    const now = q.updatedAt;54    const p = q.price;55    if (s.lastPrice != null && s.lastPrice !== 0) {56      const ret = (p - s.lastPrice) / s.lastPrice;57      const absRet = Math.abs(ret);58      const alpha = 0.05;59      s.ewmaAbsRet = s.ticks < 20 ? (s.ewmaAbsRet * s.ticks + absRet) / (s.ticks + 1) : s.ewmaAbsRet * (1 - alpha) + absRet * alpha;60      s.ewmaVar = s.ticks < 20 ? (s.ewmaVar * s.ticks + ret * ret) / (s.ticks + 1) : s.ewmaVar * (1 - alpha) + ret * ret * alpha;61      s.recentReturns.push(ret);62      if (s.recentReturns.length > 60) s.recentReturns.shift();63      // Volatility spike: realized volatility over the last 30 ticks vs long-run baseline (needs history).64      if (s.ticks > 120 && s.recentReturns.length >= 30) {65        const recent = Math.sqrt(s.recentReturns.slice(-30).reduce((a, r) => a + r * r, 0) / 30);66        const baseline = Math.sqrt(s.ewmaVar);67        if (baseline > 0 && recent > 4 * baseline && this.cooldown(s, "VOLATILITY_SPIKE", now, 10 * 60_000)) {68          this.emitDerived("VOLATILITY_SPIKE", q, name, now, "WARNING", `${q.symbol} volatility ${(recent / baseline).toFixed(1)}× baseline`, {69            ratio: round(recent / baseline, 2),70            realized: recent,71            baseline,72          });73        }74      }75    }76    // Price change events on a per-instrument adaptive threshold (≥ 0.5% and ≥ 8× typical tick move).77    const ref = s.lastEventPrice ?? s.lastPrice;78    if (ref != null && ref !== 0 && s.ticks > 60) {79      const move = (p - ref) / ref;80      const threshold = Math.max(0.01, 12 * s.ewmaAbsRet);81      if (Math.abs(move) >= threshold && this.cooldown(s, "PRICE_CHANGE", now, 5 * 60_000)) {82        this.emitDerived("PRICE_CHANGE", q, name, now, Math.abs(move) >= 0.03 ? "WARNING" : "INFO", `${q.symbol} ${move >= 0 ? "+" : ""}${(move * 100).toFixed(2)}%`, {83          from: ref,84          to: p,85          changePercent: round(move * 100, 3),86        });87        s.lastEventPrice = p;88      }89    }90    if (s.lastEventPrice == null) s.lastEventPrice = p;91    // Session highs/lows: only after meaningful history, when the previous extreme is beaten by ≥ 0.1 %, with a 15-minute cooldown.92    if (s.ticks > 200) {93      if (p > s.sessionHigh * 1.001 && Number.isFinite(s.sessionHigh) && this.cooldown(s, "SESSION_HIGH", now, 15 * 60_000)) {94        this.emitDerived("SESSION_HIGH", q, name, now, "INFO", `${q.symbol} new session high ${fmt(p)}`, { price: p, previousHigh: s.sessionHigh });95      }96      if (p < s.sessionLow * 0.999 && Number.isFinite(s.sessionLow) && this.cooldown(s, "SESSION_LOW", now, 15 * 60_000)) {97        this.emitDerived("SESSION_LOW", q, name, now, "INFO", `${q.symbol} new session low ${fmt(p)}`, { price: p, previousLow: s.sessionLow });98      }99    }100    if (p > s.sessionHigh) s.sessionHigh = p;101    if (p < s.sessionLow) s.sessionLow = p;102    // Source divergence: included sources disagree by more than 50 bps (crypto venues legitimately differ; flag ≥ 50).103    if (q.dispersionBps != null && q.dispersionBps > 100 && q.sourceCount >= 2 && now - s.lastSourceDivergenceAt > 60 * 60_000) {104      s.lastSourceDivergenceAt = now;105      this.emitDerived("SOURCE_DIVERGENCE", q, name, now, "NOTICE", `${q.symbol}: sources diverge by ${q.dispersionBps.toFixed(0)} bps`, {106        dispersionBps: q.dispersionBps,107        contributions: q.contributions.filter((c) => c.included).map((c) => ({ source: c.sourceId, value: c.value, ageMs: c.ageMs })),108      });109    }110    s.lastPrice = p;111    s.ticks++;112  }113114  private cooldown(s: InstrumentStats, type: string, now: number, ms: number): boolean {115    const last = s.lastEventAt[type] ?? 0;116    if (now - last < ms) return false;117    s.lastEventAt[type] = now;118    return true;119  }120121  private emitDerived(type: EventType, q: CanonicalQuote, name: string, ts: number, severity: Severity, title: string, data: Record<string, unknown>) {122    const minute = Math.floor(ts / 60_000);123    const fingerprint = shortHash(`${type}|${q.instrumentId}|${minute}|${JSON.stringify(data)}`, 24);124    this.publish({125      version: 1,126      id: makeEventId(ts, fingerprint),127      type,128      instrumentIds: [q.instrumentId],129      entityIds: [],130      timestamp: ts,131      severity,132      confidence: q.confidence,133      sourceCount: q.sourceCount,134      sources: q.contributions.filter((c) => c.included).map((c) => c.sourceId),135      title,136      summary: `${name} · ${q.sourceCount} independent source${q.sourceCount === 1 ? "" : "s"} · confidence ${(q.confidence * 100).toFixed(1)}%`,137      data: { ...data, symbol: q.symbol, price: q.price, currency: q.currency },138      fingerprint,139      supportingObservations: [],140    });141  }142143  /** Canonicalize an event proposed by a connector or the runtime (dedupe on dedupeKey within a 24 h window). */144  proposed(e: ProposedEvent, sourceId: string, instrumentIds: string[]): MarketEvent | null {145    const fingerprint = shortHash(`${e.type}|${e.dedupeKey}`, 24);146    const ev: MarketEvent = {147      version: 1,148      id: makeEventId(e.timestamp, fingerprint),149      type: e.type,150      instrumentIds,151      entityIds: e.entityIds ?? [],152      timestamp: e.timestamp,153      severity: e.severity ?? "INFO",154      confidence: e.confidence ?? 0.9,155      sourceCount: 1,156      sources: [sourceId],157      title: e.title,158      summary: e.summary ?? null,159      data: e.data ?? {},160      fingerprint,161      supportingObservations: [],162    };163    return this.publish(ev) ? ev : null;164  }165166  /** System events (connector failure/recovery, schema drift, market open/close). */167  system(type: EventType, key: string, ts: number, severity: Severity, title: string, data: Record<string, unknown>, sources: string[] = [], instrumentIds: string[] = []) {168    const fingerprint = shortHash(`${type}|${key}`, 24);169    this.publish({170      version: 1,171      id: makeEventId(ts, fingerprint),172      type,173      instrumentIds,174      entityIds: [],175      timestamp: ts,176      severity,177      confidence: 1,178      sourceCount: sources.length,179      sources,180      title,181      summary: null,182      data,183      fingerprint,184      supportingObservations: [],185    });186  }187188  private publish(ev: MarketEvent): boolean {189    const now = Date.now();190    const prev = this.recentFingerprints.get(ev.fingerprint);191    if (prev && now - prev < 86_400_000) {192      // Another source confirms an existing event → bump confirmation in DB, no new feed item.193      this.confirm(ev).catch(() => {});194      return false;195    }196    this.recentFingerprints.set(ev.fingerprint, now);197    if (this.recentFingerprints.size > 200_000) for (const [k, t] of this.recentFingerprints) if (now - t > 86_400_000) this.recentFingerprints.delete(k);198    this.pending.push(ev);199    bus.publish("market.event", ev);200    telemetry.inc("events_total", 1, { type: ev.type });201    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 500);202    return true;203  }204205  private async confirm(ev: MarketEvent) {206    await pool.query(207      `update market_events set source_count = source_count + 1, sources = array(select distinct unnest(sources || $2::text[])), confirmed_at = coalesce(confirmed_at, now()) where fingerprint = $1`,208      [ev.fingerprint, ev.sources],209    );210  }211212  async flush(): Promise<void> {213    this.timer = null;214    const batch = this.pending.splice(0);215    if (!batch.length) return;216    const cols = 14;217    const tuples: string[] = [];218    const values: unknown[] = [];219    batch.forEach((e, i) => {220      const b = i * cols;221      tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);222      values.push(e.id, e.type, new Date(e.timestamp).toISOString(), e.instrumentIds, e.entityIds, e.severity, e.confidence, e.sourceCount, e.sources, e.title, e.summary, JSON.stringify(e.data), e.fingerprint, e.supportingObservations);223    });224    try {225      await pool.query(226        `insert into market_events (id, type, ts, instrument_ids, entity_ids, severity, confidence, source_count, sources, title, summary, data, fingerprint, supporting_observations)227         values ${tuples.join(",")} on conflict (fingerprint) do update set source_count = market_events.source_count + 1, confirmed_at = coalesce(market_events.confirmed_at, now())`,228        values,229      );230      await pool.query(`insert into daily_stats (day, events) values (current_date, $1) on conflict (day) do update set events = daily_stats.events + excluded.events, updated_at = now()`, [batch.length]);231    } catch (err) {232      logger.error({ err, size: batch.length }, "event insert failed");233    }234  }235236  /** Warm the dedupe cache from recent DB events so restarts do not re-emit filings/halts. */237  async warm(): Promise<void> {238    const r = await pool.query<{ fingerprint: string; ts: string }>("select fingerprint, ts from market_events where ts > now() - interval '30 hours'");239    for (const row of r.rows) this.recentFingerprints.set(row.fingerprint, Date.parse(row.ts));240  }241}242243type NodeJS_Timeout = ReturnType<typeof setTimeout>;244const round = (n: number, d: number) => Math.round(n * 10 ** d) / 10 ** d;245const fmt = (n: number) => (Math.abs(n) >= 1000 ? n.toLocaleString("en-US", { maximumFractionDigits: 2 }) : n.toPrecision(6).replace(/\.?0+$/, ""));246247export const eventEngine = new EventEngine();248