# 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)`.

```ts
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()`:

| key | consumer |
|---|---|
| `observations` | pipeline → observations, consensus, quotes, events |
| `events` (`ProposedEvent`, `dedupeKey`) | event engine (halts, document changes…) |
| `instruments` (`ProposedInstrument`, `createIfMissing`) | instrument master (directories) |
| `filings` (`metadata.material`) | `filings` table + FILING_PUBLISHED for material forms |
| `holidays` | market-hours engine (`exchange_holidays`) |
| `bars` | historical backfill (`bars`, producer = connector id) |

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.