Nightly spec-drift: strict contract tests failing (since 2026-08-05)

[github-actions]

Nightly strict contract tests failed

The scheduled Spec Drift Detection run found the SDK contract suite no longer matches upstream OpenAPI/AsyncAPI.

To resolve: locally run uv run python scripts/sync_spec.py + uv run python scripts/generate.py, reconcile models/maps, and open a PR with Closes #<this issue>. This tracker auto-closes on the next green scheduled run.

To silence while a reconcile is in flight, add the spec-drift-ack label — nightly runs then stop commenting until the failing set changes.

• Failing run: https://github.com/TexasCoding/kalshi-python-sdk/actions/runs/30989054116

Failing test output (last 200 lines)

        missing: list[str] = []
        for entry in CONTRACT_MAP:
            try:
                _resolve_schema(self.spec, entry.spec_schema)
            except pytest.fail.Exception as exc:
                missing.append(f"{entry.spec_schema} — {exc}")
        if missing:
>           pytest.fail(
                "Mapped schemas that no longer resolve in the spec:\n"
                + "\n".join(f"  - {m}" for m in missing)
            )
E           Failed: Mapped schemas that no longer resolve in the spec:
E             - LookupTickersForMarketInMultivariateEventCollectionRequest — Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
E             - LookupTickersForMarketInMultivariateEventCollectionResponse — Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec

tests/test_contracts.py:880: Failed
_________ TestWsSpecDrift.test_ws_additive_drift[MultivariatePayload] __________

self = <tests.test_contracts.TestWsSpecDrift object at 0x7fa87ee17c50>
entry = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ..._lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).")

    @pytest.mark.parametrize(
        "entry",
        WS_CONTRACT_MAP,
        ids=[e.sdk_model.rsplit(".", 1)[1] for e in WS_CONTRACT_MAP],
    )
    def test_ws_additive_drift(self, entry: ContractEntry) -> None:
        """Fail on new spec-**required** WS fields the SDK lacks; soft-warn on optional ones."""
