From 2d48907b0c3f0ea7f5ffd8002ea337c4cce91970 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 08:02:09 -0500 Subject: [PATCH 1/3] feat(fix): market-settlement / post-trade reports (closes #429) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FIX market settlement — prediction only. A read-only MarketSettlementReport (35=UMS) stream on KalshiPT (default on) and on KalshiRT (opt in via ReceiveSettlementReports). This is the last sub-issue of the #402 FIX epic. - messages/settlement.py: MarketSettlementReport (UMS), inbound (fields optional, codes raw). Reuses the MarketSettlementParty group entry (from #423), which nests NoCollateralAmountChanges + NoMiscFees per party. market_result is a raw str (compare against the new MarketResult enum); settlement_price (cents) and collateral/fee amounts are DollarDecimal; positions FixedPointCount. - enums.py: MarketResult (yes/no/scalar), CollateralAmountType (BALANCE/PAYOUT). - settlement.py: SettlementReassembler — paginated batches span multiple UMS fragments whose MarketSettlementReportID differs per page, so it correlates by Symbol and accumulates parties until the terminal fragment (LastFragment Y/ absent), emitting one combined report. - config: receive_settlement_reports (bool | None) threaded into the Logon — None omits tag 20127 (gateway default: on for PT, off for RT), True opts RT in (20127=Y), False opts PT out (20127=N). The Logon field + all tags already existed from the foundation; client.post_trade() is the session factory. - dispatch registers UMS; exports from kalshi.fix / kalshi.fix.messages. Hardening from a pre-PR adversarial review (8 confirmed findings; the money path and nested decode were verified correct). The reassembler keys only on Symbol (the report id differs per page) with no batch-id in the protocol, so its contiguous-per-symbol-batch assumption + limitations (no TotNum reconciliation; interleaved same-symbol batches would coalesce — not produced by a well-formed stream; orphaned buffers persist until clear(); duplicate terminals re-emit) are now documented on the class, and clear() warns when it discards non-empty buffers. Filed #437 for a FIX dictionary<->model drift test (infra follow-up). 17 new tests: golden UMS wire, nested-group round-trip (incl. two parties each with collateral + fee groups, negative collateral, sub-cent fee), a decode of the docs example with LastFragment trailing the party group, dispatch, multi-fragment reassembly (buffer/emit/per-symbol/standalone/no-symbol/sequential-same-symbol/ clear-warns), a KalshiPT settlement-stream integration test, and the ReceiveSettlementReports logon flag across opt-in / opt-out / default. ruff + mypy --strict clean; 248 FIX tests pass. Closes #429. Part of #402. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/__init__.py | 8 + kalshi/fix/config.py | 8 + kalshi/fix/enums.py | 15 ++ kalshi/fix/messages/__init__.py | 2 + kalshi/fix/messages/dispatch.py | 3 + kalshi/fix/messages/settlement.py | 63 ++++++ kalshi/fix/session.py | 3 + kalshi/fix/settlement.py | 94 ++++++++ tests/fix/test_settlement.py | 358 ++++++++++++++++++++++++++++++ 9 files changed, 554 insertions(+) create mode 100644 kalshi/fix/messages/settlement.py create mode 100644 kalshi/fix/settlement.py create mode 100644 tests/fix/test_settlement.py diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index 48cf601..88bd004 100644 --- a/kalshi/fix/__init__.py +++ b/kalshi/fix/__init__.py @@ -44,10 +44,12 @@ from kalshi.fix.enums import ( AcceptQuoteResult, ApplVerID, + CollateralAmountType, EncryptMethod, EventResendRejectReason, ExecInst, ExecType, + MarketResult, MDEntryType, MDReqRejReason, MDUpdateAction, @@ -98,6 +100,7 @@ MarketDataRequestReject, MarketDataSnapshotFullRefresh, MarketSettlementParty, + MarketSettlementReport, MDIncrementalEntry, MDSnapshotEntry, MiscFee, @@ -134,6 +137,7 @@ ) from kalshi.fix.orderbook import BookLevel, FixOrderBook, MarketDataBook from kalshi.fix.session import FixSession, FixSessionState +from kalshi.fix.settlement import SettlementReassembler from kalshi.fix.tags import Tag # Sorted (ruff RUF022); grouping is by the imports above, not by ``__all__`` order. @@ -148,6 +152,7 @@ "BookLevel", "BusinessMessageReject", "CollateralAmountChange", + "CollateralAmountType", "EncryptMethod", "EventResendComplete", "EventResendReject", @@ -190,7 +195,9 @@ "MarketDataRequest", "MarketDataRequestReject", "MarketDataSnapshotFullRefresh", + "MarketResult", "MarketSettlementParty", + "MarketSettlementReport", "MiscFee", "MsgType", "MultivariateSelectedLeg", @@ -233,6 +240,7 @@ "SelfTradePreventionType", "SequenceReset", "SessionRejectReason", + "SettlementReassembler", "Side", "SubscriptionRequestType", "Tag", diff --git a/kalshi/fix/config.py b/kalshi/fix/config.py index cb6a447..6f38041 100644 --- a/kalshi/fix/config.py +++ b/kalshi/fix/config.py @@ -140,6 +140,10 @@ class FixConfig: # an order-entry capability. Valid on NR/RT and requires skip_pending_exec_reports. listener_session: bool = False skip_pending_exec_reports: bool = False + # Settlement-report stream (tag 20127), applies to RT/PT logon. ``None`` omits + # the flag — the gateway then applies its default (on for KalshiPT, off for + # KalshiRT). Set ``True`` to opt KalshiRT in, ``False`` to opt KalshiPT out. + receive_settlement_reports: bool | None = None # ``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 @@ -207,6 +211,7 @@ def prediction( cancel_orders_on_disconnect: bool = False, listener_session: bool = False, skip_pending_exec_reports: bool = False, + receive_settlement_reports: bool | None = None, use_dollars: bool | None = None, allow_unknown_host: bool = False, ) -> FixConfig: @@ -225,6 +230,7 @@ def prediction( cancel_orders_on_disconnect=cancel_orders_on_disconnect, listener_session=listener_session, skip_pending_exec_reports=skip_pending_exec_reports, + receive_settlement_reports=receive_settlement_reports, use_dollars=use_dollars, allow_unknown_host=allow_unknown_host, ) @@ -245,6 +251,7 @@ def margin( cancel_orders_on_disconnect: bool = False, listener_session: bool = False, skip_pending_exec_reports: bool = False, + receive_settlement_reports: bool | None = None, use_dollars: bool = True, allow_unknown_host: bool = False, ) -> FixConfig: @@ -263,6 +270,7 @@ def margin( cancel_orders_on_disconnect=cancel_orders_on_disconnect, listener_session=listener_session, skip_pending_exec_reports=skip_pending_exec_reports, + receive_settlement_reports=receive_settlement_reports, use_dollars=use_dollars, allow_unknown_host=allow_unknown_host, ) diff --git a/kalshi/fix/enums.py b/kalshi/fix/enums.py index f4c1010..5e47e1a 100644 --- a/kalshi/fix/enums.py +++ b/kalshi/fix/enums.py @@ -258,6 +258,21 @@ class OrderGroupAction(IntEnum): UPDATE = 5 +class MarketResult(StrEnum): + """Tag 20107 — market outcome on a MarketSettlementReport (35=UMS).""" + + YES_OUTCOME = "yes" + NO_OUTCOME = "no" + SCALAR_VALUE = "scalar" + + +class CollateralAmountType(StrEnum): + """Tag 1705 — kind of collateral change in a settlement collateral entry.""" + + BALANCE = "BALANCE" + PAYOUT = "PAYOUT" + + class EventResendRejectReason(IntEnum): """Tag 21004 — why a drop-copy EventResendRequest (35=U3) was rejected.""" diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index f7dbb1e..bafbd53 100644 --- a/kalshi/fix/messages/__init__.py +++ b/kalshi/fix/messages/__init__.py @@ -75,6 +75,7 @@ SequenceReset, TestRequest, ) +from kalshi.fix.messages.settlement import MarketSettlementReport __all__ = [ "APP_MESSAGE_MODELS", @@ -100,6 +101,7 @@ "MarketDataRequestReject", "MarketDataSnapshotFullRefresh", "MarketSettlementParty", + "MarketSettlementReport", "MiscFee", "MultivariateSelectedLeg", "NewOrderSingle", diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py index 258787c..50c4dcf 100644 --- a/kalshi/fix/messages/dispatch.py +++ b/kalshi/fix/messages/dispatch.py @@ -42,6 +42,7 @@ QuoteStatusReport, RFQCancelStatus, ) +from kalshi.fix.messages.settlement import MarketSettlementReport logger = logging.getLogger("kalshi.fix") @@ -70,6 +71,8 @@ MsgType.QUOTE_CANCEL_STATUS.value: QuoteCancelStatus, MsgType.RFQ_CANCEL_STATUS.value: RFQCancelStatus, MsgType.ACCEPT_QUOTE_STATUS.value: AcceptQuoteStatus, + # Market settlement (post trade) + MsgType.MARKET_SETTLEMENT_REPORT.value: MarketSettlementReport, } ) diff --git a/kalshi/fix/messages/settlement.py b/kalshi/fix/messages/settlement.py new file mode 100644 index 0000000..7ea3b69 --- /dev/null +++ b/kalshi/fix/messages/settlement.py @@ -0,0 +1,63 @@ +"""Market-settlement / post-trade FIX messages (GH #429) — prediction only. + +A :class:`MarketSettlementReport` (35=UMS) is a read-only settlement notification +streamed on the ``KalshiPT`` session (and on ``KalshiRT`` when +``ReceiveSettlementReports`` is opted in). Per-party collateral changes and fees +are nested inside the ``NoMarketSettlementPartyIDs`` group (the shared +:class:`~kalshi.fix.messages.components.MarketSettlementParty` entry, which itself +nests ``NoCollateralAmountChanges`` and ``NoMiscFees``). + +Fields are optional and code fields raw (compare against :mod:`kalshi.fix.enums`) +per the inbound convention. Large settlement batches paginate across several +fragments — correlate them by ``Symbol`` (the report id differs per page) and +reassemble with :class:`~kalshi.fix.settlement.SettlementReassembler`. +""" + +from __future__ import annotations + +from typing import Annotated + +from kalshi.fix.enums import MsgType +from kalshi.fix.messages.base import ( + FixGroupMeta, + FixMessage, + FixType, + fixfield, + groupfield, +) +from kalshi.fix.messages.components import MarketSettlementParty +from kalshi.fix.tags import Tag +from kalshi.types import DollarDecimal + + +class MarketSettlementReport(FixMessage): + """MarketSettlementReport (35=UMS) — a market's settlement detail. + + ``market_result`` is a raw str (compare against + :class:`~kalshi.fix.enums.MarketResult`). ``settlement_price`` is in cents + (e.g. ``Decimal("100.00")``). ``clearing_business_date`` is the raw + ``YYYYMMDD`` string. ``last_fragment`` is ``False`` on non-final pages of a + paginated batch, ``True`` (or absent) on the last. + """ + + MSG_TYPE = MsgType.MARKET_SETTLEMENT_REPORT + + market_settlement_report_id: str | None = fixfield( + Tag.MARKET_SETTLEMENT_REPORT_ID, FixType.STRING, default=None + ) + symbol: str | None = fixfield(Tag.SYMBOL, FixType.STRING, default=None) + clearing_business_date: str | None = fixfield( + Tag.CLEARING_BUSINESS_DATE, FixType.LOCALMKTDATE, default=None + ) + tot_num_market_settlement_reports: int | None = fixfield( + Tag.TOT_NUM_MARKET_SETTLEMENT_REPORTS, FixType.INT, default=None + ) + market_result: str | None = fixfield(Tag.MARKET_RESULT, FixType.STRING, default=None) + settlement_price: DollarDecimal | None = fixfield( + Tag.SETTLEMENT_PRICE, FixType.PRICE, default=None + ) + last_fragment: bool | None = fixfield(Tag.LAST_FRAGMENT, FixType.BOOLEAN, default=None) + parties: Annotated[ + list[MarketSettlementParty], + FixGroupMeta(Tag.NO_MARKET_SETTLEMENT_PARTY_IDS, MarketSettlementParty), + ] = groupfield() diff --git a/kalshi/fix/session.py b/kalshi/fix/session.py index cc5b041..2e097f1 100644 --- a/kalshi/fix/session.py +++ b/kalshi/fix/session.py @@ -347,6 +347,9 @@ async def _open_and_logon(self) -> 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, + # Pass through directly: None omits the tag (gateway default), True/False + # explicitly opt in (RT) / out (PT). + receive_settlement_reports=self._config.receive_settlement_reports, ) await self._send(logon) diff --git a/kalshi/fix/settlement.py b/kalshi/fix/settlement.py new file mode 100644 index 0000000..8cfbe0a --- /dev/null +++ b/kalshi/fix/settlement.py @@ -0,0 +1,94 @@ +"""Reassembly of paginated market-settlement reports (GH #429). + +A large settlement batch is delivered as several :class:`MarketSettlementReport` +(35=UMS) fragments: each non-final fragment carries ``LastFragment=N`` and a +subset of the parties, the final one ``LastFragment=Y`` (or omitted). Fragments +of one batch are correlated by ``Symbol`` — the ``MarketSettlementReportID`` +differs per page, so it cannot be used as the key. + +:class:`SettlementReassembler` accumulates each symbol's party entries across +fragments and, on the terminal fragment, returns a single +:class:`MarketSettlementReport` carrying every party. +""" + +from __future__ import annotations + +import logging + +from kalshi.fix.messages.components import MarketSettlementParty +from kalshi.fix.messages.settlement import MarketSettlementReport + +logger = logging.getLogger("kalshi.fix") + + +class SettlementReassembler: + """Reassemble paginated UMS fragments into one report per settlement batch. + + Usage:: + + reasm = SettlementReassembler() + complete = reasm.add(decode_app_message(raw)) # MarketSettlementReport | None + if complete is not None: + ... # all parties for the batch are in complete.parties + + Assumptions / limitations. ``Symbol`` is the only correlation field (the report + id differs per page), so this keys solely on it and assumes a batch's fragments + arrive as one contiguous run terminated by ``LastFragment=Y`` (or a standalone + report). It does **not** reconcile against ``TotNumMarketSettlementReports`` + (20106). Consequences: + + * Two batches for the *same* symbol must not interleave — a second batch's + terminal arriving while the first is still buffered would coalesce both into + one report. A well-formed gateway sends a symbol's fragments contiguously + (each batch ends with ``LastFragment=Y`` before the next begins), so this + does not occur in practice; sequential same-symbol batches reassemble + independently because the buffer clears on each terminal. + * If a terminal fragment never arrives (mid-batch disconnect), the partial + parties stay buffered — see :meth:`pending_symbols` — until :meth:`clear` + (call it on reconnect). + * A re-delivered terminal fragment re-emits its report, so consumers should + handle emitted reports idempotently. + """ + + def __init__(self) -> None: + self._pending: dict[str, list[MarketSettlementParty]] = {} + + def add(self, report: MarketSettlementReport) -> MarketSettlementReport | None: + """Feed one fragment; return the assembled report on the terminal fragment. + + Returns ``None`` while more fragments are expected (``LastFragment=N``). + A terminal fragment (``LastFragment`` ``True`` or absent) returns the + report with every accumulated party; a standalone report is returned + unchanged. + """ + if report.last_fragment is False: + # Non-final page: must have a Symbol to correlate the batch. + if report.symbol is None: + logger.warning("settlement fragment without Symbol; cannot reassemble") + return None + self._pending.setdefault(report.symbol, []).extend(report.parties) + return None + + # Terminal fragment (True or None). + if report.symbol is not None and report.symbol in self._pending: + parties = self._pending.pop(report.symbol) + parties.extend(report.parties) + return report.model_copy(update={"parties": parties}) + return report + + def pending_symbols(self) -> set[str]: + """Symbols with buffered fragments still awaiting their terminal page.""" + return set(self._pending) + + def clear(self) -> None: + """Drop all buffered fragments (e.g. on reconnect). + + Warns when discarding non-empty buffers so silently-dropped partial + settlement data is observable. + """ + if self._pending: + logger.warning( + "SettlementReassembler.clear() dropping buffered settlement fragments for %s", + sorted(self._pending), + ) + self._pending.clear() diff --git a/tests/fix/test_settlement.py b/tests/fix/test_settlement.py new file mode 100644 index 0000000..368f339 --- /dev/null +++ b/tests/fix/test_settlement.py @@ -0,0 +1,358 @@ +"""Tests for the market-settlement / post-trade FIX flow (GH #429).""" + +from __future__ import annotations + +import logging +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 CollateralAmountType, MarketResult +from kalshi.fix.messages import ( + CollateralAmountChange, + MarketSettlementParty, + MarketSettlementReport, + MiscFee, + decode_app_message, +) +from kalshi.fix.messages.base import FixMessage +from kalshi.fix.session import FixSession +from kalshi.fix.settlement import SettlementReassembler +from kalshi.fix.tags import Tag + +from .conftest import MockAcceptor + +SYM = "HIGHNY-23DEC31" + + +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: + return "|".join(f"{t}={v}" for t, v in msg.to_body_fields()) + + +def _party(party_id: str, role: int = 24) -> MarketSettlementParty: + return MarketSettlementParty(party_id=party_id, party_role=role) + + +# --------------------------------------------------------------------------- +# MarketSettlementReport (35=UMS) golden wire + round-trip +# --------------------------------------------------------------------------- + + +def test_settlement_report_golden_wire() -> None: + msg = MarketSettlementReport( + market_settlement_report_id="settle-123", + symbol=SYM, + clearing_business_date="20231231", + market_result="yes", + settlement_price=Decimal("100.00"), + last_fragment=True, + ) + assert _wire(msg) == ( + f"20105=settle-123|55={SYM}|715=20231231|20107=yes|730=100.00|893=Y" + ) + + +def test_settlement_report_nested_roundtrip() -> None: + party = MarketSettlementParty( + party_id="user-456", + party_role=24, + long_qty=Decimal("100"), + short_qty=Decimal("0"), + collateral_amount_changes=[ + CollateralAmountChange( + collateral_amount_change=Decimal("100.00"), collateral_amount_type="PAYOUT" + ) + ], + misc_fees=[ + MiscFee( + misc_fee_amt=Decimal("0.006"), + misc_fee_curr="USD", + misc_fee_type="4", + misc_fee_basis=0, + ) + ], + ) + msg = MarketSettlementReport( + market_settlement_report_id="settle-456", + symbol=SYM, + clearing_business_date="20231231", + market_result="yes", + settlement_price=Decimal("100.00"), + last_fragment=True, + parties=[party], + ) + assert _roundtrip(msg) == msg + + +def test_settlement_report_two_parties_with_nested_groups_roundtrip() -> None: + # Each party carries its own collateral + misc-fee groups; the positional + # decoder must keep them separated (no cross-party leakage). Exercises negative + # collateral (BALANCE) and a sub-cent fee. + def _mk(pid: str, coll: str, coll_type: str, fee: str) -> MarketSettlementParty: + return MarketSettlementParty( + party_id=pid, + party_role=24, + long_qty=Decimal("100"), + short_qty=Decimal("0"), + collateral_amount_changes=[ + CollateralAmountChange( + collateral_amount_change=Decimal(coll), collateral_amount_type=coll_type + ) + ], + misc_fees=[ + MiscFee( + misc_fee_amt=Decimal(fee), misc_fee_curr="USD", misc_fee_type="4", + misc_fee_basis=0, + ) + ], + ) + + msg = MarketSettlementReport( + market_settlement_report_id="settle-789", + symbol=SYM, + clearing_business_date="20231231", + market_result="yes", + settlement_price=Decimal("100.00"), + last_fragment=True, + parties=[_mk("u1", "100.00", "PAYOUT", "0.00"), _mk("u2", "-50.00", "BALANCE", "0.006")], + ) + rt = _roundtrip(msg) + assert rt == msg + assert isinstance(rt, MarketSettlementReport) + assert [p.party_id for p in rt.parties] == ["u1", "u2"] + assert rt.parties[1].collateral_amount_changes[0].collateral_amount_change == Decimal("-50.00") + assert rt.parties[1].misc_fees[0].misc_fee_amt == Decimal("0.006") + + +def test_settlement_report_decodes_doc_example_with_trailing_last_fragment() -> None: + # Doc example wire: LastFragment(893) comes AFTER the nested party group. + # Positional group decode must still attribute the nested fields to the party + # and read the trailing top-level 893 by tag. + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), "UMS"), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + (int(Tag.MARKET_SETTLEMENT_REPORT_ID), "settle-123"), + (int(Tag.SYMBOL), SYM), + (int(Tag.CLEARING_BUSINESS_DATE), "20231231"), + (int(Tag.MARKET_RESULT), "yes"), + (int(Tag.SETTLEMENT_PRICE), "100.00"), + (int(Tag.NO_MARKET_SETTLEMENT_PARTY_IDS), "1"), + (int(Tag.MARKET_SETTLEMENT_PARTY_ID), "user-456"), + (int(Tag.MARKET_SETTLEMENT_PARTY_ROLE), "24"), + (int(Tag.LONG_QTY), "100"), + (int(Tag.SHORT_QTY), "0"), + (int(Tag.NO_COLLATERAL_AMOUNT_CHANGES), "1"), + (int(Tag.COLLATERAL_AMOUNT_CHANGE), "100.00"), + (int(Tag.COLLATERAL_AMOUNT_TYPE), "PAYOUT"), + (int(Tag.NO_MISC_FEES), "1"), + (int(Tag.MISC_FEE_AMT), "0.00"), + (int(Tag.MISC_FEE_CURR), "USD"), + (int(Tag.MISC_FEE_TYPE), "4"), + (int(Tag.MISC_FEE_BASIS), "0"), + (int(Tag.LAST_FRAGMENT), "Y"), + ] + ) + ) + msg = decode_app_message(raw) + assert isinstance(msg, MarketSettlementReport) + assert msg.symbol == SYM + assert msg.market_result == MarketResult.YES_OUTCOME + assert msg.settlement_price == Decimal("100.00") + assert msg.last_fragment is True + assert len(msg.parties) == 1 + party = msg.parties[0] + assert party.party_id == "user-456" + assert party.long_qty == Decimal("100") + assert party.collateral_amount_changes[0].collateral_amount_type == CollateralAmountType.PAYOUT + assert party.collateral_amount_changes[0].collateral_amount_change == Decimal("100.00") + assert party.misc_fees[0].misc_fee_amt == Decimal("0.00") + assert party.misc_fees[0].misc_fee_type == "4" + + +def test_decode_app_message_settlement() -> None: + msg = MarketSettlementReport(symbol=SYM, market_result="no", last_fragment=True) + 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 isinstance(decoded, MarketSettlementReport) + assert decoded == msg + + +# --------------------------------------------------------------------------- +# Multi-fragment reassembly +# --------------------------------------------------------------------------- + + +def _frag(symbol: str | None, party_ids: list[str], last: bool | None) -> MarketSettlementReport: + return MarketSettlementReport( + symbol=symbol, + market_result="yes", + parties=[_party(p) for p in party_ids], + last_fragment=last, + ) + + +def test_reassembler_multi_fragment() -> None: + r = SettlementReassembler() + assert r.add(_frag(SYM, ["a"], False)) is None + assert r.add(_frag(SYM, ["b"], False)) is None + assert r.pending_symbols() == {SYM} + done = r.add(_frag(SYM, ["c"], True)) + assert done is not None + assert [p.party_id for p in done.parties] == ["a", "b", "c"] + assert r.pending_symbols() == set() + + +@pytest.mark.parametrize("last", [True, None]) +def test_reassembler_standalone(last: bool | None) -> None: + # A standalone report (terminal True, or LastFragment absent) returns as-is. + r = SettlementReassembler() + done = r.add(_frag(SYM, ["x"], last)) + assert done is not None + assert [p.party_id for p in done.parties] == ["x"] + assert r.pending_symbols() == set() + + +def test_reassembler_routes_by_symbol() -> None: + r = SettlementReassembler() + r.add(_frag("A", ["a1"], False)) + r.add(_frag("B", ["b1"], False)) + done_a = r.add(_frag("A", ["a2"], True)) + assert done_a is not None + assert [p.party_id for p in done_a.parties] == ["a1", "a2"] + assert r.pending_symbols() == {"B"} + done_b = r.add(_frag("B", ["b2"], True)) + assert done_b is not None + assert [p.party_id for p in done_b.parties] == ["b1", "b2"] + + +def test_reassembler_non_final_without_symbol_dropped() -> None: + r = SettlementReassembler() + assert r.add(_frag(None, ["a"], False)) is None + assert r.pending_symbols() == set() + + +def test_reassembler_sequential_same_symbol_batches() -> None: + # Two contiguous batches for the same symbol, each terminated by Y, reassemble + # independently — the buffer clears on each terminal, so they do not coalesce. + r = SettlementReassembler() + assert r.add(_frag(SYM, ["a1"], False)) is None + done1 = r.add(_frag(SYM, ["a2"], True)) + assert done1 is not None + assert [p.party_id for p in done1.parties] == ["a1", "a2"] + assert r.pending_symbols() == set() + + assert r.add(_frag(SYM, ["b1"], False)) is None + done2 = r.add(_frag(SYM, ["b2"], True)) + assert done2 is not None + assert [p.party_id for p in done2.parties] == ["b1", "b2"] + + +def test_reassembler_clear_warns_on_pending(caplog: pytest.LogCaptureFixture) -> None: + r = SettlementReassembler() + r.add(_frag(SYM, ["a"], False)) + with caplog.at_level(logging.WARNING, logger="kalshi.fix"): + r.clear() + assert r.pending_symbols() == set() + assert any("dropping buffered settlement" in rec.message for rec in caplog.records) + + +def test_reassembler_clear_silent_when_empty(caplog: pytest.LogCaptureFixture) -> None: + r = SettlementReassembler() + with caplog.at_level(logging.WARNING, logger="kalshi.fix"): + r.clear() + assert caplog.records == [] + + +# --------------------------------------------------------------------------- +# Post-trade session stream + ReceiveSettlementReports logon flag +# --------------------------------------------------------------------------- + + +async def test_post_trade_settlement_stream( + 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.POST_TRADE, on_message=on_message + ) + await session.start() + reasm = SettlementReassembler() + try: + f1 = MarketSettlementReport( + market_settlement_report_id="s1", symbol=SYM, market_result="yes", + settlement_price=Decimal("100.00"), last_fragment=False, parties=[_party("u1")], + ) + f2 = MarketSettlementReport( + market_settlement_report_id="s2", symbol=SYM, market_result="yes", + last_fragment=True, parties=[_party("u2")], + ) + await acceptor.push("UMS", f1.to_body_fields(), seq=2) + await acceptor.push("UMS", f2.to_body_fields(), seq=3) + await until(lambda: len(received) == 2) + results = [reasm.add(decode_app_message(r)) for r in received] # type: ignore[arg-type] + assert results[0] is None # first fragment buffered + assert results[1] is not None + assert [p.party_id for p in results[1].parties] == ["u1", "u2"] + finally: + await session.close() + + +@pytest.mark.parametrize(("setting", "expected"), [(True, "Y"), (False, "N"), (None, None)]) +async def test_receive_settlement_reports_logon_flag( + fix_signer: FixSigner, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], + setting: bool | None, + expected: str | 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, + receive_settlement_reports=setting, + ) + session = FixSession(fix_signer, config, FixSessionType.POST_TRADE) + await session.start() + try: + await until(lambda: acceptor.first("A") is not None) + logon = acceptor.first("A") + assert logon is not None + assert logon.get(Tag.RECEIVE_SETTLEMENT_REPORTS) == expected + finally: + await session.close() From 1e6a44b5aed69da43e564b0fd27429f0d2eac7af Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 08:07:32 -0500 Subject: [PATCH 2/3] =?UTF-8?q?fix(fix):=20address=20#438=20review=20?= =?UTF-8?q?=E2=80=94=20MarketResult=20names=20+=20settlement=20log/docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - enums: rename MarketResult members YES_OUTCOME/NO_OUTCOME/SCALAR_VALUE -> YES/NO/SCALAR to match the file's concise convention (BUY_YES, NEW, BID); the class name already supplies the "outcome" context. Pre-ship, so no break. - settlement: the symbol-less fragment warning now reports the dropped party count (actionable log); document that an assembled report carries the terminal fragment's MarketSettlementReportID (no canonical batch id — dedup off Symbol + clearing date). No action (already correct / documented per the review): tot_num is receive-only by design (see #437); APP_MESSAGE_MODELS + exports verified complete. ruff + mypy --strict clean; 248 FIX tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/enums.py | 6 +++--- kalshi/fix/settlement.py | 9 +++++++-- tests/fix/test_settlement.py | 2 +- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/kalshi/fix/enums.py b/kalshi/fix/enums.py index 5e47e1a..2597c61 100644 --- a/kalshi/fix/enums.py +++ b/kalshi/fix/enums.py @@ -261,9 +261,9 @@ class OrderGroupAction(IntEnum): class MarketResult(StrEnum): """Tag 20107 — market outcome on a MarketSettlementReport (35=UMS).""" - YES_OUTCOME = "yes" - NO_OUTCOME = "no" - SCALAR_VALUE = "scalar" + YES = "yes" + NO = "no" + SCALAR = "scalar" class CollateralAmountType(StrEnum): diff --git a/kalshi/fix/settlement.py b/kalshi/fix/settlement.py index 8cfbe0a..1f7fb5c 100644 --- a/kalshi/fix/settlement.py +++ b/kalshi/fix/settlement.py @@ -59,12 +59,17 @@ def add(self, report: MarketSettlementReport) -> MarketSettlementReport | None: Returns ``None`` while more fragments are expected (``LastFragment=N``). A terminal fragment (``LastFragment`` ``True`` or absent) returns the report with every accumulated party; a standalone report is returned - unchanged. + unchanged. The assembled report carries the *terminal* fragment's + ``MarketSettlementReportID`` (each page has its own — there is no canonical + batch id), so key any dedup off ``Symbol`` + clearing date, not that id. """ if report.last_fragment is False: # Non-final page: must have a Symbol to correlate the batch. if report.symbol is None: - logger.warning("settlement fragment without Symbol; cannot reassemble") + logger.warning( + "settlement fragment without Symbol; dropping %d parties", + len(report.parties), + ) return None self._pending.setdefault(report.symbol, []).extend(report.parties) return None diff --git a/tests/fix/test_settlement.py b/tests/fix/test_settlement.py index 368f339..4ca8739 100644 --- a/tests/fix/test_settlement.py +++ b/tests/fix/test_settlement.py @@ -173,7 +173,7 @@ def test_settlement_report_decodes_doc_example_with_trailing_last_fragment() -> msg = decode_app_message(raw) assert isinstance(msg, MarketSettlementReport) assert msg.symbol == SYM - assert msg.market_result == MarketResult.YES_OUTCOME + assert msg.market_result == MarketResult.YES assert msg.settlement_price == Decimal("100.00") assert msg.last_fragment is True assert len(msg.parties) == 1 From a4c5bb702a9a2eff160f6f86c291940a74754928 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 08:18:08 -0500 Subject: [PATCH 3/3] =?UTF-8?q?fix(fix):=20address=20#438=20review=20round?= =?UTF-8?q?=202=20=E2=80=94=20settlement=20docs=20+=20test=20polish?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - settlement.py: clarify settlement_price — the value is in cents (2dp; 100.00 = full YES payout, 30.60 = scalar), DollarDecimal is just the no-float coercion type (same cents-on-prediction convention as the order-entry price), not dollars. - tests: drop the type: ignore in the PT-stream test by narrowing each decoded message via isinstance (a misrouted dispatch now fails the test); assert the "without Symbol" warning in the no-symbol drop test; parametrize the multi-fragment terminal over True/None (exercises absent-terminal merge); add MarketResult YES/NO/SCALAR wire + round-trip coverage (SCALAR was untested). No action (already correct per the review): market_result raw-str inbound convention; tot_num receive-only (#437); receive_settlement_reports passthrough. ruff + mypy --strict clean; 252 FIX tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/messages/settlement.py | 11 ++++++---- tests/fix/test_settlement.py | 35 ++++++++++++++++++++++++++----- 2 files changed, 37 insertions(+), 9 deletions(-) diff --git a/kalshi/fix/messages/settlement.py b/kalshi/fix/messages/settlement.py index 7ea3b69..5b5e3b6 100644 --- a/kalshi/fix/messages/settlement.py +++ b/kalshi/fix/messages/settlement.py @@ -34,10 +34,13 @@ class MarketSettlementReport(FixMessage): """MarketSettlementReport (35=UMS) — a market's settlement detail. ``market_result`` is a raw str (compare against - :class:`~kalshi.fix.enums.MarketResult`). ``settlement_price`` is in cents - (e.g. ``Decimal("100.00")``). ``clearing_business_date`` is the raw - ``YYYYMMDD`` string. ``last_fragment`` is ``False`` on non-final pages of a - paginated batch, ``True`` (or absent) on the last. + :class:`~kalshi.fix.enums.MarketResult`). ``settlement_price`` rides the FIX + ``Price`` field as a no-float ``Decimal`` (``DollarDecimal`` is just the + coercion type); the value is in *cents*, 2 dp — ``Decimal("100.00")`` is 100c + (full YES payout) and ``Decimal("30.60")`` a scalar market — not dollars. + ``clearing_business_date`` is the raw ``YYYYMMDD`` string. ``last_fragment`` + is ``False`` on non-final pages of a paginated batch, ``True`` (or absent) on + the last. """ MSG_TYPE = MsgType.MARKET_SETTLEMENT_REPORT diff --git a/tests/fix/test_settlement.py b/tests/fix/test_settlement.py index 4ca8739..d519fab 100644 --- a/tests/fix/test_settlement.py +++ b/tests/fix/test_settlement.py @@ -98,6 +98,20 @@ def test_settlement_report_nested_roundtrip() -> None: assert _roundtrip(msg) == msg +@pytest.mark.parametrize( + ("result", "wire"), + [(MarketResult.YES, "yes"), (MarketResult.NO, "no"), (MarketResult.SCALAR, "scalar")], +) +def test_settlement_report_market_result_values(result: MarketResult, wire: str) -> None: + msg = MarketSettlementReport( + symbol=SYM, market_result=wire, settlement_price=Decimal("30.60"), last_fragment=True + ) + assert dict(msg.to_body_fields())[int(Tag.MARKET_RESULT)] == wire + rt = _roundtrip(msg) + assert isinstance(rt, MarketSettlementReport) + assert rt.market_result == result + + def test_settlement_report_two_parties_with_nested_groups_roundtrip() -> None: # Each party carries its own collateral + misc-fee groups; the positional # decoder must keep them separated (no cross-party leakage). Exercises negative @@ -217,12 +231,15 @@ def _frag(symbol: str | None, party_ids: list[str], last: bool | None) -> Market ) -def test_reassembler_multi_fragment() -> None: +@pytest.mark.parametrize("terminal", [True, None]) +def test_reassembler_multi_fragment(terminal: bool | None) -> None: + # The terminal page may carry LastFragment=Y or omit it (absent == Y); both + # must merge the buffered fragments. r = SettlementReassembler() assert r.add(_frag(SYM, ["a"], False)) is None assert r.add(_frag(SYM, ["b"], False)) is None assert r.pending_symbols() == {SYM} - done = r.add(_frag(SYM, ["c"], True)) + done = r.add(_frag(SYM, ["c"], terminal)) assert done is not None assert [p.party_id for p in done.parties] == ["a", "b", "c"] assert r.pending_symbols() == set() @@ -251,10 +268,14 @@ def test_reassembler_routes_by_symbol() -> None: assert [p.party_id for p in done_b.parties] == ["b1", "b2"] -def test_reassembler_non_final_without_symbol_dropped() -> None: +def test_reassembler_non_final_without_symbol_dropped( + caplog: pytest.LogCaptureFixture, +) -> None: r = SettlementReassembler() - assert r.add(_frag(None, ["a"], False)) is None + with caplog.at_level(logging.WARNING, logger="kalshi.fix"): + assert r.add(_frag(None, ["a"], False)) is None assert r.pending_symbols() == set() + assert any("without Symbol" in rec.message for rec in caplog.records) def test_reassembler_sequential_same_symbol_batches() -> None: @@ -322,7 +343,11 @@ async def on_message(raw: RawMessage) -> None: await acceptor.push("UMS", f1.to_body_fields(), seq=2) await acceptor.push("UMS", f2.to_body_fields(), seq=3) await until(lambda: len(received) == 2) - results = [reasm.add(decode_app_message(r)) for r in received] # type: ignore[arg-type] + results: list[MarketSettlementReport | None] = [] + for raw in received: + decoded = decode_app_message(raw) + assert isinstance(decoded, MarketSettlementReport) # dispatch routed UMS correctly + results.append(reasm.add(decoded)) assert results[0] is None # first fragment buffered assert results[1] is not None assert [p.party_id for p in results[1].parties] == ["u1", "u2"]