spb/market-atlas
Public
TypeScript 96.7%
SQL 1.6%
CSS 0.8%
JavaScript 0.5%
1import type { Bar, Instrument, NormalizedObservation, Observation, RawObservation } from "@market-atlas/market-model";2import { CANONICAL_VERSIONS, NormalizedObservationSchema, PUBLICLY_REDISTRIBUTABLE } from "@market-atlas/market-model";3import { observationFingerprint, type ConnectorDefinition, type NormalizedBatch } from "@market-atlas/connector-sdk";4import { config } from "../config.js";5import { pool } from "../db/pool.js";6import { logger } from "../logger.js";7import { barAggregator, upsertBars } from "./bars.js";8import { bus } from "./bus.js";9import { calendar } from "./calendar.js";10import { ConsensusEngine } from "./consensus.js";11import { eventEngine } from "./events.js";12import { health } from "./health.js";13import { instruments } from "./instruments.js";14import { observationWriter } from "./observations.js";15import { quoteStore } from "./quotes.js";16import { telemetry } from "./telemetry.js";17import { lineage } from "./lineage.js";1819export interface SourceInfo {20 family: string | null;21 isOfficial: boolean;22 realtimeStatus: Observation["realtimeStatus"];23}2425/**26 * L1→L4 pipeline for one normalized batch: validate, resolve instruments, persist observations,27 * update consensus + canonical quotes, feed the event engine and the bar aggregator, and handle28 * side outputs (proposed events, instrument master updates, filings, holidays).29 */30export class Pipeline {31 readonly consensus: ConsensusEngine;32 private sourceInfo = new Map<string, SourceInfo>();33 private recomputeQueue = new Map<string, Instrument>();34 private recomputeTimer: NodeJS.Timeout | null = null;3536 constructor() {37 this.consensus = new ConsensusEngine(38 (sourceId) => {39 const info = this.sourceInfo.get(sourceId);40 return {41 family: lineage.familyOf(sourceId) ?? info?.family ?? null,42 reliability: health.sourceReliability(sourceId),43 realtimeStatus: info?.realtimeStatus ?? "UNKNOWN",44 isOfficial: info?.isOfficial ?? false,45 };46 },47 (instrumentId) => {48 const inst = instruments.get(instrumentId);49 const ex = inst?.exchangeId ? calendar.exchange(inst.exchangeId) : undefined;50 if (!ex || ex.sessions.continuous) return null;51 return calendar.state(ex.id) === "OPEN";52 },53 (instrumentId) => instruments.get(instrumentId)?.assetClass ?? null,54 );55 bus.subscribe("canonical.quote", (q) => barAggregator.onQuote(q));56 }5758 registerSource(sourceId: string, info: SourceInfo) {59 this.sourceInfo.set(sourceId, info);60 }6162 async process(def: ConnectorDefinition, raw: RawObservation, batch: NormalizedBatch, rawRef: string | null): Promise<number> {63 const connectorId = def.metadata.id;64 const sourceId = def.metadata.sourceId;65 let accepted = 0;66 const touched = new Map<string, Instrument>();67 const ids: string[] = [];6869 if (batch.instruments?.length) await this.applyInstruments(batch, sourceId);7071 for (const n of batch.observations) {72 const parsed = NormalizedObservationSchema.safeParse(n);73 if (!parsed.success) {74 telemetry.inc("observations_invalid_total", 1, { connector: connectorId });75 continue;76 }77 const inst = n.instrumentId ? instruments.get(n.instrumentId) : await instruments.resolveOrCreate(n.symbol, sourceId, n.instrumentHint);78 if (!inst) {79 telemetry.inc("observations_unresolved_total", 1, { connector: connectorId });80 continue;81 }82 if (n.instrumentId && !instruments.resolve(n.symbol, sourceId)) await instruments.addAlias(n.symbol, sourceId, inst.id);83 if (!plausible(n, inst)) {84 telemetry.inc("observations_rejected_total", 1, { connector: connectorId, reason: "implausible" });85 continue;86 }87 const obs = this.toObservation(n, inst, raw, rawRef);88 if (this.shouldPersist(obs)) observationWriter.enqueue(obs);89 else telemetry.inc("observations_sampled_out_total");90 this.consensus.ingest(obs);91 lineage.observe(obs);92 bus.publish("normalized.observation", obs);93 touched.set(inst.id, inst);94 ids.push(inst.id);95 accepted++;96 }97 telemetry.inc("observations_accepted_total", accepted);98 const latency = batch.observations.find((o) => o.sourceTimestamp)?.sourceTimestamp;99 health.message(connectorId, latency ? raw.receivedAt - latency : null, ids);100 for (const [id, inst] of touched) this.recomputeQueue.set(id, inst);101 this.scheduleRecompute();102103 if (batch.events?.length) await this.applyEvents(batch, sourceId);104 if (batch.filings?.length) await this.applyFilings(batch, sourceId);105 if (batch.holidays?.length) await this.applyHolidays(batch, sourceId);106 if (batch.bars?.length) await this.applyBars(batch, def);107 return accepted;108 }109110 private async applyBars(batch: NormalizedBatch, def: ConnectorDefinition) {111 const out: Bar[] = [];112 for (const b of batch.bars ?? []) {113 const inst = await instruments.resolveOrCreate(b.symbol, def.metadata.sourceId, b.instrumentHint);114 if (!inst) continue;115 if (!(b.high >= b.low && b.open > 0 && b.close > 0)) continue;116 out.push({ instrumentId: inst.id, resolution: b.resolution, ts: b.ts, open: b.open, high: b.high, low: b.low, close: b.close, volume: b.volume, sourceCount: 1, producer: def.metadata.id, version: CANONICAL_VERSIONS.normalizer });117 }118 await upsertBars(out);119 if (out.length) telemetry.inc("bars_backfilled_total", out.length, { connector: def.metadata.id });120 }121122 private lastPersisted = new Map<string, number>();123124 /** REALTIME streaming ticks are stored at most once per (source, instrument, field) per interval; everything else always. */125 private shouldPersist(o: Observation): boolean {126 if (o.realtimeStatus !== "REALTIME" || config.tickPersistIntervalMs <= 0) return true;127 const key = `${o.sourceId}|${o.instrumentId}|${o.field}`;128 const last = this.lastPersisted.get(key) ?? 0;129 if (o.receivedAt - last < config.tickPersistIntervalMs) return false;130 this.lastPersisted.set(key, o.receivedAt);131 if (this.lastPersisted.size > 200_000) this.lastPersisted.clear();132 return true;133 }134135 private toObservation(n: NormalizedObservation, inst: Instrument, raw: RawObservation, rawRef: string | null): Observation {136 const fp = observationFingerprint({137 sourceId: raw.sourceId,138 instrumentId: inst.id,139 field: n.field,140 value: n.value,141 sourceTimestamp: n.sourceTimestamp,142 sequence: n.sequence ?? null,143 });144 return {145 observationId: fp,146 instrumentId: inst.id,147 symbol: n.symbol,148 field: n.field,149 value: n.value,150 currency: n.currency ?? inst.currency ?? null,151 observationType: n.observationType,152 sourceTimestamp: n.sourceTimestamp,153 timestampTrust: n.timestampTrust,154 sequence: n.sequence ?? null,155 rightsStatus: n.rightsStatus,156 realtimeStatus: n.realtimeStatus,157 confidence: n.confidence,158 meta: n.meta,159 sourceId: raw.sourceId,160 connectorId: raw.connectorId,161 receivedAt: raw.receivedAt,162 latencyMs: n.sourceTimestamp == null ? null : Math.max(0, raw.receivedAt - n.sourceTimestamp),163 rawRef,164 normalizerVersion: CANONICAL_VERSIONS.normalizer,165 };166 }167168 /** Recompute canonical quotes at most ~10×/s per instrument (UI batching happens downstream too). */169 private scheduleRecompute() {170 if (this.recomputeTimer) return;171 this.recomputeTimer = setTimeout(() => {172 this.recomputeTimer = null;173 const batch = [...this.recomputeQueue.values()];174 this.recomputeQueue.clear();175 const now = Date.now();176 for (const inst of batch) {177 const q = this.consensus.compute(inst.id, inst.symbol, now);178 if (!q) continue;179 // Public redistribution guard: values built only from restricted sources stay internal.180 quoteStore.update(q);181 if (PUBLICLY_REDISTRIBUTABLE.has(q.rightsStatus)) eventEngine.onQuote(q, inst.name);182 }183 telemetry.gauge("consensus_instruments", this.consensus.instrumentsTracked());184 }, 100);185 }186187 /** Recompute every tracked instrument (used by the freshness sweeper so stale quotes decay). */188 sweep(): void {189 for (const q of quoteStore.all()) {190 const inst = instruments.get(q.instrumentId);191 if (inst) this.recomputeQueue.set(inst.id, inst);192 }193 this.scheduleRecompute();194 }195196 private async applyInstruments(batch: NormalizedBatch, sourceId: string) {197 let created = 0;198 for (const p of batch.instruments ?? []) {199 const before = instruments.count();200 const existing = instruments.resolve(p.symbol, sourceId);201 if (!existing && p.createIfMissing === false) continue;202 const inst = existing ?? (await instruments.resolveOrCreate(p.symbol, sourceId, p.hint));203 if (!inst) continue;204 if (instruments.count() > before) created++;205 for (const a of p.aliases ?? []) await instruments.addAlias(a, "*", inst.id);206 // Enrichment of an existing instrument: company/CIK, name, exchange when previously unknown.207 if (existing) {208 let companyId = existing.companyId;209 if (!companyId && p.hint.companyName) companyId = await instruments.upsertCompany({ name: p.hint.companyName, cik: p.hint.cik ?? null, country: p.hint.country ?? null });210 else if (companyId && p.hint.cik) await instruments.upsertCompany({ name: p.hint.companyName ?? existing.name, cik: p.hint.cik, country: p.hint.country ?? null });211 const patch: Partial<Instrument> = {};212 if (companyId !== existing.companyId) patch.companyId = companyId;213 if (!existing.exchangeId && p.hint.exchangeId) patch.exchangeId = p.hint.exchangeId;214 if (!existing.mic && p.hint.mic) patch.mic = p.hint.mic;215 if (existing.name === existing.symbol && p.hint.name) patch.name = p.hint.name;216 if (p.isActive === false && existing.isActive) patch.isActive = false;217 if (Object.keys(patch).length) await instruments.upsert({ ...existing, ...patch });218 }219 }220 if (created) logger.info({ source: sourceId, created, proposed: batch.instruments?.length }, "instrument master updated");221 }222223 private async applyEvents(batch: NormalizedBatch, sourceId: string) {224 for (const e of batch.events ?? []) {225 const ids = new Set<string>(e.instrumentIds ?? []);226 for (const s of e.symbols ?? []) {227 const inst = instruments.resolve(s, sourceId) ?? instruments.bySymbolCandidates(s).find((i) => i.assetClass === "EQUITY" || i.assetClass === "ETF");228 if (inst) ids.add(inst.id);229 }230 eventEngine.proposed(e, sourceId, [...ids]);231 }232 }233234 private async applyFilings(batch: NormalizedBatch, sourceId: string) {235 let inserted = 0;236 for (const f of batch.filings ?? []) {237 const ids: string[] = [];238 if (f.cik) {239 const r = await pool.query<{ id: string }>("select i.id from instruments i join companies c on c.id = i.company_id where c.cik = $1 and i.is_active limit 5", [f.cik.replace(/^0+/, "")]);240 ids.push(...r.rows.map((x) => x.id));241 }242 for (const s of f.symbols ?? []) {243 const inst = instruments.bySymbolCandidates(s)[0];244 if (inst) ids.push(inst.id);245 }246 const res = await pool.query(247 `insert into filings (id, source_id, cik, company_name, form_type, filed_at, url, instrument_ids, metadata) values ($1,$2,$3,$4,$5,$6,$7,$8,$9) on conflict (id) do nothing`,248 [f.id, sourceId, f.cik, f.companyName, f.formType, new Date(f.filedAt).toISOString(), f.url, ids, JSON.stringify(f.metadata ?? {})],249 );250 if (res.rowCount) {251 inserted++;252 if (f.metadata?.material === false) continue;253 eventEngine.proposed(254 {255 type: "FILING_PUBLISHED",256 timestamp: f.filedAt,257 title: `${f.companyName}: ${f.formType} filing`,258 summary: null,259 severity: "INFO",260 confidence: 1,261 dedupeKey: `filing:${f.id}`,262 data: { formType: f.formType, cik: f.cik, url: f.url, companyName: f.companyName, filingId: f.id },263 },264 sourceId,265 ids,266 );267 }268 }269 if (inserted) {270 await pool.query(`insert into daily_stats (day, filings) values (current_date, $1) on conflict (day) do update set filings = daily_stats.filings + excluded.filings, updated_at = now()`, [inserted]);271 telemetry.inc("filings_total", inserted);272 }273 }274275 private async applyHolidays(batch: NormalizedBatch, sourceId: string) {276 for (const h of batch.holidays ?? []) {277 await pool.query(278 `insert into exchange_holidays (exchange_id, date, name, kind, close_time, source_id) values ($1,$2,$3,$4,$5,$6)279 on conflict (exchange_id, date) do update set name = excluded.name, kind = excluded.kind, close_time = excluded.close_time, source_id = excluded.source_id`,280 [h.exchangeId, h.date, h.name, h.kind, h.closeTime ?? null, sourceId],281 );282 calendar.addHoliday({ ...h, sourceId });283 }284 }285}286287/** Reject values that cannot be right for the field/instrument (guards against schema drift producing garbage). */288function plausible(n: NormalizedObservation, inst: Instrument): boolean {289 if (!Number.isFinite(n.value)) return false;290 switch (n.field) {291 case "LAST_PRICE":292 case "OPEN":293 case "HIGH":294 case "LOW":295 case "PREVIOUS_CLOSE":296 case "CLOSE":297 case "BID":298 case "ASK":299 case "VWAP":300 case "NAV":301 if (n.value <= 0 && inst.assetClass !== "INTEREST_RATE" && inst.assetClass !== "TREASURY") return false;302 if (n.value > 1e9) return false;303 return true;304 case "VOLUME":305 case "BID_SIZE":306 case "ASK_SIZE":307 case "OPEN_INTEREST":308 case "MARKET_CAP":309 return n.value >= 0;310 case "CHANGE_PERCENT":311 return Math.abs(n.value) < 1000;312 case "YIELD":313 case "RATE":314 return n.value > -20 && n.value < 200;315 default:316 return true;317 }318}319320export const pipeline = new Pipeline();321