>       spec_fields = _get_ws_msg_fields(self.spec, entry.spec_schema)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_contracts.py:1020: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'asyncapi': '3.0.0', 'info': {'title': 'Kalshi Market Data WebSocket API', 'version': '2.0.0', 'description': 'WebSoc...oduction Trade API WebSocket server (encrypted connection only)', ...}}, 'defaultContentType': 'application/json', ...}
schema_name = 'multivariateLookupPayload'

    def _get_ws_msg_fields(spec: dict[str, Any], schema_name: str) -> dict[str, dict[str, Any]]:
        """Extract msg sub-object fields from an AsyncAPI payload schema.
    
        AsyncAPI payloads nest data fields under .properties.msg.properties,
        unlike OpenAPI which uses .properties directly.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        schema = schemas.get(schema_name)
        if schema is None:
>           pytest.fail(f"Schema '{schema_name}' not found in AsyncAPI spec")
E           Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec

tests/test_contracts.py:595: Failed
___________________ TestWsSpecDrift.test_ws_schema_coverage ____________________

self = <tests.test_contracts.TestWsSpecDrift object at 0x7fa87ee1b800>

    def test_ws_schema_coverage(self) -> None:
        """Every mapped WS schema must exist in the AsyncAPI spec."""
        schemas = self.spec.get("components", {}).get("schemas", {})
        for entry in WS_CONTRACT_MAP:
>           assert entry.spec_schema in schemas, (
                f"WS contract map references '{entry.spec_schema}' "
                f"but it doesn't exist in the AsyncAPI spec"
            )
E           AssertionError: WS contract map references 'multivariateLookupPayload' but it doesn't exist in the AsyncAPI spec
E           assert 'multivariateLookupPayload' in {'commandId': {'type': 'integer', 'description': 'Unique ID of the command request. Generated by the client and should be unique within a WS session.\nThe simplest way to use it would be to start from 1 and then increment the value for every new command sent to the server.\nIf the id is set to 0, the server treats it the same way as if there was no id.\n', 'minimum': 0}, 'subscriptionId': {'type': 'integer', 'description': 'Server-generated subscription identifier (sid) used to identify the channel', 'minimum': 1}, 'sequenceNumber': {'type': 'integer', 'description': 'Sequential number that should be checked if you want to guarantee you received all the messages. Used for snapshot/delta consistency', 'minimum': 1}, 'marketTicker': {'type': 'string', 'description': 'Unique market identifier', 'pattern': '^[A-Z0-9-]+$', 'examples': ['FED-23DEC-T3.00', 'HIGHNY-22DEC23-B53.5']}, ...}
E            +  where 'multivariateLookupPayload' = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ignored_fields=frozenset(), notes="Aligned to spec v0.14.0 (2026-04-19): envelope type is 'multivariate_lookup' on the wire; channel name stays 'multivariate'. MultivariateLifecycleMessage (type 'multivariate_market_lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).").spec_schema

tests/test_contracts.py:1069: AssertionError
____ TestWsSpecDrift.test_ws_payload_field_type_drift[MultivariatePayload] _____

self = <tests.test_contracts.TestWsSpecDrift object at 0x7fa87efe9a30>
entry = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ..._lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).")

    @pytest.mark.parametrize(
        "entry",
        WS_CONTRACT_MAP,
        ids=[e.sdk_model.rsplit(".", 1)[1] for e in WS_CONTRACT_MAP],
    )
    def test_ws_payload_field_type_drift(self, entry: ContractEntry) -> None:
        """Hard-fail if SDK field type doesn't match AsyncAPI wire type.
    
        Targets the class of bug surfaced during v0.14.0 Task 11 integration
        tests: SDK models type ``_dollars``-aliased fields as ``int`` but demo
        sends dollar-decimal strings (``"0.0200"``), and types ``ts`` as int
        where the spec says ``string`` with ``format: date-time``. Both cases
        cause pydantic to reject real frames at ``model_validate`` time, which
        silently drops every matching message.
    
        A drift flagged here would have blocked the v0.14.0 envelope-only PR.
        """
>       spec_fields = _get_ws_msg_fields(self.spec, entry.spec_schema)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_contracts.py:1149: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'asyncapi': '3.0.0', 'info': {'title': 'Kalshi Market Data WebSocket API', 'version': '2.0.0', 'description': 'WebSoc...oduction Trade API WebSocket server (encrypted connection only)', ...}}, 'defaultContentType': 'application/json', ...}
schema_name = 'multivariateLookupPayload'

    def _get_ws_msg_fields(spec: dict[str, Any], schema_name: str) -> dict[str, dict[str, Any]]:
        """Extract msg sub-object fields from an AsyncAPI payload schema.
    
        AsyncAPI payloads nest data fields under .properties.msg.properties,
        unlike OpenAPI which uses .properties directly.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        schema = schemas.get(schema_name)
        if schema is None:
>           pytest.fail(f"Schema '{schema_name}' not found in AsyncAPI spec")
E           Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec

tests/test_contracts.py:595: Failed
_____ TestRequestBodyDrift.test_body_properties_match_spec[lookup_tickers] _____

self = <tests.test_contracts.TestRequestBodyDrift object at 0x7fa87efea930>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.multivariate.MultivariateCollectionsResource.lookup_tickers', http_me...ticker}/lookup', request_body_schema='#/components/schemas/LookupTickersForMarketInMultivariateEventCollectionRequest')

    def test_body_properties_match_spec(
        self,
        entry: MethodEndpointEntry,
    ) -> None:
        assert entry.request_body_schema is not None
        model_fqn = BODY_MODEL_MAP.get(entry.request_body_schema)
        assert model_fqn is not None, (
            f"No request model registered in BODY_MODEL_MAP for "
            f"{entry.request_body_schema!r}. Add the mapping."
        )
        model_cls = _get_model_class_from_fqn(model_fqn)
    
>       schema = _resolve_request_body_schema(
            self.spec,
            entry.path_template,
            entry.http_method,
        )

