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

[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)