Trading-ML docs krew-solutions/trading-ml … …

Transport supervisor

broker/lib/infrastructure/acl/common/transport_supervisor.ml holds the WS-primary / REST-fallback pattern that broker ACL adapters use to keep Broker.event flowing across WS disconnects. It is the answer to the port contract documented on Broker.event itself:

Adapter encapsulates the transport (WS push, REST poll, synthetic generator, replay) — the same callback fires regardless of which path delivered the event.

Concretely: when the WebSocket bridge loses its connection, the supervisor activates a REST poll fiber. When the WS reconnects, the supervisor runs one synchronous catch-up poll over the disconnect gap, then puts the poll fiber back to sleep. Downstream consumers see a continuous stream; they do not know which transport produced any given event.

#State machine

Each supervisor instance owns three pieces of state: 

mutable poll_active : bool;   (* true ⇒ REST poll fiber ticks *)
mutable last_ts : int64;      (* cursor for catch-up windows *)
mutable stopped : bool;       (* tear-down latch *)

Three states, three transitions: 

       ┌─────────────────┐   WS first connect succeeded
       │ INIT (poll on)  │ ──────────────────────────────┐
       └─────────────────┘                               │
              │                                          ▼
              │ initial WS connect failed       ┌──────────────────┐
              │ (Resilient in backoff)          │ WS_HEALTHY       │
              ▼                                 │ (poll dormant)   │
       ┌─────────────────┐                      └──────────────────┘
       │ WS_DOWN         │ ◄───────────────────────┐           ▲
       │ (poll on)       │   Resilient                         │
       └─────────────────┘   on_disconnect                     │
              │                                                │
              │ Resilient on_reconnect:                        │
              │   1) catch-up poll over (last_ts, now)         │
              │   2) flip poll dormant ────────────────────────┘

The invariant is ws_healthy ⇒ ¬poll_active — the two transports never run simultaneously by design. WS health is not inferred from traffic (a quiet channel ≠ a dead channel); it is signalled explicitly through three calls a bridge wires into its WS layer:

Caller Supervisor call Effect
WS bridge after initial connect ws_came_up poll_active ← false
Websocket.Resilient.on_disconnect ws_went_down poll_active ← true
Websocket.Resilient.on_reconnect ws_reconnected one synchronous catch-up poll over (last_ts, now), then poll_active ← false

Initial state on start is poll_active = true, last_ts = initial_since_ts. This means polling begins immediately so an adapter that comes up before its WS is connected still surfaces events through REST.

A second fiber runs the steady-state poll loop. On each tick it checks poll_active; if false, it sleeps and loops; if true, it calls the user-supplied poll_window ~since_ts:last_ts ~to_ts:now and funnels the result through the same dedup + emit pipeline as the WS branch.

#What the supervisor does not own

Three things stay with the caller, by design:

  1. The WS bridge. The supervisor never opens, closes, or reads from a socket. It only observes lifecycle transitions the bridge feeds it through ws_came_up / ws_went_down / ws_reconnected.
  2. The Stream_dedup table. The caller passes a dedup_accept : 'event -> bool closure. Typically the closure binds a Stream_dedup.t instance the adapter holds on its own type, so the same dedup state can be observed by non-supervised code paths (an OHS publisher, a snapshot query, a manual reconcile).
  3. The final emit callback. The adapter passes its own dispatch closure, which often does extra work (cumulative-sum bump, view-model projection) before producing the Broker.event.

The result is a polymorphic value: 