tests/test_contracts.py:1670: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.27.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/multivariate_event_collections/{collection_ticker}/lookup'
http_method = 'PUT', max_ref_depth = 8

    def _resolve_request_body_schema(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> dict[str, Any] | None:
        """Return the resolved request body schema for any operation with a
        ``requestBody`` (HTTP method-agnostic: POST/PUT/PATCH — and DELETE, which
        the Kalshi spec uses for ``batch_cancel``). Follows ``$ref`` pointers in
        ``content['application/json'].schema`` and resolves them via ``_resolve_ref``.
    
        Only ``application/json`` content is considered — Kalshi's spec doesn't use
        other media types for request bodies. If that changes, extend this function.
    
        Returns ``None`` when the operation has no ``requestBody`` key at all, or
        when the body has no ``application/json`` content or no ``schema`` under
        it. Returns the resolved schema dict otherwise (with any top-level ``$ref``
        already chased).
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/multivariate_event_collections/{collection_ticker}/lookup' not found in spec"

tests/_contract_support.py:1841: KeyError
=============================== warnings summary ===============================
tests/test_contracts.py::TestSpecDrift::test_additive_drift[MultivariateEventCollection]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:837: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.models.multivariate.MultivariateEventCollection: Spec field 'exchange_index' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

tests/test_contracts.py::TestAdditiveDriftClassification::test_optional_drift_survives_the_nightly_strict_gate
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:988: AdditiveOptionalDriftWarning: Additive-optional drift in tests.synthetic.Fake: Spec field 'opt_new' has no SDK mapping
    _warn_additive_optional("tests.synthetic.Fake", msgs)

tests/test_contracts.py::TestPerpsSpecDrift::test_response_drift[MarginMarket]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:2096: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.perps.models.markets.MarginMarket: Spec field 'long_leverage_estimates' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

tests/test_contracts.py::TestPerpsSpecDrift::test_response_drift[MarginMarket]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:2096: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.perps.models.markets.MarginMarket: Spec field 'short_leverage_estimates' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[LookupTickersForMarketInMultivariateEventCollectionRequest] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[LookupTickersResponse] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[LookupTickersForMarketInMultivariateEventCollectionRequest] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[LookupTickersResponse] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_schema_coverage - Failed: Mapped schemas that no longer resolve in the spec:
  - LookupTickersForMarketInMultivariateEventCollectionRequest — Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
  - LookupTickersForMarketInMultivariateEventCollectionResponse — Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_additive_drift[MultivariatePayload] - Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_schema_coverage - AssertionError: WS contract map references 'multivariateLookupPayload' but it doesn't exist in the AsyncAPI spec
assert 'multivariateLookupPayload' in {'commandId': {'type': 'integer', 'description': 'Unique ID of the command request. Generated by the client and should be unique within a WS session.\nThe simplest way to use it would be to start from 1 and then increment the value for every new command sent to the server.\nIf the id is set to 0, the server treats it the same way as if there was no id.\n', 'minimum': 0}, 'subscriptionId': {'type': 'integer', 'description': 'Server-generated subscription identifier (sid) used to identify the channel', 'minimum': 1}, 'sequenceNumber': {'type': 'integer', 'description': 'Sequential number that should be checked if you want to guarantee you received all the messages. Used for snapshot/delta consistency', 'minimum': 1}, 'marketTicker': {'type': 'string', 'description': 'Unique market identifier', 'pattern': '^[A-Z0-9-]+$', 'examples': ['FED-23DEC-T3.00', 'HIGHNY-22DEC23-B53.5']}, ...}
 +  where 'multivariateLookupPayload' = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ignored_fields=frozenset(), notes="Aligned to spec v0.14.0 (2026-04-19): envelope type is 'multivariate_lookup' on the wire; channel name stays 'multivariate'. MultivariateLifecycleMessage (type 'multivariate_market_lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).").spec_schema
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_payload_field_type_drift[MultivariatePayload] - Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec
FAILED tests/test_contracts.py::TestRequestBodyDrift::test_body_properties_match_spec[lookup_tickers] - KeyError: "path '/multivariate_event_collections/{collection_ticker}/lookup' not found in spec"
============ 9 failed, 670 passed, 4 warnings in 249.00s (0:04:09) =============

[github-actions]

Failing set changed as of 2026-08-06 (run).

Latest failing output (last 200 lines)

        schema = schemas.get(schema_name)
        if schema is None:
>           pytest.fail(f"Schema '{schema_name}' not found in AsyncAPI spec")
E           Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec

tests/test_contracts.py:595: Failed
___________________ TestWsSpecDrift.test_ws_schema_coverage ____________________

self = <tests.test_contracts.TestWsSpecDrift object at 0x7f915011f560>

    def test_ws_schema_coverage(self) -> None:
        """Every mapped WS schema must exist in the AsyncAPI spec."""
        schemas = self.spec.get("components", {}).get("schemas", {})
        for entry in WS_CONTRACT_MAP:
>           assert entry.spec_schema in schemas, (
                f"WS contract map references '{entry.spec_schema}' "
                f"but it doesn't exist in the AsyncAPI spec"
            )
E           AssertionError: WS contract map references 'multivariateLookupPayload' but it doesn't exist in the AsyncAPI spec
E           assert 'multivariateLookupPayload' in {'commandId': {'type': 'integer', 'description': 'Unique ID of the command request. Generated by the client and should be unique within a WS session.\nThe simplest way to use it would be to start from 1 and then increment the value for every new command sent to the server.\nIf the id is set to 0, the server treats it the same way as if there was no id.\n', 'minimum': 0}, 'subscriptionId': {'type': 'integer', 'description': 'Server-generated subscription identifier (sid) used to identify the channel', 'minimum': 1}, 'sequenceNumber': {'type': 'integer', 'description': 'Sequential number that should be checked if you want to guarantee you received all the messages. Used for snapshot/delta consistency', 'minimum': 1}, 'marketTicker': {'type': 'string', 'description': 'Unique market identifier', 'pattern': '^[A-Z0-9-]+$', 'examples': ['FED-23DEC-T3.00', 'HIGHNY-22DEC23-B53.5']}, ...}
E            +  where 'multivariateLookupPayload' = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ignored_fields=frozenset(), notes="Aligned to spec v0.14.0 (2026-04-19): envelope type is 'multivariate_lookup' on the wire; channel name stays 'multivariate'. MultivariateLifecycleMessage (type 'multivariate_market_lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).").spec_schema

tests/test_contracts.py:1069: AssertionError
____ TestWsSpecDrift.test_ws_payload_field_type_drift[MultivariatePayload] _____

self = <tests.test_contracts.TestWsSpecDrift object at 0x7f91502e98e0>
entry = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ..._lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).")

    @pytest.mark.parametrize(
        "entry",
        WS_CONTRACT_MAP,
        ids=[e.sdk_model.rsplit(".", 1)[1] for e in WS_CONTRACT_MAP],
    )
    def test_ws_payload_field_type_drift(self, entry: ContractEntry) -> None:
        """Hard-fail if SDK field type doesn't match AsyncAPI wire type.
    
        Targets the class of bug surfaced during v0.14.0 Task 11 integration
        tests: SDK models type ``_dollars``-aliased fields as ``int`` but demo
        sends dollar-decimal strings (``"0.0200"``), and types ``ts`` as int
        where the spec says ``string`` with ``format: date-time``. Both cases
        cause pydantic to reject real frames at ``model_validate`` time, which
        silently drops every matching message.
    
        A drift flagged here would have blocked the v0.14.0 envelope-only PR.
        """
>       spec_fields = _get_ws_msg_fields(self.spec, entry.spec_schema)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_contracts.py:1149: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'asyncapi': '3.0.0', 'info': {'title': 'Kalshi Market Data WebSocket API', 'version': '2.0.0', 'description': 'WebSoc...oduction Trade API WebSocket server (encrypted connection only)', ...}}, 'defaultContentType': 'application/json', ...}
schema_name = 'multivariateLookupPayload'

    def _get_ws_msg_fields(spec: dict[str, Any], schema_name: str) -> dict[str, dict[str, Any]]:
        """Extract msg sub-object fields from an AsyncAPI payload schema.
    
        AsyncAPI payloads nest data fields under .properties.msg.properties,
        unlike OpenAPI which uses .properties directly.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        schema = schemas.get(schema_name)
        if schema is None:
>           pytest.fail(f"Schema '{schema_name}' not found in AsyncAPI spec")
E           Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec

tests/test_contracts.py:595: Failed
___________________ test_every_spec_endpoint_is_mapped[core] ___________________

spec_loader = <function _load_spec at 0x7f914fe48400>
endpoint_map = [MethodEndpointEntry(sdk_method='kalshi.resources.markets.MarketsResource.list', http_method='GET', path_template='/ma...rkets.MarketsResource.list_trades', http_method='GET', path_template='/markets/trades', request_body_schema=None), ...]
label = 'core'

    @pytest.mark.parametrize(
        ("spec_loader", "endpoint_map", "label"),
        [
            (_load_spec, METHOD_ENDPOINT_MAP, "core"),
            (_load_perps_spec, PERPS_METHOD_ENDPOINT_MAP, "perps"),
            (_load_perps_scm_spec, PERPS_SCM_METHOD_ENDPOINT_MAP, "perps_scm"),
        ],
        ids=["core", "perps", "perps_scm"],
    )
    def test_every_spec_endpoint_is_mapped(
        spec_loader: Any,
        endpoint_map: list[MethodEndpointEntry],
        label: str,
    ) -> None:
        """Fail if a spec operation is neither mapped nor explicitly unimplemented.
    
        Closes the false-negative gap where a newly-ADDED upstream endpoint shipped
        untested because nothing asserted spec paths ⊆ ``METHOD_ENDPOINT_MAP``. The
        reverse direction (a map entry whose path the spec dropped) is caught by the
        param-drift tests, which fail loud when the path is absent from the spec.
        """
        spec_ops = _spec_operations(spec_loader())
        mapped = {(e.http_method.upper(), e.path_template) for e in endpoint_map}
        unmapped = spec_ops - mapped - set(_UNIMPLEMENTED_ENDPOINTS)
        if unmapped:
>           pytest.fail(
                f"{label}: spec operations not in the endpoint map (add a "
                f"MethodEndpointEntry, or record in _UNIMPLEMENTED_ENDPOINTS with a reason):\n"
                + "\n".join(f"  - {m} {p}" for m, p in sorted(unmapped))
            )
E           Failed: core: spec operations not in the endpoint map (add a MethodEndpointEntry, or record in _UNIMPLEMENTED_ENDPOINTS with a reason):
E             - GET /portfolio/intra_exchange_instance_transfers
E             - GET /portfolio/intra_exchange_instance_transfers/{transfer_id}

tests/test_contracts.py:1299: Failed
_____ TestRequestBodyDrift.test_body_properties_match_spec[lookup_tickers] _____

self = <tests.test_contracts.TestRequestBodyDrift object at 0x7f91502ea7e0>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.multivariate.MultivariateCollectionsResource.lookup_tickers', http_me...ticker}/lookup', request_body_schema='#/components/schemas/LookupTickersForMarketInMultivariateEventCollectionRequest')

    def test_body_properties_match_spec(
        self,
        entry: MethodEndpointEntry,
    ) -> None:
        assert entry.request_body_schema is not None
        model_fqn = BODY_MODEL_MAP.get(entry.request_body_schema)
        assert model_fqn is not None, (
            f"No request model registered in BODY_MODEL_MAP for "
            f"{entry.request_body_schema!r}. Add the mapping."
        )
        model_cls = _get_model_class_from_fqn(model_fqn)
    
