Skip to content

perps: WebSocket foundation — connection, client, subscribe/dispatch framing, sequence/orderbook reuse #397

Closed

Description

@TexasCoding

Epic: part of #387 · Depends on: #388

Summary

Build the perps WebSocket transport layer under a new kalshi/perps/ws/ package, mirroring the existing kalshi/ws/ stack. This issue delivers the connection, the user-facing client facade, command/response framing (subscribe/unsubscribe/update_subscription/list_subscriptions + control frames), and the reused sequence-gap / orderbook-reconstruction / backpressure machinery. It explicitly does not define per-channel payload models (ticker, trade, fill, user_orders, orderbook_*, order_group_updates) — those land in the separate perps: WS channels issue. The deliverable is "a perps WS connection that can authenticate, send every command frame, parse every response/control frame, track sequence gaps, and route raw frames to subscription queues."

The perps margin exchange has its own WS host, its own command grammar (extra update_subscription variants, list_subscriptions), and — critically — epoch-millisecond _ms-suffixed timestamps (ts_ms, created_ts_ms, …) rather than the event-contract WS timestamp convention. This issue stands up the transport so channel models can be slotted in next.

Endpoints / scope

WebSocket server (from /tmp/perps_asyncapi.yaml servers):

Env Host Pathname Protocol
production external-api-margin-ws.kalshi.com /trade-api/ws/v2/margin wss
demo external-api-margin-ws.demo.kalshi.co /trade-api/ws/v2/margin wss

Full URLs: prod wss://external-api-margin-ws.kalshi.com/trade-api/ws/v2/margin, demo wss://external-api-margin-ws.demo.kalshi.co/trade-api/ws/v2/margin. Auth is RSA-PSS apiKey during the handshake (same signer as event-contract WS — sign GET + ws path).

Client-to-server commands (AsyncAPI operations with action: receive, i.e. client sends):

Channel Direction Payload schema Notes
root client→server subscribeCommandPayload cmd: "subscribe"; params.channels (1+), optional market_ticker/market_tickers, send_initial_snapshot (default false), skip_ticker_ack (default false)
root client→server unsubscribeCommandPayload cmd: "unsubscribe"; params.sids (1+)
root client→server updateSubscriptionCommandPayload (add) cmd: "update_subscription"; params.action: "add_markets", sids (exactly 1) or sid, market_ticker/market_tickers, send_initial_snapshot
root client→server updateSubscriptionCommandPayload (delete) same payload; params.action: "delete_markets"
root client→server updateSubscriptionCommandPayload (single sid) same payload; uses scalar params.sid instead of sids array
root client→server listSubscriptionsCommandPayload cmd: "list_subscriptions"; params omitted
control_frames client→server ping/pong (incomingPing/incomingPong) WS control frames 0x9/0xA; empty body. Handled by the websockets lib's keepalive — see Notes

Server-to-client responses / control (AsyncAPI operations with action: send):

Channel Direction Payload schema Notes
root server→client subscribedResponsePayload type: "subscribed"; msg.channel, msg.sid
root server→client unsubscribedResponsePayload type: "unsubscribed"; top-level sid, seq, id
root server→client okResponsePayload type: "ok"; optional sid, seq, msg.market_tickers (update ack)
root server→client listSubscriptionsResponsePayload type: "ok"; msg is an array of {channel, sid}
root server→client errorResponsePayload type: "error"; msg.code (int 1–18), msg.msg, optional msg.market_ticker
control_frames server→client ping (outgoingPing body heartbeat) / pong (outgoingPong) Kalshi sends Ping every 10s with body heartbeat; client lib responds with Pong

NOTE the listSubscriptionsResponsePayload collision: list-subscriptions replies use type: "ok" (not a distinct type), disambiguated by msg being an array vs the scalar-ok update ack whose msg is an object. The framing layer must branch on msg shape, not just type.

Models

New file: kalshi/perps/ws/models/control.py (command + response/control envelopes only — channel payloads are out of scope for this issue).