val start :
  env:Eio_unix.Stdenv.base ->
  sw:Eio.Switch.t ->
  label:string ->
  poll_interval:float ->
  ts_now:(unit -> int64) ->
  poll_window:(since_ts:int64 -> to_ts:int64 -> 'event list) ->
  ts_of_event:('event -> int64) ->
  dedup_accept:('event -> bool) ->
  emit:('event -> unit) ->
  initial_since_ts:int64 ->
  'event t

The supervisor is parametric on 'event; the same type covers bars, fills, and any future per-symbol or per-account stream.

#The dedup invariant

Both branches funnel events through one closure: 

WS push ──┐
          ├─► absorb ─► dedup_accept ─► (advance last_ts) ─► emit
REST poll ┘

The bridges that wrap the supervisor (Bcs.Ws_bridge, Bcs.Order_event_bridge, Finam.Ws_bridge, Alor.Ws_bridge plus the *_broker.dispatch_ws_event routers) all route through Transport_supervisor.feed_ws, and the supervisor's poll fiber calls the same internal absorb for REST-derived events. The single shared dedup_accept closure means:

The discriminator the dedup uses must be stable across both transports. For bars this is trivially the candle's structural equality (Candle.equal). For fills both adapters dedup on trade_id:

Both broker REST surfaces expose the venue-side per-leg id, so the supervisor seam can rely on exact String.equal rather than a partial (qty, price) discriminator. (Earlier revisions of these adapters compromised on partial dedup because their wire DTOs were under-parsed; the fix in both cases was extending the parser, not the dedup logic.)

#Wiring under two socket models

Real brokers come in two transport shapes, and the supervisor adapts to both. The count of supervisors is symmetric (N for bars + 1 for fills per adapter); only the disconnect-fan-out mechanism differs.

#Socket-per-subscription: BCS market data

BCS opens one WebSocket per (instrument, timeframe). Each socket belongs to exactly one bar subscription. The Resilient.config for that socket can capture the subscription's supervisor directly in closure: 

on_disconnect = (fun () -> Transport_supervisor.ws_went_down sup);
on_reconnect  = (fun () -> Transport_supervisor.ws_reconnected sup);

The bridge keeps a parallel map supervisors : Candle.t Transport_supervisor.t SubMap.t alongside conns, keyed by the same (instrument, timeframe). Unsubscribe removes both and stops the supervisor. No separate listener registry is needed — the socket-to-supervisor mapping is 1:1.

For BCS fills the picture is even simpler: one account-wide WS, one fill supervisor. The supervisor is created inside Bcs_broker.start_live_feed and the Order_event_bridge callback closes over it directly.

#Multiplexed socket: Finam

Finam runs one WebSocket carrying every subscription — bars for any number of (instrument, timeframe) pairs plus the account-wide trades stream. A single disconnect must notify every active supervisor at once. There is no 1:1 socket-to-supervisor capture available.

Finam.Ws_bridge therefore carries an explicit listener registry: 

type listener_id = int

val register_lifecycle :
  bridge ->
  on_disconnect:(unit -> unit) ->
  on_reconnect:(unit -> unit) ->
  listener_id

val unregister_lifecycle : bridge -> listener_id -> unit

The bridge's own Resilient.config callbacks fan over the registered listeners on every transition. Each supervisor (one per bars subscription plus the fill supervisor) registers at creation and unregisters at tear-down.

Because the multiplexed on_event : Ws.event -> unit cannot know upfront which supervisor a particular Bars b belongs to, Finam_broker.dispatch_ws_event routes by lookup: 

| Bars b ->
    match SubMap.find_opt (b.instrument, b.timeframe) t.bar_supervisors with
    | Some sup -> List.iter (Transport_supervisor.feed_ws sup) b.bars
    | None     -> Log.info "[finam ws] bars for unregistered key — dropping"

So the supervisors live on the broker (finam_broker.t) rather than on the bridge: the broker is the routing point.

#Multiplexed socket: Alor

Alor is multiplexed like Finam, with the same broker-side supervisor registry and register_lifecycle fan-out. The one difference is correlation: Alor data frames carry no subscription key — a bar payload is bare OHLCV { data, guid } — so Alor.Ws_bridge keeps a guid → target (Bars (instrument, timeframe) | Trades) map and enriches each frame back into a typed Ws.event before handing it to Alor_broker.dispatch_ws_event, which then routes to the matching supervisor exactly as Finam does. On reconnect the bridge resubscribes reusing the original guids so the map stays valid.

#Catch-up windows

ws_reconnected calls poll_window ~since_ts:last_ts ~to_ts:now once, synchronously. Whether this fully covers a long outage depends on the broker's REST endpoint:

Broker Stream REST endpoint Cursor support
BCS bars Rest.bars ?from_ts ?to_ts precise cursors (BCS caps at 1440 bars per request)
BCS fills Rest.get_deals ?from_ts ?to_ts precise cursors
Finam bars Rest.bars ?from_ts ?to_ts ?n precise cursors
Finam fills Rest.get_trades ?from_ts ?to_ts precise cursors
Alor bars Rest.bars ?from_ts ?to_ts ?n precise cursors
Alor fills Rest.get_trades ~exchange session-scoped (no time cursor); dedup on trade_id filters re-observed legs

Bars are symmetric on cursor support across all three brokers. Alor's fills endpoint is the one exception — it returns the current session's trades with no time window, so the supervisor relies on Stream_dedup (keyed by trade_id) to suppress the re-observed legs rather than on a precise cursor. The steady-state poll fiber ticks every 60 s, so its window stays narrow regardless of the broker's per-request cap. The single-shot reconnect catch-up can hit the cap on long outages — for BCS bars specifically, a gap longer than ~24 h on M1 exceeds 1440 bars and the server returns 400. In that case the catch-up returns no bars and the gap is filled only by the next steady-state ticks once WS is back; some intermediate bars are lost. Acceptable for current scope; chunked catch-up requests are a possible future enhancement.

#Known limitations

#See also