>       schema = _resolve_request_body_schema(
            self.spec,
            entry.path_template,
            entry.http_method,
        )

tests/test_contracts.py:1670: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.27.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/multivariate_event_collections/{collection_ticker}/lookup'
http_method = 'PUT', max_ref_depth = 8

    def _resolve_request_body_schema(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> dict[str, Any] | None:
        """Return the resolved request body schema for any operation with a
        ``requestBody`` (HTTP method-agnostic: POST/PUT/PATCH — and DELETE, which
        the Kalshi spec uses for ``batch_cancel``). Follows ``$ref`` pointers in
        ``content['application/json'].schema`` and resolves them via ``_resolve_ref``.
    
        Only ``application/json`` content is considered — Kalshi's spec doesn't use
        other media types for request bodies. If that changes, extend this function.
    
        Returns ``None`` when the operation has no ``requestBody`` key at all, or
        when the body has no ``application/json`` content or no ``schema`` under
        it. Returns the resolved schema dict otherwise (with any top-level ``$ref``
        already chased).
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/multivariate_event_collections/{collection_ticker}/lookup' not found in spec"

tests/_contract_support.py:1841: KeyError
=============================== warnings summary ===============================
tests/test_contracts.py::TestSpecDrift::test_additive_drift[MultivariateEventCollection]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:837: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.models.multivariate.MultivariateEventCollection: Spec field 'exchange_index' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

tests/test_contracts.py::TestAdditiveDriftClassification::test_optional_drift_survives_the_nightly_strict_gate
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:988: AdditiveOptionalDriftWarning: Additive-optional drift in tests.synthetic.Fake: Spec field 'opt_new' has no SDK mapping
    _warn_additive_optional("tests.synthetic.Fake", msgs)

tests/test_contracts.py::TestPerpsSpecDrift::test_response_drift[MarginMarket]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:2096: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.perps.models.markets.MarginMarket: Spec field 'long_leverage_estimates' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

tests/test_contracts.py::TestPerpsSpecDrift::test_response_drift[MarginMarket]
  /home/runner/work/kalshi-python-sdk/kalshi-python-sdk/tests/test_contracts.py:2096: AdditiveOptionalDriftWarning: Additive-optional drift in kalshi.perps.models.markets.MarginMarket: Spec field 'short_leverage_estimates' has no SDK mapping
    _warn_additive_optional(entry.sdk_model, additive_optional)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[LookupTickersForMarketInMultivariateEventCollectionRequest] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[LookupTickersResponse] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[LookupTickersForMarketInMultivariateEventCollectionRequest] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[LookupTickersResponse] - Failed: Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_schema_coverage - Failed: Mapped schemas that no longer resolve in the spec:
  - LookupTickersForMarketInMultivariateEventCollectionRequest — Schema 'LookupTickersForMarketInMultivariateEventCollectionRequest' not found in OpenAPI spec
  - LookupTickersForMarketInMultivariateEventCollectionResponse — Schema 'LookupTickersForMarketInMultivariateEventCollectionResponse' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_additive_drift[MultivariatePayload] - Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_schema_coverage - AssertionError: WS contract map references 'multivariateLookupPayload' but it doesn't exist in the AsyncAPI spec
assert 'multivariateLookupPayload' in {'commandId': {'type': 'integer', 'description': 'Unique ID of the command request. Generated by the client and should be unique within a WS session.\nThe simplest way to use it would be to start from 1 and then increment the value for every new command sent to the server.\nIf the id is set to 0, the server treats it the same way as if there was no id.\n', 'minimum': 0}, 'subscriptionId': {'type': 'integer', 'description': 'Server-generated subscription identifier (sid) used to identify the channel', 'minimum': 1}, 'sequenceNumber': {'type': 'integer', 'description': 'Sequential number that should be checked if you want to guarantee you received all the messages. Used for snapshot/delta consistency', 'minimum': 1}, 'marketTicker': {'type': 'string', 'description': 'Unique market identifier', 'pattern': '^[A-Z0-9-]+$', 'examples': ['FED-23DEC-T3.00', 'HIGHNY-22DEC23-B53.5']}, ...}
 +  where 'multivariateLookupPayload' = ContractEntry(sdk_model='kalshi.ws.models.multivariate.MultivariatePayload', spec_schema='multivariateLookupPayload', ignored_fields=frozenset(), notes="Aligned to spec v0.14.0 (2026-04-19): envelope type is 'multivariate_lookup' on the wire; channel name stays 'multivariate'. MultivariateLifecycleMessage (type 'multivariate_market_lifecycle') is unaffected -- separate spec-aligned sibling. No direct demo capture (no active collections emitting).").spec_schema
FAILED tests/test_contracts.py::TestWsSpecDrift::test_ws_payload_field_type_drift[MultivariatePayload] - Failed: Schema 'multivariateLookupPayload' not found in AsyncAPI spec
FAILED tests/test_contracts.py::test_every_spec_endpoint_is_mapped[core] - Failed: core: spec operations not in the endpoint map (add a MethodEndpointEntry, or record in _UNIMPLEMENTED_ENDPOINTS with a reason):
  - GET /portfolio/intra_exchange_instance_transfers
  - GET /portfolio/intra_exchange_instance_transfers/{transfer_id}
FAILED tests/test_contracts.py::TestRequestBodyDrift::test_body_properties_match_spec[lookup_tickers] - KeyError: "path '/multivariate_event_collections/{collection_ticker}/lookup' not found in spec"
============ 10 failed, 669 passed, 4 warnings in 170.41s (0:02:50) ============

[TexasCoding]

Resolved by v10.0.0 via #498 (merge commit a5ef152). Nightly contract drift fixed: multivariate lookup removed, transfer-history GETs mapped, perps leverage side fields added. Release: https://github.com/TexasCoding/kalshi-python-sdk/releases/tag/v10.0.0