Pydantic models (mirror kalshi/ws/models/base.py style; all envelopes are response-side and so are plain validating models, except the command params which are request-side):

  • Command params (request-side, model_config = {"extra": "forbid"}):
    • SubscribeParams — channels: builtins.list[PerpsChannel] (1+), market_ticker: str | None, market_tickers: builtins.list[str] | None, send_initial_snapshot: bool | None, skip_ticker_ack: bool | None. Serialized via model_dump(exclude_none=True, by_alias=True, mode="json").
    • UnsubscribeParams — sids: builtins.list[int] (1+).
    • UpdateSubscriptionParams — action: UpdateSubscriptionAction, sid: int | None, sids: builtins.list[int] | None (max 1), market_ticker: str | None, market_tickers: builtins.list[str] | None, send_initial_snapshot: bool | None. The "single sid" variant sets sid; the array variant sets sids.
    • Command wrappers SubscribeCommand, UnsubscribeCommand, UpdateSubscriptionCommand, ListSubscriptionsCommand each with id: int, cmd: Literal[...] (serialization_alias-free; the const string matches the wire). ListSubscriptionsCommand has no params. All extra="forbid".
  • Response envelopes (response-side, validating):
    • SubscribedResponse — id: int | None, type: Literal["subscribed"], msg: SubscribedMsg where SubscribedMsg has channel: str, sid: int.
    • UnsubscribedResponse — id: int | None, sid: int, seq: int, type: Literal["unsubscribed"].
    • OkResponse — id: int | None, sid: int | None, seq: int | None, type: Literal["ok"], msg: OkMsg | None where OkMsg has market_tickers: builtins.list[str] | None.
    • ListSubscriptionsResponse — id: int, type: Literal["ok"], msg: builtins.list[SubscriptionEntry] where SubscriptionEntry has channel: str, sid: int.
    • PerpsErrorMsg — code: int (1–18), msg: str, market_ticker: str | None. PerpsErrorResponse — id: int | None, type: Literal["error"], msg: PerpsErrorMsg.

Enums (cite exact spec enum members):

  • PerpsChannel (from subscribeCommandPayload.params.channels.items.enum): orderbook_delta, ticker, trade, fill, user_orders, order_group_updates.
  • UpdateSubscriptionAction (from updateSubscriptionCommandPayload.params.action.enum): add_markets, delete_markets.
  • PerpsBookSide (from bookSide): bid, ask. Reuse note: the event-contract OrderbookManager is hardwired to yes/no sides — perps uses bid/ask. Flag this in the orderbook reuse decision (see below). Defined here for the framing layer even though the orderbook payload model itself lives in the channels issue.

Timestamp convention (CRITICAL — call out in models and parsing): perps WS timestamps are Unix epoch milliseconds with a _ms suffix on the field name. None appear in this issue's control envelopes (only seq/sid are integers here), but the framing/parse helpers and any shared base must NOT assume the event-contract WS timestamp shape. Document in the module docstring that channel models (next issue) will carry ts_ms, created_ts_ms, last_updated_ts_ms, expiration_ts_ms, next_funding_time_ms (all int, epoch ms), so the parse path stays int-typed and does not coerce to RFC3339 datetimes.

No DollarDecimal/FixedPointCount fields appear in the control/command envelopes (prices/counts ride only on channel payloads). Note for the channels issue: orderbook levels are priceLevelDollarsCountFp = [price_in_dollars (DollarDecimal), contract_count_fp (FixedPointCount)].

Resource methods

