Skip to content

binance.py Walkthrough

The examples/binance.py script is the canonical end-to-end demo of the worker pipeline. It runs 2 workers × 5 connections = 10 concurrent WebSocket subscriptions to Binance's depth streams, dispatches the relevant commands to each worker, and publishes aggregated orderbook updates to the worker's response topic. This page walks through what the script does and why, line by line at the section level.

What binance.py does

  1. Discovers the worker's nupkg URL by reading build_version.txt next to the locally-built nupkg and constructing the URL from config.json's script_url_template.
  2. Asserts the runtime (the host's arch/OS) is in the nupkg's .supported-rids sidecar (the runtime arch check — fails fast with a clear error if the local RID isn't supported).
  3. Creates two workers via the WorkManager's CreateWorker gRPC, both pointing at the same nupkg URL.
  4. Starts the workers via StartWorker gRPC.
  5. For each of 10 connections (5 per worker), sends three commands in sequence:
    • create — opens a WebSocket connection to Binance's combined-stream endpoint
    • subscribe — subscribes the open connection to a chunk of currency-pair depth streams
    • publish — starts the worker publishing orderbook updates for the open connection
  6. Listens on the response topic for command results. Each command is sent with a correlationid so the matching response can be identified.
  7. Reports when setup is complete (=== Setup complete. Consuming ... ===), then subscribes to the orderbook topic via the API gateway's Pubsub.Subscribe and prints one log line per depth event (stream name, bid/ask level counts) until Ctrl+C. The workers keep running in the WorkManager after the script stops.

Configuration

config.json (in the worker root) provides:

  • script_url_template — the URL pattern for the nupkg. Default: the Gitea NuGet feed at https://nuget.haenerconsulting.com/....
  • mime_type — application/x-dotnet-dll (the script targets the managed engine).
  • topic — the WorkManager command topic the script publishes to (command.websocketmanagercontroller).

The handler flow

The script doesn't talk to the worker directly — it goes through the WorkManager via the API gateway, which routes to the appropriate engine (managed in this case). Each command is a CloudEvent published to the topic; the WM dispatches it to the worker instance; the worker dispatches the command in its handler; the result is published back on the response topic.

The script is a client of the platform, not a participant. The interesting code is in the worker handler (WebSocketManagerControllerLogic.HandleCommandAsync in the worker's source) — the script just orchestrates the orchestration.

Where the workers do their work

After the script exits, the workers keep running in the WorkManager. The WM is the long-lived process; the workers are the entities that hold the WebSocket connections to Binance and publish orderbook updates. The script is a one-shot setup; the workers are persistent.

This matches the platform's overall pattern: short-lived clients (scripts, tests, ad-hoc commands) interact with long-lived services (the WM, the API gateway) that hold state and connections.

Why this is the canonical demo

  • End-to-end: covers the full path from script → API gateway → Dapr pub/sub → WorkManager → worker → WebSocketManager → Binance → back.
  • Multi-instance: runs 2 workers and 10 connections concurrently, exercising the WM's per-worker AssemblyLoadContext and the per-connection WebSocket state inside the WebSocketManager.
  • Content-addressed versioning: the script reads the version from the locally-built nupkg's build_version.txt and downloads the matching version from the Gitea feed. No version is hard-coded.
  • Error surface: if the WM is down, the gRPC call fails with a clear Gateway invoke failed: ... message. If the nupkg can't be fetched, the script exits before creating any worker. If the runtime arch doesn't match, the runtime check raises with a clear UnsupportedWorkerRuntimeError.

What the script doesn't do

  • It doesn't manage the workers after setup. After setup the script only consumes the output stream; killing the script does NOT kill the workers — they're in the WorkManager. To stop them, you'd need to call DeleteWorker for each (the example doesn't demonstrate this; see scripts/delete_worker.py in the worker root).
  • It doesn't render an order book. The script subscribes to act.exchange.binance.orderbook.update and prints one log line per event (stream name, bid/ask counts) — it doesn't maintain book state. See binance_clob_example.py in virtufin-examples for a consumer that builds a local order book.
  • It doesn't run on the worker's host. The script can run anywhere — it talks to the WorkManager via the API gateway's gRPC port (localhost:5002 by default).

The handler protocol in detail

Each command in binance.py follows the WebSocketManagerController command protocol:

Command Required fields Effect
create url, auto_reconnect, api_host Open a WebSocket connection to the url (typically wss://stream.binance.com:9443/stream); return a connection id
subscribe id, params (array of stream names like btcusdt@depth@100ms) Subscribe the connection to the streams; subsequent messages are forwarded to the worker's act.exchange.binance.* pubsub topic
publish id, topic Start the worker publishing the connection's inbound messages to topic
unsubscribe id, params Unsubscribes a chunk of streams from the connection
stop_publish id Stops publishing (but keeps the connection open)
send_raw id, message, content_type Sends a raw payload to the upstream (used for SUBSCRIBE/UNSUBSCRIBE in JSON-RPC form)
list (none) Lists active connections
disconnect id Closes a single connection
stop (none) Closes all connections and stops the worker

The script uses create, subscribe, publish for setup and disconnect would be the inverse. The other commands are useful for runtime introspection and control.

A few fields in the protocol table deserve explanation:

  • api_host: the API Gateway host:port the managed shim uses to configure its ApiClient. The managed ApiCommandWorker<T>.EnsureClientFromCommand reads this field on every handler invocation. The AOT shim doesn't need it — AotGatewayAdapter pulls the gateway from AotNative<T>.CurrentContext, which the engine sets per invocation.
  • send_raw: sends arbitrary bytes through the WebSocket connection identified by id. Binance's stream subscription protocol is JSON-RPC over WebSocket — the worker constructs {"method":"SUBSCRIBE","params":["btcusdt@depth@100ms",...],"id":1} and sends it via send_raw with content-type application/json. The message parameter is base64-encoded in the CloudEvent data payload and decoded by the worker before forwarding to the upstream.

See also