diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index d20a506..1e231ea 100644 --- a/kalshi/fix/__init__.py +++ b/kalshi/fix/__init__.py @@ -64,32 +64,56 @@ KalshiFixError, ) from kalshi.fix.messages import ( + APP_MESSAGE_MODELS, + BusinessMessageReject, + CollateralAmountChange, + ExecutionReport, + FixGroup, + FixGroupMeta, FixMessage, Heartbeat, Logon, Logout, + MarketSettlementParty, + MiscFee, + MultivariateSelectedLeg, + NewOrderSingle, + OrderCancelReject, + OrderCancelReplaceRequest, + OrderCancelRequest, + OrderMassCancelReport, + OrderMassCancelRequest, + Party, Reject, ResendRequest, SequenceReset, TestRequest, + decode_app_message, + groupfield, ) from kalshi.fix.session import FixSession, FixSessionState from kalshi.fix.tags import Tag # Sorted (ruff RUF022); grouping is by the imports above, not by ``__all__`` order. __all__ = [ + "APP_MESSAGE_MODELS", "BEGIN_STRING_FIXT11", "SOH", "ApplVerID", + "BusinessMessageReject", + "CollateralAmountChange", "EncryptMethod", "ExecInst", "ExecType", + "ExecutionReport", "FixClient", "FixCodecError", "FixConfig", "FixConnection", "FixConnectionError", "FixEnvironment", + "FixGroup", + "FixGroupMeta", "FixLogonError", "FixMessage", "FixParser", @@ -105,9 +129,19 @@ "KalshiFixError", "Logon", "Logout", + "MarketSettlementParty", + "MiscFee", "MsgType", + "MultivariateSelectedLeg", + "NewOrderSingle", "OrdStatus", "OrdType", + "OrderCancelReject", + "OrderCancelReplaceRequest", + "OrderCancelRequest", + "OrderMassCancelReport", + "OrderMassCancelRequest", + "Party", "RawMessage", "Reject", "ResendRequest", @@ -119,5 +153,7 @@ "TestRequest", "TimeInForce", "decode", + "decode_app_message", "encode", + "groupfield", ] diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index 9b4f7f7..7713a4e 100644 --- a/kalshi/fix/messages/__init__.py +++ b/kalshi/fix/messages/__init__.py @@ -1,9 +1,9 @@ """Typed FIX message models (FIX Dictionary v1.03). Exposes the base framework (scalar + repeating-group fields), the session-layer -(admin) messages, and the shared repeating-group entry components. Application -messages (order entry, drop copy, market data, RFQ/settlement) are added in later -phases — see GH #402. +(admin) messages, the shared repeating-group entry components, and the +order-entry message flow. Market-data, drop-copy, and RFQ/settlement flows are +added in later phases — see GH #402. """ from __future__ import annotations @@ -23,6 +23,18 @@ MultivariateSelectedLeg, Party, ) +from kalshi.fix.messages.order_entry import ( + APP_MESSAGE_MODELS, + BusinessMessageReject, + ExecutionReport, + NewOrderSingle, + OrderCancelReject, + OrderCancelReplaceRequest, + OrderCancelRequest, + OrderMassCancelReport, + OrderMassCancelRequest, + decode_app_message, +) from kalshi.fix.messages.session import ( Heartbeat, Logon, @@ -34,7 +46,10 @@ ) __all__ = [ + "APP_MESSAGE_MODELS", + "BusinessMessageReject", "CollateralAmountChange", + "ExecutionReport", "FixGroup", "FixGroupMeta", "FixMessage", @@ -45,11 +60,18 @@ "MarketSettlementParty", "MiscFee", "MultivariateSelectedLeg", + "NewOrderSingle", + "OrderCancelReject", + "OrderCancelReplaceRequest", + "OrderCancelRequest", + "OrderMassCancelReport", + "OrderMassCancelRequest", "Party", "Reject", "ResendRequest", "SequenceReset", "TestRequest", + "decode_app_message", "fixfield", "groupfield", ] diff --git a/kalshi/fix/messages/base.py b/kalshi/fix/messages/base.py index 7242321..1f43d6f 100644 --- a/kalshi/fix/messages/base.py +++ b/kalshi/fix/messages/base.py @@ -25,6 +25,7 @@ from __future__ import annotations +import logging from dataclasses import dataclass from decimal import Decimal from enum import StrEnum @@ -37,6 +38,8 @@ from kalshi.fix.enums import MsgType from kalshi.fix.tags import DATA_LENGTH_FIELDS +logger = logging.getLogger("kalshi.fix") + # Reverse of DATA_LENGTH_FIELDS: data_tag -> length_tag. Used by to_body_fields # to auto-emit the length field immediately before a data field. _DATA_TO_LENGTH: dict[int, int] = {data: length for length, data in DATA_LENGTH_FIELDS.items()} @@ -77,7 +80,7 @@ class FixGroupMeta: """ num_in_group_tag: int - entry_model: type[_FixFieldModel] + entry_model: type[FixGroup] @dataclass(frozen=True) @@ -91,7 +94,7 @@ class _ScalarSpec: class _GroupSpec: name: str count_tag: int - entry_model: type[_FixFieldModel] + entry_model: type[FixGroup] _FieldSpec = _ScalarSpec | _GroupSpec @@ -173,8 +176,8 @@ def _parse_group( pairs: list[tuple[int, str]], start: int, count: int, - entry_model: type[_FixFieldModel], -) -> tuple[list[_FixFieldModel], int]: + entry_model: type[FixGroup], +) -> tuple[list[FixGroup], int]: """Parse up to ``count`` group entries from ``pairs`` beginning at ``start``. Each entry begins at the entry model's delimiter (first) tag and runs until @@ -184,7 +187,7 @@ def _parse_group( """ delim = entry_model._first_tag() member_tags = entry_model._member_tags() - entries: list[_FixFieldModel] = [] + entries: list[FixGroup] = [] i = start n = len(pairs) while len(entries) < count and i < n and pairs[i][0] == delim: @@ -194,6 +197,14 @@ def _parse_group( entry_pairs.append(pairs[i]) i += 1 entries.append(entry_model._from_pairs(entry_pairs)) + if len(entries) < count: + # Malformed FIX: NumInGroup promised more entries than were delivered. + logger.debug( + "FIX group %s: NumInGroup=%d but only %d entries parsed", + entry_model.__name__, + count, + len(entries), + ) return entries, i @@ -258,7 +269,12 @@ def _member_tags(cls) -> frozenset[int]: @classmethod def _first_tag(cls) -> int: - """The delimiter tag — the first declared field's tag (group entry start).""" + """The delimiter tag — the first declared field's tag (group entry start). + + Falls back to a leading group's ``NumInGroup`` tag, but no FIX group + places a nested group as an entry's first field, so in practice this is + always a scalar tag. + """ for spec in cls._layout(): return spec.tag if isinstance(spec, _ScalarSpec) else spec.count_tag raise ValueError(f"{cls.__name__} declares no FIX fields") @@ -309,6 +325,11 @@ def _from_pairs(cls, pairs: list[tuple[int, str]]) -> Self: try: count = int(pairs[idx][1]) except ValueError: + logger.debug( + "FIX group %s: non-integer NumInGroup %r", + spec.entry_model.__name__, + pairs[idx][1], + ) continue entries, _ = _parse_group(pairs, idx + 1, count, spec.entry_model) kwargs[spec.name] = entries diff --git a/kalshi/fix/messages/order_entry.py b/kalshi/fix/messages/order_entry.py new file mode 100644 index 0000000..6186f62 --- /dev/null +++ b/kalshi/fix/messages/order_entry.py @@ -0,0 +1,268 @@ +"""Order-entry FIX messages (GH #424). + +Outbound requests (NewOrderSingle / OrderCancelRequest / OrderCancelReplaceRequest +/ OrderMassCancelRequest) use the typed enum vocabulary and keep required fields +required, so a caller error fails at construction. Inbound reports +(ExecutionReport / OrderCancelReject / OrderMassCancelReport / +BusinessMessageReject) keep their fields optional and carry char/int code fields +as raw ``str``/``int`` (compare against :mod:`kalshi.fix.enums`) so a slightly +off-spec report from the server parses cleanly rather than rejecting. + +Pricing: ``Price`` (44) rides the FIX ``PRICE`` field; its units follow the +session's ``UseDollars``: integer cents (1-99) on the prediction product by +default, fixed-point dollars on margin (and on prediction with UseDollars=Y). +Quantities are fractional decimals. +""" + +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, + MsgType, + OrdType, + SelfTradePreventionType, + Side, + TimeInForce, +) +from kalshi.fix.messages.base import ( + FixGroupMeta, + FixMessage, + FixType, + fixfield, + groupfield, +) +from kalshi.fix.messages.components import CollateralAmountChange, MiscFee, Party +from kalshi.fix.tags import Tag +from kalshi.types import DollarDecimal, FixedPointCount + +logger = logging.getLogger("kalshi.fix") + +# --------------------------------------------------------------------------- +# Outbound requests +# --------------------------------------------------------------------------- + + +class NewOrderSingle(FixMessage): + """NewOrderSingle (35=D) — submit a new order. + + ``TransactTime`` (60) is intentionally not sent: the Kalshi dictionary v1.03 + does not include it on 35=D (the gateway stamps its own time), unlike the + generic FIX 5.0SP2 message. + """ + + MSG_TYPE = MsgType.NEW_ORDER_SINGLE + + cl_ord_id: str = fixfield(Tag.CL_ORD_ID, FixType.STRING) + exec_inst: ExecInst | None = fixfield(Tag.EXEC_INST, FixType.MULTIPLEVALUESTRING, default=None) + order_qty: FixedPointCount = fixfield(Tag.ORDER_QTY, FixType.QTY) + ord_type: OrdType = fixfield(Tag.ORD_TYPE, FixType.CHAR, default=OrdType.LIMIT) + price: DollarDecimal = fixfield(Tag.PRICE, FixType.PRICE) + side: Side = fixfield(Tag.SIDE, FixType.CHAR) + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + time_in_force: TimeInForce | None = fixfield(Tag.TIME_IN_FORCE, FixType.CHAR, default=None) + expire_time: datetime | None = fixfield(Tag.EXPIRE_TIME, FixType.UTCTIMESTAMP, default=None) + self_trade_prevention_type: SelfTradePreventionType | None = fixfield( + Tag.SELF_TRADE_PREVENTION_TYPE, FixType.INT, default=None + ) + parties: Annotated[list[Party], FixGroupMeta(Tag.NO_PARTY_IDS, Party)] = groupfield() + secondary_cl_ord_id: str | None = fixfield( + Tag.SECONDARY_CL_ORD_ID, FixType.STRING, default=None + ) + order_group_id: str | None = fixfield(Tag.ORDER_GROUP_ID, FixType.STRING, default=None) + cancel_order_on_pause: bool | None = fixfield( + Tag.CANCEL_ORDER_ON_PAUSE, FixType.BOOLEAN, default=None + ) + max_execution_cost: DollarDecimal | None = fixfield( + Tag.MAX_EXECUTION_COST, FixType.DECIMAL, default=None + ) + # Kalshi dictionary v1.03 types AllocAccount (79) as INT — the subaccount + # number (0 primary, 1-32) — not the FIX-standard STRING. + alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None) + + +class OrderCancelReplaceRequest(FixMessage): + """OrderCancelReplaceRequest (35=G) — amend an order's price/quantity.""" + + MSG_TYPE = MsgType.ORDER_CANCEL_REPLACE_REQUEST + + cl_ord_id: str = fixfield(Tag.CL_ORD_ID, FixType.STRING) + order_id: str | None = fixfield(Tag.ORDER_ID, FixType.STRING, default=None) + orig_cl_ord_id: str = fixfield(Tag.ORIG_CL_ORD_ID, FixType.STRING) + order_qty: FixedPointCount = fixfield(Tag.ORDER_QTY, FixType.QTY) + ord_type: OrdType = fixfield(Tag.ORD_TYPE, FixType.CHAR, default=OrdType.LIMIT) + price: DollarDecimal | None = fixfield(Tag.PRICE, FixType.PRICE, default=None) + side: Side = fixfield(Tag.SIDE, FixType.CHAR) + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + parties: Annotated[list[Party], FixGroupMeta(Tag.NO_PARTY_IDS, Party)] = groupfield() + order_group_id: str | None = fixfield(Tag.ORDER_GROUP_ID, FixType.STRING, default=None) + alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None) + + +class OrderCancelRequest(FixMessage): + """OrderCancelRequest (35=F) — cancel an order's remaining quantity.""" + + MSG_TYPE = MsgType.ORDER_CANCEL_REQUEST + + cl_ord_id: str = fixfield(Tag.CL_ORD_ID, FixType.STRING) + order_id: str | None = fixfield(Tag.ORDER_ID, FixType.STRING, default=None) + orig_cl_ord_id: str = fixfield(Tag.ORIG_CL_ORD_ID, FixType.STRING) + side: Side = fixfield(Tag.SIDE, FixType.CHAR) + symbol: str = fixfield(Tag.SYMBOL, FixType.STRING) + parties: Annotated[list[Party], FixGroupMeta(Tag.NO_PARTY_IDS, Party)] = groupfield() + alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None) + + +class OrderMassCancelRequest(FixMessage): + """OrderMassCancelRequest (35=q) — cancel all orders for the session (NR only).""" + + MSG_TYPE = MsgType.ORDER_MASS_CANCEL_REQUEST + + cl_ord_id: str = fixfield(Tag.CL_ORD_ID, FixType.STRING) + mass_cancel_request_type: MassCancelRequestType = fixfield( + Tag.MASS_CANCEL_REQUEST_TYPE, FixType.CHAR, default=MassCancelRequestType.CANCEL_FOR_SESSION + ) + + +# --------------------------------------------------------------------------- +# Inbound reports (fields optional; codes raw for robustness) +# --------------------------------------------------------------------------- + + +class ExecutionReport(FixMessage): + """ExecutionReport (35=8) — order state change from the exchange. + + ``exec_type`` / ``ord_status`` / ``side`` are raw chars (compare against + :class:`~kalshi.fix.enums.ExecType` / ``OrdStatus`` / ``Side``); rejection and + STP codes are raw ints. Position/fee/collateral detail is present on + ``ExecType=Trade`` via the misc-fees and collateral groups. + """ + + MSG_TYPE = MsgType.EXECUTION_REPORT + + order_id: str | None = fixfield(Tag.ORDER_ID, FixType.STRING, default=None) + cl_ord_id: str | None = fixfield(Tag.CL_ORD_ID, FixType.STRING, default=None) + # Kalshi compound "int;int" exec id (e.g. "4;7"; "-1;-1" for PENDING reports); + # kept as the raw string, not normalized. + exec_id: str | None = fixfield(Tag.EXEC_ID, FixType.STRING, default=None) + exec_type: str | None = fixfield(Tag.EXEC_TYPE, FixType.CHAR, default=None) + ord_status: str | None = fixfield(Tag.ORD_STATUS, FixType.CHAR, default=None) + side: str | None = fixfield(Tag.SIDE, FixType.CHAR, default=None) + symbol: str | None = fixfield(Tag.SYMBOL, FixType.STRING, default=None) + leaves_qty: FixedPointCount | None = fixfield(Tag.LEAVES_QTY, FixType.QTY, default=None) + cum_qty: FixedPointCount | None = fixfield(Tag.CUM_QTY, FixType.QTY, default=None) + avg_px: DollarDecimal | None = fixfield(Tag.AVG_PX, FixType.PRICE, default=None) + order_qty: FixedPointCount | None = fixfield(Tag.ORDER_QTY, FixType.QTY, default=None) + last_px: DollarDecimal | None = fixfield(Tag.LAST_PX, FixType.PRICE, default=None) + last_qty: FixedPointCount | None = fixfield(Tag.LAST_QTY, FixType.QTY, default=None) + ord_rej_reason: int | None = fixfield(Tag.ORD_REJ_REASON, FixType.INT, default=None) + orig_cl_ord_id: str | None = fixfield(Tag.ORIG_CL_ORD_ID, FixType.STRING, default=None) + price: DollarDecimal | None = fixfield(Tag.PRICE, FixType.PRICE, default=None) + transact_time: datetime | None = fixfield(Tag.TRANSACT_TIME, FixType.UTCTIMESTAMP, default=None) + text: str | None = fixfield(Tag.TEXT, FixType.STRING, default=None) + expire_time: datetime | None = fixfield(Tag.EXPIRE_TIME, FixType.UTCTIMESTAMP, default=None) + exec_restatement_reason: int | None = fixfield( + Tag.EXEC_RESTATEMENT_REASON, FixType.INT, default=None + ) + misc_fees: Annotated[list[MiscFee], FixGroupMeta(Tag.NO_MISC_FEES, MiscFee)] = groupfield() + parties: Annotated[list[Party], FixGroupMeta(Tag.NO_PARTY_IDS, Party)] = groupfield() + alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None) + long_qty: FixedPointCount | None = fixfield(Tag.LONG_QTY, FixType.QTY, default=None) + short_qty: FixedPointCount | None = fixfield(Tag.SHORT_QTY, FixType.QTY, default=None) + collateral_amount_changes: Annotated[ + list[CollateralAmountChange], + FixGroupMeta(Tag.NO_COLLATERAL_AMOUNT_CHANGES, CollateralAmountChange), + ] = groupfield() + trd_match_id: str | None = fixfield(Tag.TRD_MATCH_ID, FixType.STRING, default=None) + aggressor_indicator: bool | None = fixfield( + Tag.AGGRESSOR_INDICATOR, FixType.BOOLEAN, default=None + ) + self_trade_prevention_type: int | None = fixfield( + Tag.SELF_TRADE_PREVENTION_TYPE, FixType.INT, default=None + ) + + +class OrderCancelReject(FixMessage): + """OrderCancelReject (35=9) — an amend/cancel was rejected by the exchange.""" + + MSG_TYPE = MsgType.ORDER_CANCEL_REJECT + + order_id: str | None = fixfield(Tag.ORDER_ID, FixType.STRING, default=None) + cl_ord_id: str | None = fixfield(Tag.CL_ORD_ID, FixType.STRING, default=None) + orig_cl_ord_id: str | None = fixfield(Tag.ORIG_CL_ORD_ID, FixType.STRING, default=None) + ord_status: str | None = fixfield(Tag.ORD_STATUS, FixType.CHAR, default=None) + cxl_rej_response_to: str | None = fixfield(Tag.CXL_REJ_RESPONSE_TO, FixType.CHAR, default=None) + cxl_rej_reason: int | None = fixfield(Tag.CXL_REJ_REASON, FixType.INT, default=None) + text: str | None = fixfield(Tag.TEXT, FixType.STRING, default=None) + + +class OrderMassCancelReport(FixMessage): + """OrderMassCancelReport (35=r) — response to a mass-cancel request.""" + + MSG_TYPE = MsgType.ORDER_MASS_CANCEL_REPORT + + cl_ord_id: str | None = fixfield(Tag.CL_ORD_ID, FixType.STRING, default=None) + order_id: str | None = fixfield(Tag.ORDER_ID, FixType.STRING, default=None) + mass_cancel_response: str | None = fixfield( + Tag.MASS_CANCEL_RESPONSE, FixType.CHAR, default=None + ) + mass_cancel_reject_reason: int | None = fixfield( + Tag.MASS_CANCEL_REJECT_REASON, FixType.INT, default=None + ) + + +class BusinessMessageReject(FixMessage): + """BusinessMessageReject (35=j) — application-level rejection of a message.""" + + MSG_TYPE = MsgType.BUSINESS_MESSAGE_REJECT + + ref_seq_num: int | None = fixfield(Tag.REF_SEQ_NUM, FixType.SEQNUM, default=None) + ref_msg_type: str | None = fixfield(Tag.REF_MSG_TYPE, FixType.STRING, default=None) + business_reject_ref_id: str | None = fixfield( + Tag.BUSINESS_REJECT_REF_ID, FixType.STRING, default=None + ) + business_reject_reason: int | None = fixfield( + 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/tests/fix/test_groups.py b/tests/fix/test_groups.py index 9714cda..9f5b360 100644 --- a/tests/fix/test_groups.py +++ b/tests/fix/test_groups.py @@ -5,7 +5,7 @@ from decimal import Decimal from typing import Annotated -from kalshi.fix.codec import SOH, decode, encode +from kalshi.fix.codec import SOH, RawMessage, decode, encode from kalshi.fix.enums import MsgType from kalshi.fix.messages.base import ( FixGroupMeta, @@ -161,6 +161,39 @@ def test_nested_group_roundtrip() -> None: assert back.parties[1].collateral_amount_changes[0].collateral_amount_type == "PAYOUT" +def test_short_numingroup_parses_available_entries() -> None: + # NumInGroup promises 3 entries but only 1 is delivered — parse the one present + # (the short-count debug path in _parse_group) and keep the trailing scalar. + raw = RawMessage( + [ + (int(Tag.MSG_TYPE), "D"), + (int(Tag.CL_ORD_ID), "x"), + (int(Tag.NO_PARTY_IDS), "3"), + (int(Tag.PARTY_ID), "0"), + (int(Tag.PARTY_ROLE), "24"), + (int(Tag.SYMBOL), "Y"), + ] + ) + msg = _OrderWithParties.from_raw(raw) + assert len(msg.parties) == 1 + assert msg.symbol == "Y" + + +def test_non_integer_numingroup_drops_group() -> None: + # A non-integer NumInGroup drops the group (debug path) but keeps scalars. + raw = RawMessage( + [ + (int(Tag.MSG_TYPE), "D"), + (int(Tag.CL_ORD_ID), "x"), + (int(Tag.NO_PARTY_IDS), "abc"), + (int(Tag.SYMBOL), "Y"), + ] + ) + msg = _OrderWithParties.from_raw(raw) + assert msg.parties == [] + assert msg.symbol == "Y" + + def test_nested_group_wire_structure() -> None: msg = _Settlement( report_id="R1", diff --git a/tests/fix/test_order_entry.py b/tests/fix/test_order_entry.py new file mode 100644 index 0000000..6502391 --- /dev/null +++ b/tests/fix/test_order_entry.py @@ -0,0 +1,346 @@ +"""Tests for the order-entry FIX message flow (GH #424).""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from decimal import Decimal + +from kalshi.fix.auth import FixSigner +from kalshi.fix.codec import RawMessage, decode, encode +from kalshi.fix.config import FixConfig, FixSessionType +from kalshi.fix.enums import ( + ExecInst, + ExecType, + OrdStatus, + OrdType, + SelfTradePreventionType, + Side, + TimeInForce, +) +from kalshi.fix.messages import ( + BusinessMessageReject, + CollateralAmountChange, + ExecutionReport, + MiscFee, + NewOrderSingle, + OrderCancelReject, + OrderCancelReplaceRequest, + OrderCancelRequest, + OrderMassCancelReport, + OrderMassCancelRequest, + Party, + 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_new_order_single_roundtrip_with_parties() -> None: + order = NewOrderSingle( + cl_ord_id="abc", + order_qty=Decimal("10"), + price=Decimal("0.7500"), + side=Side.BUY_YES, + symbol="KXTEST-25", + time_in_force=TimeInForce.GTC, + exec_inst=ExecInst.POST_ONLY, + self_trade_prevention_type=SelfTradePreventionType.MAKER, + parties=[Party(party_id="0", party_role=24)], + ) + back = _roundtrip(order) + assert back == order + assert back.side is Side.BUY_YES + assert back.ord_type is OrdType.LIMIT # default + assert back.parties[0].party_id == "0" + + +def test_new_order_enum_and_price_wire_values() -> None: + order = NewOrderSingle( + cl_ord_id="x", + order_qty=Decimal("3"), + price=Decimal("0.7500"), + side=Side.SELL_NO, + symbol="Y", + time_in_force=TimeInForce.IOC, + exec_inst=ExecInst.POST_ONLY, + self_trade_prevention_type=SelfTradePreventionType.TAKER_AT_CROSS, + ) + body = dict(order.to_body_fields()) + assert body[int(Tag.SIDE)] == "2" # SELL_NO + assert body[int(Tag.ORD_TYPE)] == "2" # LIMIT + assert body[int(Tag.TIME_IN_FORCE)] == "3" # IOC + assert body[int(Tag.EXEC_INST)] == "6" # POST_ONLY + assert body[int(Tag.SELF_TRADE_PREVENTION_TYPE)] == "1" # TAKER_AT_CROSS + assert body[int(Tag.PRICE)] == "0.7500" # fixed-point dollars preserved + + +def test_price_cents_vs_dollars_encoding() -> None: + # The Decimal value is emitted verbatim; units follow the session UseDollars. + cents = NewOrderSingle( + cl_ord_id="c", order_qty=Decimal("1"), price=Decimal("75"), side=Side.BUY_YES, symbol="Y" + ) + assert dict(cents.to_body_fields())[int(Tag.PRICE)] == "75" + dollars = NewOrderSingle( + cl_ord_id="d", order_qty=Decimal("1"), price=Decimal("19.5"), side=Side.BUY_YES, symbol="Y" + ) + assert dict(dollars.to_body_fields())[int(Tag.PRICE)] == "19.5" + + +def test_cancel_request_roundtrip() -> None: + msg = OrderCancelRequest( + cl_ord_id="c1", orig_cl_ord_id="o1", side=Side.BUY_YES, symbol="KXTEST", alloc_account=3 + ) + assert _roundtrip(msg) == msg + + +def test_cancel_replace_roundtrip() -> None: + msg = OrderCancelReplaceRequest( + cl_ord_id="r1", + orig_cl_ord_id="o1", + order_qty=Decimal("5"), + price=Decimal("0.6000"), + side=Side.SELL_NO, + symbol="KXTEST", + order_group_id="grp-1", + ) + back = _roundtrip(msg) + assert back == msg + assert back.price == Decimal("0.6000") + + +def test_mass_cancel_request_roundtrip() -> None: + msg = OrderMassCancelRequest(cl_ord_id="m1") + body = dict(msg.to_body_fields()) + assert body[int(Tag.MASS_CANCEL_REQUEST_TYPE)] == "6" # CANCEL_FOR_SESSION + assert _roundtrip(msg) == msg + + +def test_execution_report_trade_roundtrip() -> None: + report = ExecutionReport( + order_id="OID-1", + cl_ord_id="abc", + exec_id="4;7", + exec_type=ExecType.TRADE.value, + ord_status=OrdStatus.PARTIALLY_FILLED.value, + side=Side.BUY_YES.value, + symbol="KXTEST", + leaves_qty=Decimal("5"), + cum_qty=Decimal("5"), + avg_px=Decimal("0.6600"), + order_qty=Decimal("10"), + last_px=Decimal("0.7000"), + last_qty=Decimal("5"), + long_qty=Decimal("5"), + short_qty=Decimal("0"), + trd_match_id="T-1", + aggressor_indicator=True, + misc_fees=[ + MiscFee( + misc_fee_amt=Decimal("0.02"), + misc_fee_curr="USD", + misc_fee_type="4", + misc_fee_basis=0, + ) + ], + collateral_amount_changes=[ + CollateralAmountChange( + collateral_amount_change=Decimal("1.50"), collateral_amount_type="BALANCE" + ) + ], + parties=[Party(party_id="0", party_role=24)], + ) + back = _roundtrip(report) + assert back == report + # Raw char codes compare against the enums (StrEnum equality with str). + assert back.exec_type == ExecType.TRADE + assert back.ord_status == OrdStatus.PARTIALLY_FILLED + assert back.misc_fees[0].misc_fee_amt == Decimal("0.02") + assert back.collateral_amount_changes[0].collateral_amount_type == "BALANCE" + assert back.aggressor_indicator is True + + +def test_inbound_report_tolerates_minimal_fields() -> None: + # A sparse report (e.g. PENDING_NEW) still parses; absent fields are None. + report = ExecutionReport( + cl_ord_id="abc", + exec_id="-1;-1", + exec_type=ExecType.PENDING_NEW.value, + ord_status=OrdStatus.PENDING_NEW.value, + ) + back = _roundtrip(report) + assert back == report + assert back.avg_px is None + assert back.misc_fees == [] + + +def test_order_cancel_reject_roundtrip() -> None: + msg = OrderCancelReject( + order_id="OID-1", + cl_ord_id="c1", + orig_cl_ord_id="o1", + ord_status=OrdStatus.NEW.value, + cxl_rej_reason=1, + text="UNKNOWN_ORDER", + ) + assert _roundtrip(msg) == msg + + +def test_mass_cancel_report_roundtrip() -> None: + msg = OrderMassCancelReport(cl_ord_id="m1", order_id="op-1", mass_cancel_response="6") + assert _roundtrip(msg) == msg + + +def test_business_message_reject_roundtrip() -> None: + msg = BusinessMessageReject( + ref_seq_num=5, ref_msg_type="D", business_reject_reason=3, text="UNSUPPORTED_MESSAGE_TYPE" + ) + assert _roundtrip(msg) == msg + + +def test_decode_app_message_dispatch() -> None: + report = ExecutionReport( + cl_ord_id="abc", exec_type=ExecType.NEW.value, ord_status=OrdStatus.NEW.value + ) + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), "8"), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + *report.to_body_fields(), + ] + ) + ) + decoded = decode_app_message(raw) + assert isinstance(decoded, ExecutionReport) + assert decoded.cl_ord_id == "abc" + # Admin / unregistered message types return None. + heartbeat = RawMessage([(int(Tag.MSG_TYPE), "0"), (int(Tag.MSG_SEQ_NUM), "3")]) + assert decode_app_message(heartbeat) is None + # A message with no MsgType at all also returns None (no dispatch key). + assert decode_app_message(RawMessage([])) is None + + +def test_decode_app_message_all_inbound_types() -> None: + for msg in ( + OrderCancelReject(cl_ord_id="c", cxl_rej_reason=1, text="UNKNOWN_ORDER"), + OrderMassCancelReport(cl_ord_id="m", mass_cancel_response="6"), + BusinessMessageReject(business_reject_reason=3, text="x"), + ): + 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_decode_app_message_returns_none_on_malformed() -> None: + # A malformed inbound payload is swallowed (logged) rather than raised into + # the consumer's on_message — bad bool (ValueError) and bad Decimal + # (ArithmeticError) both yield None. + bad_bool = RawMessage([(int(Tag.MSG_TYPE), "8"), (int(Tag.AGGRESSOR_INDICATOR), "X")]) + assert decode_app_message(bad_bool) is None + bad_decimal = RawMessage([(int(Tag.MSG_TYPE), "8"), (int(Tag.AVG_PX), "notanumber")]) + assert decode_app_message(bad_decimal) is None + + +def test_cancel_replace_without_price_is_qty_only() -> None: + # Kalshi allows a quantity-only amend (Price is required only when changing it). + msg = OrderCancelReplaceRequest( + cl_ord_id="r", orig_cl_ord_id="o", order_qty=Decimal("5"), side=Side.BUY_YES, symbol="Y" + ) + assert int(Tag.PRICE) not in {t for t, _ in msg.to_body_fields()} + back = _roundtrip(msg) + assert back == msg + assert back.price is None + + +def test_new_order_parties_default_empty() -> None: + order = NewOrderSingle( + cl_ord_id="x", + order_qty=Decimal("1"), + price=Decimal("0.5"), + side=Side.BUY_YES, + symbol="Y", + ) + assert order.parties == [] + assert int(Tag.NO_PARTY_IDS) not in {t for t, _ in order.to_body_fields()} + + +async def test_send_order_and_receive_execution_report( + 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.ORDER_ENTRY_NR, on_message=on_message + ) + await session.start() + try: + seq = await session.send( + NewOrderSingle( + cl_ord_id="abc", + order_qty=Decimal("10"), + price=Decimal("0.7500"), + side=Side.BUY_YES, + symbol="KXTEST-25", + ) + ) + assert seq == 2 # logon consumed 1 + await until(lambda: acceptor.first("D") is not None) + sent = acceptor.first("D") + assert sent is not None + assert sent.get(Tag.CL_ORD_ID) == "abc" + assert sent.get(Tag.SIDE) == "1" + assert sent.get(Tag.PRICE) == "0.7500" + + # Server streams back an ExecutionReport; the consumer decodes it typed. + report = ExecutionReport( + order_id="OID-1", + cl_ord_id="abc", + exec_id="1;1", + exec_type=ExecType.NEW.value, + ord_status=OrdStatus.NEW.value, + side=Side.BUY_YES.value, + symbol="KXTEST-25", + leaves_qty=Decimal("10"), + cum_qty=Decimal("0"), + avg_px=Decimal("0"), + order_qty=Decimal("10"), + ) + await acceptor.push("8", report.to_body_fields(), seq=2) + await until(lambda: len(received) == 1) + decoded = decode_app_message(received[0]) + assert isinstance(decoded, ExecutionReport) + assert decoded.order_id == "OID-1" + assert decoded.ord_status == OrdStatus.NEW + assert decoded.leaves_qty == Decimal("10") + finally: + await session.close()