diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index 88bd004..f4be120 100644 --- a/kalshi/fix/__init__.py +++ b/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", ] diff --git a/kalshi/fix/client.py b/kalshi/fix/client.py index e9be68a..4b734f6 100644 --- a/kalshi/fix/client.py +++ b/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,13 +104,17 @@ 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( @@ -111,10 +122,14 @@ def drop_copy( *, 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( @@ -122,10 +137,14 @@ def market_data( *, 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,10 +185,14 @@ 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( @@ -177,8 +200,12 @@ def rfq( *, 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, ) diff --git a/kalshi/fix/errors.py b/kalshi/fix/errors.py index b7f40eb..463ce47 100644 --- a/kalshi/fix/errors.py +++ b/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. diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index bafbd53..7579b29 100644 --- a/kalshi/fix/messages/__init__.py +++ b/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", ] diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py index 50c4dcf..735045f 100644 --- a/kalshi/fix/messages/dispatch.py +++ b/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 diff --git a/kalshi/fix/session.py b/kalshi/fix/session.py index 2e097f1..5f5edb0 100644 --- a/kalshi/fix/session.py +++ b/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) diff --git a/tests/fix/test_decode_routing.py b/tests/fix/test_decode_routing.py new file mode 100644 index 0000000..196eed9 --- /dev/null +++ b/tests/fix/test_decode_routing.py @@ -0,0 +1,211 @@ +"""Tests for inbound decode-error routing (GH #432). + +``decode_app_message`` collapses "no registered model" and "registered but +malformed" to ``None``; ``decode_app_message_strict`` and ``FixSession``'s +``on_decode_error`` hook make the malformed case observable. +""" + +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.errors import FixDecodeError +from kalshi.fix.messages import ( + ExecutionReport, + decode_app_message, + decode_app_message_strict, +) +from kalshi.fix.session import FixSession +from kalshi.fix.tags import Tag + +from .conftest import MockAcceptor + + +def _exec_report_raw(*body: tuple[int, str]) -> RawMessage: + return decode( + encode( + [ + (int(Tag.MSG_TYPE), "8"), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + *body, + ] + ) + ) + + +# --------------------------------------------------------------------------- +# decode_app_message_strict vs decode_app_message +# --------------------------------------------------------------------------- + + +def test_strict_decode_valid_message() -> None: + msg = decode_app_message_strict(_exec_report_raw((int(Tag.AVG_PX), "0.66"))) + assert isinstance(msg, ExecutionReport) + assert msg.avg_px == Decimal("0.66") + + +def test_strict_decode_malformed_raises() -> None: + raw = _exec_report_raw((int(Tag.AVG_PX), "notanumber")) # bad Decimal + with pytest.raises(FixDecodeError) as ei: + decode_app_message_strict(raw) + assert ei.value.msg_type == "8" + assert ei.value.raw is raw + assert ei.value.__cause__ is not None # underlying validation error chained + + +def test_strict_decode_unregistered_returns_none() -> None: + # An admin / unregistered MsgType has no model — None, not raised. + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), "0"), # Heartbeat + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + ] + ) + ) + assert decode_app_message_strict(raw) is None + + +def test_decode_app_message_still_swallows_malformed() -> None: + # The lenient contract is unchanged: malformed -> None (logged, not raised). + assert decode_app_message(_exec_report_raw((int(Tag.AVG_PX), "notanumber"))) is None + + +# --------------------------------------------------------------------------- +# FixSession on_decode_error routing +# --------------------------------------------------------------------------- + + +async def test_session_routes_decode_error( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + received: list[RawMessage] = [] + errors: list[tuple[RawMessage, FixDecodeError]] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + errors.append((raw, exc)) + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(errors)) + err_raw, exc = errors[0] + assert isinstance(exc, FixDecodeError) + assert exc.msg_type == "8" + assert err_raw.msg_type == "8" + # The raw message is still delivered to on_message — nothing silently lost. + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + finally: + await session.close() + + +async def test_session_no_decode_error_on_valid_message( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + received: list[RawMessage] = [] + errors: list[tuple[RawMessage, FixDecodeError]] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + errors.append((raw, exc)) + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "0.66")], seq=2) + await until(lambda: bool(received)) + assert errors == [] # a valid ExecutionReport does not trigger the hook + finally: + await session.close() + + +async def test_session_survives_raising_decode_error_callback( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + # A buggy on_decode_error must not tear down the session — the raw is still + # delivered to on_message and the session stays active. + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + raise RuntimeError("buggy consumer hook") + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + assert session.state.value == "active" + finally: + await session.close() + + +async def test_session_without_hook_delivers_raw_on_malformed( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + # No on_decode_error: behaviour is unchanged — the raw still reaches + # on_message and the session survives the malformed payload. + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + session = FixSession( + fix_signer, fix_config, FixSessionType.ORDER_ENTRY_NR, on_message=on_message + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + assert session.state.value == "active" + finally: + await session.close()