The WS layer is not a REST resource, so there are no sync Resource methods. The transport mirrors kalshi/ws/ modules. New files under kalshi/perps/ws/:

  • connection.py — PerpsConnectionManager mirroring ConnectionManager. Same state machine (ConnectionState — reuse the existing enum from kalshi.ws.connection, do not duplicate), same _open_socket/connect/reconnect/close/send/recv, same RSA-PSS _build_auth_headers (sign GET + ws path), same AWS-full-jitter reconnect. Reads ws_base_url/ws_max_retries/retry_*/ws_ping_interval/ws_close_timeout from PerpsConfig (delivered by the foundation issue). Signature parity:
    • async def connect(self) -> None
    • async def reconnect(self) -> None
    • async def close(self) -> None
    • async def send(self, msg: dict[str, Any]) -> None
    • async def recv(self) -> str
  • channels.py — PerpsSubscriptionManager mirroring SubscriptionManager, but with the perps command grammar:
    • async def subscribe(self, channel: str, params: dict[str, Any] | None = None) -> int (returns client_id; sends cmd: "subscribe")
    • async def unsubscribe(self, client_id: int) -> None (sends cmd: "unsubscribe", params.sids)
    • async def update_subscription(self, client_id: int, action: str, market_tickers: builtins.list[str] | None = None, send_initial_snapshot: bool | None = None) -> None — array-sids form
    • async def update_subscription_single_sid(self, client_id: int, action: str, market_tickers: builtins.list[str] | None = None, send_initial_snapshot: bool | None = None) -> None — scalar-sid form (the updateSubscriptionSingleSidCommand variant)
    • async def list_subscriptions(self) -> builtins.list[SubscriptionEntry] (sends cmd: "list_subscriptions", parses the array msg)
    • async def resubscribe_all(self) -> None
    • Inside this class annotate list returns/params with builtins.list[T] (the list builtin is shadowed by .list()-style methods on the surrounding API).
  • dispatch.py — PerpsMessageDispatcher mirroring MessageDispatcher. CONTROL_TYPES = {"subscribed", "unsubscribed", "ok", "error"}. Same orphan-subscribed / server-unsubscribe handling. MESSAGE_MODELS is left empty/stub here (populated by the channels issue) — this issue only needs control-frame routing + the "unknown message type" path.
  • sequence.py — reuse kalshi.ws.sequence.SequenceTracker + SequenceGap unchanged. Only the channel set differs: define PERPS_SEQUENCED_CHANNELS = {"orderbook_delta", "orderbook_snapshot", "order_group_updates"} (these are the perps payloads carrying seq per spec; unsubscribed also carries seq but is a control frame). Pass the perps set into the tracker (parameterize SEQUENCED_CHANNELS or subclass) rather than copying the class.
  • orderbook.py — reuse-with-caveat. The existing OrderbookManager hardcodes yes/no sides and consumes OrderbookSnapshotMessage/OrderbookDeltaMessage whose sides are yes/no. Perps uses bid/ask (PerpsBookSide) and priceLevelDollarsCountFp levels. Decision to make in this issue (flag explicitly): either (a) parameterize OrderbookManager over its two side names + level message types, or (b) fork a PerpsOrderbookManager. Recommend (a) — extract the side-name pair as a constructor arg so the Decimal price-indexed dict logic, lazy materialization, remove_by_sid, and cache invalidation are shared verbatim. The actual perps orderbook payload models land in the channels issue; this issue wires the manager seam and leaves a stub/typed-protocol for the message types.
  • backpressure.py — reuse kalshi.ws.backpressure.MessageQueue + OverflowStrategy verbatim (no perps-specific behavior).
  • client.py — PerpsWebSocket facade mirroring KalshiWebSocket: connect() -> _PerpsWebSocketSession async context manager, _start/_stop/recv-loop, on_state_change/on_error hooks, cooperative-pause recv loop, _PERMANENT_CLOSE_CODES fast-fail. Constructor (auth: KalshiAuth, config: PerpsConfig, heartbeat_timeout: float = 30.0, on_state_change=..., on_error=...). Per-channel subscribe_* convenience methods are deferred to the channels issue; this issue exposes the generic subscribe(channel, params) / unsubscribe / update_subscription / list_subscriptions surface only.

Pagination via list_all() / AsyncIterator is N/A (WS has no REST cursor).

Contract-test wiring

The AsyncAPI WS spec is not covered by METHOD_ENDPOINT_MAP / BODY_MODEL_MAP (those are REST-only). No new entries there. Wiring required for this issue:

  • Command-frame ↔ model parity test (new, perps-specific): add a lightweight test that loads /tmp/perps_asyncapi.yaml (committed to specs/perps_asyncapi.yaml by the foundation issue — flag this dependency) and asserts every components.messages.* command/response payload.$ref schema has a corresponding model in kalshi/perps/ws/models/control.py, and that the command cmd consts (subscribe, unsubscribe, update_subscription, list_subscriptions) and response type consts (subscribed, unsubscribed, ok, error) match the Literal values on the models. This is the WS analog of the REST drift test.
  • EXCLUSIONS: none in the REST EXCLUSIONS map. If the new WS parity test needs to skip the listSubscriptionsResponsePayload type: "ok" collision (shares type with okResponsePayload), encode the disambiguation (branch on msg array-vs-object) in the test rather than excluding it; document the rationale inline.

(No path_template/sdk_method/request_body_schema rows are added — those are REST constructs.)

