spb/market-atlas

1import type { AssetClass, CanonicalQuote, ComparabilityClass, Observation, ObservationType, RealtimeStatus, RightsStatus, SourceContribution } from "@market-atlas/market-model";2import { CANONICAL_VERSIONS, COMPARABILITY, PROXY_TYPES, PUBLICLY_REDISTRIBUTABLE } from "@market-atlas/market-model";3import { config } from "../config.js";45export interface SourceProfile {6  family: string | null;7  reliability: number; // 0..1 operational reliability (from health engine)8  realtimeStatus: RealtimeStatus;9  isOfficial: boolean;10}1112interface Latest {13  obs: Observation;14}1516/** Fresh window per realtime class: realtime ticks age out fast, EOD values live for days. */17const FRESH_WINDOW_MS: Record<RealtimeStatus, number> = {18  REALTIME: config.consensusFreshWindowMs,19  DELAYED: 30 * 60_000,20  INDICATIVE: 60 * 60_000,21  END_OF_DAY: 10 * 86_400_000, // daily/weekly-refreshed official series stay "last close" over long weekends and weekly lake reloads22  STALE: 0,23  UNKNOWN: 60_000,24};2526const REALTIME_RANK: Record<RealtimeStatus, number> = { REALTIME: 4, DELAYED: 3, INDICATIVE: 2, END_OF_DAY: 1, UNKNOWN: 0, STALE: -1 };27/** Same-class observations must be close in time to be compared: live within 2 minutes, fixings/closes on the same UTC day. */28const LIVE_TOLERANCE_MS = 120_000;2930/** Infer the observation type when a connector did not declare one. */31export function inferObservationType(o: Pick<Observation, "realtimeStatus" | "observationType" | "field">): ObservationType {32  if (o.observationType) return o.observationType;33  if (o.field === "PREVIOUS_CLOSE") return "EOD_CLOSE";34  switch (o.realtimeStatus) {35    case "END_OF_DAY":36      return "EOD_CLOSE";37    case "INDICATIVE":38      return "INDICATIVE";39    default:40      return "TRADE";41  }42}4344interface Candidate {45  value: number;46  weight: number;47  family: string;48  obs: Observation;49  type: ObservationType;50  proxy: boolean;51  validator: boolean;52  contrib: SourceContribution;53}5455/**56 * Per-instrument consensus. Keeps the latest observation of each (source, field), keeps only57 * *comparable* observations (same class — live / official fix / end-of-day — and close in time),58 * computes a weighted median (one vote per upstream family, liquidity-aware for crypto), rejects59 * outliers, uses restricted-rights sources and proxies as confirmations only, and exposes the full60 * contribution list ("Why this price?").61 */62export class ConsensusEngine {63  private latest = new Map<string, Map<string, Latest>>(); // instrumentId -> `${sourceId}|${field}` -> latest64  private session = new Map<string, { high: number; low: number; startedAt: number }>();6566  /**67   * @param profiles     source reliability/family lookup68   * @param marketOpen   true/false when the instrument's venue has a session calendar, null for continuous or unknown venues69   * @param assetClassOf asset class of an instrument (liquidity weighting for crypto, official dominance for rates)70   */71  constructor(72    private profiles: (sourceId: string) => SourceProfile,73    private marketOpen: (instrumentId: string) => boolean | null = () => null,74    private assetClassOf: (instrumentId: string) => AssetClass | null = () => null,75  ) {}7677  ingest(o: Observation): void {78    let m = this.latest.get(o.instrumentId);79    if (!m) {80      m = new Map();81      this.latest.set(o.instrumentId, m);82    }83    const key = `${o.sourceId}|${o.field}`;84    const prev = m.get(key);85    // Ordering: prefer source timestamp, fall back to receive time. Ignore out-of-order values.86    if (prev) {87      const pt = prev.obs.sourceTimestamp ?? prev.obs.receivedAt;88      const nt = o.sourceTimestamp ?? o.receivedAt;89      if (nt < pt) return;90    }91    m.set(key, { obs: o });92  }9394  resetSession(instrumentId: string, at: number) {95    this.session.set(instrumentId, { high: -Infinity, low: Infinity, startedAt: at });96  }9798  sessionOf(instrumentId: string) {99    return this.session.get(instrumentId);100  }101102  instrumentsTracked(): number {103    return this.latest.size;104  }105106  sourcesFor(instrumentId: string): string[] {107    const m = this.latest.get(instrumentId);108    if (!m) return [];109    return [...new Set([...m.keys()].map((k) => k.split("|")[0]!))];110  }111112  /** Latest LAST_PRICE observation per source (for lineage inference and coverage). */113  latestPrices(instrumentId: string): Observation[] {114    const m = this.latest.get(instrumentId);115    if (!m) return [];116    return [...m.entries()].filter(([k]) => k.endsWith("|LAST_PRICE")).map(([, l]) => l.obs);117  }118119  compute(instrumentId: string, symbol: string, now = Date.now()): CanonicalQuote | null {120    const m = this.latest.get(instrumentId);121    if (!m) return null;122    const closed = this.marketOpen(instrumentId) === false;123    const assetClass = this.assetClassOf(instrumentId);124    const price = this.aggregate(m, "LAST_PRICE", now, closed, assetClass);125    const pick = (field: string) => this.aggregate(m, field, now, closed, assetClass);126    const open = pick("OPEN");127    const high = pick("HIGH");128    const low = pick("LOW");129    const prevClose = pick("PREVIOUS_CLOSE");130    const volume = pick("VOLUME");131    const bid = pick("BID");132    const ask = pick("ASK");133    const changeObs = pick("CHANGE");134    const changePctObs = pick("CHANGE_PERCENT");135    const p = price.value;136    let change = changeObs.value;137    let changePercent = changePctObs.value;138    if (p != null && prevClose.value != null && prevClose.value !== 0) {139      change = p - prevClose.value;140      changePercent = (change / prevClose.value) * 100;141    } else if (p != null && change == null && open.value != null && open.value !== 0) {142      // Continuous markets (crypto) publish a rolling 24 h open instead of a previous close.143      change = p - open.value;144      changePercent = (change / open.value) * 100;145    }146    let sess = this.session.get(instrumentId);147    if (p != null) {148      if (!sess) {149        sess = { high: p, low: p, startedAt: now };150        this.session.set(instrumentId, sess);151      } else {152        if (p > sess.high) sess.high = p;153        if (p < sess.low) sess.low = p;154      }155    }156    const all = [...m.values()].map((l) => l.obs);157    const currency = all.find((o) => o.currency)?.currency ?? null;158    const anyNonPublic = price.contributions.some((c) => c.included && !PUBLICLY_REDISTRIBUTABLE.has(c.rightsStatus));159    const rightsBasis = price.contributions.find((c) => c.included) ?? [...price.contributions].sort((a, b) => a.ageMs - b.ageMs)[0];160    const rights: RightsStatus = anyNonPublic ? "INTERNAL_ONLY" : (rightsBasis?.rightsStatus ?? "UNKNOWN");161    return {162      instrumentId,163      symbol,164      price: p,165      open: open.value,166      high: high.value ?? (sess && Number.isFinite(sess.high) ? sess.high : null),167      low: low.value ?? (sess && Number.isFinite(sess.low) ? sess.low : null),168      previousClose: prevClose.value,169      change: change ?? null,170      changePercent: changePercent ?? null,171      volume: volume.value,172      bid: bid.value,173      ask: ask.value,174      currency,175      sourceCount: price.independentFamilies,176      observationCount: price.observationCount,177      proxyCount: price.proxyCount,178      validatorCount: price.validatorCount,179      comparability: price.comparability,180      dispersionBps: price.dispersionBps,181      confidence: price.confidence,182      freshnessMs: price.freshnessMs,183      realtimeStatus: price.realtimeStatus,184      rightsStatus: rights,185      updatedAt: now,186      sourceTimestamp: price.sourceTimestamp,187      contributions: price.contributions,188      consensusVersion: CANONICAL_VERSIONS.consensus,189      sessionHigh: sess && Number.isFinite(sess.high) ? sess.high : null,190      sessionLow: sess && Number.isFinite(sess.low) ? sess.low : null,191    };192  }193194  private aggregate(m: Map<string, Latest>, field: string, now: number, marketClosed: boolean, assetClass: AssetClass | null) {195    const contributions: SourceContribution[] = [];196    const fresh: Candidate[] = [];197    const empty = {198      value: null as number | null,199      confidence: 0,200      dispersionBps: null as number | null,201      independentFamilies: 0,202      observationCount: 0,203      proxyCount: 0,204      validatorCount: 0,205      comparability: null as ComparabilityClass | null,206      freshnessMs: null as number | null,207      realtimeStatus: "STALE" as RealtimeStatus,208      sourceTimestamp: null as number | null,209      contributions,210    };211212    for (const [key, l] of m) {213      if (!key.endsWith(`|${field}`)) continue;214      const o = l.obs;215      const prof = this.profiles(o.sourceId);216      const type = inferObservationType(o);217      const eventTs = o.sourceTimestamp ?? o.receivedAt;218      const ageMs = Math.max(0, now - eventTs);219      // While the venue is closed, the last session value is the truth for up to 4 days (long weekends).220      const window = marketClosed && (o.realtimeStatus === "REALTIME" || o.realtimeStatus === "DELAYED") ? 4 * 86_400_000 : (FRESH_WINDOW_MS[o.realtimeStatus] ?? 60_000);221      const isFresh = ageMs <= window;222      const base = 0.35 + 0.65 * prof.reliability;223      const tsQuality = o.timestampTrust === "EXCHANGE" ? 1 : o.timestampTrust === "SOURCE" ? 0.9 : 0.75;224      const rtQuality = o.realtimeStatus === "REALTIME" ? 1 : o.realtimeStatus === "DELAYED" ? 0.6 : o.realtimeStatus === "INDICATIVE" ? 0.5 : 0.4;225      const official = prof.isOfficial ? (assetClass === "TREASURY" || assetClass === "INTEREST_RATE" || assetClass === "BOND" ? 1.6 : 1.15) : 1;226      const decay = isFresh ? Math.exp(-ageMs / Math.max(1000, window)) : 0;227      const weight = base * tsQuality * rtQuality * official * (0.4 + 0.6 * decay) * (o.confidence ?? 1);228      const contrib: SourceContribution = {229        sourceId: o.sourceId,230        connectorId: o.connectorId,231        value: o.value,232        sourceTimestamp: o.sourceTimestamp,233        receivedAt: o.receivedAt,234        ageMs,235        weight: 0,236        included: false,237        observationType: type,238        deltaBps: null,239        realtimeStatus: isFresh ? o.realtimeStatus : "STALE",240        rightsStatus: o.rightsStatus,241        ...(isFresh ? {} : { reason: "stale" }),242      };243      contributions.push(contrib);244      if (isFresh) fresh.push({ value: o.value, weight, family: prof.family ?? o.sourceId, obs: o, type, proxy: PROXY_TYPES.has(type), validator: !PUBLICLY_REDISTRIBUTABLE.has(o.rightsStatus), contrib });245    }246    if (!fresh.length) {247      // Fall back to the most recent stale value so the UI can show "last known", clearly labelled STALE.248      const newest = [...contributions].sort((a, b) => a.ageMs - b.ageMs)[0];249      return { ...empty, value: newest?.value ?? null, freshnessMs: newest?.ageMs ?? null, sourceTimestamp: newest?.sourceTimestamp ?? null, contributions: contributions.sort((a, b) => a.ageMs - b.ageMs) };250    }251252    // 1. Comparability class: live values from real markets win, then official fixings, then end-of-day closes,253    //    then live proxies alone (stablecoin markets over a weekend). Other classes are reported, never mixed.254    const cls = (c: Candidate) => COMPARABILITY[c.type];255    const hasRealLive = fresh.some((c) => cls(c) === "LIVE" && !c.proxy);256    const chosen: ComparabilityClass = hasRealLive ? "LIVE" : fresh.some((c) => cls(c) === "FIX") ? "FIX" : fresh.some((c) => cls(c) === "EOD") ? "EOD" : "LIVE";257    let pool = fresh.filter((c) => cls(c) === chosen);258    for (const c of fresh) if (cls(c) !== chosen) exclude(c, "not_comparable");259260    // 2. Temporal tolerance inside the class.261    const tsOf = (c: Candidate) => c.obs.sourceTimestamp ?? c.obs.receivedAt;262    const freshestTs = Math.max(...pool.map(tsOf));263    pool = pool.filter((c) => {264      const ok = chosen === "LIVE" ? freshestTs - tsOf(c) <= LIVE_TOLERANCE_MS : sameUtcDay(tsOf(c), freshestTs);265      if (!ok) exclude(c, "temporal_mismatch");266      return ok;267    });268269    // 3. Roles: restricted-rights sources validate only; proxies confirm with reduced weight when real sources exist.270    const validators = pool.filter((c) => c.validator);271    for (const c of validators) exclude(c, "validation_only");272    let voters = pool.filter((c) => !c.validator);273    const realVoters = voters.filter((c) => !c.proxy);274    const proxies = voters.filter((c) => c.proxy);275    if (realVoters.length) for (const c of proxies) c.weight *= 0.35;276277    // 4. One vote per upstream family (the freshest keeps its weight, others are discounted).278    const seenFamily = new Set<string>();279    for (const c of [...voters].sort((a, b) => tsOf(b) - tsOf(a))) {280      if (seenFamily.has(c.family)) c.weight *= 0.25;281      else seenFamily.add(c.family);282    }283    // 5. Crypto: liquidity-aware — a venue's weight scales with its share of the 24 h volume among voters.284    if (assetClass === "CRYPTO" && voters.length > 1) {285      const vols = new Map<string, number>();286      for (const c of voters) {287        const v = m.get(`${c.obs.sourceId}|VOLUME`)?.obs.value;288        if (v != null && v > 0) vols.set(c.obs.sourceId, v * c.value); // notional289      }290      const max = Math.max(0, ...vols.values());291      if (max > 0) for (const c of voters) c.weight *= 0.6 + 0.8 * ((vols.get(c.obs.sourceId) ?? 0) / max);292    }293294    // 6. Outlier rejection (> 2 % from the weighted median of real voters, when at least 3 voters).295    const reference = realVoters.length ? realVoters : voters;296    let median = weightedMedian(reference.map((c) => ({ v: c.value, w: c.weight })));297    if (voters.length >= 3 && median !== 0) {298      const kept = voters.filter((c) => Math.abs(c.value - median) / Math.abs(median) <= 0.02);299      for (const c of voters) if (!kept.includes(c)) exclude(c, "outlier");300      if (kept.length) voters = kept;301      median = weightedMedian((kept.filter((c) => !c.proxy).length ? kept.filter((c) => !c.proxy) : kept).map((c) => ({ v: c.value, w: c.weight })));302    }303    const finalVoters = voters;304    for (const c of finalVoters) {305      c.contrib.included = true;306      c.contrib.weight = round(c.weight, 4);307    }308    // Deltas vs canonical for every fresh contribution (included or not).309    for (const c of fresh) c.contrib.deltaBps = median ? round(((c.value - median) / Math.abs(median)) * 10_000, 2) : null;310311    const realIncluded = finalVoters.filter((c) => !c.proxy);312    const spreadBasis = realIncluded.length ? realIncluded : finalVoters;313    const values = spreadBasis.map((c) => c.value);314    const spread = values.length > 1 ? (Math.max(...values) - Math.min(...values)) / Math.abs(median || 1) : 0;315    const families = new Set(realIncluded.map((c) => c.family)).size;316    const proxyCount = finalVoters.filter((c) => c.proxy && Math.abs(c.contrib.deltaBps ?? 1e9) <= 50).length;317    const validatorCount = validators.filter((c) => Math.abs(c.contrib.deltaBps ?? 1e9) <= 25).length;318    for (const c of validators) if (Math.abs(c.contrib.deltaBps ?? 0) > 50) c.contrib.reason = "validator_disagrees";319    const newest = Math.min(...finalVoters.map((c) => Math.max(0, now - tsOf(c))));320    let bestRt = finalVoters.map((c) => c.obs.realtimeStatus).sort((a, b) => REALTIME_RANK[b] - REALTIME_RANK[a])[0] ?? "UNKNOWN";321    if (chosen === "LIVE" && !realIncluded.length) bestRt = "INDICATIVE"; // proxies only322    // Confidence: agreement (dispersion), redundancy (independent families + confirmations), freshness, reliability. Never claims > 0.995.323    const agreement = Math.max(0, 1 - Math.min(1, spread / 0.01)); // 1% spread → 0324    const redundancy = 1 - Math.exp(-(families + 0.5 * proxyCount + 0.5 * validatorCount) / 2);325    const freshness = marketClosed ? 1 : Math.exp(-newest / 60_000);326    const avgReliability = finalVoters.reduce((a, c) => a + this.profiles(c.obs.sourceId).reliability, 0) / finalVoters.length;327    const confidence = round(Math.min(0.995, 0.15 + 0.35 * agreement + 0.25 * redundancy + 0.1 * freshness + 0.15 * avgReliability), 3);328    return {329      value: median,330      confidence,331      dispersionBps: round(spread * 10_000, 2),332      independentFamilies: families,333      observationCount: fresh.length,334      proxyCount,335      validatorCount,336      comparability: chosen,337      freshnessMs: Math.round(newest),338      realtimeStatus: bestRt,339      sourceTimestamp: finalVoters.map((c) => c.obs.sourceTimestamp).filter((t): t is number => t != null).sort((a, b) => b - a)[0] ?? null,340      contributions: contributions.sort((a, b) => Number(b.included) - Number(a.included) || a.ageMs - b.ageMs),341    };342  }343}344345function exclude(c: Candidate, reason: string) {346  c.contrib.included = false;347  c.contrib.weight = 0;348  if (!c.contrib.reason) c.contrib.reason = reason;349}350351function sameUtcDay(a: number, b: number): boolean {352  return Math.floor(a / 86_400_000) === Math.floor(b / 86_400_000);353}354355export function weightedMedian(items: Array<{ v: number; w: number }>): number {356  if (!items.length) return NaN;357  const sorted = [...items].sort((a, b) => a.v - b.v);358  const total = sorted.reduce((a, b) => a + Math.max(b.w, 1e-9), 0);359  let acc = 0;360  for (const it of sorted) {361    acc += Math.max(it.w, 1e-9);362    if (acc >= total / 2) return it.v;363  }364  return sorted[sorted.length - 1]!.v;365}366367export const round = (n: number, d: number) => Math.round(n * 10 ** d) / 10 ** d;368