Connect to your API's /ws endpoint and pass Authorization: Bearer YOUR_API_KEY during the upgrade. Your API key must allow the channels you request. See Authorization.
This command checks the HTTP upgrade. It is not a complete WebSocket application client:
curl --http1.1 -i \
-H "Authorization: Bearer $C2_API_KEY" \
-H "Connection: Upgrade" \
-H "Upgrade: websocket" \
-H "Sec-WebSocket-Version: 13" \
-H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
https://api4-general.collective2.com/wsHTTP 101 Switching Protocols means the upgrade succeeded. An invalid API key returns HTTP 401. After the upgrade, wait for connection.connected and begin receiving continuously. connection.disconnected is reserved and is not currently emitted; detect socket closure.
For interactive inspection:
websocat -H "Authorization: Bearer $C2_API_KEY" wss://api4-general.collective2.com/wsAn interactive client still needs to acknowledge application heartbeats with a current timestamp.
The deployed application heartbeat interval is 30 seconds. Reply promptly to every heartbeat with heartbeat.ack and a current UTC timestamp. These JSON messages are separate from WebSocket protocol Ping/Pong frames; a library's automatic Pong response does not acknowledge a JSON heartbeat.
Missing heartbeats or consuming messages too slowly can close the connection. Close timing varies; do not assume an exact deadline. Keep the receive loop responsive even while doing REST reconciliation or slow application work.
Recommended The server supports negotiated permessage-deflate compression. A client must offer it during the upgrade, and it is used only if negotiation succeeds. The Python example requests it with compression="deflate". Continue to handle uncompressed connections when negotiation does not select the extension. HTTP response compression does not replace WebSocket extension negotiation.
Install websockets in your Python environment. Set C2_API_KEY, and optionally C2_WS_URL, C2_ACCOUNT and C2_BROKER_ID. This complete connection example keeps one receive loop alive, acknowledges heartbeats, correlates acknowledgements by request ID and subscribes to all four account channels. It prints channel data; use the account consumer when maintaining local trading state.
import asyncio
import json
import os
import random
import uuid
from datetime import datetime, timezone
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed
API_KEY = os.environ["C2_API_KEY"]
WS_URL = os.getenv("C2_WS_URL", "wss://api4-general.collective2.com/ws")
ACCOUNT = os.getenv("C2_ACCOUNT", "DU123456")
BROKER_ID = int(os.getenv("C2_BROKER_ID", "111"))
CHANNELS = ("account.orders", "account.positions", "account.balance", "account.state")
def utc_now():
return datetime.now(timezone.utc).isoformat()
async def receive(ws, pending, connected):
async for raw in ws:
message = json.loads(raw)
kind = message["msg_type"]
if kind == "heartbeat":
await ws.send(json.dumps({"msg_type": "heartbeat.ack", "timestamp": utc_now()}))
elif kind == "connection.connected":
if not connected.done():
connected.set_result(None)
elif kind == "subscription.ack":
future = pending.pop(message.get("request_id"), None)
if future is not None and not future.done():
future.set_result(message["payload"])
elif message["payload"]["action"] == "unsubscribe":
# Unsolicited revocation: stop using this channel's cached state.
print("Server unsubscribed:", message["payload"])
elif kind == "error":
future = pending.pop(message.get("request_id"), None)
if future is not None and not future.done():
future.set_exception(ValueError(json.dumps(message["payload"])))
else:
print("Server error:", message["payload"])
else:
# Keep this handler short. Queue slow work outside the receive loop.
print(kind, message.get("account"), message.get("broker_id"), message.get("payload"))
async def run():
delay = 1.0
while True:
pending = {}
receiver = None
try:
async with connect(WS_URL, additional_headers={"Authorization": f"Bearer {API_KEY}"},
compression="deflate") as ws:
# Account state is stale after reconnect. Wait for replacement snapshots.
connected = asyncio.get_running_loop().create_future()
receiver = asyncio.create_task(receive(ws, pending, connected))
await asyncio.wait_for(connected, timeout=10)
acknowledgements = []
for channel in CHANNELS:
request_id = uuid.uuid4().hex
future = asyncio.get_running_loop().create_future()
pending[request_id] = future
acknowledgements.append(future)
await ws.send(json.dumps({
"msg_type": "subscription.request", "request_id": request_id,
"timestamp": utc_now(),
"payload": {"action": "subscribe", "channel": channel,
"account": ACCOUNT, "broker_id": BROKER_ID}
}))
# The receiver remains active while acknowledgements/snapshots are prepared.
await asyncio.wait_for(asyncio.gather(*acknowledgements), timeout=35)
await receiver
except (ConnectionClosed, OSError, TimeoutError) as error:
print("Reconnecting after:", error)
finally:
if receiver is not None:
receiver.cancel()
await asyncio.gather(receiver, return_exceptions=True)
for future in pending.values():
future.cancel()
await asyncio.sleep(delay + random.uniform(0, delay / 4))
delay = min(30.0, delay * 2)
asyncio.run(run())A correlated subscription rejection raises ValueError; correct its permission/field cause before restarting. Adjust client timeouts for your deployment. Acknowledgements confirm the subscription, and each account channel's initial data message establishes its baseline. Unsolicited unsubscribe acknowledgements can arrive without request_id after authority is revoked. There is no need to close a healthy connection solely because another subscription was rejected.
When done, closing the connection releases its subscriptions. For an individual channel, send action: "unsubscribe" with the same identity and correlate its acknowledgement. See Errors and Recovery.