# Writing a Market Atlas connector

A connector is one technical integration of a source (connectors/src/sources.ts). It declares metadata (type, rights, real-time class, family), optionally seeds instruments, and implements either a streaming lifecycle (start/stop) or a polling one (poll + schedule), plus a pure normalize(raw).

export const example = defineConnector({
  metadata: { id: "example-xhr", name: "…", version: "1.0.0", sourceId: "example", organization: "…", sourceType: "XHR",
    jurisdiction: "US", rightsStatus: "PUBLIC_ATTRIBUTED", realtimeStatus: "DELAYED", expectedLatencyMs: 900_000,
    supportsStreaming: false, supportsHistorical: false, assetClasses: ["EQUITY"], exchanges: ["xnys"], homepage: "https://…",
    description: "What it observes, how often, what is delayed.", rightsNotes: "Attribution…", termsUrl: "https://…",
    sourceFamily: "example", enabled: true },
  seeds: [{ symbol: "ABC", hint: { assetClass: "EQUITY", name: "ABC Corp", exchangeId: "xnys", mic: "XNYS", currency: "USD", country: "US" } }],
  defaultSymbols: ["ABC"],
  rateLimits: { "api.example.com": 2 },                       // requests/second, shared bucket
  schedule: { openMs: 60_000, closedMs: 900_000, weekendMs: 3_600_000, exchangeId: "xnys" }, // or { intervalMs }
  async poll(ctx) { const { data, response } = await ctx.http.getJson(url); return response.notModified ? [] : [raw(id, sourceId, "quote", data)]; },
  normalize(r) { /* payload → { observations, events?, instruments?, filings?, holidays?, bars? } */ },
  fixturesDir: "fixtures",
});

Rules (enforced by tests and the runtime):

1. Metadata is complete — ConnectorMetadataSchema; rightsStatus must not be UNKNOWN for production connectors; describe delays honestly.
2. normalize() is pure — no network, no clock other than raw.receivedAt; deterministic so raw payloads can be replayed (ma replay).
3. Timestamps — always convert to UTC ms; declare timestampTrust (EXCHANGE / SOURCE / CONNECTOR). Naive local times → zonedTimeToUtc(local, tz).
4. 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.
5. Never sleep in connector code; use rateLimits and the shared ctx.http. One WebSocket carries many symbols.
6. Fixtures + tests — at least one valid payload, one malformed/unexpected payload; tests run offline (makeTestContext, fakeFetch).
7. README per connector: endpoint, type, rights, real-time class, fields, quirks, schedule.
8. Register in connectors/src/index.ts and, for a new source, in sources.ts (family = upstream provider when known).

Side outputs of normalize():

Production connectors (v0.2, 22): coinbase-ws, kraken-ws (+12 fiat pairs), binance-ws, okx-ws, bitstamp-ws (+EUR/USD, GBP/USD), gemini-ws, bitfinex-ws (+EUR/GBP/JPY vs USD), bybit-ws, gate-ws, cryptocom-ws, kucoin-ws (WEBSOCKET) · cboe-delayed-quotes, nasdaq-quote-api (validator) (XHR) · ecb-frankfurter, bank-of-canada-valet, hfmarketdata-daily (OFFICIAL_API) · us-treasury-yield-curve (XML) · sec-edgar-filings, nasdaq-trade-halts (RSS) · sec-company-tickers, nasdaq-symbol-directory (BULK_FILE) · nasdaq-market-calendar (HTML). All keyless (hfmarketdata accepts an optional key of the sister platform). Declare observationType on every observation (TRADE / QUOTE / OFFICIAL_FIX / REFERENCE_RATE / EOD_CLOSE / STABLECOIN_PROXY…); use _shared/pairs.ts (planPair, pairObservations) for venue pairs so fiat/fiat markets become FOREX TRADE observations and stable/fiat markets become STABLECOIN_PROXY observations (inverted when needed). See each README.md.

Onboarding workflow for a discovered source: POST /v1/admin/discovery {url} → candidate report (endpoints, embedded state, price/symbol-like fields) → manual rights review → connector + fixtures → staging (enabled: false or MA_DISABLED_CONNECTORS) → production.