perps: WebSocket channels & payload models — orderbook_delta, ticker(+funding), trade, fill, user_orders, order_group_updates #398

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

Summary

Add Pydantic payload/envelope models and typed per-channel subscribe helpers for the six Kalshi Perps (margin) WebSocket data channels: orderbook_delta (snapshot + delta), ticker (carries perps-unique funding_rate + reference/mark tickerPrice objects), trade, fill, user_orders, and order_group_updates. These models are wire-distinct from the existing equities/event WS models — perps uses bid/ask book sides (not yes/no), single-sided price/bid/ask fields, and exclusively epoch-millisecond ts_ms timestamps — so they live under the new kalshi/perps/ws/ package rather than reusing kalshi/ws/models/.

This issue depends on the perps WS foundation (the PerpsWebSocket client, connection manager, auth, subscription manager, and dispatcher). It adds the six channel models, wires them into the dispatcher's type→model map, and adds the typed subscribe helpers on the perps WS client.

Endpoints / scope

Source: /tmp/perps_asyncapi.yaml (committed at specs/perps_asyncapi.yaml per the foundation issue). AsyncAPI v3.0.0, server 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).

Explicitly out of scope (document in code comments + PR body): the perps AsyncAPI defines only these six data channels. The equities WS channels market_positions, market_lifecycle_v2 / event_fee_update, multivariate, multivariate_market_lifecycle, and communications have no perps counterpart and MUST NOT get perps subscribe helpers. The foundation's per-channel param allow-set (_CHANNEL_PARAMS equivalent) must list only the six channels above.

Models

New file: kalshi/perps/ws/models/__init__.py (re-exports) plus one module per channel under kalshi/perps/ws/models/. Mirror kalshi/ws/models/ conventions: every model sets model_config = {"extra": "allow", "populate_by_name": True}; use DollarDecimal for dollar prices and FixedPointCount for _fp counts (both from kalshi.types); envelope models carry type (Literal[...]), sid: int, optional/required seq, and msg: <Payload>.

These are inbound/read models — extra="allow" per existing convention, NOT request bodies, so do NOT use extra="forbid" here. Subscribe-command payloads belong to the foundation issue.

Shared enums / literals (define once, e.g. kalshi/perps/ws/models/_common.py)

• PerpsBookSide = Literal["bid", "ask"] — from schema bookSide. Note the difference from equities (yes/no); do not reuse kalshi.models.orders.BookSideLiteral.
• PerpsSelfTradePreventionType = Literal["taker_at_cross", "maker"] — from selfTradePreventionType.
• PerpsLastUpdateReason = Literal["", "Decrease", "Amend", "MarginCancel", "SelfTradeCancel", "ExpiryCancel", "Trade", "PostOnlyCrossCancel"] — from lastUpdateReason.
• OrderGroupEventType = Literal["created", "triggered", "reset", "deleted", "limit_updated"] — from orderGroupUpdatesPayload.msg.event_type.

kalshi/perps/ws/models/orderbook.py

MarginOrderbookSnapshotPayload (from marginOrderbookSnapshotPayload.msg):

• market_ticker: str (required)
• bid, ask — arrays of priceLevelDollarsCountFp = [price_in_dollars, contract_count_fp] string pairs. Mirror kalshi/ws/models/orderbook_delta.py's PriceCountMap (Annotated[dict[Decimal, Decimal], BeforeValidator(_levels_to_dict)]) so each [price, count] pair collapses to a dict[Decimal, Decimal] in one walk. Per the perps spec these two arrays are NOT required on msg, so default each to an empty map (this differs from the equities #268 both-required rule — call it out in the docstring). Field aliasing: validation_alias accepting both bid/ask (wire names already match short names, so no _dollars/_fp suffix mismatch here).

MarginOrderbookDeltaPayload (from marginOrderbookDeltaPayload.msg):

• market_ticker: str, price: DollarDecimal (required), delta: FixedPointCount (required, may be negative), side: PerpsBookSide (required)
• last_update_reason: PerpsLastUpdateReason | None = None, client_order_id: str | None = None, subaccount: int | None = None, ts_ms: int | None = None (epoch ms; format: int64)

