spb/market-atlas

1# Writing a Market Atlas connector23A connector is one technical integration of a **source** (`connectors/src/sources.ts`). It declares metadata (type, rights,4real-time class, family), optionally seeds instruments, and implements either a streaming lifecycle (`start/stop`) or a polling5one (`poll` + `schedule`), plus a **pure** `normalize(raw)`.67```ts8export const example = defineConnector({9  metadata: { id: "example-xhr", name: "…", version: "1.0.0", sourceId: "example", organization: "…", sourceType: "XHR",10    jurisdiction: "US", rightsStatus: "PUBLIC_ATTRIBUTED", realtimeStatus: "DELAYED", expectedLatencyMs: 900_000,11    supportsStreaming: false, supportsHistorical: false, assetClasses: ["EQUITY"], exchanges: ["xnys"], homepage: "https://…",12    description: "What it observes, how often, what is delayed.", rightsNotes: "Attribution…", termsUrl: "https://…",13    sourceFamily: "example", enabled: true },14  seeds: [{ symbol: "ABC", hint: { assetClass: "EQUITY", name: "ABC Corp", exchangeId: "xnys", mic: "XNYS", currency: "USD", country: "US" } }],15  defaultSymbols: ["ABC"],16  rateLimits: { "api.example.com": 2 },                       // requests/second, shared bucket17  schedule: { openMs: 60_000, closedMs: 900_000, weekendMs: 3_600_000, exchangeId: "xnys" }, // or { intervalMs }18  async poll(ctx) { const { data, response } = await ctx.http.getJson(url); return response.notModified ? [] : [raw(id, sourceId, "quote", data)]; },19  normalize(r) { /* payload → { observations, events?, instruments?, filings?, holidays?, bars? } */ },20  fixturesDir: "fixtures",21});22```2324Rules (enforced by tests and the runtime):25261. **Metadata is complete** — `ConnectorMetadataSchema`; `rightsStatus` must not be UNKNOWN for production connectors; describe delays honestly.272. **normalize() is pure** — no network, no clock other than `raw.receivedAt`; deterministic so raw payloads can be replayed (`ma replay`).283. **Timestamps** — always convert to UTC ms; declare `timestampTrust` (EXCHANGE / SOURCE / CONNECTOR). Naive local times → `zonedTimeToUtc(local, tz)`.294. **Symbols** — emit the source-native symbol; provide an `instrumentHint` so the resolver can create the instrument deterministically (`makeInstrumentId`). Aliases (`BRK.B` / `BRK-B`) are declared in seeds/proposals.305. **Never sleep** in connector code; use `rateLimits` and the shared `ctx.http`. One WebSocket carries many symbols.316. **Fixtures + tests** — at least one valid payload, one malformed/unexpected payload; tests run offline (`makeTestContext`, `fakeFetch`).327. **README** per connector: endpoint, type, rights, real-time class, fields, quirks, schedule.338. Register in `connectors/src/index.ts` and, for a new source, in `sources.ts` (family = upstream provider when known).3435Side outputs of `normalize()`:3637| key | consumer |38|---|---|39| `observations` | pipeline → observations, consensus, quotes, events |40| `events` (`ProposedEvent`, `dedupeKey`) | event engine (halts, document changes…) |41| `instruments` (`ProposedInstrument`, `createIfMissing`) | instrument master (directories) |42| `filings` (`metadata.material`) | `filings` table + FILING_PUBLISHED for material forms |43| `holidays` | market-hours engine (`exchange_holidays`) |44| `bars` | historical backfill (`bars`, producer = connector id) |4546Production connectors (v0.2, 22): coinbase-ws, kraken-ws (+12 fiat pairs), binance-ws, okx-ws, bitstamp-ws (+EUR/USD, GBP/USD), gemini-ws,47bitfinex-ws (+EUR/GBP/JPY vs USD), bybit-ws, gate-ws, cryptocom-ws, kucoin-ws (WEBSOCKET) · cboe-delayed-quotes, nasdaq-quote-api (validator)48(XHR) · ecb-frankfurter, bank-of-canada-valet, hfmarketdata-daily (OFFICIAL_API) · us-treasury-yield-curve (XML) · sec-edgar-filings,49nasdaq-trade-halts (RSS) · sec-company-tickers, nasdaq-symbol-directory (BULK_FILE) · nasdaq-market-calendar (HTML). All keyless50(hfmarketdata accepts an optional key of the sister platform). Declare `observationType` on every observation (TRADE / QUOTE / OFFICIAL_FIX /51REFERENCE_RATE / EOD_CLOSE / STABLECOIN_PROXY…); use `_shared/pairs.ts` (`planPair`, `pairObservations`) for venue pairs so fiat/fiat markets52become FOREX TRADE observations and stable/fiat markets become STABLECOIN_PROXY observations (inverted when needed). See each `README.md`.5354Onboarding workflow for a *discovered* source: `POST /v1/admin/discovery {url}` → candidate report (endpoints, embedded state, price/symbol-like55fields) → manual rights review → connector + fixtures → staging (`enabled: false` or `MA_DISABLED_CONNECTORS`) → production.56