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";67interface Row {8  id: string;9  symbol: string;10  name: string;11  asset_class: Instrument["assetClass"];12  exchange_id: string | null;13  mic: string | null;14  currency: string | null;15  country: string | null;16  company_id: string | null;17  isin: string | null;18  security_type: string | null;19  base: string | null;20  quote: string | null;21  is_active: boolean;22  metadata: Record<string, unknown>;23}2425export const rowToInstrument = (r: Row): Instrument => ({26  id: r.id,27  symbol: r.symbol,28  name: r.name,29  assetClass: r.asset_class,30  exchangeId: r.exchange_id,31  mic: r.mic,32  currency: r.currency,33  country: r.country,34  companyId: r.company_id,35  isin: r.isin,36  securityType: r.security_type,37  base: r.base,38  quote: r.quote,39  isActive: r.is_active,40  metadata: r.metadata ?? {},41});4243/**44 * Instrument master + resolver. Keeps an in-memory index (id, alias per source, canonical symbol)45 * so that the hot path never queries Postgres per tick.46 */47export class InstrumentStore {48  private byId = new Map<string, Instrument>();49  private byAlias = new Map<string, string>(); // `${sourceId}|${canonicalAlias}` or `*|alias` -> id50  private bySymbol = new Map<string, string[]>(); // canonical symbol -> ids51  private exchangesById = new Map<string, Exchange>();52  private creating = new Map<string, Promise<Instrument>>();5354  async load(): Promise<void> {55    const [inst, aliases, ex] = await Promise.all([56      query<Row>("select * from instruments"),57      query<{ alias: string; source_id: string; instrument_id: string }>("select alias, source_id, instrument_id from symbol_aliases"),58      query<any>("select * from exchanges"),59    ]);60    this.byId.clear();61    this.byAlias.clear();62    this.bySymbol.clear();63    for (const r of inst.rows) this.index(rowToInstrument(r));64    for (const a of aliases.rows) this.byAlias.set(`${a.source_id}|${canonicalSymbol(a.alias)}`, a.instrument_id);65    this.exchangesById.clear();66    for (const e of ex.rows) this.exchangesById.set(e.id, rowToExchange(e));67    telemetry.gauge("instruments_total", this.byId.size);68    logger.info({ instruments: this.byId.size, aliases: this.byAlias.size, exchanges: this.exchangesById.size }, "instrument master loaded");69  }7071  private index(i: Instrument) {72    this.byId.set(i.id, i);73    const cs = canonicalSymbol(i.symbol);74    const arr = this.bySymbol.get(cs) ?? [];75    if (!arr.includes(i.id)) arr.push(i.id);76    this.bySymbol.set(cs, arr);77    this.byAlias.set(`*|${cs}`, i.id);78  }7980  get(id: string): Instrument | undefined {81    return this.byId.get(id);82  }8384  all(): Instrument[] {85    return [...this.byId.values()];86  }8788  count(): number {89    return this.byId.size;90  }9192  exchange(id: string | null | undefined): Exchange | undefined {93    return id ? this.exchangesById.get(id) : undefined;94  }9596  exchanges(): Exchange[] {97    return [...this.exchangesById.values()];98  }99100  /** Resolve a source-native symbol. Source-specific aliases win over global ones. */101  resolve(symbol: string, sourceId: string): Instrument | undefined {102    const cs = canonicalSymbol(symbol);103    const id = this.byAlias.get(`${sourceId}|${cs}`) ?? this.byAlias.get(`*|${cs}`);104    return id ? this.byId.get(id) : undefined;105  }106107  /** Ids sharing a canonical symbol (e.g. AAPL on several venues). */108  bySymbolCandidates(symbol: string): Instrument[] {109    return (this.bySymbol.get(canonicalSymbol(symbol)) ?? []).map((id) => this.byId.get(id)!).filter(Boolean);110  }111112  /** Resolve, or create from a hint (deterministic id). Concurrent creations of the same id are coalesced. */113  async resolveOrCreate(symbol: string, sourceId: string, hint: InstrumentHint | undefined): Promise<Instrument | undefined> {114    const found = this.resolve(symbol, sourceId);115    if (found) return found;116    if (!hint) return undefined;117    const id = makeInstrumentId(hint.assetClass, { symbol, ...hint });118    const existing = this.byId.get(id);119    if (existing) {120      await this.addAlias(symbol, sourceId, id);121      return existing;122    }123    let p = this.creating.get(id);124    if (!p) {125      p = this.create(id, symbol, sourceId, hint).finally(() => this.creating.delete(id));126      this.creating.set(id, p);127    }128    return p;129  }130131  private async create(id: string, symbol: string, sourceId: string, hint: InstrumentHint): Promise<Instrument> {132    let companyId: string | null = null;133    if (hint.companyName) companyId = await this.upsertCompany({ name: hint.companyName, cik: hint.cik ?? null, country: hint.country ?? null });134    const inst: Instrument = {135      id,136      symbol: symbol.replace(/^[_^]/, ""),137      name: hint.name ?? symbol,138      assetClass: hint.assetClass,139      exchangeId: hint.exchangeId ?? null,140      mic: hint.mic ?? null,141      currency: hint.currency ?? null,142      country: hint.country ?? null,143      companyId,144      isin: null,145      securityType: hint.securityType ?? null,146      base: hint.base ?? null,147      quote: hint.quote ?? null,148      isActive: true,149      metadata: hint.metadata ?? {},150    };151    await this.upsert(inst);152    await this.addAlias(symbol, sourceId, id);153    telemetry.inc("instruments_created_total");154    return inst;155  }156157  async upsert(i: Instrument, client: Queryable = pool): Promise<void> {158    await query(159      `insert into instruments (id, symbol, name, asset_class, exchange_id, mic, currency, country, company_id, isin, security_type, base, quote, is_active, metadata)160       values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)161       on conflict (id) do update set162         name = case when excluded.name <> excluded.symbol then excluded.name else instruments.name end,163         exchange_id = coalesce(excluded.exchange_id, instruments.exchange_id),164         mic = coalesce(excluded.mic, instruments.mic),165         currency = coalesce(excluded.currency, instruments.currency),166         country = coalesce(excluded.country, instruments.country),167         company_id = coalesce(excluded.company_id, instruments.company_id),168         isin = coalesce(excluded.isin, instruments.isin),169         security_type = coalesce(excluded.security_type, instruments.security_type),170         is_active = excluded.is_active,171         metadata = instruments.metadata || excluded.metadata,172         updated_at = now()`,173      [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 ?? {})],174      client,175    );176    const merged = { ...(this.byId.get(i.id) ?? i), ...stripNulls(i) } as Instrument;177    this.index(merged);178    telemetry.gauge("instruments_total", this.byId.size);179  }180181  async addAlias(alias: string, sourceId: string, instrumentId: string): Promise<void> {182    const key = `${sourceId}|${canonicalSymbol(alias)}`;183    if (this.byAlias.get(key) === instrumentId) return;184    this.byAlias.set(key, instrumentId);185    await query(186      `insert into symbol_aliases (alias, source_id, instrument_id) values ($1,$2,$3)187       on conflict (alias, source_id) do update set instrument_id = excluded.instrument_id`,188      [alias, sourceId, instrumentId],189    );190  }191192  aliasesOf(instrumentId: string): SymbolAlias[] {193    const out: SymbolAlias[] = [];194    for (const [k, v] of this.byAlias) {195      if (v !== instrumentId) continue;196      const [sourceId, alias] = k.split("|");197      out.push({ alias: alias!, sourceId: sourceId === "*" ? null : sourceId!, instrumentId });198    }199    return out;200  }201202  async upsertCompany(c: { name: string; cik: string | null; country: string | null; sector?: string | null; website?: string | null }): Promise<string> {203    const id = makeCompanyId(c.name);204    await query(205      `insert into companies (id, name, cik, country, sector, website) values ($1,$2,$3,$4,$5,$6)206       on conflict (id) do update set cik = coalesce(excluded.cik, companies.cik), country = coalesce(excluded.country, companies.country),207         sector = coalesce(excluded.sector, companies.sector), website = coalesce(excluded.website, companies.website), updated_at = now()`,208      [id, c.name, c.cik, c.country, c.sector ?? null, c.website ?? null],209    );210    return id;211  }212213  async getCompany(id: string): Promise<Company | null> {214    const r = await query<any>("select * from companies where id = $1", [id]);215    const c = r.rows[0];216    return c ? { id: c.id, name: c.name, cik: c.cik, country: c.country, sector: c.sector, industry: c.industry, website: c.website } : null;217  }218}219220function stripNulls<T extends object>(o: T): Partial<T> {221  return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== null && v !== undefined)) as Partial<T>;222}223224export function rowToExchange(e: any): Exchange {225  return {226    id: e.id,227    mic: e.mic,228    name: e.name,229    operator: e.operator,230    country: e.country,231    city: e.city,232    timezone: e.timezone,233    currency: e.currency,234    website: e.website,235    sessions: e.sessions,236    lat: e.lat,237    lon: e.lon,238    assetClasses: e.asset_classes ?? [],239  };240}241242export const instruments = new InstrumentStore();243