Repository navigation
Description
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 viamodel_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 setssid; the array variant setssids.- Command wrappers
SubscribeCommand,UnsubscribeCommand,UpdateSubscriptionCommand,ListSubscriptionsCommandeach withid: int,cmd: Literal[...](serialization_alias-free; the const string matches the wire).ListSubscriptionsCommandhas noparams. Allextra="forbid".
- Response envelopes (response-side, validating):
SubscribedResponse—id: int | None,type: Literal["subscribed"],msg: SubscribedMsgwhereSubscribedMsghaschannel: 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 | NonewhereOkMsghasmarket_tickers: builtins.list[str] | None.ListSubscriptionsResponse—id: int,type: Literal["ok"],msg: builtins.list[SubscriptionEntry]whereSubscriptionEntryhaschannel: 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(fromsubscribeCommandPayload.params.channels.items.enum):orderbook_delta,ticker,trade,fill,user_orders,order_group_updates.UpdateSubscriptionAction(fromupdateSubscriptionCommandPayload.params.action.enum):add_markets,delete_markets.PerpsBookSide(frombookSide):bid,ask. Reuse note: the event-contractOrderbookManageris hardwired toyes/nosides — perps usesbid/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—PerpsConnectionManagermirroringConnectionManager. Same state machine (ConnectionState— reuse the existing enum fromkalshi.ws.connection, do not duplicate), same_open_socket/connect/reconnect/close/send/recv, same RSA-PSS_build_auth_headers(signGET+ ws path), same AWS-full-jitter reconnect. Readsws_base_url/ws_max_retries/retry_*/ws_ping_interval/ws_close_timeoutfromPerpsConfig(delivered by the foundation issue). Signature parity:async def connect(self) -> Noneasync def reconnect(self) -> Noneasync def close(self) -> Noneasync def send(self, msg: dict[str, Any]) -> Noneasync def recv(self) -> str
channels.py—PerpsSubscriptionManagermirroringSubscriptionManager, but with the perps command grammar:async def subscribe(self, channel: str, params: dict[str, Any] | None = None) -> int(returns client_id; sendscmd: "subscribe")async def unsubscribe(self, client_id: int) -> None(sendscmd: "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-sidsformasync 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-sidform (theupdateSubscriptionSingleSidCommandvariant)async def list_subscriptions(self) -> builtins.list[SubscriptionEntry](sendscmd: "list_subscriptions", parses the arraymsg)async def resubscribe_all(self) -> None- Inside this class annotate list returns/params with
builtins.list[T](thelistbuiltin is shadowed by.list()-style methods on the surrounding API).
dispatch.py—PerpsMessageDispatchermirroringMessageDispatcher.CONTROL_TYPES = {"subscribed", "unsubscribed", "ok", "error"}. Same orphan-subscribed / server-unsubscribe handling.MESSAGE_MODELSis left empty/stub here (populated by the channels issue) — this issue only needs control-frame routing + the "unknown message type" path.sequence.py— reusekalshi.ws.sequence.SequenceTracker+SequenceGapunchanged. Only the channel set differs: definePERPS_SEQUENCED_CHANNELS = {"orderbook_delta", "orderbook_snapshot", "order_group_updates"}(these are the perps payloads carryingseqper spec;unsubscribedalso carriesseqbut is a control frame). Pass the perps set into the tracker (parameterizeSEQUENCED_CHANNELSor subclass) rather than copying the class.orderbook.py— reuse-with-caveat. The existingOrderbookManagerhardcodesyes/nosides and consumesOrderbookSnapshotMessage/OrderbookDeltaMessagewhose sides areyes/no. Perps usesbid/ask(PerpsBookSide) andpriceLevelDollarsCountFplevels. Decision to make in this issue (flag explicitly): either (a) parameterizeOrderbookManagerover its two side names + level message types, or (b) fork aPerpsOrderbookManager. 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— reusekalshi.ws.backpressure.MessageQueue+OverflowStrategyverbatim (no perps-specific behavior).client.py—PerpsWebSocketfacade mirroringKalshiWebSocket:connect() -> _PerpsWebSocketSessionasync context manager,_start/_stop/recv-loop,on_state_change/on_errorhooks, cooperative-pause recv loop,_PERMANENT_CLOSE_CODESfast-fail. Constructor(auth: KalshiAuth, config: PerpsConfig, heartbeat_timeout: float = 30.0, on_state_change=..., on_error=...). Per-channelsubscribe_*convenience methods are deferred to the channels issue; this issue exposes the genericsubscribe(channel, params)/unsubscribe/update_subscription/list_subscriptionssurface 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 tospecs/perps_asyncapi.yamlby the foundation issue — flag this dependency) and asserts everycomponents.messages.*command/responsepayload.$refschema has a corresponding model inkalshi/perps/ws/models/control.py, and that the commandcmdconsts (subscribe,unsubscribe,update_subscription,list_subscriptions) and responsetypeconsts (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
EXCLUSIONSmap. If the new WS parity test needs to skip thelistSubscriptionsResponsePayloadtype: "ok"collision (sharestypewithokResponsePayload), encode the disambiguation (branch onmsgarray-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,KalshiConnectionErrorwith ws path but no query string); edge case (reconnect honors AWS full-jitter and fast-fails on_PERMANENT_CLOSE_CODESlike 4001).PerpsSubscriptionManager.subscribe: happy (sendscmd:"subscribe"with correctparams, resolves onsubscribedack, installs sid→client mapping); error (errorresponse →KalshiSubscriptionError); edge (orphansubscribedack with no client mapping → auto-unsubscribe).unsubscribe: happy (sendsparams.sids, serverunsubscribedack 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 rightactionandsidvssidsshape, resolves onok); error (no active sub →KalshiSubscriptionError); edge (okack carryingmsg.market_tickers).list_subscriptions: happy (sendscmd:"list_subscriptions", parses arraymsgintoSubscriptionEntrylist); edge (empty list); error (errorresponse).- Framing/parse: every response envelope round-trips through its model via
respx-free directmodel_validate; thetype:"ok"array-vs-object disambiguation picksListSubscriptionsResponsefor arrays andOkResponsefor objects. - Sequence reuse: a
orderbook_delta/order_group_updatesframe with a forward seq gap fireson_gap; non-sequenced perps channels (ticker/trade/fill/user_order) pass through untracked. - Orderbook seam: a
bid/asksnapshot+delta applied through the parameterized manager produces the expected price-indexed book (guards against theyes/nohardcode regression). - Command models reject unknown keys (
extra="forbid"): buildingSubscribeParams(bogus=1)raises.
Acceptance criteria
-
kalshi/perps/ws/package created withconnection.py,channels.py,dispatch.py,sequence.py(reusing/parameterizingSequenceTracker),orderbook.py(reusing/parameterizingOrderbookManagerover side-name pair),backpressure.py(re-exportingMessageQueue/OverflowStrategy),client.py, andmodels/control.py. -
PerpsWebSocketconnects to the margin host with RSA-PSS apiKey auth and exposes genericsubscribe/unsubscribe/update_subscription/update_subscription_single_sid/list_subscriptions. - Every command frame (
subscribe,unsubscribe,update_subscriptionadd/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 viamodel_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/askside handling verified (noyes/nohardcode leak); sequence tracking usesPERPS_SEQUENCED_CHANNELS. -
PerpsWebSocketexported fromkalshi/perps/ws/__init__.py→kalshi/perps/__init__.py→kalshi/__init__.py. - mypy strict clean (
builtins.list[T]used insidePerpsSubscriptionManager); ruff clean. -
pytestgreen including the new WS command/response parity test loadingspecs/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.yamlandspecs/perps_asyncapi.yaml, parameterize the contract-test harness to load the perps spec files)
Activity