spb/market-atlas

1import type { Company, Exchange, Instrument, InstrumentHint, SymbolAlias } from "@market-atlas/market-model";2import { canonicalSymbol, makeCompanyId, makeInstrumentId } from "@market-atlas/market-model";3import { pool, query, type Queryable } from "../db/pool.js";4import { logger } from "../logger.js";5import { telemetry } from "./telemetry.js";6import { masterMutex } from "./mutex.js";78interface Row {9  id: string;10  symbol: string;11  name: string;12  asset_class: Instrument["assetClass"];13  exchange_id: string | null;14  mic: string | null;15  currency: string | null;16  country: string | null;17  company_id: string | null;18  isin: string | null;19  security_type: string | null;20  base: string | null;21  quote: string | null;22  is_active: boolean;23  metadata: Record<string, unknown>;24}2526export const rowToInstrument = (r: Row): Instrument => ({27  id: r.id,28  symbol: r.symbol,29  name: r.name,30  assetClass: r.asset_class,31  exchangeId: r.exchange_id,32  mic: r.mic,33  currency: r.currency,34  country: r.country,35  companyId: r.company_id,36  isin: r.isin,37  securityType: r.security_type,38  base: r.base,39  quote: r.quote,40  isActive: r.is_active,41  metadata: r.metadata ?? {},42});4344/**45 * Instrument master + resolver. Keeps an in-memory index (id, alias per source, canonical symbol)46 * so that the hot path never queries Postgres per tick.47 */48export class InstrumentStore {49  private byId = new Map<string, Instrument>();50  private byAlias = new Map<string, string>(); // `${sourceId}|${canonicalAlias}` or `*|alias` -> id51  private bySymbol = new Map<string, string[]>(); // canonical symbol -> ids52  private exchangesById = new Map<string, Exchange>();53  private creating = new Map<string, Promise<Instrument>>();5455  async load(): Promise<void> {56    const [inst, aliases, ex] = await Promise.all([57      query<Row>("select * from instruments"),58      query<{ alias: string; source_id: string; instrument_id: string }>("select alias, source_id, instrument_id from symbol_aliases"),59      query<any>("select * from exchanges"),60    ]);61    this.byId.clear();62    this.byAlias.clear();63    this.bySymbol.clear();64    for (const r of inst.rows) this.index(rowToInstrument(r));65    for (const a of aliases.rows) this.byAlias.set(`${a.source_id}|${canonicalSymbol(a.alias)}`, a.instrument_id);66    this.exchangesById.clear();67    for (const e of ex.rows) this.exchangesById.set(e.id, rowToExchange(e));68    telemetry.gauge("instruments_total", this.byId.size);69    logger.info({ instruments: this.byId.size, aliases: this.byAlias.size, exchanges: this.exchangesById.size }, "instrument master loaded");70  }7172  private index(i: Instrument) {73    this.byId.set(i.id, i);74    const cs = canonicalSymbol(i.symbol);75    const arr = this.bySymbol.get(cs) ?? [];76    if (!arr.includes(i.id)) arr.push(i.id);77    this.bySymbol.set(cs, arr);78    this.byAlias.set(`*|${cs}`, i.id);79  }8081  get(id: string): Instrument | undefined {82    return this.byId.get(id);83  }8485  all(): Instrument[] {86    return [...this.byId.values()];87  }8889  count(): number {90    return this.byId.size;91  }9293  exchange(id: string | null | undefined): Exchange | undefined {94    return id ? this.exchangesById.get(id) : undefined;95  }9697  exchanges(): Exchange[] {98    return [...this.exchangesById.values()];99  }100101  /** Resolve a source-native symbol. Source-specific aliases win over global ones. */102  resolve(symbol: string, sourceId: string): Instrument | undefined {103    const cs = canonicalSymbol(symbol);104    const id = this.byAlias.get(`${sourceId}|${cs}`) ?? this.byAlias.get(`*|${cs}`);105    return id ? this.byId.get(id) : undefined;106  }107108  /** Ids sharing a canonical symbol (e.g. AAPL on several venues). */109  bySymbolCandidates(symbol: string): Instrument[] {110    return (this.bySymbol.get(canonicalSymbol(symbol)) ?? []).map((id) => this.byId.get(id)!).filter(Boolean);111  }112113  /** Resolve, or create from a hint (deterministic id). Concurrent creations of the same id are coalesced. */114  async resolveOrCreate(symbol: string, sourceId: string, hint: InstrumentHint | undefined): Promise<Instrument | undefined> {115    const found = this.resolve(symbol, sourceId);116    if (found) return found;117    if (!hint) return undefined;118    const id = makeInstrumentId(hint.assetClass, { symbol, ...hint });119    const existing = this.byId.get(id);120    if (existing) {121      await this.addAlias(symbol, sourceId, id);122      return existing;123    }124    let p = this.creating.get(id);125    if (!p) {126      p = this.create(id, symbol, sourceId, hint).finally(() => this.creating.delete(id));127      this.creating.set(id, p);128    }129    return p;130  }131132  private async create(id: string, symbol: string, sourceId: string, hint: InstrumentHint): Promise<Instrument> {133    let companyId: string | null = null;134    if (hint.companyName) companyId = await this.upsertCompany({ name: hint.companyName, cik: hint.cik ?? null, country: hint.country ?? null });135    const inst: Instrument = {136      id,137      symbol: symbol.replace(/^[_^]/, ""),138      name: hint.name ?? symbol,139      assetClass: hint.assetClass,140      exchangeId: hint.exchangeId ?? null,141      mic: hint.mic ?? null,142      currency: hint.currency ?? null,143      country: hint.country ?? null,144      companyId,145      isin: null,146      securityType: hint.securityType ?? null,147      base: hint.base ?? null,148      quote: hint.quote ?? null,149      isActive: true,150      metadata: hint.metadata ?? {},151    };152    await this.upsert(inst);153    await this.addAlias(symbol, sourceId, id);154    telemetry.inc("instruments_created_total");155    return inst;156  }157158  async upsert(i: Instrument, client: Queryable = pool): Promise<void> {159    await masterMutex.run(() => query(160      `insert into instruments (id, symbol, name, asset_class, exchange_id, mic, currency, country, company_id, isin, security_type, base, quote, is_active, metadata)161       values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)162       on conflict (id) do update set163         name = case when excluded.name <> excluded.symbol then excluded.name else instruments.name end,164         exchange_id = coalesce(excluded.exchange_id, instruments.exchange_id),165         mic = coalesce(excluded.mic, instruments.mic),166         currency = coalesce(excluded.currency, instruments.currency),167         country = coalesce(excluded.country, instruments.country),168         company_id = coalesce(excluded.company_id, instruments.company_id),169         isin = coalesce(excluded.isin, instruments.isin),170         security_type = coalesce(excluded.security_type, instruments.security_type),171         is_active = excluded.is_active,172         metadata = instruments.metadata || excluded.metadata,173         updated_at = now()`,174      [i.id, i.symbol, i.name, i.assetClass, i.exchangeId, i.mic, i.currency, i.country, i.companyId, i.isin, i.securityType, i.base, i.quote, i.isActive, JSON.stringify(i.metadata ?? {})],175      client,176    ));177    const merged = { ...(this.byId.get(i.id) ?? i), ...stripNulls(i) } as Instrument;178    this.index(merged);179    telemetry.gauge("instruments_total", this.byId.size);180  }181182  async addAlias(alias: string, sourceId: string, instrumentId: string): Promise<void> {183    const key = `${sourceId}|${canonicalSymbol(alias)}`;184    if (this.byAlias.get(key) === instrumentId) return;185    this.byAlias.set(key, instrumentId);186    await masterMutex.run(() =>187      query(188        `insert into symbol_aliases (alias, source_id, instrument_id) values ($1,$2,$3)189         on conflict (alias, source_id) do update set instrument_id = excluded.instrument_id`,190        [alias, sourceId, instrumentId],191      ),192    );193  }194195  aliasesOf(instrumentId: string): SymbolAlias[] {196    const out: SymbolAlias[] = [];197    for (const [k, v] of this.byAlias) {198      if (v !== instrumentId) continue;199      const [sourceId, alias] = k.split("|");200      out.push({ alias: alias!, sourceId: sourceId === "*" ? null : sourceId!, instrumentId });201    }202    return out;203  }204205  async upsertCompany(c: { name: string; cik: string | null; country: string | null; sector?: string | null; website?: string | null }): Promise<string> {206    const id = makeCompanyId(c.name);207    await masterMutex.run(() => query(208      `insert into companies (id, name, cik, country, sector, website) values ($1,$2,$3,$4,$5,$6)209       on conflict (id) do update set cik = coalesce(excluded.cik, companies.cik), country = coalesce(excluded.country, companies.country),210         sector = coalesce(excluded.sector, companies.sector), website = coalesce(excluded.website, companies.website), updated_at = now()`,211      [id, c.name, c.cik, c.country, c.sector ?? null, c.website ?? null],212    ));213    return id;214  }215216  async getCompany(id: string): Promise<Company | null> {217    const r = await query<any>("select * from companies where id = $1", [id]);218    const c = r.rows[0];219    return c ? { id: c.id, name: c.name, cik: c.cik, country: c.country, sector: c.sector, industry: c.industry, website: c.website } : null;220  }221}222223function stripNulls<T extends object>(o: T): Partial<T> {224  return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== null && v !== undefined)) as Partial<T>;225}226227export function rowToExchange(e: any): Exchange {228  return {229    id: e.id,230    mic: e.mic,231    name: e.name,232    operator: e.operator,233    country: e.country,234    city: e.city,235    timezone: e.timezone,236    currency: e.currency,237    website: e.website,238    sessions: e.sessions,239    lat: e.lat,240    lon: e.lon,241    assetClasses: e.asset_classes ?? [],242  };243}244245export const instruments = new InstrumentStore();246