Tests

New tests/perps/ws/ directory, mirroring tests/ws/. Use a fake/echo WS server or monkeypatched connect (mirror existing tests/ws/ fixtures); reuse conftest RSA fixtures for the auth signer.

  • PerpsConnectionManager: happy path (connect builds RSA-PSS headers, transitions DISCONNECTED→CONNECTING→CONNECTED); error path (connect failure → CLOSED, KalshiConnectionError with ws path but no query string); edge case (reconnect honors AWS full-jitter and fast-fails on _PERMANENT_CLOSE_CODES like 4001).
  • PerpsSubscriptionManager.subscribe: happy (sends cmd:"subscribe" with correct params, resolves on subscribed ack, installs sid→client mapping); error (error response → KalshiSubscriptionError); edge (orphan subscribed ack with no client mapping → auto-unsubscribe).
  • unsubscribe: happy (sends params.sids, server unsubscribed ack tears down state); error (unknown client_id); edge (server-initiated unsubscribe reaps seq + orderbook state).
  • update_subscription / update_subscription_single_sid: happy (each sends the right action and sid vs sids shape, resolves on ok); error (no active sub → KalshiSubscriptionError); edge (ok ack carrying msg.market_tickers).
  • list_subscriptions: happy (sends cmd:"list_subscriptions", parses array msg into SubscriptionEntry list); edge (empty list); error (error response).
  • Framing/parse: every response envelope round-trips through its model via respx-free direct model_validate; the type:"ok" array-vs-object disambiguation picks ListSubscriptionsResponse for arrays and OkResponse for objects.
  • Sequence reuse: a orderbook_delta/order_group_updates frame with a forward seq gap fires on_gap; non-sequenced perps channels (ticker/trade/fill/user_order) pass through untracked.
  • Orderbook seam: a bid/ask snapshot+delta applied through the parameterized manager produces the expected price-indexed book (guards against the yes/no hardcode regression).
  • Command models reject unknown keys (extra="forbid"): building SubscribeParams(bogus=1) raises.

Acceptance criteria

  • kalshi/perps/ws/ package created with connection.py, channels.py, dispatch.py, sequence.py (reusing/parameterizing SequenceTracker), orderbook.py (reusing/parameterizing OrderbookManager over side-name pair), backpressure.py (re-exporting MessageQueue/OverflowStrategy), client.py, and models/control.py.
  • PerpsWebSocket connects to the margin host with RSA-PSS apiKey auth and exposes generic subscribe/unsubscribe/update_subscription/update_subscription_single_sid/list_subscriptions.
  • Every command frame (subscribe, unsubscribe, update_subscription add/delete/single-sid, list_subscriptions) and every response/control frame (subscribed, unsubscribed, ok, list_subscriptions-as-ok-array, error, ping/pong) is built/parsed.
  • Command params models use model_config = {"extra": "forbid"} and serialize via model_dump(exclude_none=True, by_alias=True, mode="json").
  • Epoch-ms _ms-suffix timestamp convention documented in the models module; parse path stays int-typed (no RFC3339 coercion).
  • Orderbook bid/ask side handling verified (no yes/no hardcode leak); sequence tracking uses PERPS_SEQUENCED_CHANNELS.
  • PerpsWebSocket exported from kalshi/perps/ws/__init__.py → kalshi/perps/__init__.py → kalshi/__init__.py.
  • mypy strict clean (builtins.list[T] used inside PerpsSubscriptionManager); ruff clean.
  • pytest green including the new WS command/response parity test loading specs/perps_asyncapi.yaml.
  • AsyncAPI WS parity test added and passing (no silent schema drift between spec and models/control.py).

Dependencies

  • perps: foundation (PerpsClient/AsyncPerpsClient, PerpsConfig with margin REST + WS base URLs, reuse of KalshiAuth + transports, commit specs/perps_openapi.yaml and specs/perps_asyncapi.yaml, parameterize the contract-test harness to load the perps spec files)

Activity

  1. added this to the Perps API milestone on Jun 4, 2026
  2. added
    enhancementNew feature or request
    wsWebSocket-related
    perpsPerps / margin (perpetual futures) API
    on Jun 4, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or requestperpsPerps / margin (perpetual futures) APIwsWebSocket-related

    Projects

    No projects

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions