feat(fix): surface inbound decode failures (closes #432)

kalshi/fix/__init__.py

@@ -73,6 +73,7 @@
 from kalshi.fix.errors import (
     FixCodecError,
     FixConnectionError,
+    FixDecodeError,
     FixLogonError,
     FixRejectError,
     FixSequenceError,
@@ -133,6 +134,7 @@
     SequenceReset,
     TestRequest,
     decode_app_message,
+    decode_app_message_strict,
     groupfield,
 )
 from kalshi.fix.orderbook import BookLevel, FixOrderBook, MarketDataBook
@@ -166,6 +168,7 @@
     "FixConfig",
     "FixConnection",
     "FixConnectionError",
+    "FixDecodeError",
     "FixEnvironment",
     "FixGroup",
     "FixGroupMeta",
@@ -248,6 +251,7 @@
     "TimeInForce",
     "decode",
     "decode_app_message",
+    "decode_app_message_strict",
     "encode",
     "groupfield",
 ]

kalshi/fix/client.py

@@ -15,7 +15,12 @@
 from kalshi.auth import KalshiAuth
 from kalshi.fix.auth import FixSigner
 from kalshi.fix.config import FixConfig, FixEnvironment, FixProduct, FixSessionType
-from kalshi.fix.session import FixSession, MessageHandler, StateChangeHandler
+from kalshi.fix.session import (
+    DecodeErrorHandler,
+    FixSession,
+    MessageHandler,
+    StateChangeHandler,
+)
 
 
 class _BaseFixClient:
@@ -80,6 +85,7 @@ def session(
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """Construct (but do not start) a :class:`FixSession` for ``session_type``."""
         return FixSession(
@@ -88,6 +94,7 @@ def session(
             session_type,
             on_message=on_message,
             on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
             ssl_context=self._ssl_context,
         )
 
@@ -97,35 +104,47 @@ def order_entry(
         retransmission: bool = False,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """An order-entry session (KalshiNR, or KalshiRT when ``retransmission``)."""
         session_type = (
             FixSessionType.ORDER_ENTRY_RT if retransmission else FixSessionType.ORDER_ENTRY_NR
         )
         return self.session(
-            session_type, on_message=on_message, on_state_change=on_state_change
+            session_type,
+            on_message=on_message,
+            on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
         )
 
     def drop_copy(
         self,
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """A drop-copy (KalshiDC) session for historical execution-report queries."""
         return self.session(
-            FixSessionType.DROP_COPY, on_message=on_message, on_state_change=on_state_change
+            FixSessionType.DROP_COPY,
+            on_message=on_message,
+            on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
         )
 
     def market_data(
         self,
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """A market-data (KalshiMD) session for order-book snapshots and updates."""
         return self.session(
-            FixSessionType.MARKET_DATA, on_message=on_message, on_state_change=on_state_change
+            FixSessionType.MARKET_DATA,
+            on_message=on_message,
+            on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
         )
 
 
@@ -166,19 +185,27 @@ def post_trade(
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """A post-trade (KalshiPT) session for market-settlement reports."""
         return self.session(
-            FixSessionType.POST_TRADE, on_message=on_message, on_state_change=on_state_change
+            FixSessionType.POST_TRADE,
+            on_message=on_message,
+            on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
         )
 
     def rfq(
         self,
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
     ) -> FixSession:
         """An RFQ (KalshiRFQ) market-maker session."""
         return self.session(
-            FixSessionType.RFQ, on_message=on_message, on_state_change=on_state_change
+            FixSessionType.RFQ,
+            on_message=on_message,
+            on_state_change=on_state_change,
+            on_decode_error=on_decode_error,
         )

kalshi/fix/errors.py

@@ -9,8 +9,13 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from kalshi.errors import KalshiError
 
+if TYPE_CHECKING:
+    from kalshi.fix.codec import RawMessage
+
 
 class KalshiFixError(KalshiError):
     """Base exception for all Kalshi FIX errors.
@@ -95,6 +100,25 @@ def __init__(
         super().__init__(message)
 
 
+class FixDecodeError(KalshiFixError):
+    """A registered inbound application message failed schema validation.
+
+    Distinguishes a *malformed* known message (a real message with an off-spec
+    field — e.g. a bad ``DollarDecimal`` / ``FixedPointCount``) from an
+    *unregistered* message type. :func:`~kalshi.fix.messages.decode_app_message`
+    collapses both to ``None``;
+    :func:`~kalshi.fix.messages.decode_app_message_strict` raises this for the
+    former so the failure is observable (see ``FixSession``'s ``on_decode_error``
+    hook). ``raw`` carries the offending message and ``msg_type`` its ``MsgType``
+    (both always present); the underlying validation error chains via ``__cause__``.
+    """
+
+    def __init__(self, message: str, *, raw: RawMessage, msg_type: str) -> None:
+        self.raw = raw
+        self.msg_type = msg_type
+        super().__init__(message)
+
+
 class FixRejectError(KalshiFixError):
     """The acceptor rejected a message we sent.
 

kalshi/fix/messages/__init__.py

@@ -23,7 +23,11 @@
     MultivariateSelectedLeg,
     Party,
 )
-from kalshi.fix.messages.dispatch import APP_MESSAGE_MODELS, decode_app_message
+from kalshi.fix.messages.dispatch import (
+    APP_MESSAGE_MODELS,
+    decode_app_message,
+    decode_app_message_strict,
+)
 from kalshi.fix.messages.drop_copy import (
     EventResendComplete,
     EventResendReject,
@@ -132,6 +136,7 @@
     "SequenceReset",
     "TestRequest",
     "decode_app_message",
+    "decode_app_message_strict",
     "fixfield",
     "groupfield",
 ]

kalshi/fix/messages/dispatch.py

@@ -16,6 +16,7 @@
 
 from kalshi.fix.codec import RawMessage
 from kalshi.fix.enums import MsgType
+from kalshi.fix.errors import FixDecodeError
 from kalshi.fix.messages.base import FixMessage
 from kalshi.fix.messages.drop_copy import EventResendComplete, EventResendReject
 from kalshi.fix.messages.market_data import (
@@ -77,21 +78,41 @@
 )
 
 
+def decode_app_message_strict(raw: RawMessage) -> FixMessage | None:
+    """Decode an inbound application :class:`RawMessage`, raising on a malformed one.
+
+    Returns ``None`` only for an *unregistered* message type (an admin message or
+    a not-yet-implemented flow). A registered model whose payload fails schema
+    validation raises :class:`~kalshi.fix.errors.FixDecodeError` (chaining the
+    underlying error) rather than returning ``None``, so a genuine message lost to
+    a single off-spec field is observable. Used by ``FixSession``'s
+    ``on_decode_error`` hook; direct callers wanting the distinction can call this.
+    """
+    mt = raw.msg_type or ""
+    model = APP_MESSAGE_MODELS.get(mt)
+    if model is None:
+        return None
+    try:
+        return model.from_raw(raw)
+    except (ValidationError, ValueError, ArithmeticError) as exc:
+        # ValueError: bad bool / int; ArithmeticError: bad Decimal (InvalidOperation).
+        # mt is the registered key here, so it is a real (non-empty) MsgType.
+        raise FixDecodeError(f"failed to decode inbound {mt}", raw=raw, msg_type=mt) from exc
+
+
 def decode_app_message(raw: RawMessage) -> FixMessage | None:
     """Decode an inbound application :class:`RawMessage` to its typed model.
 
     Returns ``None`` for message types without a registered model (an admin
     message or a not-yet-implemented application flow), and also ``None`` if the
     payload fails schema validation — a malformed inbound message is logged and
-    swallowed rather than raised into the consumer's ``on_message`` handler.
-    See GH #432 for surfacing decode failures to consumers.
+    swallowed rather than raised into the consumer's ``on_message`` handler. To
+    distinguish those two cases (e.g. to route a malformed known message to a
+    dead-letter), use :func:`decode_app_message_strict` or ``FixSession``'s
+    ``on_decode_error`` hook.
     """
-    model = APP_MESSAGE_MODELS.get(raw.msg_type or "")
-    if model is None:
-        return None
     try:
-        return model.from_raw(raw)
-    except (ValidationError, ValueError, ArithmeticError):
-        # ValueError: bad bool / int; ArithmeticError: bad Decimal (InvalidOperation).
+        return decode_app_message_strict(raw)
+    except FixDecodeError:
         logger.warning("failed to decode inbound %s; returning None", raw.msg_type, exc_info=True)
         return None

kalshi/fix/session.py

@@ -42,8 +42,15 @@
 from kalshi.fix.config import FixConfig, FixSessionType
 from kalshi.fix.connection import FixConnection
 from kalshi.fix.enums import MsgType
-from kalshi.fix.errors import FixConnectionError, FixLogonError, FixSequenceError, FixSessionError
+from kalshi.fix.errors import (
+    FixConnectionError,
+    FixDecodeError,
+    FixLogonError,
+    FixSequenceError,
+    FixSessionError,
+)
 from kalshi.fix.messages.base import FixMessage
+from kalshi.fix.messages.dispatch import decode_app_message_strict
 from kalshi.fix.messages.session import (
     Heartbeat,
     Logon,
@@ -59,6 +66,7 @@
 
 MessageHandler = Callable[[RawMessage], Awaitable[None]]
 StateChangeHandler = Callable[["FixSessionState", "FixSessionState"], Awaitable[None]]
+DecodeErrorHandler = Callable[[RawMessage, FixDecodeError], Awaitable[None]]
 
 
 def _heartbeat_due(now: float, last_send: float, interval: float) -> bool:
@@ -102,6 +110,16 @@ class FixSession:
                              on_message=handle)
         async with session:
             ...  # send order-entry messages, receive execution reports
+
+    ``on_message`` receives every inbound application message as a raw
+    :class:`~kalshi.fix.codec.RawMessage` (decode it with
+    :func:`~kalshi.fix.messages.decode_app_message`). Setting ``on_decode_error``
+    additionally routes a *registered-but-malformed* message (a real message lost
+    to one off-spec field) — but it makes the session run a full Pydantic
+    validation on **every** inbound application message to detect failures, so on a
+    high-rate session that is real per-message overhead (and the consumer's own
+    decode in ``on_message`` is then a second pass). Leave it unset unless you
+    need malformed messages surfaced; there is no cost when it is ``None``.
     """
 
     def __init__(
@@ -112,6 +130,7 @@ def __init__(
         *,
         on_message: MessageHandler | None = None,
         on_state_change: StateChangeHandler | None = None,
+        on_decode_error: DecodeErrorHandler | None = None,
         ssl_context: ssl.SSLContext | None = None,
     ) -> None:
         config._check_session(session_type)
@@ -122,6 +141,9 @@ def __init__(
         self._supports_retransmission = config.supports_retransmission(session_type)
         self._on_message = on_message
         self._on_state_change = on_state_change
+        # Routes registered-but-malformed inbound messages; see the class docstring
+        # for the per-message decode cost this opts into. GH #432.
+        self._on_decode_error = on_decode_error
         self._ssl_context = ssl_context
 
         self._hb = config.heartbeat_interval
@@ -619,6 +641,16 @@ async def _process(self, raw: RawMessage, mt: str | None) -> None:
         # Application message. Isolate consumer callback failures so a buggy
         # handler cannot tear down the session (the message's seq is already
         # accounted for by the caller).
+        if self._on_decode_error is not None:
+            # Detect a registered-but-malformed message and route it, so a fill
+            # lost to one off-spec field is observable rather than silently None.
+            try:
+                decode_app_message_strict(raw)
+            except FixDecodeError as exc:
+                try:
+                    await self._on_decode_error(raw, exc)
+                except Exception:
+                    logger.exception("FIX on_decode_error callback raised; continuing session")
         if self._on_message is not None:
             try:
                 await self._on_message(raw)