diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index 1e231ea..aaf2d47 100644 --- a/kalshi/fix/__init__.py +++ b/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", diff --git a/kalshi/fix/config.py b/kalshi/fix/config.py index ce8328e..cb6a447 100644 --- a/kalshi/fix/config.py +++ b/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, ) diff --git a/kalshi/fix/enums.py b/kalshi/fix/enums.py index 3f12498..107fbe2 100644 --- a/kalshi/fix/enums.py +++ b/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 diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index 7713a4e..4b6fb2d 100644 --- a/kalshi/fix/messages/__init__.py +++ b/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", diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py new file mode 100644 index 0000000..6b3d70a --- /dev/null +++ b/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 diff --git a/kalshi/fix/messages/drop_copy.py b/kalshi/fix/messages/drop_copy.py new file mode 100644 index 0000000..f9508f9 --- /dev/null +++ b/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 + ) diff --git a/kalshi/fix/messages/order_entry.py b/kalshi/fix/messages/order_entry.py index 6186f62..358fa0b 100644 --- a/kalshi/fix/messages/order_entry.py +++ b/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 diff --git a/kalshi/fix/session.py b/kalshi/fix/session.py index b2f984d..cc5b041 100644 --- a/kalshi/fix/session.py +++ b/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) diff --git a/tests/fix/test_drop_copy.py b/tests/fix/test_drop_copy.py new file mode 100644 index 0000000..a562450 --- /dev/null +++ b/tests/fix/test_drop_copy.py @@ -0,0 +1,167 @@ +"""Tests for the drop-copy FIX flow (GH #425).""" + +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, FixEnvironment, FixSessionType +from kalshi.fix.enums import EventResendRejectReason, ExecType, OrdStatus, Side +from kalshi.fix.messages import ( + EventResendComplete, + EventResendReject, + EventResendRequest, + ExecutionReport, + decode_app_message, +) +from kalshi.fix.messages.base import FixMessage +from kalshi.fix.session import FixSession +from kalshi.fix.tags import Tag + +from .conftest import MockAcceptor + + +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 test_event_resend_request_roundtrip() -> None: + msg = EventResendRequest(begin_exec_id="12345;67890", end_exec_id="12350;67895") + body = dict(msg.to_body_fields()) + assert body[int(Tag.BEGIN_EXEC_ID)] == "12345;67890" + assert body[int(Tag.END_EXEC_ID)] == "12350;67895" + assert _roundtrip(msg) == msg + + +def test_event_resend_request_begin_only() -> None: + msg = EventResendRequest(begin_exec_id="12345;67890") + assert int(Tag.END_EXEC_ID) not in {t for t, _ in msg.to_body_fields()} + assert _roundtrip(msg) == msg + + +def test_event_resend_complete_roundtrip() -> None: + msg = EventResendComplete(ref_seq_num=7, resend_event_count=12) + assert _roundtrip(msg) == msg + + +def test_event_resend_reject_roundtrip() -> None: + msg = EventResendReject(ref_seq_num=7, event_resend_reject_reason=3) + back = _roundtrip(msg) + assert back == msg + assert back.event_resend_reject_reason == EventResendRejectReason.BEGIN_EXECID_TOO_SMALL + + +def test_decode_app_message_drop_copy_types() -> None: + for msg in ( + EventResendComplete(ref_seq_num=7, resend_event_count=3), + EventResendReject(ref_seq_num=7, event_resend_reject_reason=1), + ): + 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 + + +def test_listener_session_requires_skip_pending() -> None: + # Both factories feed the same __post_init__ validator. + with pytest.raises(ValueError, match="skip_pending_exec_reports"): + FixConfig.prediction(listener_session=True) + with pytest.raises(ValueError, match="skip_pending_exec_reports"): + FixConfig.margin(listener_session=True) + + +async def test_drop_copy_query_returns_execution_reports( + 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.DROP_COPY, on_message=on_message + ) + await session.start() + try: + await session.send(EventResendRequest(begin_exec_id="1;1", end_exec_id="9;9")) + await until(lambda: acceptor.first("U1") is not None) + assert acceptor.first("U1").get(Tag.BEGIN_EXEC_ID) == "1;1" # type: ignore[union-attr] + + # The gateway replays matching ExecutionReports then an EventResendComplete. + report = ExecutionReport( + order_id="OID-1", + cl_ord_id="abc", + exec_id="4;7", + exec_type=ExecType.TRADE.value, + ord_status=OrdStatus.FILLED.value, + side=Side.BUY_YES.value, + symbol="KXTEST", + leaves_qty=Decimal("0"), + cum_qty=Decimal("10"), + avg_px=Decimal("0.66"), + order_qty=Decimal("10"), + ) + await acceptor.push("8", report.to_body_fields(), seq=2) + await acceptor.push( + "U2", EventResendComplete(ref_seq_num=2, resend_event_count=1).to_body_fields(), seq=3 + ) + await until(lambda: len(received) == 2) + decoded = [decode_app_message(r) for r in received] + assert isinstance(decoded[0], ExecutionReport) + assert decoded[0].exec_id == "4;7" + assert isinstance(decoded[1], EventResendComplete) + assert decoded[1].resend_event_count == 1 + finally: + await session.close() + + +async def test_listener_logon_emits_flags( + fix_signer: FixSigner, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + config = FixConfig.prediction( + environment=FixEnvironment.DEMO, + host="127.0.0.1", + port=acceptor.port, + use_tls=False, + heartbeat_interval=4, + connect_timeout=2.0, + listener_session=True, + skip_pending_exec_reports=True, + ) + session = FixSession(fix_signer, config, FixSessionType.ORDER_ENTRY_NR) + await session.start() + try: + # start() returns only after the Logon round-trips (the acceptor records + # before it responds), so the Logon is already landed — poll anyway to + # match the file's idiom and stay robust to acceptor changes. + await until(lambda: acceptor.first("A") is not None) + logon = acceptor.first("A") + assert logon is not None + assert logon.get(Tag.LISTENER_SESSION) == "Y" + assert logon.get(Tag.SKIP_PENDING_EXEC_REPORTS) == "Y" + finally: + await session.close()