When your client disconnects and reconnects, signals may have been published while you were offline. The WebSocket API provides a built-in mechanism to replay those missed signals.
Every strategy.signal message includes a seq_id — a monotonically increasing integer that uniquely identifies each signal in the stream. Always store the seq_id of the last signal you successfully processed:
last_seq_id = 0
async for raw in ws:
msg = json.loads(raw)
if msg["msg_type"] == "strategy.signal":
process_signal(msg["payload"])
last_seq_id = msg["seq_id"] # persist this valuePersist last_seq_id to durable storage (a database or file) so it survives process restarts.
When you resubscribe after a disconnection, pass your last seq_id as resume_from:
{
"msg_type": "subscription.request",
"timestamp": "2024-01-01T12:00:00Z",
"payload": {
"action": "subscribe",
"channel": "strategy.signal",
"strategy_id": 12345678,
"resume_from": 98765
}
}The server will:
- Send a
subscription.ack - Replay all signals with
seq_id > resume_fromin order (up to 1000 messages) - Begin live delivery once replay is complete
Replayed messages are indistinguishable from live messages — same envelope, same seq_id. If your processing is idempotent on seq_id, no deduplication is needed.
If more than 1000 signals were published while you were disconnected, the server cannot replay them all. After sending the maximum number of replayed messages, it sends a gap message:
{
"msg_type": "gap",
"id": "msg_a1b2c3d4e5f6a1b2c3d4e5f6a7b8",
"timestamp": "2024-01-01T12:00:00Z",
"v": 4,
"payload": {
"strategy_id": null,
"channel": "strategy.signal:12345678",
"from": "98765",
"message": "Replay limit of 1000 messages exceeded"
}
}| Field | Description |
|---|---|
| channel | The channel key — you can parse the strategy ID from here (e.g. "strategy.signal:12345678" → strategy ID 12345678) |
| from | The last replayed seq_id as a string |
| strategy_id | Always null in the current implementation — use channel to identify the strategy |
| message | Human-readable description |
When you receive a gap message, the missed signals cannot be replayed individually via REST — REST endpoints are time-based, not seq_id-based. Instead, snapshot the current state of the strategy using the REST API and reconcile it against your local state:
GET /Strategies/GetStrategyOpenPositions?StrategyIds=12345678— returns all currently open positions. Reconcile your local positions against this to find any that were opened or closed during the gap.GET /Strategies/GetStrategyActiveOrders?StrategyIds=12345678— returns all currently working orders. Reconcile your local orders against this to find any that were placed or canceled during the gap.
After reconciling, continue receiving live signals from the WebSocket as normal. The from field in the gap payload is the last seq_id the server replayed — use it as resume_from if you reconnect again.
import asyncio
import json
import websockets
from datetime import datetime, timezone
API_KEY = "YOUR_API_KEY"
WS_URL = "wss://api4-general.collective2.com/ws"
STRATEGY_ID = 12345678
async def run(last_seq_id: int = 0):
headers = {"Authorization": f"Bearer {API_KEY}"}
while True:
try:
async with websockets.connect(WS_URL, additional_headers=headers) as ws:
# Wait for connection.connected
await ws.recv()
# Subscribe, resuming from last known position
request = {
"msg_type": "subscription.request",
"timestamp": datetime.now(timezone.utc).isoformat(),
"payload": {
"action": "subscribe",
"channel": "strategy.signal",
"strategy_id": STRATEGY_ID,
"resume_from": last_seq_id
}
}
await ws.send(json.dumps(request))
async for raw in ws:
msg = json.loads(raw)
if msg["msg_type"] == "heartbeat":
ack = {
"msg_type": "heartbeat.ack",
"timestamp": datetime.now(timezone.utc).isoformat()
}
await ws.send(json.dumps(ack))
elif msg["msg_type"] == "strategy.signal":
process_signal(msg["payload"])
last_seq_id = msg["seq_id"]
elif msg["msg_type"] == "gap":
cursor = msg["payload"]["from"]
# Fetch remaining signals via REST and advance last_seq_id
last_seq_id = fetch_historical_signals(STRATEGY_ID, cursor)
except websockets.ConnectionClosed:
print("Disconnected, reconnecting in 5 seconds...")
await asyncio.sleep(5)
def process_signal(payload):
print(f"Signal: {payload}")
def fetch_historical_signals(strategy_id: int, from_cursor: str) -> int:
# REST endpoints are time-based, not seq_id-based — you cannot replay by seq_id.
# Instead, snapshot current state and reconcile:
# GET /Strategies/GetStrategyOpenPositions?StrategyIds={strategy_id}
# GET /Strategies/GetStrategyActiveOrders?StrategyIds={strategy_id}
# After reconciling, return from_cursor so the caller knows where the gap ended.
return int(from_cursor)
asyncio.run(run())