MarginOrderbookSnapshotMessage: type: Literal["orderbook_snapshot"], sid: int, seq: int (required), msg: MarginOrderbookSnapshotPayload.
MarginOrderbookDeltaMessage: type: Literal["orderbook_delta"], sid: int, seq: int (required), msg: MarginOrderbookDeltaPayload.

kalshi/perps/ws/models/ticker.py

FundingRate (from fundingRate schema — perps-unique):

• rate: MultiplierDecimal (wire is type: number, format: double — use MultiplierDecimal from kalshi.types to avoid binary-float drift per the #225 invariant, matching how Series.fee_multiplier is handled)
• next_funding_time_ms: int (epoch ms, format: int64), ts_ms: int (epoch ms)

TickerPrice (from tickerPrice schema):

• price: DollarDecimal (wire string, 4 decimals), ts_ms: int (epoch ms)

MarginTickerPayload (from marginTickerPayload.msg):

• market_ticker: str, price: DollarDecimal, bid: DollarDecimal, ask: DollarDecimal (all required; single-sided — note no yes_/no_ prefixes unlike equities)
• bid_size: FixedPointCount (alias bid_size_fp), ask_size: FixedPointCount (alias ask_size_fp), last_trade_size: FixedPointCount (alias last_trade_size_fp) — use validation_alias=AliasChoices("bid_size_fp", "bid_size") etc.
• volume: FixedPointCount, volume_24h: FixedPointCount, open_interest: FixedPointCount (wire names are bare strings, no _fp suffix per spec — but they're fixed-point counts, so type as FixedPointCount; no alias needed)
• reference_price: TickerPrice | None = None, settlement_mark_price: TickerPrice | None = None, liquidation_mark_price: TickerPrice | None = None, funding_rate: FundingRate | None = None
• ts_ms: int (required, epoch ms)

MarginTickerMessage: type: Literal["ticker"], sid: int, seq: int | None = None (NOT required per spec), msg: MarginTickerPayload.

kalshi/perps/ws/models/trade.py

MarginTradePayload (from marginTradePayload.msg, all listed required):

• trade_id: str (format: uuid), market_ticker: str, price: DollarDecimal, count: FixedPointCount, taker_side: PerpsBookSide, ts_ms: int

MarginTradeMessage: type: Literal["trade"], sid: int, seq: int | None = None, msg: MarginTradePayload.

kalshi/perps/ws/models/fill.py

MarginFillPayload (from marginFillPayload.msg):

• Required: trade_id: str, order_id: str, market_ticker: str, is_taker: bool, side: PerpsBookSide, ts_ms: int, price: DollarDecimal, count: FixedPointCount, fee_cost: DollarDecimal, post_position: FixedPointCount
• Optional: client_order_id: str | None = None, subaccount: int | None = None

MarginFillMessage: type: Literal["fill"], sid: int, seq: int | None = None, msg: MarginFillPayload.

kalshi/perps/ws/models/user_orders.py

MarginUserOrderPayload (from marginUserOrderPayload.msg):

• Required: order_id: str (uuid), user_id: str (uuid), client_order_id: str, ticker: str (note field is ticker, not market_ticker), side: PerpsBookSide, price: DollarDecimal, fill_count: FixedPointCount, remaining_count: FixedPointCount, created_ts_ms: int
• Optional: self_trade_prevention_type: PerpsSelfTradePreventionType | None = None, order_group_id: str | None = None, expiration_ts_ms: int | None = None, last_updated_ts_ms: int | None = None, subaccount_number: int | None = None

MarginUserOrderMessage: type: Literal["user_order"], sid: int, seq: int | None = None, msg: MarginUserOrderPayload.

kalshi/perps/ws/models/order_group.py

OrderGroupUpdatesPayload (from orderGroupUpdatesPayload.msg):

• Required: event_type: OrderGroupEventType, order_group_id: str, ts_ms: int (matching-engine epoch ms)
• Optional: contracts_limit: FixedPointCount | None = None — validation_alias=AliasChoices("contracts_limit_fp", "contracts_limit") (present only for created and limit_updated events)

OrderGroupUpdatesMessage: type: Literal["order_group_updates"], sid: int, seq: int (REQUIRED — sequenced channel, mirror equities OrderGroupMessage), msg: OrderGroupUpdatesPayload.

Resource methods

This is WS, not REST — the "resource methods" are typed subscribe helpers on the perps WS client. Add to the perps WS client created by the foundation issue: kalshi/perps/ws/client.py (PerpsWebSocket). Mirror the async-only signatures in kalshi/ws/client.py (lines ~760–829). Each returns an AsyncIterator[<Message>]; iterate via async for msg in await client.subscribe_*(...).

async def subscribe_orderbook_delta(
    self, *, tickers: builtins.list[str] | None = None, maxsize: int = 1000,
) -> AsyncIterator[MarginOrderbookSnapshotMessage | MarginOrderbookDeltaMessage]: ...
    # sets params["send_initial_snapshot"] = True; market_tickers from `tickers`;
    # overflow=OverflowStrategy.ERROR (gap-sensitive sequenced channel)

async def subscribe_ticker(
    self, *, tickers: builtins.list[str] | None = None,
    send_initial_snapshot: bool = False, maxsize: int = 1000,
) -> AsyncIterator[MarginTickerMessage]: ...
    # market_tickers from `tickers`; overflow=DROP_OLDEST

async def subscribe_trade(
    self, *, tickers: builtins.list[str] | None = None, maxsize: int = 1000,
) -> AsyncIterator[MarginTradeMessage]: ...   # overflow=DROP_OLDEST

async def subscribe_fill(
    self, *, tickers: builtins.list[str] | None = None, maxsize: int = 1000,
) -> AsyncIterator[MarginFillMessage]: ...    # overflow=DROP_OLDEST

async def subscribe_user_orders(
    self, *, tickers: builtins.list[str] | None = None, maxsize: int = 1000,
) -> AsyncIterator[MarginUserOrderMessage]: ...  # overflow=DROP_OLDEST

async def subscribe_order_group(
    self, *, maxsize: int = 1000,
) -> AsyncIterator[OrderGroupUpdatesMessage]: ...  # market spec ignored; overflow=ERROR

Use builtins.list[T] for the tickers param annotation inside the client class (the list builtin is shadowed by .list() elsewhere in resource classes; apply the same rule here for consistency). No list_all()/cursor pagination — WS streams are unbounded iterators, not paged.

Dispatcher wiring (in the foundation's perps dispatch module, mirror kalshi/ws/dispatch.py MESSAGE_MODELS): register the envelope type → model map:
"orderbook_snapshot" → MarginOrderbookSnapshotMessage, "orderbook_delta" → MarginOrderbookDeltaMessage, "ticker" → MarginTickerMessage, "trade" → MarginTradeMessage, "fill" → MarginFillMessage, "user_order" → MarginUserOrderMessage, "order_group_updates" → OrderGroupUpdatesMessage. The orderbook channel feeds the perps OrderbookManager; the sequenced channels (orderbook_delta, order_group_updates) participate in the seq-watermark / gap-recovery path.

Contract-test wiring

The Kalshi REST contract harness (METHOD_ENDPOINT_MAP / BODY_MODEL_MAP) does not cover WebSocket messages — there are no WS entries in those maps today, so no METHOD_ENDPOINT_MAP / BODY_MODEL_MAP / EXCLUSIONS rows are added by this issue.

Instead add a perps-WS payload-parity test that asserts each model parses the AsyncAPI example/required fields, loading the committed specs/perps_asyncapi.yaml:

• New tests/perps/ws/test_perps_ws_models.py. For each componentsmessages payload schema (marginOrderbookSnapshotPayload, marginOrderbookDeltaPayload, marginTickerPayload, marginTradePayload, marginFillPayload, marginUserOrderPayload, orderGroupUpdatesPayload), build a fixture frame from the schema's required fields + the spec's inline examples (e.g. the orderGroupLimitUpdated example at orderGroupUpdates message: {type: order_group_updates, sid: 21, seq: 7, msg: {event_type: limit_updated, order_group_id: og_123, contracts_limit_fp: "150.00"}}) and assert the corresponding Margin*Message model validates it and that decimals land as Decimal.
• A drift guard that fails if the AsyncAPI adds a required field on any owned payload schema that the model lacks (iterate required lists against model_fields) — mirrors the spirit of TestRequestParamDrift but for inbound WS payloads. Confirm the exact harness shape with the foundation issue (it commits the spec + the perps test package); if the foundation defines a shared WS-drift base, extend it rather than duplicating.

Tests

tests/perps/ws/ — reuse conftest RSA-key/auth/config fixtures; use respx.mock only where the foundation's connection mock needs HTTP. Per channel:

• subscribe_orderbook_delta — happy: snapshot frame yields MarginOrderbookSnapshotMessage with bid/ask as dict[Decimal, Decimal], then a delta frame with side="bid", negative delta; error: malformed snapshot (non-numeric level) raises ValidationError; edge: snapshot with bid/ask omitted → empty maps (perps allows partial book, unlike equities #268).
• subscribe_ticker — happy: ticker with funding_rate + all three tickerPrice mark fields parses; rate lands as Decimal (no float drift); edge: ticker with funding_rate/mark prices absent → all None; seq absent → None; error: missing required ts_ms raises.
• subscribe_trade — happy: taker_side="ask", count Decimal, ts_ms int; error: taker_side="yes" (equities value) rejected by PerpsBookSide; edge: trade_id uuid string accepted.
• subscribe_fill — happy: taker fill with fee_cost/post_position decimals; edge: optional client_order_id/subaccount omitted; error: missing post_position raises.
• subscribe_user_orders — happy: order with order_group_id + self_trade_prevention_type="maker"; edge: ticker field (not market_ticker) maps correctly; optional *_ts_ms omitted; error: missing required created_ts_ms raises.
• subscribe_order_group — happy: event_type="limit_updated" with contracts_limit_fp="150.00" → contracts_limit Decimal; edge: event_type="deleted" with contracts_limit absent → None; error: seq omitted raises (required on this channel); event_type="foo" rejected.
• Out-of-scope guard: assert PerpsWebSocket has no subscribe_market_positions / subscribe_multivariate / subscribe_communications / subscribe_market_lifecycle attributes.

Acceptance criteria

• Six payload modules created under kalshi/perps/ws/models/ with all envelope + payload models and shared enums above, matching spec field/required-ness exactly.
• Perps book sides typed Literal["bid", "ask"] (NOT reusing equities yes/no); all timestamps are ts_ms/*_ts_ms epoch-ms ints; funding_rate.rate uses MultiplierDecimal.
• Six typed subscribe_* helpers on PerpsWebSocket returning AsyncIterator[Margin*Message]; builtins.list[T] used for tickers params.
• Dispatcher type→model map registers all seven envelope types; sequenced channels (orderbook_delta, order_group_updates) require seq.
• Models exported from kalshi/perps/ws/models/__init__.py → kalshi/perps/__init__.py → kalshi/__init__.py (Margin*Message/*Payload, FundingRate, TickerPrice, and the four enums).
• uv run mypy kalshi/ strict clean (incl. builtins.list[T] inside the client class).
• uv run ruff check . clean.
• uv run pytest tests/perps/ws/ -v green, incl. the payload-parity + drift guard; full uv run pytest tests/ -v green (no regression to existing equities WS drift tests).
• Out-of-scope channels (market_positions, market_lifecycle_v2/event_fee_update, multivariate, multivariate_market_lifecycle, communications) documented as having no perps counterpart and given no subscribe helpers.

Dependencies

• perps: WebSocket foundation — PerpsWebSocket client, connection, auth, subscribe/dispatch machinery (provides the WS client, connection manager, subscription manager, dispatcher with the type→model registry, the perps WS base URL / PerpsConfig, reuse of KalshiAuth, the committed specs/perps_asyncapi.yaml, and the tests/perps/ws/ package + any shared WS-drift test base this issue extends).