WebSocket integration

[ainergiz]

I saw your LinkedIn post and congrats for taking on this impossible task :)

I've created a similar thing over the summer for learning purposes. It was in Python with aim to transition into cpp once I figured all moving parts. That didn't happen because I never had a chance to test the Python library in production but it was getting close.

My Python project was using websocket data and I was creating a local copy of the orderbook which was incredibly hard to keep it in performant and reliable way. There are couple of things with Binance websocket which I'm sure you already know such as max 24 hours for each connection. I'll copy the doc I've created for the websocket implementation in that library at the end of this and I'd be happy to contribute on websocket implementation and see how far this library can be pushed.

Probably first step should be outlining steps and making architecture choices since it'll affect the whole thing.

1. websocket library: I'm not familiar with websocket libraries in cpp but quick search shows this as a good choice: https://github.com/zaphoyd/websocketpp

2. since websocket is going to send ~10 updates each second with 5000 level data json parsin is an important area to consider. simdjson is a good one which I was planning to use but there can be other better options

3. again because of the update frequency we need a message buffer. a lock-free ring buffer between the transport thread and the book logic would solve that: the WebSocket reader can push raw frames (or pre-parsed structs) into an SPSC buffer and the consumer can use it at its own pace. it'll also give us control over backpressure. when the ring fills we can drop oldest frames, trigger a resync, or escalate metrics instead of silently falling behind. So before we wire in the new feed, we should introduce a lightweight ring buffer abstraction tailored to our threading model (eg. single producer, single consumer) that we're going to use

4. another thing to decide is whether there should be an option to connect multiple streams or just limit it with one. I guess in the initial phase keeping it limited to BTC/USDT futures stream might be a good trade-off and can be scaled later

5. there are probably couple other things such as whether current ordered maps is the most efficient way to store the full orderbook which can be over 50,000+ level with BTC/USDT futures stream and if there is any other more special cpp alternatives.

the websocket implementation doc from my python library:

1. Connection Layer – Streaming Data In

File: crypto/connection_helpers/single_connection_manager.py

The OptimizedConnectionManager maintains one low-latency WebSocket per trading symbol. It aggressively reconnects, rotates connections every 24 hours, and tags each message with timing metadata before handing it upstream.

Key responsibilities:

• Establish WebSocket sessions and subscribe to stream suffixes (e.g. @depth@100ms).
• Detect rate limiting or stale feeds and trigger controlled reconnections.
• Extract the Binance "e" event type without fully decoding JSON so routing stays cheap.

manager = OptimizedConnectionManager(
    symbols=["btcusdt"],
    market_type="spot",
    stream_suffixes=["@depth@100ms"],
    stream_routing_config={"depthUpdate": "orderbook_manager"},
    message_callback=self.message_handler,
)

await manager.start()  # Spawns _run_websocket() + health monitor tasks

Each incoming frame runs through _handle_message(), which timestamps the payload and forwards it to the message_handler defined in async_main.py.

2. Message Router – Dispatching Work

File: crypto/async_main.py

The AsyncOrderbookDownloader.message_handler function routes messages to specialized processors based on the stream_routing_config supplied above. Depth updates flow directly to the orderbook orchestration layer, while other event types (aggTrades, bookTickers) can be parked in dedicated processors.

async def message_handler(self, symbol, market_type, raw, ws_receive_time=None, event_type=None, processor_target=None):
    callback_time = time.time()

    if processor_target == "orderbook_manager":
        task = asyncio.create_task(
            self.orderbook_manager.process_message(symbol, market_type, raw, ws_receive_time)
        )
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)
    elif processor_target == "message_processor":
        self.message_processor.queue_message(symbol, raw, market_type, ws_receive_time, callback_time)
    # … additional routing redacted …

3. Orchestrator – Coordinating Snapshots and Deltas

File: crypto/orderbook/orderbook_orchestrator.py

