From 44bc18b1c085c5c4380030031d31094a8a44d474 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Fri, 5 Jun 2026 20:54:35 -0500 Subject: [PATCH 1/3] feat(fix): market-data session + order-book reconstruction (closes #426) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Market data rides the dedicated KalshiMD session (port 8233, non-retransmission) and is identical on both products — only the price units differ (prediction dollars vs margin fixed-point dollars under UseDollars), and both ride the FIX Price field, so DollarDecimal parses either without float drift. NB: the bundled FIX dictionary v1.03 predates market data, so the message layouts were derived from docs.kalshi.com/fix/market-data (+ the margin variant), which carry the authoritative tag tables and wire examples. Messages (kalshi/fix/messages/market_data.py): - MarketDataRequest (35=V, outbound) with snapshot()/subscribe()/unsubscribe()/ unsubscribe_all() constructors that encode the 263/146/55 rules (cancel-all omits NoRelatedSym/Symbol) - SecurityStatusRequest (35=e, outbound) with subscribe()/unsubscribe() - MarketDataSnapshotFullRefresh (35=W), MarketDataIncrementalRefresh (35=X), MarketDataRequestReject (35=Y), SecurityStatus (35=f) inbound; code fields kept raw (compare against enums), matching the order-entry convention - new enums: SubscriptionRequestType, MDEntryType, MDUpdateAction, MDReqRejReason, SecurityTradingStatus; new market-data tags Order-book reconstruction (kalshi/fix/orderbook.py): - FixOrderBook builds aggregated bid/offer books from a W snapshot + X incrementals (absolute sizes: Change sets, Delete removes). Snapshot replaces a market's book; incrementals route per-entry by Symbol; an incremental for an un-seeded market is dropped. clear()/remove() support reconnect re-subscribe (KalshiMD has no retransmission, so a gap reconnects and re-snapshots). Inbound dispatch registers W/X/Y/f. The client.market_data() session factory and the MD host/port/non-retransmission config already existed from the foundation. Hardening from a pre-PR adversarial review: apply_incremental drops an unknown MDUpdateAction instead of applying it as a Change; apply_snapshot drops 0-size levels (parity with Delete); the per-market snapshot-replacement contract (caller must clear() before re-subscribing post-gap) is documented on the public API. 34 new tests: golden exact-wire fixtures + round-trips for all six messages, dispatch, order-book snapshot/Change/Delete/zero-size/empty-snapshot/unknown- action/before-snapshot/symbol-routing/stale-survival, a KalshiMD subscription integration test, and a drop-connection gap-recovery integration test (both against the mock acceptor). ruff + mypy --strict clean; 165 FIX tests pass. Closes #426. Part of #402. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/__init__.py | 32 ++ kalshi/fix/enums.py | 43 ++ kalshi/fix/messages/__init__.py | 20 + kalshi/fix/messages/dispatch.py | 10 + kalshi/fix/messages/market_data.py | 230 ++++++++++ kalshi/fix/orderbook.py | 190 ++++++++ kalshi/fix/tags.py | 10 + tests/fix/test_market_data.py | 670 +++++++++++++++++++++++++++++ 8 files changed, 1205 insertions(+) create mode 100644 kalshi/fix/messages/market_data.py create mode 100644 kalshi/fix/orderbook.py create mode 100644 tests/fix/test_market_data.py diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index aaf2d47..8911f44 100644 --- a/kalshi/fix/__init__.py +++ b/kalshi/fix/__init__.py @@ -47,12 +47,17 @@ EventResendRejectReason, ExecInst, ExecType, + MDEntryType, + MDReqRejReason, + MDUpdateAction, MsgType, OrdStatus, OrdType, + SecurityTradingStatus, SelfTradePreventionType, SessionRejectReason, Side, + SubscriptionRequestType, TimeInForce, ) from kalshi.fix.errors import ( @@ -78,7 +83,13 @@ Heartbeat, Logon, Logout, + MarketDataIncrementalRefresh, + MarketDataRequest, + MarketDataRequestReject, + MarketDataSnapshotFullRefresh, MarketSettlementParty, + MDIncrementalEntry, + MDSnapshotEntry, MiscFee, MultivariateSelectedLeg, NewOrderSingle, @@ -89,12 +100,16 @@ OrderMassCancelRequest, Party, Reject, + RelatedSymbol, ResendRequest, + SecurityStatus, + SecurityStatusRequest, SequenceReset, TestRequest, decode_app_message, groupfield, ) +from kalshi.fix.orderbook import BookLevel, FixOrderBook, MarketDataBook from kalshi.fix.session import FixSession, FixSessionState from kalshi.fix.tags import Tag @@ -104,6 +119,7 @@ "BEGIN_STRING_FIXT11", "SOH", "ApplVerID", + "BookLevel", "BusinessMessageReject", "CollateralAmountChange", "EncryptMethod", @@ -124,6 +140,7 @@ "FixGroupMeta", "FixLogonError", "FixMessage", + "FixOrderBook", "FixParser", "FixProduct", "FixRejectError", @@ -137,6 +154,16 @@ "KalshiFixError", "Logon", "Logout", + "MDEntryType", + "MDIncrementalEntry", + "MDReqRejReason", + "MDSnapshotEntry", + "MDUpdateAction", + "MarketDataBook", + "MarketDataIncrementalRefresh", + "MarketDataRequest", + "MarketDataRequestReject", + "MarketDataSnapshotFullRefresh", "MarketSettlementParty", "MiscFee", "MsgType", @@ -152,11 +179,16 @@ "Party", "RawMessage", "Reject", + "RelatedSymbol", "ResendRequest", + "SecurityStatus", + "SecurityStatusRequest", + "SecurityTradingStatus", "SelfTradePreventionType", "SequenceReset", "SessionRejectReason", "Side", + "SubscriptionRequestType", "Tag", "TestRequest", "TimeInForce", diff --git a/kalshi/fix/enums.py b/kalshi/fix/enums.py index 107fbe2..eb7ef2e 100644 --- a/kalshi/fix/enums.py +++ b/kalshi/fix/enums.py @@ -255,3 +255,46 @@ class EventResendRejectReason(IntEnum): SERVER_ERROR = 2 BEGIN_EXECID_TOO_SMALL = 3 END_EXECID_TOO_LARGE = 4 + + +class SubscriptionRequestType(StrEnum): + """Tag 263 — market-data / security-status subscription action. + + For ``MarketDataRequest`` (35=V): ``0``=snapshot, ``1``=snapshot plus updates, + ``2``=disable a previous subscription. For ``SecurityStatusRequest`` (35=e) + only ``1`` (subscribe) and ``2`` (unsubscribe) are valid. + """ + + SNAPSHOT = "0" + SNAPSHOT_PLUS_UPDATES = "1" + DISABLE = "2" + + +class MDEntryType(StrEnum): + """Tag 269 — side of a market-data book level.""" + + BID = "0" + OFFER = "1" + + +class MDUpdateAction(StrEnum): + """Tag 279 — action for an incremental market-data entry (35=X).""" + + CHANGE = "1" + DELETE = "2" + + +class MDReqRejReason(StrEnum): + """Tag 281 — why a MarketDataRequest (35=V) was rejected (35=Y).""" + + INSUFFICIENT_BANDWIDTH = "2" + UNSUPPORTED_SUBSCRIPTION_REQUEST_TYPE = "4" + + +class SecurityTradingStatus(IntEnum): + """Tag 326 — per-market trading-status change streamed via SecurityStatus (35=f).""" + + TRADING_HALT = 2 + RESUME = 3 + KALSHI_DETERMINED = 100 + KALSHI_SETTLED = 101 diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index 4b6fb2d..6413d3f 100644 --- a/kalshi/fix/messages/__init__.py +++ b/kalshi/fix/messages/__init__.py @@ -29,6 +29,17 @@ EventResendReject, EventResendRequest, ) +from kalshi.fix.messages.market_data import ( + MarketDataIncrementalRefresh, + MarketDataRequest, + MarketDataRequestReject, + MarketDataSnapshotFullRefresh, + MDIncrementalEntry, + MDSnapshotEntry, + RelatedSymbol, + SecurityStatus, + SecurityStatusRequest, +) from kalshi.fix.messages.order_entry import ( BusinessMessageReject, ExecutionReport, @@ -64,6 +75,12 @@ "Heartbeat", "Logon", "Logout", + "MDIncrementalEntry", + "MDSnapshotEntry", + "MarketDataIncrementalRefresh", + "MarketDataRequest", + "MarketDataRequestReject", + "MarketDataSnapshotFullRefresh", "MarketSettlementParty", "MiscFee", "MultivariateSelectedLeg", @@ -75,7 +92,10 @@ "OrderMassCancelRequest", "Party", "Reject", + "RelatedSymbol", "ResendRequest", + "SecurityStatus", + "SecurityStatusRequest", "SequenceReset", "TestRequest", "decode_app_message", diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py index 6b3d70a..bd336cf 100644 --- a/kalshi/fix/messages/dispatch.py +++ b/kalshi/fix/messages/dispatch.py @@ -18,6 +18,12 @@ from kalshi.fix.enums import MsgType from kalshi.fix.messages.base import FixMessage from kalshi.fix.messages.drop_copy import EventResendComplete, EventResendReject +from kalshi.fix.messages.market_data import ( + MarketDataIncrementalRefresh, + MarketDataRequestReject, + MarketDataSnapshotFullRefresh, + SecurityStatus, +) from kalshi.fix.messages.order_entry import ( BusinessMessageReject, ExecutionReport, @@ -36,6 +42,10 @@ MsgType.BUSINESS_MESSAGE_REJECT.value: BusinessMessageReject, MsgType.EVENT_RESEND_COMPLETE.value: EventResendComplete, MsgType.EVENT_RESEND_REJECT.value: EventResendReject, + MsgType.MARKET_DATA_SNAPSHOT_FULL_REFRESH.value: MarketDataSnapshotFullRefresh, + MsgType.MARKET_DATA_INCREMENTAL_REFRESH.value: MarketDataIncrementalRefresh, + MsgType.MARKET_DATA_REQUEST_REJECT.value: MarketDataRequestReject, + MsgType.SECURITY_STATUS.value: SecurityStatus, } ) diff --git a/kalshi/fix/messages/market_data.py b/kalshi/fix/messages/market_data.py new file mode 100644 index 0000000..63c27e2 --- /dev/null +++ b/kalshi/fix/messages/market_data.py @@ -0,0 +1,230 @@ +"""Market-data FIX messages (GH #426). + +Market data rides the dedicated ``KalshiMD`` session (port 8233). It is identical +on both products; only the price *units* differ — prediction quotes dollars +(e.g. ``0.3500``) and margin quotes fixed-point dollars under ``UseDollars`` — +and both ride the FIX ``Price`` field, so :data:`~kalshi.types.DollarDecimal` +parses either without float drift. + +Outbound (client -> ``KalshiMD``): + +* :class:`MarketDataRequest` (35=V) — request a book snapshot, a snapshot-plus- + updates subscription, or cancel a subscription. Build via the + :meth:`~MarketDataRequest.snapshot` / :meth:`~MarketDataRequest.subscribe` / + :meth:`~MarketDataRequest.unsubscribe` / :meth:`~MarketDataRequest.unsubscribe_all` + helpers, which encode the ``SubscriptionRequestType`` / ``NoRelatedSym`` rules. +* :class:`SecurityStatusRequest` (35=e) — subscribe/unsubscribe a single market's + trading-status stream. + +Inbound (``KalshiMD`` -> client; code fields kept raw for robustness, compare +against :mod:`kalshi.fix.enums`): + +* :class:`MarketDataSnapshotFullRefresh` (35=W) — the full aggregated book. +* :class:`MarketDataIncrementalRefresh` (35=X) — subsequent level changes. +* :class:`MarketDataRequestReject` (35=Y) — a request could not be accepted. +* :class:`SecurityStatus` (35=f) — a market's trading-status change. + +:class:`~kalshi.fix.orderbook.FixOrderBook` reconstructs a live book from a W +snapshot plus X incrementals. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import Annotated + +from kalshi.fix.enums import MsgType, SubscriptionRequestType +from kalshi.fix.messages.base import ( + FixGroup, + FixGroupMeta, + FixMessage, + FixType, + fixfield, + groupfield, +) +from kalshi.fix.tags import Tag +from kalshi.types import DollarDecimal, FixedPointCount + +# --------------------------------------------------------------------------- +# Repeating-group entries +# --------------------------------------------------------------------------- + + +class RelatedSymbol(FixGroup): + """One ``NoRelatedSym`` (146) entry — delimiter ``Symbol`` (55).""" + + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + + +class MDSnapshotEntry(FixGroup): + """One ``NoMDEntries`` (268) entry in a snapshot — delimiter ``MDEntryType`` (269). + + ``md_entry_type`` is a raw char (compare against + :class:`~kalshi.fix.enums.MDEntryType`: ``0``=bid, ``1``=offer). + """ + + md_entry_type: str = fixfield(Tag.MD_ENTRY_TYPE, FixType.CHAR) + md_entry_px: DollarDecimal = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE) + md_entry_size: FixedPointCount = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY) + + +class MDIncrementalEntry(FixGroup): + """One ``NoMDEntries`` (268) entry in an incremental — delimiter ``MDUpdateAction`` (279). + + ``md_update_action`` / ``md_entry_type`` are raw chars (compare against + :class:`~kalshi.fix.enums.MDUpdateAction` / ``MDEntryType``). ``md_entry_size`` + is the *new* absolute size at the level (``0`` when the level is deleted). + """ + + md_update_action: str = fixfield(Tag.MD_UPDATE_ACTION, FixType.CHAR) + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + md_entry_type: str = fixfield(Tag.MD_ENTRY_TYPE, FixType.CHAR) + md_entry_px: DollarDecimal = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE) + md_entry_size: FixedPointCount = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY) + + +# --------------------------------------------------------------------------- +# Outbound requests +# --------------------------------------------------------------------------- + + +class MarketDataRequest(FixMessage): + """MarketDataRequest (35=V) — request order-book snapshots / updates. + + Prefer the :meth:`snapshot` / :meth:`subscribe` / :meth:`unsubscribe` / + :meth:`unsubscribe_all` constructors: ``NoRelatedSym``/``Symbol`` are required + for snapshot and subscribe, and a cancel-all request omits them entirely. + """ + + MSG_TYPE = MsgType.MARKET_DATA_REQUEST + + subscription_request_type: SubscriptionRequestType = fixfield( + Tag.SUBSCRIPTION_REQUEST_TYPE, FixType.CHAR + ) + related_symbols: Annotated[ + list[RelatedSymbol], FixGroupMeta(Tag.NO_RELATED_SYM, RelatedSymbol) + ] = groupfield() + + @classmethod + def _for( + cls, request_type: SubscriptionRequestType, symbols: Iterable[str] + ) -> MarketDataRequest: + return cls( + subscription_request_type=request_type, + related_symbols=[RelatedSymbol(symbol=s) for s in symbols], + ) + + @classmethod + def snapshot(cls, symbols: Iterable[str]) -> MarketDataRequest: + """A one-shot snapshot request (263=0) for the given market tickers.""" + req = cls._for(SubscriptionRequestType.SNAPSHOT, symbols) + if not req.related_symbols: + raise ValueError("snapshot() requires at least one symbol") + return req + + @classmethod + def subscribe(cls, symbols: Iterable[str]) -> MarketDataRequest: + """A snapshot-plus-updates subscription (263=1) for the given tickers.""" + req = cls._for(SubscriptionRequestType.SNAPSHOT_PLUS_UPDATES, symbols) + if not req.related_symbols: + raise ValueError("subscribe() requires at least one symbol") + return req + + @classmethod + def unsubscribe(cls, symbols: Iterable[str]) -> MarketDataRequest: + """Cancel the subscriptions (263=2) for the given tickers.""" + req = cls._for(SubscriptionRequestType.DISABLE, symbols) + if not req.related_symbols: + raise ValueError("unsubscribe() requires at least one symbol; " + "use unsubscribe_all() to cancel everything") + return req + + @classmethod + def unsubscribe_all(cls) -> MarketDataRequest: + """Cancel every subscription on the session (263=2 with no symbols).""" + return cls(subscription_request_type=SubscriptionRequestType.DISABLE) + + +class SecurityStatusRequest(FixMessage): + """SecurityStatusRequest (35=e) — subscribe/unsubscribe a market's status stream.""" + + MSG_TYPE = MsgType.SECURITY_STATUS_REQUEST + + subscription_request_type: SubscriptionRequestType = fixfield( + Tag.SUBSCRIPTION_REQUEST_TYPE, FixType.CHAR + ) + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + + @classmethod + def subscribe(cls, symbol: str) -> SecurityStatusRequest: + """Subscribe (263=1) to ``symbol``'s trading-status changes.""" + return cls( + subscription_request_type=SubscriptionRequestType.SNAPSHOT_PLUS_UPDATES, + symbol=symbol, + ) + + @classmethod + def unsubscribe(cls, symbol: str) -> SecurityStatusRequest: + """Unsubscribe (263=2) from ``symbol``'s trading-status changes.""" + return cls(subscription_request_type=SubscriptionRequestType.DISABLE, symbol=symbol) + + +# --------------------------------------------------------------------------- +# Inbound messages (fields optional; codes raw for robustness) +# --------------------------------------------------------------------------- + + +class MarketDataSnapshotFullRefresh(FixMessage): + """MarketDataSnapshotFullRefresh (35=W) — the full aggregated book for a market. + + Correlate by ``symbol``. An empty ``entries`` list is a valid empty book + (the server returns one for a symbol it has no order book for). + """ + + MSG_TYPE = MsgType.MARKET_DATA_SNAPSHOT_FULL_REFRESH + + symbol: str | None = fixfield(Tag.SYMBOL, FixType.STRING, default=None) + entries: Annotated[ + list[MDSnapshotEntry], FixGroupMeta(Tag.NO_MD_ENTRIES, MDSnapshotEntry) + ] = groupfield() + + +class MarketDataIncrementalRefresh(FixMessage): + """MarketDataIncrementalRefresh (35=X) — changed book levels. + + Each entry carries its own ``symbol`` (the group spans markets). + """ + + MSG_TYPE = MsgType.MARKET_DATA_INCREMENTAL_REFRESH + + entries: Annotated[ + list[MDIncrementalEntry], FixGroupMeta(Tag.NO_MD_ENTRIES, MDIncrementalEntry) + ] = groupfield() + + +class MarketDataRequestReject(FixMessage): + """MarketDataRequestReject (35=Y) — a MarketDataRequest could not be accepted. + + ``md_req_rej_reason`` is a raw char (compare against + :class:`~kalshi.fix.enums.MDReqRejReason`). + """ + + MSG_TYPE = MsgType.MARKET_DATA_REQUEST_REJECT + + md_req_rej_reason: str | None = fixfield(Tag.MD_REQ_REJ_REASON, FixType.CHAR, default=None) + text: str | None = fixfield(Tag.TEXT, FixType.STRING, default=None) + + +class SecurityStatus(FixMessage): + """SecurityStatus (35=f) — a market's trading-status change. + + ``security_trading_status`` is a raw int (compare against + :class:`~kalshi.fix.enums.SecurityTradingStatus`). + """ + + MSG_TYPE = MsgType.SECURITY_STATUS + + symbol: str | None = fixfield(Tag.SYMBOL, FixType.STRING, default=None) + security_trading_status: int | None = fixfield( + Tag.SECURITY_TRADING_STATUS, FixType.INT, default=None + ) diff --git a/kalshi/fix/orderbook.py b/kalshi/fix/orderbook.py new file mode 100644 index 0000000..9574c83 --- /dev/null +++ b/kalshi/fix/orderbook.py @@ -0,0 +1,190 @@ +"""Local order-book reconstruction from FIX market data (GH #426). + +:class:`FixOrderBook` maintains aggregated per-market books from a +:class:`~kalshi.fix.messages.market_data.MarketDataSnapshotFullRefresh` (35=W) +plus :class:`~kalshi.fix.messages.market_data.MarketDataIncrementalRefresh` +(35=X) stream. Unlike the WebSocket book (signed yes/no *deltas*), FIX market +data is bid/offer with *absolute* level sizes: a snapshot replaces a market's +book, an incremental ``Change`` sets a level's new size, and ``Delete`` removes +the level. + +Levels are price-indexed (``dict[Decimal, Decimal]``) so an incremental update is +O(1); the sorted public view is materialized lazily in :meth:`get`. + +Gap recovery: ``KalshiMD`` does not support retransmission, so a sequence gap +tears the session down and the client reconnects and re-subscribes. Snapshot +replacement is **per market** (keyed on ``Symbol``), so the caller MUST call +:meth:`clear` before re-subscribing after any teardown/gap: a market that was +seeded before the gap but is not immediately re-snapshotted (a different/smaller +re-subscribe set, or a read in the window before its fresh snapshot arrives) +would otherwise keep serving its stale pre-gap book. An incremental for a market +with no snapshot yet is dropped (the book is not seeded). +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from decimal import Decimal + +from kalshi.fix.enums import MDEntryType, MDUpdateAction +from kalshi.fix.messages.market_data import ( + MarketDataIncrementalRefresh, + MarketDataSnapshotFullRefresh, +) + +logger = logging.getLogger("kalshi.fix") + + +@dataclass(frozen=True) +class BookLevel: + """One aggregated price level: a price and its total size.""" + + price: Decimal + size: Decimal + + +@dataclass(frozen=True) +class MarketDataBook: + """An immutable view of one market's book. + + ``bids`` are ordered best-first (price descending) and ``offers`` best-first + (price ascending). + """ + + symbol: str + bids: tuple[BookLevel, ...] + offers: tuple[BookLevel, ...] + + +@dataclass +class _BookState: + """Mutable per-market state: price-indexed bid/offer levels.""" + + symbol: str + bids: dict[Decimal, Decimal] = field(default_factory=dict) + offers: dict[Decimal, Decimal] = field(default_factory=dict) + + def side(self, md_entry_type: str) -> dict[Decimal, Decimal] | None: + """The level dict for a wire ``MDEntryType`` char, or ``None`` if unknown.""" + if md_entry_type == MDEntryType.BID.value: + return self.bids + if md_entry_type == MDEntryType.OFFER.value: + return self.offers + return None + + def to_view(self) -> MarketDataBook: + return MarketDataBook( + symbol=self.symbol, + bids=tuple( + BookLevel(price=p, size=s) for p, s in sorted(self.bids.items(), reverse=True) + ), + offers=tuple(BookLevel(price=p, size=s) for p, s in sorted(self.offers.items())), + ) + + +class FixOrderBook: + """Reconstructs aggregated market books from FIX W snapshots + X incrementals. + + Usage:: + + book = FixOrderBook() + book.apply(decode_app_message(raw)) # W or X (others ignored) + view = book.get("KXNBAGAME-...") # MarketDataBook | None + """ + + def __init__(self) -> None: + self._books: dict[str, _BookState] = {} + + def apply_snapshot(self, msg: MarketDataSnapshotFullRefresh) -> None: + """Replace a market's book from a full snapshot (35=W). + + A snapshot without a ``symbol`` is ignored (nothing to key on). Unknown + ``MDEntryType`` entries are skipped, as are non-positive-size levels (a + ``0`` size means "no level" — parity with an incremental ``Delete``). An + empty snapshot is a valid empty book. + """ + symbol = msg.symbol + if symbol is None: + logger.warning("FIX MD snapshot without Symbol; ignoring") + return + state = _BookState(symbol=symbol) + for entry in msg.entries: + levels = state.side(entry.md_entry_type) + if levels is None: + logger.debug( + "FIX MD snapshot %s: unknown MDEntryType %r", symbol, entry.md_entry_type + ) + continue + if entry.md_entry_size <= 0: + continue + levels[entry.md_entry_px] = entry.md_entry_size + self._books[symbol] = state + + def apply_incremental(self, msg: MarketDataIncrementalRefresh) -> int: + """Apply incremental level changes (35=X). Returns the number applied. + + Each entry routes by its own ``symbol``. ``Change`` sets the level's new + absolute size (a non-positive size removes the level); ``Delete`` removes + it. An entry for a market with no snapshot yet, or with an unknown + ``MDUpdateAction``/``MDEntryType``, is skipped without mutating the book. + """ + applied = 0 + for entry in msg.entries: + state = self._books.get(entry.symbol) + if state is None: + logger.warning( + "FIX MD incremental for %s with no snapshot yet; dropping entry", entry.symbol + ) + continue + levels = state.side(entry.md_entry_type) + if levels is None: + logger.debug( + "FIX MD incremental %s: unknown MDEntryType %r", + entry.symbol, + entry.md_entry_type, + ) + continue + action = entry.md_update_action + if action == MDUpdateAction.DELETE.value or entry.md_entry_size <= 0: + levels.pop(entry.md_entry_px, None) + elif action == MDUpdateAction.CHANGE.value: + levels[entry.md_entry_px] = entry.md_entry_size + else: + # Out-of-spec action: don't silently apply it as a Change. + logger.debug( + "FIX MD incremental %s: unknown MDUpdateAction %r; dropping entry", + entry.symbol, + action, + ) + continue + applied += 1 + return applied + + def apply(self, msg: object) -> None: + """Apply a decoded W or X message; anything else is ignored. + + Convenience for feeding :func:`~kalshi.fix.messages.decode_app_message` + output straight in without a type switch at the call site. + """ + if isinstance(msg, MarketDataSnapshotFullRefresh): + self.apply_snapshot(msg) + elif isinstance(msg, MarketDataIncrementalRefresh): + self.apply_incremental(msg) + + def get(self, symbol: str) -> MarketDataBook | None: + """The current book view for ``symbol``, or ``None`` if not seeded.""" + state = self._books.get(symbol) + return state.to_view() if state is not None else None + + def symbols(self) -> set[str]: + """The set of markets with a seeded book.""" + return set(self._books) + + def remove(self, symbol: str) -> None: + """Drop a market's book (e.g. on unsubscribe or settlement).""" + self._books.pop(symbol, None) + + def clear(self) -> None: + """Drop all books (e.g. before re-subscribing after a reconnect).""" + self._books.clear() diff --git a/kalshi/fix/tags.py b/kalshi/fix/tags.py index 5435ffe..7aa505e 100644 --- a/kalshi/fix/tags.py +++ b/kalshi/fix/tags.py @@ -189,6 +189,16 @@ class Tag(IntEnum): RESEND_EVENT_COUNT = 21003 EVENT_RESEND_REJECT_REASON = 21004 + # --- Market data --- (NoRelatedSym/Symbol/Text reuse the tags above) + SUBSCRIPTION_REQUEST_TYPE = 263 + NO_MD_ENTRIES = 268 + MD_ENTRY_TYPE = 269 + MD_ENTRY_PX = 270 + MD_ENTRY_SIZE = 271 + MD_UPDATE_ACTION = 279 + MD_REQ_REJ_REASON = 281 + SECURITY_TRADING_STATUS = 326 + # Length-prefixed data fields: ``length_tag -> data_tag``. The data field's value # may legally contain the SOH (\x01) delimiter, so the decoder reads exactly diff --git a/tests/fix/test_market_data.py b/tests/fix/test_market_data.py new file mode 100644 index 0000000..b315e34 --- /dev/null +++ b/tests/fix/test_market_data.py @@ -0,0 +1,670 @@ +"""Tests for the market-data FIX flow + order-book reconstruction (GH #426).""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from decimal import Decimal + +import pytest + +from kalshi.fix.auth import FixSigner +from kalshi.fix.codec import RawMessage, decode, encode +from kalshi.fix.config import FixConfig, FixSessionType +from kalshi.fix.enums import ( + MDEntryType, + MDReqRejReason, + MDUpdateAction, + SecurityTradingStatus, + SubscriptionRequestType, +) +from kalshi.fix.messages import ( + MarketDataIncrementalRefresh, + MarketDataRequest, + MarketDataRequestReject, + MarketDataSnapshotFullRefresh, + MDIncrementalEntry, + MDSnapshotEntry, + SecurityStatus, + SecurityStatusRequest, + decode_app_message, +) +from kalshi.fix.messages.base import FixMessage +from kalshi.fix.orderbook import FixOrderBook +from kalshi.fix.session import FixSession, FixSessionState +from kalshi.fix.tags import Tag + +from .conftest import MockAcceptor + +TICKER = "KXNBAGAME-26MAY25NYKCLE-NYK" +TICKER2 = "KXTEST-OTHER" + + +def _roundtrip(msg: FixMessage) -> FixMessage: + full = [ + (int(Tag.MSG_TYPE), msg.MSG_TYPE.value), + (int(Tag.MSG_SEQ_NUM), "1"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + *msg.to_body_fields(), + ] + return type(msg).from_raw(decode(encode(full))) + + +def _wire(msg: FixMessage) -> str: + """The message body as a ``tag=value|...`` string (delimiter rendered ``|``). + + Pins exact field order and repeating-group layout — which round-trip equality + cannot, since a symmetric encode/decode bug would pass a round-trip. + """ + return "|".join(f"{t}={v}" for t, v in msg.to_body_fields()) + + +# --------------------------------------------------------------------------- +# Golden wire fixtures (pinned to the docs.kalshi.com/fix/market-data examples) +# --------------------------------------------------------------------------- + + +def test_golden_wire_market_data_request() -> None: + # docs market-data.md: snapshot request body, and cancel-all body. + assert _wire(MarketDataRequest.snapshot([TICKER])) == f"263=0|146=1|55={TICKER}" + assert _wire(MarketDataRequest.unsubscribe_all()) == "263=2" + + +def test_golden_wire_snapshot_full_refresh() -> None: + msg = MarketDataSnapshotFullRefresh( + symbol=TICKER, + entries=[ + MDSnapshotEntry( + md_entry_type="0", md_entry_px=Decimal("0.3500"), md_entry_size=Decimal("10.00") + ), + MDSnapshotEntry( + md_entry_type="1", md_entry_px=Decimal("0.6500"), md_entry_size=Decimal("5.00") + ), + ], + ) + assert _wire(msg) == f"55={TICKER}|268=2|269=0|270=0.3500|271=10.00|269=1|270=0.6500|271=5.00" + + +def test_golden_wire_incremental_refresh() -> None: + msg = MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action="1", + symbol=TICKER, + md_entry_type="0", + md_entry_px=Decimal("0.3500"), + md_entry_size=Decimal("15.00"), + ) + ] + ) + assert _wire(msg) == f"268=1|279=1|55={TICKER}|269=0|270=0.3500|271=15.00" + + +def test_golden_wire_request_reject() -> None: + msg = MarketDataRequestReject(md_req_rej_reason="4", text="bad") + assert _wire(msg) == "281=4|58=bad" + + +def test_golden_wire_security_status_request() -> None: + assert _wire(SecurityStatusRequest.subscribe(TICKER)) == f"263=1|55={TICKER}" + + +def test_golden_wire_security_status() -> None: + assert _wire(SecurityStatus(symbol=TICKER, security_trading_status=2)) == f"55={TICKER}|326=2" + + +# --------------------------------------------------------------------------- +# Outbound: MarketDataRequest (35=V) + SecurityStatusRequest (35=e) +# --------------------------------------------------------------------------- + + +def test_market_data_request_snapshot() -> None: + msg = MarketDataRequest.snapshot([TICKER]) + body = msg.to_body_fields() + assert body[0] == (int(Tag.SUBSCRIPTION_REQUEST_TYPE), SubscriptionRequestType.SNAPSHOT.value) + assert (int(Tag.NO_RELATED_SYM), "1") in body + assert (int(Tag.SYMBOL), TICKER) in body + assert _roundtrip(msg) == msg + + +def test_market_data_request_subscribe_multi() -> None: + msg = MarketDataRequest.subscribe([TICKER, TICKER2]) + body = dict(msg.to_body_fields()) + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.SNAPSHOT_PLUS_UPDATES + assert body[int(Tag.NO_RELATED_SYM)] == "2" + rt = _roundtrip(msg) + assert isinstance(rt, MarketDataRequest) + assert [e.symbol for e in rt.related_symbols] == [TICKER, TICKER2] + + +def test_market_data_request_unsubscribe() -> None: + msg = MarketDataRequest.unsubscribe([TICKER]) + body = dict(msg.to_body_fields()) + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE.value + assert body[int(Tag.NO_RELATED_SYM)] == "1" + + +def test_market_data_request_unsubscribe_all_omits_symbols() -> None: + msg = MarketDataRequest.unsubscribe_all() + tags = {t for t, _ in msg.to_body_fields()} + assert int(Tag.SUBSCRIPTION_REQUEST_TYPE) in tags + # Cancel-all carries no NoRelatedSym / Symbol per the spec. + assert int(Tag.NO_RELATED_SYM) not in tags + assert int(Tag.SYMBOL) not in tags + + +@pytest.mark.parametrize("ctor", ["snapshot", "subscribe", "unsubscribe"]) +def test_market_data_request_requires_symbol(ctor: str) -> None: + with pytest.raises(ValueError, match="symbol"): + getattr(MarketDataRequest, ctor)([]) + + +def test_security_status_request_subscribe() -> None: + msg = SecurityStatusRequest.subscribe(TICKER) + body = dict(msg.to_body_fields()) + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.SNAPSHOT_PLUS_UPDATES + assert body[int(Tag.SYMBOL)] == TICKER + assert _roundtrip(msg) == msg + + +def test_security_status_request_unsubscribe() -> None: + msg = SecurityStatusRequest.unsubscribe(TICKER) + body = dict(msg.to_body_fields()) + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE.value + assert body[int(Tag.SYMBOL)] == TICKER + + +# --------------------------------------------------------------------------- +# Inbound: W / X / Y / f round-trips + dispatch +# --------------------------------------------------------------------------- + + +def test_snapshot_full_refresh_roundtrip() -> None: + msg = MarketDataSnapshotFullRefresh( + symbol=TICKER, + entries=[ + MDSnapshotEntry( + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.3500"), + md_entry_size=Decimal("10.00"), + ), + MDSnapshotEntry( + md_entry_type=MDEntryType.OFFER.value, + md_entry_px=Decimal("0.6500"), + md_entry_size=Decimal("5.00"), + ), + ], + ) + body = dict(msg.to_body_fields()) + assert body[int(Tag.SYMBOL)] == TICKER + assert body[int(Tag.NO_MD_ENTRIES)] == "2" + assert _roundtrip(msg) == msg + + +def test_incremental_refresh_roundtrip() -> None: + msg = MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.3500"), + md_entry_size=Decimal("15.00"), + ), + MDIncrementalEntry( + md_update_action=MDUpdateAction.DELETE.value, + symbol=TICKER, + md_entry_type=MDEntryType.OFFER.value, + md_entry_px=Decimal("0.6500"), + md_entry_size=Decimal("0.00"), + ), + ], + ) + rt = _roundtrip(msg) + assert rt == msg + assert isinstance(rt, MarketDataIncrementalRefresh) + assert [e.symbol for e in rt.entries] == [TICKER, TICKER] + + +def test_market_data_request_reject_roundtrip() -> None: + msg = MarketDataRequestReject( + md_req_rej_reason=MDReqRejReason.UNSUPPORTED_SUBSCRIPTION_REQUEST_TYPE.value, + text="bad request type", + ) + back = _roundtrip(msg) + assert back == msg + assert isinstance(back, MarketDataRequestReject) + assert back.md_req_rej_reason == MDReqRejReason.UNSUPPORTED_SUBSCRIPTION_REQUEST_TYPE + + +def test_security_status_roundtrip() -> None: + msg = SecurityStatus(symbol=TICKER, security_trading_status=SecurityTradingStatus.TRADING_HALT) + back = _roundtrip(msg) + assert back == msg + assert isinstance(back, SecurityStatus) + assert back.security_trading_status == SecurityTradingStatus.TRADING_HALT + + +def test_decode_app_message_md_types() -> None: + for msg in ( + MarketDataSnapshotFullRefresh(symbol=TICKER), + MarketDataIncrementalRefresh(), + MarketDataRequestReject(md_req_rej_reason="2"), + SecurityStatus(symbol=TICKER, security_trading_status=3), + ): + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), msg.MSG_TYPE.value), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + *msg.to_body_fields(), + ] + ) + ) + decoded = decode_app_message(raw) + assert type(decoded) is type(msg) + assert decoded == msg + + +# --------------------------------------------------------------------------- +# Order-book reconstruction +# --------------------------------------------------------------------------- + + +def _snapshot_for(symbol: str, *entries: tuple[str, str, str]) -> MarketDataSnapshotFullRefresh: + return MarketDataSnapshotFullRefresh( + symbol=symbol, + entries=[ + MDSnapshotEntry(md_entry_type=t, md_entry_px=Decimal(px), md_entry_size=Decimal(sz)) + for t, px, sz in entries + ], + ) + + +def _snapshot(*entries: tuple[str, str, str]) -> MarketDataSnapshotFullRefresh: + return _snapshot_for(TICKER, *entries) + + +def _incr(symbol: str, action: str, side: str, px: str, sz: str) -> MarketDataIncrementalRefresh: + return MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=action, + symbol=symbol, + md_entry_type=side, + md_entry_px=Decimal(px), + md_entry_size=Decimal(sz), + ) + ] + ) + + +def test_orderbook_snapshot_orders_levels() -> None: + book = FixOrderBook() + book.apply_snapshot( + _snapshot( + (MDEntryType.BID.value, "0.30", "10"), + (MDEntryType.BID.value, "0.35", "20"), + (MDEntryType.OFFER.value, "0.70", "5"), + (MDEntryType.OFFER.value, "0.65", "8"), + ) + ) + view = book.get(TICKER) + assert view is not None + # Bids best-first (descending), offers best-first (ascending). + assert [(lvl.price, lvl.size) for lvl in view.bids] == [ + (Decimal("0.35"), Decimal("20")), + (Decimal("0.30"), Decimal("10")), + ] + assert [(lvl.price, lvl.size) for lvl in view.offers] == [ + (Decimal("0.65"), Decimal("8")), + (Decimal("0.70"), Decimal("5")), + ] + + +def test_orderbook_incremental_change_and_delete() -> None: + book = FixOrderBook() + book.apply_snapshot( + _snapshot( + (MDEntryType.BID.value, "0.35", "20"), + (MDEntryType.OFFER.value, "0.65", "8"), + ) + ) + applied = book.apply_incremental( + MarketDataIncrementalRefresh( + entries=[ + # Change the bid size. + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.35"), + md_entry_size=Decimal("25"), + ), + # Add a new bid level. + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.34"), + md_entry_size=Decimal("12"), + ), + # Delete the offer level. + MDIncrementalEntry( + md_update_action=MDUpdateAction.DELETE.value, + symbol=TICKER, + md_entry_type=MDEntryType.OFFER.value, + md_entry_px=Decimal("0.65"), + md_entry_size=Decimal("0"), + ), + ] + ) + ) + assert applied == 3 + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [ + (Decimal("0.35"), Decimal("25")), + (Decimal("0.34"), Decimal("12")), + ] + assert view.offers == () + + +def test_orderbook_change_to_zero_removes_level() -> None: + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + book.apply_incremental( + MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.35"), + md_entry_size=Decimal("0"), + ) + ] + ) + ) + view = book.get(TICKER) + assert view is not None + assert view.bids == () + + +def test_orderbook_incremental_before_snapshot_dropped() -> None: + book = FixOrderBook() + applied = book.apply_incremental( + MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.35"), + md_entry_size=Decimal("20"), + ) + ] + ) + ) + assert applied == 0 + assert book.get(TICKER) is None + + +def test_orderbook_snapshot_replaces_stale_book() -> None: + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + book.apply_incremental( + MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.34"), + md_entry_size=Decimal("99"), + ) + ] + ) + ) + # Gap recovery: clear + fresh snapshot fully replaces prior state. + book.clear() + assert book.get(TICKER) is None + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.40", "7"))) + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [(Decimal("0.40"), Decimal("7"))] + + +def test_orderbook_incremental_routes_by_symbol() -> None: + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) # TICKER only + applied = book.apply_incremental( + MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.36"), + md_entry_size=Decimal("3"), + ), + # Entry for an un-seeded symbol is dropped. + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER2, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.10"), + md_entry_size=Decimal("1"), + ), + ] + ) + ) + assert applied == 1 + assert book.symbols() == {TICKER} + + +def test_orderbook_apply_ignores_non_md_message() -> None: + book = FixOrderBook() + book.apply(SecurityStatus(symbol=TICKER, security_trading_status=2)) + book.apply(MarketDataRequestReject(md_req_rej_reason="2")) + assert book.symbols() == set() + + +def test_orderbook_snapshot_drops_zero_size_levels() -> None: + # A 0-size level in a snapshot is "no level" (parity with an incremental Delete). + book = FixOrderBook() + book.apply_snapshot( + _snapshot((MDEntryType.BID.value, "0.35", "0"), (MDEntryType.BID.value, "0.30", "10")) + ) + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [(Decimal("0.30"), Decimal("10"))] + + +def test_orderbook_empty_snapshot_clears_book() -> None: + # The server sends an empty snapshot for a market it has no book for. + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + book.apply_snapshot(MarketDataSnapshotFullRefresh(symbol=TICKER)) + view = book.get(TICKER) + assert view is not None + assert view.bids == () + assert view.offers == () + assert TICKER in book.symbols() # still seeded, just empty + + +def test_orderbook_incremental_unknown_action_dropped() -> None: + # An out-of-spec MDUpdateAction must not be silently applied as a Change. + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + applied = book.apply_incremental(_incr(TICKER, "9", MDEntryType.BID.value, "0.35", "5")) + assert applied == 0 + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [(Decimal("0.35"), Decimal("20"))] + + +def test_orderbook_resubscribe_without_clear_leaves_stale_book() -> None: + # Snapshot replacement is per-market: re-subscribing to a subset after a gap + # WITHOUT clear() leaves stale books for markets not re-seeded — clear() is the + # documented remedy. + book = FixOrderBook() + book.apply_snapshot(_snapshot_for(TICKER, (MDEntryType.BID.value, "0.35", "20"))) + book.apply_snapshot(_snapshot_for(TICKER2, (MDEntryType.BID.value, "0.40", "5"))) + # Re-snapshot only TICKER (as if re-subscribing to a smaller set post-gap). + book.apply_snapshot(_snapshot_for(TICKER, (MDEntryType.BID.value, "0.36", "9"))) + assert book.get(TICKER2) is not None # stale book survives — hazard the contract warns of + + book.clear() + book.apply_snapshot(_snapshot_for(TICKER, (MDEntryType.BID.value, "0.36", "9"))) + assert book.get(TICKER2) is None + assert book.get(TICKER) is not None + + +def test_orderbook_apply_dispatches_snapshot_and_incremental() -> None: + book = FixOrderBook() + book.apply(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + book.apply( + MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.35"), + md_entry_size=Decimal("21"), + ) + ] + ) + ) + view = book.get(TICKER) + assert view is not None + assert view.bids[0].size == Decimal("21") + + +# --------------------------------------------------------------------------- +# Session integration against the mock acceptor +# --------------------------------------------------------------------------- + + +async def test_market_data_subscription_flow( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + session = FixSession( + fix_signer, fix_config, FixSessionType.MARKET_DATA, on_message=on_message + ) + await session.start() + book = FixOrderBook() + try: + await session.send(MarketDataRequest.subscribe([TICKER])) + await until(lambda: acceptor.first("V") is not None) + req = acceptor.first("V") + assert req is not None + assert req.get(Tag.SUBSCRIPTION_REQUEST_TYPE) == ( + SubscriptionRequestType.SNAPSHOT_PLUS_UPDATES + ) + assert req.get(Tag.SYMBOL) == TICKER + + snapshot = MarketDataSnapshotFullRefresh( + symbol=TICKER, + entries=[ + MDSnapshotEntry( + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.35"), + md_entry_size=Decimal("10"), + ), + MDSnapshotEntry( + md_entry_type=MDEntryType.OFFER.value, + md_entry_px=Decimal("0.65"), + md_entry_size=Decimal("5"), + ), + ], + ) + incremental = MarketDataIncrementalRefresh( + entries=[ + MDIncrementalEntry( + md_update_action=MDUpdateAction.CHANGE.value, + symbol=TICKER, + md_entry_type=MDEntryType.BID.value, + md_entry_px=Decimal("0.36"), + md_entry_size=Decimal("4"), + ) + ] + ) + await acceptor.push("W", snapshot.to_body_fields(), seq=2) + await acceptor.push("X", incremental.to_body_fields(), seq=3) + await until(lambda: len(received) == 2) + + for raw in received: + book.apply(decode_app_message(raw)) + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [ + (Decimal("0.36"), Decimal("4")), + (Decimal("0.35"), Decimal("10")), + ] + assert [(lvl.price, lvl.size) for lvl in view.offers] == [(Decimal("0.65"), Decimal("5"))] + finally: + await session.close() + + +async def test_market_data_gap_recovery_resubscribe( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + """A dropped KalshiMD connection reconnects; clear() + re-subscribe rebuilds. + + KalshiMD has no retransmission, so a gap tears the session down and the + client reconnects (fresh logon, ResetSeqNumFlag=Y). The caller clears the + stale book and re-subscribes; the fresh snapshot rebuilds it. + """ + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + session = FixSession( + fix_signer, fix_config, FixSessionType.MARKET_DATA, on_message=on_message + ) + await session.start() + book = FixOrderBook() + try: + await session.send(MarketDataRequest.subscribe([TICKER])) + await until(lambda: acceptor.first("V") is not None) + snap1 = _snapshot((MDEntryType.BID.value, "0.35", "20")) + await acceptor.push("W", snap1.to_body_fields(), seq=2) + await until(lambda: len(received) == 1) + book.apply(decode_app_message(received[0])) + assert book.get(TICKER) is not None + + # Gap: drop the connection. The MD session reconnects and re-logs-on. + acceptor.drop_connection() + await until( + lambda: acceptor.connection_count >= 2 and session.state is FixSessionState.ACTIVE + ) + + # Caller's reconnect handling: clear stale state, then re-subscribe. + book.clear() + assert book.get(TICKER) is None + await session.send(MarketDataRequest.subscribe([TICKER])) + # Fresh snapshot on the new connection (seq reset to 1 by ResetSeqNumFlag=Y, + # so the first server app message after the logon response is seq 2). + snap2 = _snapshot((MDEntryType.BID.value, "0.40", "7")) + await acceptor.push("W", snap2.to_body_fields(), seq=2) + await until(lambda: len(received) == 2) + book.apply(decode_app_message(received[1])) + view = book.get(TICKER) + assert view is not None + assert [(lvl.price, lvl.size) for lvl in view.bids] == [(Decimal("0.40"), Decimal("7"))] + finally: + await session.close() From cd9822f67f82b28f152b0ebec5d13d07dac875b1 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 06:22:38 -0500 Subject: [PATCH 2/3] =?UTF-8?q?fix(fix):=20address=20#434=20review=20?= =?UTF-8?q?=E2=80=94=20MD=20incremental=20action=20ordering=20+=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - orderbook.apply_incremental: check MDUpdateAction BEFORE the size guard. An unknown action carrying size 0 was hitting the leading `size <= 0 -> pop` branch and silently deleting the level instead of being dropped — the same silent-mutation class the unknown-action branch was added to prevent, just via Delete. Now: Delete -> pop; Change -> (size<=0 ? pop : set); unknown -> drop. Regression test parametrized over size 5 and 0. - Trim the market_data.py module docstring to match the sibling FIX modules (drop the per-message table + code block; the PR/docs carry that). Not changed, with rationale: - MDReqID (262): intentionally omitted, now documented on MarketDataRequest. The Kalshi MD docs do not define tag 262 on 35=V/W/X/Y (0 occurrences in the dictionary; the docs state subscriptions/rejects correlate by Symbol<55>), unlike generic FIX 5.0SP2. Adding it would emit an undefined tag. (FIX messages are not part of METHOD_ENDPOINT_MAP, so the EXCLUSIONS mechanism does not apply.) - f-string prefix nit: false positive — TICKER is interpolated, so the f-prefix is required; ruff (F541) is clean. ruff + mypy --strict clean; 166 FIX tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/messages/market_data.py | 40 +++++++++++------------------- kalshi/fix/orderbook.py | 11 +++++--- tests/fix/test_market_data.py | 8 +++--- 3 files changed, 27 insertions(+), 32 deletions(-) diff --git a/kalshi/fix/messages/market_data.py b/kalshi/fix/messages/market_data.py index 63c27e2..dec67ff 100644 --- a/kalshi/fix/messages/market_data.py +++ b/kalshi/fix/messages/market_data.py @@ -1,31 +1,15 @@ """Market-data FIX messages (GH #426). -Market data rides the dedicated ``KalshiMD`` session (port 8233). It is identical -on both products; only the price *units* differ — prediction quotes dollars -(e.g. ``0.3500``) and margin quotes fixed-point dollars under ``UseDollars`` — -and both ride the FIX ``Price`` field, so :data:`~kalshi.types.DollarDecimal` -parses either without float drift. - -Outbound (client -> ``KalshiMD``): - -* :class:`MarketDataRequest` (35=V) — request a book snapshot, a snapshot-plus- - updates subscription, or cancel a subscription. Build via the - :meth:`~MarketDataRequest.snapshot` / :meth:`~MarketDataRequest.subscribe` / - :meth:`~MarketDataRequest.unsubscribe` / :meth:`~MarketDataRequest.unsubscribe_all` - helpers, which encode the ``SubscriptionRequestType`` / ``NoRelatedSym`` rules. -* :class:`SecurityStatusRequest` (35=e) — subscribe/unsubscribe a single market's - trading-status stream. - -Inbound (``KalshiMD`` -> client; code fields kept raw for robustness, compare -against :mod:`kalshi.fix.enums`): - -* :class:`MarketDataSnapshotFullRefresh` (35=W) — the full aggregated book. -* :class:`MarketDataIncrementalRefresh` (35=X) — subsequent level changes. -* :class:`MarketDataRequestReject` (35=Y) — a request could not be accepted. -* :class:`SecurityStatus` (35=f) — a market's trading-status change. - -:class:`~kalshi.fix.orderbook.FixOrderBook` reconstructs a live book from a W -snapshot plus X incrementals. +Market data rides the dedicated ``KalshiMD`` session and is identical on both +products; only the price units differ (prediction dollars vs margin fixed-point +dollars under ``UseDollars``), and both ride the FIX ``Price`` field, so +:data:`~kalshi.types.DollarDecimal` parses either without float drift. + +Outbound: :class:`MarketDataRequest` (35=V), :class:`SecurityStatusRequest` +(35=e). Inbound (code fields kept raw — compare against :mod:`kalshi.fix.enums`): +:class:`MarketDataSnapshotFullRefresh` (35=W), :class:`MarketDataIncrementalRefresh` +(35=X), :class:`MarketDataRequestReject` (35=Y), :class:`SecurityStatus` (35=f). +:class:`~kalshi.fix.orderbook.FixOrderBook` reconstructs a book from W + X. """ from __future__ import annotations @@ -94,6 +78,10 @@ class MarketDataRequest(FixMessage): Prefer the :meth:`snapshot` / :meth:`subscribe` / :meth:`unsubscribe` / :meth:`unsubscribe_all` constructors: ``NoRelatedSym``/``Symbol`` are required for snapshot and subscribe, and a cancel-all request omits them entirely. + + ``MDReqID`` (262) is intentionally omitted: the Kalshi MD docs do not define + it on 35=V/W/X/Y (subscriptions and rejects correlate by ``Symbol``, not a + request id), unlike generic FIX 5.0SP2. """ MSG_TYPE = MsgType.MARKET_DATA_REQUEST diff --git a/kalshi/fix/orderbook.py b/kalshi/fix/orderbook.py index 9574c83..6d0544f 100644 --- a/kalshi/fix/orderbook.py +++ b/kalshi/fix/orderbook.py @@ -145,13 +145,18 @@ def apply_incremental(self, msg: MarketDataIncrementalRefresh) -> int: entry.md_entry_type, ) continue + # Check the action FIRST: an unknown action carrying size 0 must be + # dropped, not routed into Delete by a leading size guard. action = entry.md_update_action - if action == MDUpdateAction.DELETE.value or entry.md_entry_size <= 0: + if action == MDUpdateAction.DELETE.value: levels.pop(entry.md_entry_px, None) elif action == MDUpdateAction.CHANGE.value: - levels[entry.md_entry_px] = entry.md_entry_size + if entry.md_entry_size <= 0: + levels.pop(entry.md_entry_px, None) # a 0-size Change clears the level + else: + levels[entry.md_entry_px] = entry.md_entry_size else: - # Out-of-spec action: don't silently apply it as a Change. + # Out-of-spec action: don't silently mutate the book. logger.debug( "FIX MD incremental %s: unknown MDUpdateAction %r; dropping entry", entry.symbol, diff --git a/tests/fix/test_market_data.py b/tests/fix/test_market_data.py index b315e34..d791833 100644 --- a/tests/fix/test_market_data.py +++ b/tests/fix/test_market_data.py @@ -493,11 +493,13 @@ def test_orderbook_empty_snapshot_clears_book() -> None: assert TICKER in book.symbols() # still seeded, just empty -def test_orderbook_incremental_unknown_action_dropped() -> None: - # An out-of-spec MDUpdateAction must not be silently applied as a Change. +@pytest.mark.parametrize("size", ["5", "0"]) +def test_orderbook_incremental_unknown_action_dropped(size: str) -> None: + # An out-of-spec MDUpdateAction must not mutate the book — neither applied as + # a Change (size>0) nor routed into Delete by a leading size guard (size=0). book = FixOrderBook() book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) - applied = book.apply_incremental(_incr(TICKER, "9", MDEntryType.BID.value, "0.35", "5")) + applied = book.apply_incremental(_incr(TICKER, "9", MDEntryType.BID.value, "0.35", size)) assert applied == 0 view = book.get(TICKER) assert view is not None From eeeef4b1fade12bc65b44ce94975489884be9d40 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 06:32:13 -0500 Subject: [PATCH 3/3] =?UTF-8?q?fix(fix):=20address=20#434=20review=20round?= =?UTF-8?q?=202=20=E2=80=94=20inbound=20MD=20entry=20robustness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MDSnapshotEntry / MDIncrementalEntry: make md_entry_px / md_entry_size optional (default=None), matching the order-entry inbound convention (parse a slightly off-spec server entry rather than rejecting the whole message). FIX permits a Delete to omit MDEntryPx/MDEntrySize; the parser now tolerates it. - orderbook: handle the now-optional fields — apply_snapshot skips a level lacking price or positive size; apply_incremental drops an entry without MDEntryPx, and a Change with absent/0 size clears the level. Regression test decodes an X Delete that omits tag 271 and confirms it parses + removes. - enums: document that MDUpdateAction intentionally omits FIX New=0 (Kalshi MD doesn't emit it; action='0' is dropped as unknown). - tests: add the missing subscribe (263=1) golden-wire assertion; make the unsubscribe assertions enum-direct (drop .value) for consistency. Not changed, with rationale: - (GH #426) module-docstring tags: kept. The cited CLAUDE.md rule is not present in the repo's CLAUDE.md/AGENTS.md, and all sibling FIX modules (order_entry #424, drop_copy #425, components #423) carry the same provenance tag. - to_view() per-get sort: left as-is (books are tiny; no speculative cache), consistent with the reviewer flagging it as non-blocking. ruff + mypy --strict clean; 167 FIX tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/enums.py | 6 ++++- kalshi/fix/messages/market_data.py | 14 +++++++----- kalshi/fix/orderbook.py | 19 +++++++++++----- tests/fix/test_market_data.py | 36 +++++++++++++++++++++++++++--- 4 files changed, 61 insertions(+), 14 deletions(-) diff --git a/kalshi/fix/enums.py b/kalshi/fix/enums.py index eb7ef2e..44a3b5b 100644 --- a/kalshi/fix/enums.py +++ b/kalshi/fix/enums.py @@ -278,7 +278,11 @@ class MDEntryType(StrEnum): class MDUpdateAction(StrEnum): - """Tag 279 — action for an incremental market-data entry (35=X).""" + """Tag 279 — action for an incremental market-data entry (35=X). + + FIX 5.0SP2 also defines New=0; Kalshi MD does not emit it, so it is + intentionally absent (an action='0' is dropped as unknown). + """ CHANGE = "1" DELETE = "2" diff --git a/kalshi/fix/messages/market_data.py b/kalshi/fix/messages/market_data.py index dec67ff..51d1b80 100644 --- a/kalshi/fix/messages/market_data.py +++ b/kalshi/fix/messages/market_data.py @@ -44,12 +44,14 @@ class MDSnapshotEntry(FixGroup): """One ``NoMDEntries`` (268) entry in a snapshot — delimiter ``MDEntryType`` (269). ``md_entry_type`` is a raw char (compare against - :class:`~kalshi.fix.enums.MDEntryType`: ``0``=bid, ``1``=offer). + :class:`~kalshi.fix.enums.MDEntryType`: ``0``=bid, ``1``=offer). Price/size + are optional for inbound robustness (parse an off-spec entry rather than + rejecting the whole message), matching the order-entry inbound convention. """ md_entry_type: str = fixfield(Tag.MD_ENTRY_TYPE, FixType.CHAR) - md_entry_px: DollarDecimal = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE) - md_entry_size: FixedPointCount = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY) + md_entry_px: DollarDecimal | None = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE, default=None) + md_entry_size: FixedPointCount | None = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY, default=None) class MDIncrementalEntry(FixGroup): @@ -58,13 +60,15 @@ class MDIncrementalEntry(FixGroup): ``md_update_action`` / ``md_entry_type`` are raw chars (compare against :class:`~kalshi.fix.enums.MDUpdateAction` / ``MDEntryType``). ``md_entry_size`` is the *new* absolute size at the level (``0`` when the level is deleted). + Price/size are optional for inbound robustness (a ``Delete`` that omits them + still parses), matching the order-entry inbound convention. """ md_update_action: str = fixfield(Tag.MD_UPDATE_ACTION, FixType.CHAR) symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) md_entry_type: str = fixfield(Tag.MD_ENTRY_TYPE, FixType.CHAR) - md_entry_px: DollarDecimal = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE) - md_entry_size: FixedPointCount = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY) + md_entry_px: DollarDecimal | None = fixfield(Tag.MD_ENTRY_PX, FixType.PRICE, default=None) + md_entry_size: FixedPointCount | None = fixfield(Tag.MD_ENTRY_SIZE, FixType.QTY, default=None) # --------------------------------------------------------------------------- diff --git a/kalshi/fix/orderbook.py b/kalshi/fix/orderbook.py index 6d0544f..66ad63f 100644 --- a/kalshi/fix/orderbook.py +++ b/kalshi/fix/orderbook.py @@ -116,7 +116,9 @@ def apply_snapshot(self, msg: MarketDataSnapshotFullRefresh) -> None: "FIX MD snapshot %s: unknown MDEntryType %r", symbol, entry.md_entry_type ) continue - if entry.md_entry_size <= 0: + # A level needs a price and a positive size; a 0/absent size is "no + # level" (parity with an incremental Delete). + if entry.md_entry_px is None or entry.md_entry_size is None or entry.md_entry_size <= 0: continue levels[entry.md_entry_px] = entry.md_entry_size self._books[symbol] = state @@ -145,16 +147,23 @@ def apply_incremental(self, msg: MarketDataIncrementalRefresh) -> int: entry.md_entry_type, ) continue + px = entry.md_entry_px + if px is None: + logger.debug( + "FIX MD incremental %s: entry without MDEntryPx; dropping", entry.symbol + ) + continue # Check the action FIRST: an unknown action carrying size 0 must be # dropped, not routed into Delete by a leading size guard. action = entry.md_update_action if action == MDUpdateAction.DELETE.value: - levels.pop(entry.md_entry_px, None) + levels.pop(px, None) elif action == MDUpdateAction.CHANGE.value: - if entry.md_entry_size <= 0: - levels.pop(entry.md_entry_px, None) # a 0-size Change clears the level + size = entry.md_entry_size + if size is None or size <= 0: + levels.pop(px, None) # absent / 0 size clears the level else: - levels[entry.md_entry_px] = entry.md_entry_size + levels[px] = size else: # Out-of-spec action: don't silently mutate the book. logger.debug( diff --git a/tests/fix/test_market_data.py b/tests/fix/test_market_data.py index d791833..913728b 100644 --- a/tests/fix/test_market_data.py +++ b/tests/fix/test_market_data.py @@ -64,8 +64,9 @@ def _wire(msg: FixMessage) -> str: def test_golden_wire_market_data_request() -> None: - # docs market-data.md: snapshot request body, and cancel-all body. + # docs market-data.md: snapshot, subscribe, and cancel-all request bodies. assert _wire(MarketDataRequest.snapshot([TICKER])) == f"263=0|146=1|55={TICKER}" + assert _wire(MarketDataRequest.subscribe([TICKER])) == f"263=1|146=1|55={TICKER}" assert _wire(MarketDataRequest.unsubscribe_all()) == "263=2" @@ -139,7 +140,7 @@ def test_market_data_request_subscribe_multi() -> None: def test_market_data_request_unsubscribe() -> None: msg = MarketDataRequest.unsubscribe([TICKER]) body = dict(msg.to_body_fields()) - assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE.value + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE assert body[int(Tag.NO_RELATED_SYM)] == "1" @@ -169,7 +170,7 @@ def test_security_status_request_subscribe() -> None: def test_security_status_request_unsubscribe() -> None: msg = SecurityStatusRequest.unsubscribe(TICKER) body = dict(msg.to_body_fields()) - assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE.value + assert body[int(Tag.SUBSCRIPTION_REQUEST_TYPE)] == SubscriptionRequestType.DISABLE assert body[int(Tag.SYMBOL)] == TICKER @@ -506,6 +507,35 @@ def test_orderbook_incremental_unknown_action_dropped(size: str) -> None: assert [(lvl.price, lvl.size) for lvl in view.bids] == [(Decimal("0.35"), Decimal("20"))] +def test_orderbook_incremental_delete_without_size_parses_and_applies() -> None: + # FIX permits a Delete to omit MDEntrySize(271); the inbound entry must parse + # (price/size optional) and the level must still be removed. + book = FixOrderBook() + book.apply_snapshot(_snapshot((MDEntryType.BID.value, "0.35", "20"))) + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), "X"), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + (int(Tag.NO_MD_ENTRIES), "1"), + (int(Tag.MD_UPDATE_ACTION), MDUpdateAction.DELETE.value), + (int(Tag.SYMBOL), TICKER), + (int(Tag.MD_ENTRY_TYPE), MDEntryType.BID.value), + (int(Tag.MD_ENTRY_PX), "0.35"), + # MDEntrySize (271) intentionally omitted. + ] + ) + ) + msg = decode_app_message(raw) + assert isinstance(msg, MarketDataIncrementalRefresh) + assert msg.entries[0].md_entry_size is None + assert book.apply_incremental(msg) == 1 + view = book.get(TICKER) + assert view is not None + assert view.bids == () + + def test_orderbook_resubscribe_without_clear_leaves_stale_book() -> None: # Snapshot replacement is per-market: re-subscribing to a subset after a gap # WITHOUT clear() leaves stale books for markets not re-seeded — clear() is the