feat(fix): repeating-group support in the message framework (closes #423)

kalshi/fix/messages/__init__.py

@@ -1,13 +1,28 @@
 """Typed FIX message models (FIX Dictionary v1.03).
 
-Foundation phase exposes the session-layer (admin) messages plus the base
-framework. Application messages (order entry, drop copy, market data,
-RFQ/settlement) are added in later phases — see GH #402.
+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.
 """
 
 from __future__ import annotations
 
-from kalshi.fix.messages.base import FixMessage, FixType, fixfield
+from kalshi.fix.messages.base import (
+    FixGroup,
+    FixGroupMeta,
+    FixMessage,
+    FixType,
+    fixfield,
+    groupfield,
+)
+from kalshi.fix.messages.components import (
+    CollateralAmountChange,
+    MarketSettlementParty,
+    MiscFee,
+    MultivariateSelectedLeg,
+    Party,
+)
 from kalshi.fix.messages.session import (
     Heartbeat,
     Logon,
@@ -19,14 +34,22 @@
 )
 
 __all__ = [
+    "CollateralAmountChange",
+    "FixGroup",
+    "FixGroupMeta",
     "FixMessage",
     "FixType",
     "Heartbeat",
     "Logon",
     "Logout",
+    "MarketSettlementParty",
+    "MiscFee",
+    "MultivariateSelectedLeg",
+    "Party",
     "Reject",
     "ResendRequest",
     "SequenceReset",
     "TestRequest",
     "fixfield",
+    "groupfield",
 ]

kalshi/fix/messages/base.py

@@ -1,27 +1,31 @@
 """Typed FIX message framework: Pydantic models that round-trip to wire fields.
 
-Each message subclasses :class:`FixMessage` and declares its body fields with
-:func:`fixfield`, attaching the FIX tag number and wire type to the Pydantic
-field via ``json_schema_extra``. The base then drives:
-
-* :meth:`FixMessage.to_body_fields` — ordered ``(tag, value)`` pairs for the
-  message body (everything after the standard header), with length-prefixed
-  data fields (``RawData``/``Signature``) auto-emitting their length field.
-* :meth:`FixMessage.from_raw` — build a typed model from a decoded
-  :class:`~kalshi.fix.codec.RawMessage`, reading only the tags it declares
-  (unknown/header tags are ignored, so inbound messages with extra fields
-  parse cleanly).
+Each message subclasses :class:`FixMessage` and declares its fields with
+:func:`fixfield` (scalars) and :func:`groupfield` (repeating groups). Scalar
+metadata (tag + wire type) rides on ``json_schema_extra``; group metadata (the
+``NumInGroup`` tag + entry model) rides on an :class:`Annotated`
+:class:`FixGroupMeta`. The shared base (:class:`_FixFieldModel`) drives:
+
+* :meth:`to_body_fields` — ordered ``(tag, value)`` pairs in declaration order,
+  with length-prefixed data fields auto-emitting their length field and repeating
+  groups expanding to ``NumInGroup`` + each entry's fields (nesting supported).
+* :meth:`from_raw` — build a typed model from a decoded
+  :class:`~kalshi.fix.codec.RawMessage`. Scalars are read by tag; groups are
+  parsed *positionally* (each entry delimited by its first field's tag), which
+  recurses for nested groups.
 
 The standard header (``SenderCompID``/``TargetCompID``/``MsgSeqNum``/
 ``SendingTime``) and ``MsgType`` are owned by the session/connection layer, not
 the message body — see :mod:`kalshi.fix.session`.
 
-Scope note: scalar fields only. Repeating groups (``NoPartyIDs``, ``NoMiscFees``,
-…) land with the order-entry / settlement message phases; see GH #402.
+Positional group decoding assumes group-member tags are distinct from top-level
+scalar tags (true for the Kalshi FIX dictionary v1.03), so a scalar is found by
+first-occurrence without confusion with a same-tagged field inside a group.
 """
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from decimal import Decimal
 from enum import StrEnum
 from typing import Any, ClassVar, Self
@@ -63,18 +67,55 @@ class FixType(StrEnum):
 _DECIMAL_TYPES = frozenset({FixType.PRICE, FixType.QTY, FixType.AMT, FixType.DECIMAL})
 
 
+@dataclass(frozen=True)
+class FixGroupMeta:
+    """``Annotated`` metadata marking a repeating-group field.
+
+    Declare a group as ``Annotated[list[EntryModel], FixGroupMeta(count_tag,
+    EntryModel)]`` where ``count_tag`` is the ``NumInGroup`` tag and
+    ``EntryModel`` is the :class:`FixGroup` subclass describing one entry.
+    """
+
+    num_in_group_tag: int
+    entry_model: type[_FixFieldModel]
+
+
+@dataclass(frozen=True)
+class _ScalarSpec:
+    name: str
+    tag: int
+    fix_type: FixType
+
+
+@dataclass(frozen=True)
+class _GroupSpec:
+    name: str
+    count_tag: int
+    entry_model: type[_FixFieldModel]
+
+
+_FieldSpec = _ScalarSpec | _GroupSpec
+
+
 def fixfield(tag: int, fix_type: FixType, *, default: Any = ..., **kwargs: Any) -> Any:
-    """Declare a FIX-mapped Pydantic field.
+    """Declare a scalar FIX field.
 
     ``tag`` is the FIX tag number, ``fix_type`` its wire type. Omit ``default``
-    for a required field (Pydantic treats ``...`` as required). The metadata
-    rides on ``json_schema_extra`` so the base can introspect tag + type in
-    declaration order.
+    for a required field (Pydantic treats ``...`` as required).
     """
     extra: dict[str, Any] = {"fix_tag": int(tag), "fix_type": fix_type.value}
     return Field(default=default, json_schema_extra=extra, **kwargs)
 
 
+def groupfield() -> Any:
+    """Declare a repeating-group field (defaults to an empty list).
+
+    The ``NumInGroup`` tag and entry model ride on the field's
+    :class:`Annotated` :class:`FixGroupMeta`; this only supplies the default.
+    """
+    return Field(default_factory=list)
+
+
 def _to_wire(value: Any, fix_type: FixType) -> str:
     """Convert a Python field value to its FIX wire string."""
     if fix_type is FixType.BOOLEAN:
@@ -112,77 +153,177 @@ def _from_wire(raw: str, fix_type: FixType) -> Any:
     return raw
 
 
-# Per-class cache of the introspected FIX field map. The map is deterministic
-# per message class and, once application messages land, is read on every
-# encode/decode of every order and execution report — so caching it avoids
-# re-walking ``model_fields`` on the hot path. Classes live for the process
-# lifetime, so a plain dict keyed by class is fine.
-_FIX_FIELDS_CACHE: dict[type[FixMessage], list[tuple[str, int, FixType]]] = {}
+def _first_value(pairs: list[tuple[int, str]], tag: int) -> str | None:
+    """First value for ``tag`` in ``pairs``, or ``None``."""
+    for t, v in pairs:
+        if t == tag:
+            return v
+    return None
 
 
-class FixMessage(BaseModel):
-    """Base for all typed FIX messages.
+def _index_of(pairs: list[tuple[int, str]], tag: int) -> int | None:
+    """Index of the first pair with ``tag``, or ``None``."""
+    for i, (t, _) in enumerate(pairs):
+        if t == tag:
+            return i
+    return None
 
-    Subclasses set the ``MSG_TYPE`` class var and declare body fields with
-    :func:`fixfield`. ``extra="forbid"`` makes a typo in an outbound message
-    fail at construction; inbound messages are built via :meth:`from_raw`, which
-    only passes declared tags, so server-added fields never reach the
-    constructor.
-    """
 
-    MSG_TYPE: ClassVar[MsgType]
+def _parse_group(
+    pairs: list[tuple[int, str]],
+    start: int,
+    count: int,
+    entry_model: type[_FixFieldModel],
+) -> tuple[list[_FixFieldModel], 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
+    the next delimiter or a tag the entry does not own — which naturally captures
+    nested groups (their tags are part of the entry's member set). Returns the
+    parsed entries and the index just past the group.
+    """
+    delim = entry_model._first_tag()
+    member_tags = entry_model._member_tags()
+    entries: list[_FixFieldModel] = []
+    i = start
+    n = len(pairs)
+    while len(entries) < count and i < n and pairs[i][0] == delim:
+        entry_pairs = [pairs[i]]
+        i += 1
+        while i < n and pairs[i][0] != delim and pairs[i][0] in member_tags:
+            entry_pairs.append(pairs[i])
+            i += 1
+        entries.append(entry_model._from_pairs(entry_pairs))
+    return entries, i
+
+
+# Per-class caches of the introspected layout / member-tag set. Both are
+# deterministic per class and read on every encode/decode, so caching avoids
+# re-walking ``model_fields``. Classes live for the process lifetime, so plain
+# dicts keyed by class are fine.
+_LAYOUT_CACHE: dict[type[_FixFieldModel], list[_FieldSpec]] = {}
+_MEMBER_TAGS_CACHE: dict[type[_FixFieldModel], frozenset[int]] = {}
+
+
+class _FixFieldModel(BaseModel):
+    """Shared machinery for FIX messages and repeating-group entries.
+
+    ``extra="forbid"`` makes a typo in an outbound model fail at construction;
+    inbound models are built via :meth:`from_raw` / :meth:`_from_pairs`, which
+    pass only declared tags, so server-added fields never reach the constructor.
+    """
 
     model_config = ConfigDict(extra="forbid")
 
     @classmethod
-    def _fix_fields(cls) -> list[tuple[str, int, FixType]]:
-        """``(attr_name, tag, fix_type)`` for each FIX-mapped field, in declaration order.
-
-        Cached per class (see :data:`_FIX_FIELDS_CACHE`). Callers iterate the
-        result and must not mutate it.
-        """
-        cached = _FIX_FIELDS_CACHE.get(cls)
+    def _layout(cls) -> list[_FieldSpec]:
+        """Ordered scalar/group field specs in declaration order (cached)."""
+        cached = _LAYOUT_CACHE.get(cls)
         if cached is not None:
             return cached
-        out: list[tuple[str, int, FixType]] = []
+        specs: list[_FieldSpec] = []
         for name, field in cls.model_fields.items():
+            group_meta = next(
+                (m for m in field.metadata if isinstance(m, FixGroupMeta)), None
+            )
+            if group_meta is not None:
+                specs.append(_GroupSpec(name, group_meta.num_in_group_tag, group_meta.entry_model))
+                continue
             extra = field.json_schema_extra
             if isinstance(extra, dict) and "fix_tag" in extra:
                 tag = int(extra["fix_tag"])  # type: ignore[arg-type]
-                fix_type = FixType(str(extra["fix_type"]))
-                out.append((name, tag, fix_type))
-        _FIX_FIELDS_CACHE[cls] = out
-        return out
+                specs.append(_ScalarSpec(name, tag, FixType(str(extra["fix_type"]))))
+        _LAYOUT_CACHE[cls] = specs
+        return specs
+
+    @classmethod
+    def _member_tags(cls) -> frozenset[int]:
+        """Every tag this model can contain, transitively through nested groups."""
+        cached = _MEMBER_TAGS_CACHE.get(cls)
+        if cached is not None:
+            return cached
+        tags: set[int] = set()
+        for spec in cls._layout():
+            if isinstance(spec, _GroupSpec):
+                tags.add(spec.count_tag)
+                tags |= spec.entry_model._member_tags()
+            else:
+                tags.add(spec.tag)
+                length_tag = _DATA_TO_LENGTH.get(spec.tag)
+                if length_tag is not None:
+                    tags.add(length_tag)
+        frozen = frozenset(tags)
+        _MEMBER_TAGS_CACHE[cls] = frozen
+        return frozen
+
+    @classmethod
+    def _first_tag(cls) -> int:
+        """The delimiter tag — the first declared field's tag (group entry start)."""
+        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")
 
     def to_body_fields(self) -> list[tuple[int, str]]:
-        """Ordered ``(tag, value)`` body fields (excludes the standard header).
+        """Ordered ``(tag, value)`` fields in declaration order.
 
-        ``None`` values are omitted. A data field auto-emits its length field
-        immediately before it.
+        ``None`` scalars and empty groups are omitted; a data field auto-emits
+        its length field; a group emits ``NumInGroup`` then each entry's fields.
         """
         out: list[tuple[int, str]] = []
-        for name, tag, fix_type in self._fix_fields():
-            value = getattr(self, name)
-            if value is None:
-                continue
-            wire = _to_wire(value, fix_type)
-            length_tag = _DATA_TO_LENGTH.get(tag)
-            if length_tag is not None:
-                out.append((length_tag, str(len(wire.encode("latin-1")))))
-            out.append((tag, wire))
+        for spec in self._layout():
+            value = getattr(self, spec.name)
+            if isinstance(spec, _GroupSpec):
+                if not value:
+                    continue
+                out.append((spec.count_tag, str(len(value))))
+                for entry in value:
+                    out.extend(entry.to_body_fields())
+            else:
+                if value is None:
+                    continue
+                wire = _to_wire(value, spec.fix_type)
+                length_tag = _DATA_TO_LENGTH.get(spec.tag)
+                if length_tag is not None:
+                    out.append((length_tag, str(len(wire.encode("latin-1")))))
+                out.append((spec.tag, wire))
         return out
 
     @classmethod
     def from_raw(cls, raw: RawMessage) -> Self:
-        """Build a typed message from a decoded :class:`RawMessage`.
+        """Build a typed model from a decoded :class:`RawMessage`."""
+        return cls._from_pairs(raw.pairs)
+
+    @classmethod
+    def _from_pairs(cls, pairs: list[tuple[int, str]]) -> Self:
+        """Build a typed model from an ordered ``(tag, value)`` slice.
 
-        Reads only the tags this model declares; all other fields (header,
-        trailer, server extensions) are ignored.
+        Scalars are read by tag; groups are parsed positionally from the
+        ``NumInGroup`` field onward (recursing for nested groups).
         """
         kwargs: dict[str, Any] = {}
-        for name, tag, fix_type in cls._fix_fields():
-            value = raw.get(tag)
-            if value is None:
-                continue
-            kwargs[name] = _from_wire(value, fix_type)
+        for spec in cls._layout():
+            if isinstance(spec, _GroupSpec):
+                idx = _index_of(pairs, spec.count_tag)
+                if idx is None:
+                    continue
+                try:
+                    count = int(pairs[idx][1])
+                except ValueError:
+                    continue
+                entries, _ = _parse_group(pairs, idx + 1, count, spec.entry_model)
+                kwargs[spec.name] = entries
+            else:
+                value = _first_value(pairs, spec.tag)
+                if value is not None:
+                    kwargs[spec.name] = _from_wire(value, spec.fix_type)
         return cls(**kwargs)
+
+
+class FixMessage(_FixFieldModel):
+    """Base for all typed FIX messages. Subclasses set the ``MSG_TYPE`` class var."""
+
+    MSG_TYPE: ClassVar[MsgType]
+
+
+class FixGroup(_FixFieldModel):
+    """Base for one entry of a repeating group (no ``MsgType`` of its own)."""