feat(fix): order-group lifecycle messages (closes #427)

kalshi/fix/__init__.py

@@ -51,6 +51,7 @@
     MDReqRejReason,
     MDUpdateAction,
     MsgType,
+    OrderGroupAction,
     OrdStatus,
     OrdType,
     SecurityTradingStatus,
@@ -96,6 +97,8 @@
     OrderCancelReject,
     OrderCancelReplaceRequest,
     OrderCancelRequest,
+    OrderGroupRequest,
+    OrderGroupResponse,
     OrderMassCancelReport,
     OrderMassCancelRequest,
     Party,
@@ -174,6 +177,9 @@
     "OrderCancelReject",
     "OrderCancelReplaceRequest",
     "OrderCancelRequest",
+    "OrderGroupAction",
+    "OrderGroupRequest",
+    "OrderGroupResponse",
     "OrderMassCancelReport",
     "OrderMassCancelRequest",
     "Party",

kalshi/fix/enums.py

@@ -248,6 +248,16 @@ class PartyRole(IntEnum):
     CUSTOMER_ACCOUNT = 24
 
 
+class OrderGroupAction(IntEnum):
+    """Tag 20131 — order-group lifecycle operation (35=UOG)."""
+
+    CREATE = 1
+    RESET = 2
+    DELETE = 3
+    TRIGGER_CANCEL = 4
+    UPDATE = 5
+
+
 class EventResendRejectReason(IntEnum):
     """Tag 21004 — why a drop-copy EventResendRequest (35=U3) was rejected."""
 

kalshi/fix/messages/__init__.py

@@ -50,6 +50,7 @@
     OrderMassCancelReport,
     OrderMassCancelRequest,
 )
+from kalshi.fix.messages.order_groups import OrderGroupRequest, OrderGroupResponse
 from kalshi.fix.messages.session import (
     Heartbeat,
     Logon,
@@ -88,6 +89,8 @@
     "OrderCancelReject",
     "OrderCancelReplaceRequest",
     "OrderCancelRequest",
+    "OrderGroupRequest",
+    "OrderGroupResponse",
     "OrderMassCancelReport",
     "OrderMassCancelRequest",
     "Party",

kalshi/fix/messages/dispatch.py

@@ -30,6 +30,7 @@
     OrderCancelReject,
     OrderMassCancelReport,
 )
+from kalshi.fix.messages.order_groups import OrderGroupResponse
 
 logger = logging.getLogger("kalshi.fix")
 
@@ -46,6 +47,7 @@
         MsgType.MARKET_DATA_INCREMENTAL_REFRESH.value: MarketDataIncrementalRefresh,
         MsgType.MARKET_DATA_REQUEST_REJECT.value: MarketDataRequestReject,
         MsgType.SECURITY_STATUS.value: SecurityStatus,
+        MsgType.ORDER_GROUP_RESPONSE.value: OrderGroupResponse,
     }
 )
 

kalshi/fix/messages/order_groups.py

@@ -0,0 +1,109 @@
+"""Order-group FIX messages (GH #427).
+
+Order groups ride the order-entry session (no dedicated session type) and manage
+automatic position limits: an ``OrderGroupRequest`` (35=UOG) creates / resets /
+deletes / triggers / updates a group, and the gateway replies with an
+``OrderGroupResponse`` (35=UOH). ``OrderGroupID`` (20130) is then carried on
+``NewOrderSingle`` / ``OrderCancelReplaceRequest`` to bind orders to the group.
+
+Groups are scoped per subaccount: a follow-up action (reset/delete/trigger/
+update) must pass the same ``AllocAccount`` (79) the group was created under, or
+the exchange returns a ``BusinessMessageReject`` (35=j). Build requests via the
+:meth:`~OrderGroupRequest.create` / :meth:`~OrderGroupRequest.reset` /
+:meth:`~OrderGroupRequest.delete` / :meth:`~OrderGroupRequest.trigger` /
+:meth:`~OrderGroupRequest.update` helpers, which encode each action's required
+fields (Create omits ``OrderGroupID`` — the server generates it).
+"""
+
+from __future__ import annotations
+
+from kalshi.fix.enums import MsgType, OrderGroupAction
+from kalshi.fix.messages.base import FixMessage, FixType, fixfield
+from kalshi.fix.tags import Tag
+
+
+class OrderGroupRequest(FixMessage):
+    """OrderGroupRequest (35=UOG) — create/reset/delete/trigger/update a group.
+
+    Field order follows the docs examples (action, id, contracts-limit, account);
+    FIX scalar field order is not significant, so the helpers below are the
+    intended construction path.
+    """
+
+    MSG_TYPE = MsgType.ORDER_GROUP_REQUEST
+
+    order_group_action: OrderGroupAction = fixfield(Tag.ORDER_GROUP_ACTION, FixType.INT)
+    order_group_id: str | None = fixfield(Tag.ORDER_GROUP_ID, FixType.STRING, default=None)
+    order_group_contracts_limit: int | None = fixfield(
+        Tag.ORDER_GROUP_CONTRACTS_LIMIT, FixType.INT, default=None
+    )
+    # Kalshi types AllocAccount (79) as INT — the subaccount number (0 primary,
+    # 1-32). Scopes the action to the group owned by that subaccount.
+    alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None)
+
+    @classmethod
+    def create(
+        cls, contracts_limit: int, *, alloc_account: int | None = None
+    ) -> OrderGroupRequest:
+        """Create a group with a contracts limit (the server assigns the id)."""
+        return cls(
+            order_group_action=OrderGroupAction.CREATE,
+            order_group_contracts_limit=contracts_limit,
+            alloc_account=alloc_account,
+        )
+
+    @classmethod
+    def reset(cls, order_group_id: str, *, alloc_account: int | None = None) -> OrderGroupRequest:
+        """Reset a group's rolling contract count."""
+        return cls(
+            order_group_action=OrderGroupAction.RESET,
+            order_group_id=order_group_id,
+            alloc_account=alloc_account,
+        )
+
+    @classmethod
+    def delete(cls, order_group_id: str, *, alloc_account: int | None = None) -> OrderGroupRequest:
+        """Delete a group (cancels all resting orders in it)."""
+        return cls(
+            order_group_action=OrderGroupAction.DELETE,
+            order_group_id=order_group_id,
+            alloc_account=alloc_account,
+        )
+
+    @classmethod
+    def trigger(cls, order_group_id: str, *, alloc_account: int | None = None) -> OrderGroupRequest:
+        """Trigger a group — immediately cancel all its orders (TriggerCancel=4)."""
+        return cls(
+            order_group_action=OrderGroupAction.TRIGGER_CANCEL,
+            order_group_id=order_group_id,
+            alloc_account=alloc_account,
+        )
+
+    @classmethod
+    def update(
+        cls, order_group_id: str, contracts_limit: int, *, alloc_account: int | None = None
+    ) -> OrderGroupRequest:
+        """Update a group's contracts limit."""
+        return cls(
+            order_group_action=OrderGroupAction.UPDATE,
+            order_group_id=order_group_id,
+            order_group_contracts_limit=contracts_limit,
+            alloc_account=alloc_account,
+        )
+
+
+class OrderGroupResponse(FixMessage):
+    """OrderGroupResponse (35=UOH) — the gateway's reply to an OrderGroupRequest.
+
+    ``order_group_contracts_limit`` is echoed only on Create and Update responses.
+    Fields are optional for inbound robustness; business-logic errors arrive as a
+    ``BusinessMessageReject`` (35=j), malformed fields as a session Reject (35=3).
+    """
+
+    MSG_TYPE = MsgType.ORDER_GROUP_RESPONSE
+
+    order_group_id: str | None = fixfield(Tag.ORDER_GROUP_ID, FixType.STRING, default=None)
+    order_group_contracts_limit: int | None = fixfield(
+        Tag.ORDER_GROUP_CONTRACTS_LIMIT, FixType.INT, default=None
+    )
+    alloc_account: int | None = fixfield(Tag.ALLOC_ACCOUNT, FixType.INT, default=None)

tests/fix/test_order_groups.py

@@ -0,0 +1,173 @@
+"""Tests for the order-group FIX flow (GH #427)."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+import pytest
+
+from kalshi.fix.auth import FixSigner
+from kalshi.fix.codec import RawMessage, decode, encode
+from kalshi.fix.config import FixConfig, FixSessionType
+from kalshi.fix.enums import OrderGroupAction
+from kalshi.fix.messages import (
+    OrderGroupRequest,
+    OrderGroupResponse,
+    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
+
+GID = "770e8400-e29b-41d4-a716-446655440002"
+
+
+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())
+
+
+# ---------------------------------------------------------------------------
+# Outbound: OrderGroupRequest (35=UOG) — golden wire per the docs examples
+# ---------------------------------------------------------------------------
+
+
+def test_order_group_create_golden_wire() -> None:
+    msg = OrderGroupRequest.create(5000)
+    # Create omits OrderGroupID (the server assigns it).
+    assert _wire(msg) == "20131=1|20132=5000"
+    assert int(Tag.ORDER_GROUP_ID) not in {t for t, _ in msg.to_body_fields()}
+    assert _roundtrip(msg) == msg
+
+
+def test_order_group_create_with_subaccount_golden_wire() -> None:
+    assert _wire(OrderGroupRequest.create(5000, alloc_account=2)) == "20131=1|20132=5000|79=2"
+
+
+def test_order_group_reset_golden_wire() -> None:
+    assert _wire(OrderGroupRequest.reset(GID)) == f"20131=2|20130={GID}"
+
+
+def test_order_group_delete_golden_wire() -> None:
+    assert _wire(OrderGroupRequest.delete(GID)) == f"20131=3|20130={GID}"
+
+
+def test_order_group_trigger_golden_wire() -> None:
+    assert _wire(OrderGroupRequest.trigger(GID)) == f"20131=4|20130={GID}"
+
+
+def test_order_group_update_golden_wire() -> None:
+    assert _wire(OrderGroupRequest.update(GID, 2500)) == f"20131=5|20130={GID}|20132=2500"
+
+
+@pytest.mark.parametrize(
+    ("msg", "action"),
+    [
+        (OrderGroupRequest.create(1), OrderGroupAction.CREATE),
+        (OrderGroupRequest.reset(GID), OrderGroupAction.RESET),
+        (OrderGroupRequest.delete(GID), OrderGroupAction.DELETE),
+        (OrderGroupRequest.trigger(GID), OrderGroupAction.TRIGGER_CANCEL),
+        (OrderGroupRequest.update(GID, 9), OrderGroupAction.UPDATE),
+    ],
+)
+def test_order_group_request_roundtrip(msg: OrderGroupRequest, action: OrderGroupAction) -> None:
+    rt = _roundtrip(msg)
+    assert rt == msg
+    assert isinstance(rt, OrderGroupRequest)
+    assert rt.order_group_action == action
+
+
+# ---------------------------------------------------------------------------
+# Inbound: OrderGroupResponse (35=UOH)
+# ---------------------------------------------------------------------------
+
+
+def test_order_group_response_roundtrip() -> None:
+    msg = OrderGroupResponse(order_group_id=GID, order_group_contracts_limit=5000, alloc_account=2)
+    back = _roundtrip(msg)
+    assert back == msg
+    assert isinstance(back, OrderGroupResponse)
+    assert back.order_group_id == GID
+    assert back.order_group_contracts_limit == 5000
+
+
+def test_order_group_response_minimal() -> None:
+    # Reset/Delete/Trigger responses echo only the id (no contracts limit).
+    msg = OrderGroupResponse(order_group_id=GID)
+    back = _roundtrip(msg)
+    assert back == msg
+    assert back.order_group_contracts_limit is None
+
+
+def test_decode_app_message_order_group_response() -> None:
+    msg = OrderGroupResponse(order_group_id=GID, order_group_contracts_limit=5000)
+    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, OrderGroupResponse)
+    assert decoded == msg
+
+
+# ---------------------------------------------------------------------------
+# Lifecycle against the mock acceptor
+# ---------------------------------------------------------------------------
+
+
+async def test_order_group_lifecycle(
+    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:
+        # Create: the gateway assigns and returns the OrderGroupID.
+        await session.send(OrderGroupRequest.create(5000))
+        await until(lambda: acceptor.first("UOG") is not None)
+        uog = acceptor.first("UOG")
+        assert uog is not None
+        assert uog.get(Tag.ORDER_GROUP_ACTION) == str(int(OrderGroupAction.CREATE))
+        assert uog.get(Tag.ORDER_GROUP_CONTRACTS_LIMIT) == "5000"
+
+        resp = OrderGroupResponse(order_group_id=GID, order_group_contracts_limit=5000)
+        await acceptor.push("UOH", resp.to_body_fields(), seq=2)
+        await until(lambda: len(received) == 1)
+        decoded = decode_app_message(received[0])
+        assert isinstance(decoded, OrderGroupResponse)
+        assert decoded.order_group_id == GID
+
+        # Delete the returned group.
+        await session.send(OrderGroupRequest.delete(decoded.order_group_id or ""))
+        await until(lambda: len([m for m in acceptor.received if m.msg_type == "UOG"]) == 2)
+        delete_req = [m for m in acceptor.received if m.msg_type == "UOG"][1]
+        assert delete_req.get(Tag.ORDER_GROUP_ACTION) == str(int(OrderGroupAction.DELETE))
+        assert delete_req.get(Tag.ORDER_GROUP_ID) == GID
+    finally:
+        await session.close()