OrderbookOrchestrator owns one SymbolOrderBook per (symbol, market_type) and is responsible for:

• Buffering depth updates until the first snapshot lands.
• Sequencing live updates against Binance rules 4–6.
• Fetching and applying REST snapshots via SnapshotFetcher.
• Delegating live change-sets to OrderbookUpdater for application and persistence.

async def process_message(self, symbol_name, market_type, message_str):
    data = orjson.loads(message_str)
    symbol_book = self._ensure_symbol_book_present_and_schedule_init(symbol_name, market_type)

    if data.get("e") == "depthUpdate":
        async with self.symbol_locks[symbol_book.sm_key]:
            if not symbol_book.is_initialized or symbol_book.snapshot_pending:
                symbol_book.buffer_event(data)
                if not symbol_book.snapshot_pending:
                    self._trigger_resync_for_book_nolock(symbol_book)
            else:
                await self._handle_live_update(symbol_book, data)

Snapshot coordination uses an asyncio semaphore so only a fixed number of symbols fetch REST data in parallel. After a successful fetch, buffered events are replayed (still under the per-symbol lock) to bridge the REST/WebSocket gap.

4. Symbol-Level State – SortedDict Orderbook

File: crypto/orderbook/symbol_order_book.py

SymbolOrderBook stores bids and asks in SortedDict instances to keep best levels at the map front with O(log N) inserts/erases and O(K) top-N extraction. It tracks Binance sequencing metadata (last_update_id, processed_any_event_post_snapshot, etc.) and buffers out-of-order events.

def _apply_event_update_logic(self, data: dict) -> None:
    with self.read_lock:
        for price_str, qty_str in data.get("b", []):
            price, qty = Decimal(price_str), Decimal(qty_str)
            if qty == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = qty

        for price_str, qty_str in data.get("a", []):
            price, qty = Decimal(price_str), Decimal(qty_str)
            if qty == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = qty

        self.last_update_id = data["u"]
        self.timestamp = data.get("E", int(time.time() * 1000))

Rule helpers (check_event_age_rule, check_event_against_snapshot_rules, check_subsequent_event_rules) enforce Binance’s documented guarantees and decide when a resync is required.

5. Live Update Application & Persistence

File: crypto/orderbook/book_updater.py

OrderbookUpdater times each apply, drops stale updates, and pushes top-N snapshots onto an async queue for batched persistence. It returns (applied_successfully, resync_triggered) so the orchestrator can decide whether to schedule a resync.

applied, resync = await self.updater.handle_live_update(symbol_book, depth_update)
if resync:
    self._trigger_resync_for_book_nolock(symbol_book)

6. Snapshot Fetching and Resynchronization

File: crypto/orderbook/snapshot_fetcher.py

SnapshotFetcher issues REST calls (spot or futures endpoints) using the shared aiohttp.ClientSession. It recognizes rate limiting responses and can invoke a shutdown callback so upstream components can back off.

During resync:

1. SymbolOrderBook is marked snapshot_pending and its buffer cleared.
2. _fetch_apply_and_process_buffer acquires the global semaphore and downloads the snapshot.
3. If lastUpdateId is compatible with the first buffered event, the snapshot is applied and buffered updates replayed.
4. Otherwise a resync is re-triggered, potentially with exponential backoff governed by the orchestrator.

7. End-to-End Flow

1. Connect: OptimizedConnectionManager.start() opens the WebSocket for each symbol.
2. Route: AsyncOrderbookDownloader.message_handler inspects the event type and forwards depth updates.
3. Initialize: OrderbookOrchestrator.trigger_book_initialization fetches the initial snapshot while buffering live deltas.
4. Apply: SymbolOrderBook.apply_snapshot seeds the book, OrderbookUpdater.handle_live_update applies live changes, and rule helpers ensure continuity.
5. Persist & Monitor: Top-N snapshots and raw messages go to disk queues, while metrics surfaces exposure to rate limits or stale feeds.