feat(fix): drop-copy message flow + listener sessions (closes #425)

kalshi/fix/__init__.py

@@ -44,6 +44,7 @@
 from kalshi.fix.enums import (
     ApplVerID,
     EncryptMethod,
+    EventResendRejectReason,
     ExecInst,
     ExecType,
     MsgType,
@@ -67,6 +68,9 @@
     APP_MESSAGE_MODELS,
     BusinessMessageReject,
     CollateralAmountChange,
+    EventResendComplete,
+    EventResendReject,
+    EventResendRequest,
     ExecutionReport,
     FixGroup,
     FixGroupMeta,
@@ -103,6 +107,10 @@
     "BusinessMessageReject",
     "CollateralAmountChange",
     "EncryptMethod",
+    "EventResendComplete",
+    "EventResendReject",
+    "EventResendRejectReason",
+    "EventResendRequest",
     "ExecInst",
     "ExecType",
     "ExecutionReport",

kalshi/fix/config.py

@@ -136,6 +136,10 @@ class FixConfig:
     retry_base_delay: float = 0.5
     retry_max_delay: float = 30.0
     cancel_orders_on_disconnect: bool = False
+    # Listener (read-only streaming) session: receive execution reports without
+    # an order-entry capability. Valid on NR/RT and requires skip_pending_exec_reports.
+    listener_session: bool = False
+    skip_pending_exec_reports: bool = False
     # ``None`` derives from the product: margin always uses fixed-point dollars,
     # prediction defaults to integer cents (opt into dollars by setting True).
     use_dollars: bool | None = None
@@ -152,6 +156,11 @@ def __post_init__(self) -> None:
             )
         if self.port is not None and not (0 < self.port < 65536):
             raise ValueError(f"FixConfig.port must be 1..65535, got {self.port}")
+        if self.listener_session and not self.skip_pending_exec_reports:
+            raise ValueError(
+                "FixConfig.listener_session=True requires skip_pending_exec_reports=True "
+                "(per the Kalshi FIX spec)."
+            )
 
         allow_unknown = self.allow_unknown_host or (
             os.environ.get("KALSHI_FIX_ALLOW_UNKNOWN_HOST", "").strip() == "1"
@@ -196,6 +205,8 @@ def prediction(
         retry_base_delay: float = 0.5,
         retry_max_delay: float = 30.0,
         cancel_orders_on_disconnect: bool = False,
+        listener_session: bool = False,
+        skip_pending_exec_reports: bool = False,
         use_dollars: bool | None = None,
         allow_unknown_host: bool = False,
     ) -> FixConfig:
@@ -212,6 +223,8 @@ def prediction(
             retry_base_delay=retry_base_delay,
             retry_max_delay=retry_max_delay,
             cancel_orders_on_disconnect=cancel_orders_on_disconnect,
+            listener_session=listener_session,
+            skip_pending_exec_reports=skip_pending_exec_reports,
             use_dollars=use_dollars,
             allow_unknown_host=allow_unknown_host,
         )
@@ -230,6 +243,8 @@ def margin(
         retry_base_delay: float = 0.5,
         retry_max_delay: float = 30.0,
         cancel_orders_on_disconnect: bool = False,
+        listener_session: bool = False,
+        skip_pending_exec_reports: bool = False,
         use_dollars: bool = True,
         allow_unknown_host: bool = False,
     ) -> FixConfig:
@@ -246,6 +261,8 @@ def margin(
             retry_base_delay=retry_base_delay,
             retry_max_delay=retry_max_delay,
             cancel_orders_on_disconnect=cancel_orders_on_disconnect,
+            listener_session=listener_session,
+            skip_pending_exec_reports=skip_pending_exec_reports,
             use_dollars=use_dollars,
             allow_unknown_host=allow_unknown_host,
         )

kalshi/fix/enums.py

@@ -246,3 +246,12 @@ class PartyRole(IntEnum):
     """Tag 452 — role of a party in the NoPartyIDs group."""
 
     CUSTOMER_ACCOUNT = 24
+
+
+class EventResendRejectReason(IntEnum):
+    """Tag 21004 — why a drop-copy EventResendRequest (35=U3) was rejected."""
+
+    RATE_LIMITED = 1
+    SERVER_ERROR = 2
+    BEGIN_EXECID_TOO_SMALL = 3
+    END_EXECID_TOO_LARGE = 4

kalshi/fix/messages/__init__.py

@@ -23,8 +23,13 @@
     MultivariateSelectedLeg,
     Party,
 )
+from kalshi.fix.messages.dispatch import APP_MESSAGE_MODELS, decode_app_message
+from kalshi.fix.messages.drop_copy import (
+    EventResendComplete,
+    EventResendReject,
+    EventResendRequest,
+)
 from kalshi.fix.messages.order_entry import (
-    APP_MESSAGE_MODELS,
     BusinessMessageReject,
     ExecutionReport,
     NewOrderSingle,
@@ -33,7 +38,6 @@
     OrderCancelRequest,
     OrderMassCancelReport,
     OrderMassCancelRequest,
-    decode_app_message,
 )
 from kalshi.fix.messages.session import (
     Heartbeat,
@@ -49,6 +53,9 @@
     "APP_MESSAGE_MODELS",
     "BusinessMessageReject",
     "CollateralAmountChange",
+    "EventResendComplete",
+    "EventResendReject",
+    "EventResendRequest",
     "ExecutionReport",
     "FixGroup",
     "FixGroupMeta",

kalshi/fix/messages/dispatch.py

@@ -0,0 +1,60 @@
+"""Inbound application-message dispatch: ``MsgType`` -> typed model.
+
+Central registry spanning the application message flows (order entry, drop copy,
+and later market data / RFQ / settlement). A consumer turns an inbound
+:class:`~kalshi.fix.codec.RawMessage` (delivered to ``FixSession.on_message``)
+into its typed model via :func:`decode_app_message`.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+from types import MappingProxyType
+
+from pydantic import ValidationError
+
+from kalshi.fix.codec import RawMessage
+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.order_entry import (
+    BusinessMessageReject,
+    ExecutionReport,
+    OrderCancelReject,
+    OrderMassCancelReport,
+)
+
+logger = logging.getLogger("kalshi.fix")
+
+# Read-only (MappingProxyType) so application code cannot corrupt dispatch.
+APP_MESSAGE_MODELS: Mapping[str, type[FixMessage]] = MappingProxyType(
+    {
+        MsgType.EXECUTION_REPORT.value: ExecutionReport,
+        MsgType.ORDER_CANCEL_REJECT.value: OrderCancelReject,
+        MsgType.ORDER_MASS_CANCEL_REPORT.value: OrderMassCancelReport,
+        MsgType.BUSINESS_MESSAGE_REJECT.value: BusinessMessageReject,
+        MsgType.EVENT_RESEND_COMPLETE.value: EventResendComplete,
+        MsgType.EVENT_RESEND_REJECT.value: EventResendReject,
+    }
+)
+
+
+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.
+    """
+    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).
+        logger.warning("failed to decode inbound %s; returning None", raw.msg_type, exc_info=True)
+        return None

kalshi/fix/messages/drop_copy.py

@@ -0,0 +1,52 @@
+"""Drop-copy FIX messages (GH #425).
+
+Kalshi drop copy is a request/response history query, not a streaming feed:
+send an :class:`EventResendRequest` (35=U1) for an ExecID range on a ``KalshiDC``
+session; the gateway replays the matching :class:`~kalshi.fix.messages.order_entry.ExecutionReport`
+messages (35=8, rejects and pending ``-1;-1`` excluded, 3-hour lookback) and
+terminates the batch with :class:`EventResendComplete` (35=U2) or
+:class:`EventResendReject` (35=U3). For a real-time stream use a listener session
+on KalshiRT instead (see ``FixConfig.listener_session``).
+
+ExecIDs use the Kalshi compound ``"int;int"`` format (e.g. ``"12345;67890"``);
+resent reports carry new FIX sequence numbers, so reconcile by ExecID.
+"""
+
+from __future__ import annotations
+
+from kalshi.fix.enums import MsgType
+from kalshi.fix.messages.base import FixMessage, FixType, fixfield
+from kalshi.fix.tags import Tag
+
+
+class EventResendRequest(FixMessage):
+    """EventResendRequest (35=U1) — request execution reports in an ExecID range."""
+
+    MSG_TYPE = MsgType.EVENT_RESEND_REQUEST
+
+    begin_exec_id: str = fixfield(Tag.BEGIN_EXEC_ID, FixType.STRING)
+    end_exec_id: str | None = fixfield(Tag.END_EXEC_ID, FixType.STRING, default=None)
+
+
+class EventResendComplete(FixMessage):
+    """EventResendComplete (35=U2) — all requested events have been resent."""
+
+    MSG_TYPE = MsgType.EVENT_RESEND_COMPLETE
+
+    ref_seq_num: int | None = fixfield(Tag.REF_SEQ_NUM, FixType.SEQNUM, default=None)
+    resend_event_count: int | None = fixfield(Tag.RESEND_EVENT_COUNT, FixType.INT, default=None)
+
+
+class EventResendReject(FixMessage):
+    """EventResendReject (35=U3) — the resend request could not be fulfilled.
+
+    ``event_resend_reject_reason`` is a raw int (compare against
+    :class:`~kalshi.fix.enums.EventResendRejectReason`).
+    """
+
+    MSG_TYPE = MsgType.EVENT_RESEND_REJECT
+
+    ref_seq_num: int | None = fixfield(Tag.REF_SEQ_NUM, FixType.SEQNUM, default=None)
+    event_resend_reject_reason: int | None = fixfield(
+        Tag.EVENT_RESEND_REJECT_REASON, FixType.INT, default=None
+    )

kalshi/fix/messages/order_entry.py

@@ -16,15 +16,9 @@
 
 from __future__ import annotations
 
-import logging
-from collections.abc import Mapping
 from datetime import datetime
-from types import MappingProxyType
 from typing import Annotated
 
-from pydantic import ValidationError
-
-from kalshi.fix.codec import RawMessage
 from kalshi.fix.enums import (
     ExecInst,
     MassCancelRequestType,
@@ -45,8 +39,6 @@
 from kalshi.fix.tags import Tag
 from kalshi.types import DollarDecimal, FixedPointCount
 
-logger = logging.getLogger("kalshi.fix")
-
 # ---------------------------------------------------------------------------
 # Outbound requests
 # ---------------------------------------------------------------------------
@@ -234,35 +226,3 @@ class BusinessMessageReject(FixMessage):
         Tag.BUSINESS_REJECT_REASON, FixType.INT, default=None
     )
     text: str | None = fixfield(Tag.TEXT, FixType.STRING, default=None)
-
-
-# Inbound application-message dispatch: MsgType -> model. Lets a consumer turn an
-# inbound RawMessage (delivered to FixSession.on_message) into a typed model.
-# Read-only (MappingProxyType) so application code cannot corrupt dispatch.
-APP_MESSAGE_MODELS: Mapping[str, type[FixMessage]] = MappingProxyType(
-    {
-        MsgType.EXECUTION_REPORT.value: ExecutionReport,
-        MsgType.ORDER_CANCEL_REJECT.value: OrderCancelReject,
-        MsgType.ORDER_MASS_CANCEL_REPORT.value: OrderMassCancelReport,
-        MsgType.BUSINESS_MESSAGE_REJECT.value: BusinessMessageReject,
-    }
-)
-
-
-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.
-    """
-    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).
-        logger.warning("failed to decode inbound %s; returning None", raw.msg_type, exc_info=True)
-        return None

kalshi/fix/session.py

@@ -345,6 +345,8 @@ async def _open_and_logon(self) -> None:
             reset_seq_num_flag=True if reset else None,
             use_dollars=True if self._config.effective_use_dollars else None,
             cancel_orders_on_disconnect=True if self._config.cancel_orders_on_disconnect else None,
+            listener_session=True if self._config.listener_session else None,
+            skip_pending_exec_reports=True if self._config.skip_pending_exec_reports else None,
         )
         await self._send(logon)