Skip to content

Ingestion framework

Family: SDK-crate (atelier-connect::framework) · Min version: atelier-connect 0.0.10

The ingestion framework is a set of composable seams that let one shared WebSocket transport and one shared reconstruction runtime serve every exchange. A venue is integrated by filling in the parts that actually differ between exchanges — subscribe framing, heartbeat, symbol spelling, and how a book is kept in sync — behind stable traits; everything else is written once.

The two generations

Two ingestion generations live in atelier-connect. They differ in where per-venue behaviour lives, not in what they produce.

Generation Per-venue surface Shared surface Reconstruction
Legacy path A full per-source WSS client + event enum per venue Reconnect/backoff plumbing Ad hoc, per client
Framework path ProtocolHooks + Normalizer + a ReconstructionModel choice, registered as one ExchangeAdapter WssTransport, the drive loop, the reconstruction runtime, the checksum kit, the symbol codec SourcedOrderBook / SourcedTradeBook, model-selected

The framework path collapses the per-venue surface to the pieces that vary and promotes reconstruction into shared, model-selected types. Both generations decode the same wire types and emit the same normalized Orderbook and Trade, so downstream consumers see one schema regardless of which path produced a stream.

Turning the framework on

A worker runs the framework path when framework_ingest = true is set under [collect] in its config. MarketWorker::run_framework then resolves the requested venues against the registry and drives them through the shared transport and reconstruction runtime.

The adapter model

A venue adapter is a single ExchangeAdapter that composes four seams. The adapter names its venue id, carries a static ExchangeProfile (symbol codec, connection budget, version pins), chooses a ReconstructionModel per channel, and spawns one connection carrying a set of symbols.

flowchart LR
    A["ExchangeAdapter<br/>(one per venue)"] --> H["ProtocolHooks<br/>subscribe · heartbeat ·<br/>control · codec · prepare"]
    A --> N["Normalizer<br/>decoded frame →<br/>DomainEvent(s)"]
    A --> M["ReconstructionModel<br/>per channel"]
    A --> P["ExchangeProfile<br/>SymbolCodec ·<br/>ConnectionBudget · versions"]
    H --> T["WssTransport<br/>(shared)"]
    N --> T
    T --> D["drive loop<br/>(shared)"]
    D --> R["SourceRuntime +<br/>SourcedOrderBook /<br/>SourcedTradeBook<br/>(shared)"]
    M --> R

Diagram: one per-venue ExchangeAdapter supplies ProtocolHooks, a Normalizer, a per-channel ReconstructionModel, and an ExchangeProfile; the WssTransport, drive loop, and reconstruction runtime are shared across every venue.

ProtocolHooks — the WSS protocol seam

ProtocolHooks captures the connection behaviour that varies across exchanges. The WssTransport calls into it at each stage of the connect → subscribe → read loop.

Hook Purpose Default
endpoint Base WSS URL required
subscribe_frames Frames that subscribe a set of symbols on one socket required
heartbeat Keep-alive strategy (None · WsPong · Text · Resubscribe) None (passive)
on_inbound_control React to a control frame — Ignore or Reply Ignore
frame_codec Frame encoding — PlainText or Gzip PlainText
prepare Pre-subscribe bootstrap: token fetch, dynamic URL, login no-op

prepare returns a Prepared that may override the endpoint and heartbeat and inject extra frames, covering venues with a dynamic URL or an in-band seed request without special-casing the transport.

WssTransport — the shared connect-and-read loop

WssTransport<H, D> is the exchange-agnostic pump, monomorphized per venue over the ProtocolHooks H and the frame decoder D. It runs one lifecycle: bootstrap via prepare → resolve the endpoint → connect → send subscribe frames → start the heartbeat → read (inflate → control → decode) → exit. Each inbound frame is inflated per the venue codec, offered to on_inbound_control, then decoded to the venue event type and forwarded with its local receipt timestamp. A WS-protocol RTT ping runs independently so the connection round-trip is measured with no per-venue code.

The loop returns a WssExitReason, which the drive loop maps onto a TaskExit (Completed · DrainTimedOut · Failed) that the worker uses to drive the task to its terminal state and to decide reconnection.

Normalizer — decoded frame to DomainEvent

A per-venue Normalizer maps one decoded WSS event into zero or more DomainEvents — Book(NormalizedDelta) or Trade. Returning a list means a batched-trade frame yields one event per trade rather than dropping all but the first. The normalizer derives the canonical pair from the event's own venue symbol, since a socket is multi-symbol.

BookReconstructor — model-selected sync

Reconstruction is not per-venue code; it is a ReconstructionModel choice consumed by the shared SourcedOrderBook and SourcedTradeBook. The adapter picks a model per channel (a venue may run one model on its book channel and another elsewhere), and the shared apply path handles seeding, continuity checks, and gap detection. See Reconstruction models.

The registry

Every compiled-in venue is listed once in register_all, keyed by its stable venue id and resolved into a process-wide registry at boot. resolve is a fail-fast self-check: it turns a request for an unknown venue into a startup error listing the missing ids, rather than a silently empty stream.

let adapter = registry()
    .get("binance")
    .expect("binance is compiled in");
let model = adapter.book_model("orders");

Adding a venue is one line in register_all plus the adapter that implements ExchangeAdapter over its ProtocolHooks, Normalizer, and chosen ReconstructionModel.

Building a venue adapter

The steps to integrate a new public order-book + trades venue:

  1. Choose the SymbolCodec for the venue's wire symbol spelling and set the ConnectionBudget on its ExchangeProfile.
  2. Implement ProtocolHooks: endpoint, subscribe_frames, and only the other hooks the venue needs (heartbeat, control-frame reply, gzip codec, or a prepare bootstrap).
  3. Implement a frame decoder and a Normalizer that maps its events to DomainEvents.
  4. Choose a ReconstructionModel per channel (see Reconstruction models); provide a RestSnapshot seeder if the model needs one.
  5. Register the adapter in register_all.
impl ProtocolHooks for BinanceHooks {
    fn endpoint(&self) -> String {
        "wss://stream.binance.com:9443/ws".to_string()
    }
    fn subscribe_frames(&self, symbols: &[String]) -> Vec<Message> {
        vec![subscribe_frame(symbols)]
    }
    fn heartbeat(&self) -> Heartbeat {
        Heartbeat::WsPong { every: Duration::from_secs(15) }
    }
}

See also