From a40ccd318aa2022e13538fe1967a0999b0e7bce5 Mon Sep 17 00:00:00 2001 From: audit-w3b Date: Wed, 20 May 2026 23:08:46 -0500 Subject: [PATCH] polish: cursor guard O(1); batch bytes path; bench scripts; close terminality; env isolation; from_env test - _list_all cursor-loop guard switched from set[str] to single last_cursor: O(1) memory. - Resource _base: new _post_json / _delete_with_body_json bytes helpers. Batch order paths use model_dump_json + httpx content= to skip one full dict-walk pass per call. - Transport.request: optional content + headers kwargs (errors if json+content both set). - KalshiAuth.close() is now terminal: subsequent sign_request_async raises RuntimeError instead of silently respawning the executor (pairs with #210 ownership tracking). - tests/integration/conftest.py: env-bridging moved from import-time mutation to a session-scoped fixture with monkeypatch. - tests/test_auth.py: KALSHI_PRIVATE_KEY takes precedence over KALSHI_PRIVATE_KEY_PATH in from_env / try_from_env (pinning current behavior). - New scripts/bench_ws_recv.py, scripts/bench_orderbook_delta.py, scripts/bench_request_hot_path.py for perf regression visibility. Closes #223 Refs #224 --- kalshi/_base_client.py | 30 +++++++- kalshi/auth.py | 21 +++++- kalshi/resources/_base.py | 87 ++++++++++++++++++++--- kalshi/resources/orders.py | 24 ++++--- scripts/bench_orderbook_delta.py | 84 ++++++++++++++++++++++ scripts/bench_request_hot_path.py | 90 +++++++++++++++++++++++ scripts/bench_ws_recv.py | 104 +++++++++++++++++++++++++++ tests/integration/conftest.py | 38 +++++++--- tests/test_async_orders.py | 33 +++++---- tests/test_auth.py | 114 +++++++++++++++++++++++++++++- tests/test_base_helpers.py | 30 +++++++- tests/test_orders.py | 104 +++++++++++++++++++++++---- 12 files changed, 698 insertions(+), 61 deletions(-) create mode 100644 scripts/bench_orderbook_delta.py create mode 100644 scripts/bench_request_hot_path.py create mode 100644 scripts/bench_ws_recv.py diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index 22083757..b3b0101c 100644 --- a/kalshi/_base_client.py +++ b/kalshi/_base_client.py @@ -178,6 +178,8 @@ def request( *, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, + content: bytes | None = None, + headers: dict[str, str] | None = None, extra_headers: dict[str, str] | None = None, ) -> httpx.Response: """Make an authenticated HTTP request with retry logic. @@ -187,6 +189,8 @@ def request( client-default one), but the ``KALSHI-ACCESS-*`` auth headers always win — callers cannot forge them via ``extra_headers``. """ + if json is not None and content is not None: + raise TypeError("request() accepts `json=` or `content=`, not both.") # P1.6: canonicalize trailing slash BEFORE both signing and httpx call # so the wire path and the signed payload stay byte-identical. Root "/" # is preserved. @@ -200,11 +204,18 @@ def request( total_timeout = self._config.total_timeout config_extra = self._config.extra_headers or {} per_call_extra = extra_headers or {} + body_headers = headers or {} for attempt in range(self._config.max_retries + 1): auth_headers = self._auth.sign_request(method.upper(), sign_path) if self._auth else {} - # Order matters: config defaults < per-request overrides < signed auth. - merged_headers = {**config_extra, **per_call_extra, **auth_headers} + # Order matters: config defaults < per-request overrides + # < body-helper headers < signed auth. + merged_headers = { + **config_extra, + **per_call_extra, + **body_headers, + **auth_headers, + } logger.debug( "Request: %s %s (attempt %d/%d)", @@ -220,6 +231,7 @@ def request( url=path, params=params, json=json, + content=content, headers=merged_headers, ) except httpx.PoolTimeout as e: @@ -349,6 +361,8 @@ async def request( *, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, + content: bytes | None = None, + headers: dict[str, str] | None = None, extra_headers: dict[str, str] | None = None, ) -> httpx.Response: """Make an authenticated async HTTP request with retry logic. @@ -357,6 +371,9 @@ async def request( """ import asyncio + if json is not None and content is not None: + raise TypeError("request() accepts `json=` or `content=`, not both.") + # P1.6: canonicalize trailing slash BEFORE both signing and httpx call. if path != "/" and path.endswith("/"): path = path.rstrip("/") or "/" @@ -368,13 +385,19 @@ async def request( total_timeout = self._config.total_timeout config_extra = self._config.extra_headers or {} per_call_extra = extra_headers or {} + body_headers = headers or {} for attempt in range(self._config.max_retries + 1): if self._auth: auth_headers = await self._auth.sign_request_async(method.upper(), sign_path) else: auth_headers = {} - merged_headers = {**config_extra, **per_call_extra, **auth_headers} + merged_headers = { + **config_extra, + **per_call_extra, + **body_headers, + **auth_headers, + } logger.debug( "Async request: %s %s (attempt %d/%d)", @@ -390,6 +413,7 @@ async def request( url=path, params=params, json=json, + content=content, headers=merged_headers, ) except httpx.PoolTimeout as e: diff --git a/kalshi/auth.py b/kalshi/auth.py index e79df9c6..1c595291 100644 --- a/kalshi/auth.py +++ b/kalshi/auth.py @@ -71,6 +71,9 @@ def __init__(self, key_id: str, private_key: rsa.RSAPrivateKey) -> None: # await-latency rather than inline CPU. (#178) self._sign_executor: ThreadPoolExecutor | None = None self._sign_executor_lock = threading.Lock() + # Set in close(); checked by _get_sign_executor so async signs after + # close() raise instead of silently respawning the executor. + self._closed: bool = False @classmethod def from_key_path( @@ -278,7 +281,17 @@ def _get_sign_executor(self) -> ThreadPoolExecutor: Thread-safe lazy init: the lock guards the assignment, and the double-checked pattern keeps the fast path lock-free after first use. + + Raises ``RuntimeError`` after :meth:`close` so a closed auth does + not silently respawn the executor (which would defeat the point of + ``close()`` — the old executor's threads stay shut down but a new + ThreadPoolExecutor would be allocated on the next async sign). """ + if self._closed: + raise RuntimeError( + "KalshiAuth has been closed; create a new instance to " + "sign further requests." + ) if self._sign_executor is None: with self._sign_executor_lock: if self._sign_executor is None: @@ -312,14 +325,20 @@ async def sign_request_async( ) def close(self) -> None: - """Shut down the sign-offload executor if one was created. + """Shut down the sign-offload executor and mark this auth closed. Idempotent. Safe to call without an executor having been initialised. Long-lived clients ordinarily rely on interpreter-shutdown cleanup (atexit-joined executor threads); this hook is for tests and other callers that want deterministic teardown. + + Terminal: a subsequent :meth:`sign_request_async` raises + ``RuntimeError`` rather than silently respawning a fresh executor. + The synchronous :meth:`sign_request` is unaffected — it never used + the executor — but callers should treat the instance as retired. """ with self._sign_executor_lock: + self._closed = True if self._sign_executor is not None: self._sign_executor.shutdown(wait=False) self._sign_executor = None diff --git a/kalshi/resources/_base.py b/kalshi/resources/_base.py index fad20559..b4459f09 100644 --- a/kalshi/resources/_base.py +++ b/kalshi/resources/_base.py @@ -150,6 +150,26 @@ def _post( result: dict[str, Any] = response.json() return result + def _post_json( + self, + path: str, + *, + content: bytes, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """POST a pre-serialized JSON body (bytes). + + Skips the dict-walk pass httpx's ``json=`` does. Use this when + a Pydantic model already produced the wire bytes via + ``model_dump_json(...).encode()``. + """ + response = self._transport.request( + "POST", path, params=params, content=content, + headers={"Content-Type": "application/json"}, + ) + result: dict[str, Any] = response.json() + return result + def _put( self, path: str, @@ -181,6 +201,23 @@ def _delete_with_body( result: dict[str, Any] = response.json() return result + def _delete_with_body_json( + self, + path: str, + *, + content: bytes, + params: dict[str, Any] | None = None, + ) -> dict[str, Any] | None: + """DELETE with a pre-serialized JSON body (bytes). See :meth:`_post_json`.""" + response = self._transport.request( + "DELETE", path, params=params, content=content, + headers={"Content-Type": "application/json"}, + ) + if response.status_code == 204: + return None + result: dict[str, Any] = response.json() + return result + def _list( self, path: str, @@ -225,11 +262,13 @@ def _list_all( cursor-repeat guard below provides the safety net against infinite loops. - Raises ``KalshiError`` if a cursor value repeats, which indicates - a server-side pagination bug. + Raises ``KalshiError`` if the previous page's cursor repeats — + the realistic server-pagination-bug shape (last page replayed + forever). Only the most recent cursor is retained, so memory is + O(1) regardless of page count. """ current_params = dict(params) if params else {} - seen_cursors: set[str] = set() + last_cursor: str | None = None pages_fetched = 0 while max_pages is None or pages_fetched < max_pages: page = self._list( @@ -240,13 +279,13 @@ def _list_all( pages_fetched += 1 if not page.cursor: break - if page.cursor in seen_cursors: + if page.cursor == last_cursor: raise KalshiError( f"Cursor loop detected on {path}: server returned " f"cursor {page.cursor!r} which was already used. " f"This indicates a server-side pagination bug." ) - seen_cursors.add(page.cursor) + last_cursor = page.cursor current_params["cursor"] = page.cursor @@ -282,6 +321,21 @@ async def _post( result: dict[str, Any] = response.json() return result + async def _post_json( + self, + path: str, + *, + content: bytes, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Async :meth:`SyncResource._post_json`.""" + response = await self._transport.request( + "POST", path, params=params, content=content, + headers={"Content-Type": "application/json"}, + ) + result: dict[str, Any] = response.json() + return result + async def _put( self, path: str, @@ -313,6 +367,23 @@ async def _delete_with_body( result: dict[str, Any] = response.json() return result + async def _delete_with_body_json( + self, + path: str, + *, + content: bytes, + params: dict[str, Any] | None = None, + ) -> dict[str, Any] | None: + """Async :meth:`SyncResource._delete_with_body_json`.""" + response = await self._transport.request( + "DELETE", path, params=params, content=content, + headers={"Content-Type": "application/json"}, + ) + if response.status_code == 204: + return None + result: dict[str, Any] = response.json() + return result + async def _list( self, path: str, @@ -344,7 +415,7 @@ async def _list_all( Raises ``KalshiError`` on repeated cursor; see sync docstring. """ current_params = dict(params) if params else {} - seen_cursors: set[str] = set() + last_cursor: str | None = None pages_fetched = 0 while max_pages is None or pages_fetched < max_pages: page = await self._list( @@ -356,11 +427,11 @@ async def _list_all( pages_fetched += 1 if not page.cursor: break - if page.cursor in seen_cursors: + if page.cursor == last_cursor: raise KalshiError( f"Cursor loop detected on {path}: server returned " f"cursor {page.cursor!r} which was already used. " f"This indicates a server-side pagination bug." ) - seen_cursors.add(page.cursor) + last_cursor = page.cursor current_params["cursor"] = page.cursor diff --git a/kalshi/resources/orders.py b/kalshi/resources/orders.py index ffa4a06c..11b25df4 100644 --- a/kalshi/resources/orders.py +++ b/kalshi/resources/orders.py @@ -124,7 +124,13 @@ def _build_create_order_body( def _build_batch_create_body( request: BatchCreateOrdersRequest | None, orders: Sequence[CreateOrderRequest] | None, -) -> dict[str, Any]: +) -> bytes: + """Serialize the batch-create body directly to JSON bytes. + + Skips the intermediate ``model_dump(mode="json")`` dict-walk: with up + to 100 orders x ~10 Decimal fields, the dict version pays the + serializer cost twice (once here, once in httpx's json encoder). + """ _check_request_exclusive(request, orders=orders) if request is None: if orders is None: @@ -132,13 +138,15 @@ def _build_batch_create_body( "batch_create() requires `orders` (or pass `request=...`)" ) request = BatchCreateOrdersRequest(orders=list(orders)) - return request.model_dump(exclude_none=True, by_alias=True, mode="json") + return request.model_dump_json(exclude_none=True, by_alias=True).encode() def _build_batch_cancel_body( request: BatchCancelOrdersRequest | None, orders: Sequence[BatchCancelOrdersRequestOrder | str] | None, -) -> dict[str, Any]: +) -> bytes: + """Serialize the batch-cancel body directly to JSON bytes. See + :func:`_build_batch_create_body` for the perf rationale.""" _check_request_exclusive(request, orders=orders) if request is None: if orders is None: @@ -152,7 +160,7 @@ def _build_batch_cancel_body( for o in orders ] request = BatchCancelOrdersRequest(orders=normalized) - return request.model_dump(exclude_none=True, by_alias=True, mode="json") + return request.model_dump_json(exclude_none=True, by_alias=True).encode() def _build_amend_body( @@ -475,7 +483,7 @@ def batch_create( """ self._require_auth() body = _build_batch_create_body(request, orders) - data = self._post("/portfolio/orders/batched", json=body) + data = self._post_json("/portfolio/orders/batched", content=body) return BatchCreateOrdersResponse.model_validate(data) @overload @@ -513,7 +521,7 @@ def batch_cancel( """ self._require_auth() body = _build_batch_cancel_body(request, orders) - data = self._delete_with_body("/portfolio/orders/batched", json=body) + data = self._delete_with_body_json("/portfolio/orders/batched", content=body) if data is None: raise KalshiError( "Expected BatchCancelOrdersResponse body, got 204 No Content." @@ -929,7 +937,7 @@ async def batch_create( """Place a batch of orders. See :meth:`OrdersResource.batch_create`.""" self._require_auth() body = _build_batch_create_body(request, orders) - data = await self._post("/portfolio/orders/batched", json=body) + data = await self._post_json("/portfolio/orders/batched", content=body) return BatchCreateOrdersResponse.model_validate(data) @overload @@ -952,7 +960,7 @@ async def batch_cancel( """Batch-cancel orders. See :meth:`OrdersResource.batch_cancel`.""" self._require_auth() body = _build_batch_cancel_body(request, orders) - data = await self._delete_with_body("/portfolio/orders/batched", json=body) + data = await self._delete_with_body_json("/portfolio/orders/batched", content=body) if data is None: raise KalshiError( "Expected BatchCancelOrdersResponse body, got 204 No Content." diff --git a/scripts/bench_orderbook_delta.py b/scripts/bench_orderbook_delta.py new file mode 100644 index 00000000..cb166677 --- /dev/null +++ b/scripts/bench_orderbook_delta.py @@ -0,0 +1,84 @@ +"""Microbench for :meth:`OrderbookManager._apply_delta_inplace` (#199 hot path). + +Seeds one book at a given depth, then drives N alternating positive / +negative deltas against the inner price levels. Reports updates/sec. + +Usage:: + + uv run python scripts/bench_orderbook_delta.py [--depth D] [--updates N] +""" + +from __future__ import annotations + +import argparse +import gc +import time +from decimal import Decimal + +from kalshi.ws.models.orderbook_delta import ( + OrderbookDeltaMessage, + OrderbookDeltaPayload, + OrderbookSnapshotMessage, + OrderbookSnapshotPayload, +) +from kalshi.ws.orderbook import OrderbookManager + + +def _build_snapshot(ticker: str, depth: int) -> OrderbookSnapshotMessage: + yes = [(Decimal("0." + f"{i:04d}"), Decimal("100.00")) for i in range(1, depth + 1)] + no = [(Decimal("0." + f"{i:04d}"), Decimal("100.00")) for i in range(1, depth + 1)] + return OrderbookSnapshotMessage( + sid=1, seq=1, + msg=OrderbookSnapshotPayload( + market_ticker=ticker, market_id=ticker, yes=yes, no=no, + ), + ) + + +def _build_delta( + ticker: str, seq: int, price: Decimal, delta: Decimal, side: str, +) -> OrderbookDeltaMessage: + return OrderbookDeltaMessage( + sid=1, seq=seq, + msg=OrderbookDeltaPayload( + market_ticker=ticker, market_id=ticker, + price=price, delta=delta, side=side, # type: ignore[arg-type] + ), + ) + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--depth", type=int, default=100) + p.add_argument("--updates", type=int, default=100_000) + args = p.parse_args() + + ticker = "BENCH-DELTA" + mgr = OrderbookManager() + mgr._apply_snapshot_inplace(_build_snapshot(ticker, args.depth)) + + # Pre-build deltas so the loop only measures _apply_delta_inplace. + prices = [Decimal("0." + f"{(i % args.depth) + 1:04d}") for i in range(args.updates)] + deltas = [_build_delta( + ticker, seq=i + 2, + price=prices[i], + delta=Decimal("1.00") if i % 2 == 0 else Decimal("-1.00"), + side="yes" if i % 2 == 0 else "no", + ) for i in range(args.updates)] + + gc.collect() + t0 = time.perf_counter() + applied = 0 + for d in deltas: + if mgr._apply_delta_inplace(d): + applied += 1 + elapsed = time.perf_counter() - t0 + + rate = args.updates / elapsed if elapsed > 0 else float("inf") + print(f"depth={args.depth} updates={args.updates} applied={applied}") + print(f" elapsed={elapsed * 1000:.1f} ms rate={rate:,.0f} updates/sec") + + +if __name__ == "__main__": + main() diff --git a/scripts/bench_request_hot_path.py b/scripts/bench_request_hot_path.py new file mode 100644 index 00000000..bdca9f8a --- /dev/null +++ b/scripts/bench_request_hot_path.py @@ -0,0 +1,90 @@ +"""Microbench for the REST request hot path. + +Wires a respx mock transport so every ``client.markets.list(limit=10)`` +short-circuits without touching the network, and measures requests/sec +for the sync and async clients. Catches regressions in per-request +allocations / sign overhead. + +Usage:: + + uv run python scripts/bench_request_hot_path.py [--requests N] +""" + +from __future__ import annotations + +import argparse +import asyncio +import time + +import httpx +import respx +from cryptography.hazmat.primitives.asymmetric import rsa + +from kalshi.async_client import AsyncKalshiClient +from kalshi.auth import KalshiAuth +from kalshi.client import KalshiClient +from kalshi.config import KalshiConfig + +_MOCK_RESPONSE = {"markets": [], "cursor": ""} + + +def _make_auth() -> KalshiAuth: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return KalshiAuth(key_id="bench", private_key=key) + + +def _config() -> KalshiConfig: + return KalshiConfig( + base_url="https://bench.kalshi.com/trade-api/v2", + max_retries=0, + timeout=5.0, + ) + + +def _bench_sync(requests_n: int) -> float: + with respx.mock: + respx.get("https://bench.kalshi.com/trade-api/v2/markets").mock( + return_value=httpx.Response(200, json=_MOCK_RESPONSE) + ) + client = KalshiClient(auth=_make_auth(), config=_config()) + try: + t0 = time.perf_counter() + for _ in range(requests_n): + client.markets.list(limit=10) + return time.perf_counter() - t0 + finally: + client.close() + + +async def _bench_async(requests_n: int) -> float: + with respx.mock: + respx.get("https://bench.kalshi.com/trade-api/v2/markets").mock( + return_value=httpx.Response(200, json=_MOCK_RESPONSE) + ) + client = AsyncKalshiClient(auth=_make_auth(), config=_config()) + try: + t0 = time.perf_counter() + for _ in range(requests_n): + await client.markets.list(limit=10) + return time.perf_counter() - t0 + finally: + await client.close() + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--requests", type=int, default=2000) + args = p.parse_args() + + sync_elapsed = _bench_sync(args.requests) + print(f"sync : {args.requests} reqs in {sync_elapsed * 1000:.1f} ms " + f"({args.requests / sync_elapsed:,.0f} req/s)") + + async_elapsed = asyncio.run(_bench_async(args.requests)) + print(f"async : {args.requests} reqs in {async_elapsed * 1000:.1f} ms " + f"({args.requests / async_elapsed:,.0f} req/s)") + + +if __name__ == "__main__": + main() diff --git a/scripts/bench_ws_recv.py b/scripts/bench_ws_recv.py new file mode 100644 index 00000000..a1f60e2c --- /dev/null +++ b/scripts/bench_ws_recv.py @@ -0,0 +1,104 @@ +"""Microbench for the WS recv-loop parse+apply hot path. + +Feeds N raw orderbook_snapshot + orderbook_delta frames through the +same json-loads → model_validate → ``_apply_*_inplace`` sequence +:meth:`KalshiWebSocket._process_frame` runs, without standing up the +dispatcher / connection / subscription manager. Reports frames/sec +and GC counts before/after. + +Usage:: + + uv run python scripts/bench_ws_recv.py [--frames N] [--depth D] +""" + +from __future__ import annotations + +import argparse +import gc +import json +import time + +from kalshi.ws.models.orderbook_delta import ( + OrderbookDeltaMessage, + OrderbookSnapshotMessage, +) +from kalshi.ws.orderbook import OrderbookManager + + +def _snapshot_frame(ticker: str, depth: int) -> str: + return json.dumps({ + "type": "orderbook_snapshot", + "sid": 1, + "seq": 1, + "msg": { + "market_ticker": ticker, + "market_id": ticker, + "yes_dollars_fp": [ + [f"0.{i:04d}", "100.00"] for i in range(1, depth + 1) + ], + "no_dollars_fp": [ + [f"0.{i:04d}", "100.00"] for i in range(1, depth + 1) + ], + }, + }) + + +def _delta_frame(ticker: str, seq: int, depth: int) -> str: + price_idx = (seq % depth) + 1 + return json.dumps({ + "type": "orderbook_delta", + "sid": 1, + "seq": seq, + "msg": { + "market_ticker": ticker, + "market_id": ticker, + "price_dollars": f"0.{price_idx:04d}", + "delta_fp": "1.00" if seq % 2 == 0 else "-1.00", + "side": "yes" if seq % 2 == 0 else "no", + }, + }) + + +def _process(raw: str, mgr: OrderbookManager) -> None: + """Mirror of ``KalshiWebSocket._process_frame``'s orderbook fast path.""" + data = json.loads(raw) + msg_type = data.get("type") + if msg_type == "orderbook_snapshot": + snap = OrderbookSnapshotMessage.model_validate(data) + mgr._apply_snapshot_inplace(snap, sid=snap.sid) + elif msg_type == "orderbook_delta": + delta = OrderbookDeltaMessage.model_validate(data) + mgr._apply_delta_inplace(delta) + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--frames", type=int, default=10_000) + p.add_argument("--depth", type=int, default=100, choices=[10, 100, 500]) + args = p.parse_args() + + ticker = "BENCH-WSRECV" + payloads = [_snapshot_frame(ticker, args.depth)] + payloads.extend( + _delta_frame(ticker, seq=i + 2, depth=args.depth) + for i in range(args.frames - 1) + ) + + mgr = OrderbookManager() + gc.collect() + gc_before = gc.get_count() + t0 = time.perf_counter() + for raw in payloads: + _process(raw, mgr) + elapsed = time.perf_counter() - t0 + gc_after = gc.get_count() + + rate = args.frames / elapsed if elapsed > 0 else float("inf") + print(f"frames={args.frames} depth={args.depth}") + print(f" elapsed={elapsed * 1000:.1f} ms rate={rate:,.0f} frames/sec") + print(f" gc_count before={gc_before} after={gc_after}") + + +if __name__ == "__main__": + main() diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index fa7ec2f9..682a8e04 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -60,23 +60,39 @@ def pytest_collection_modifyitems( DEMO_HOST = "demo-api.kalshi.co" -def _bridge_env_vars() -> None: +@pytest.fixture(scope="session") +def _monkeypatch_session() -> Iterator[pytest.MonkeyPatch]: + """Session-scoped ``pytest.MonkeyPatch`` (pytest ships only a + function-scoped fixture). All mutations restore at session end.""" + with pytest.MonkeyPatch.context() as mp: + yield mp + + +@pytest.fixture(scope="session", autouse=True) +def _bridge_env_vars( + _monkeypatch_session: pytest.MonkeyPatch, +) -> None: """Map KALSHI_DEMO_* env vars to KALSHI_* for from_env() compatibility. Supports both naming conventions: .env style: KALSHI_DEMO_KEY_ID, KALSHI_DEMO_PRIVATE_KEY_PATH from_env(): KALSHI_KEY_ID, KALSHI_PRIVATE_KEY_PATH, KALSHI_DEMO=true - """ - if os.environ.get("KALSHI_DEMO_KEY_ID") and not os.environ.get("KALSHI_KEY_ID"): - os.environ["KALSHI_KEY_ID"] = os.environ["KALSHI_DEMO_KEY_ID"] - if os.environ.get("KALSHI_DEMO_PRIVATE_KEY_PATH") and not os.environ.get( - "KALSHI_PRIVATE_KEY_PATH" - ): - os.environ["KALSHI_PRIVATE_KEY_PATH"] = os.environ["KALSHI_DEMO_PRIVATE_KEY_PATH"] - os.environ.setdefault("KALSHI_DEMO", "true") - -_bridge_env_vars() + Bridged through ``_monkeypatch_session`` so the mutations are + automatically rolled back when the integration session ends. + Previously the same logic ran at conftest import time, leaking + demo credentials into any unit test that read ``os.environ`` + without an explicit override (e.g. running ``pytest tests/`` would + silently bridge demo creds for the whole process). + """ + demo_key_id = os.environ.get("KALSHI_DEMO_KEY_ID") + if demo_key_id and not os.environ.get("KALSHI_KEY_ID"): + _monkeypatch_session.setenv("KALSHI_KEY_ID", demo_key_id) + demo_path = os.environ.get("KALSHI_DEMO_PRIVATE_KEY_PATH") + if demo_path and not os.environ.get("KALSHI_PRIVATE_KEY_PATH"): + _monkeypatch_session.setenv("KALSHI_PRIVATE_KEY_PATH", demo_path) + if "KALSHI_DEMO" not in os.environ: + _monkeypatch_session.setenv("KALSHI_DEMO", "true") def _credentials_available() -> bool: diff --git a/tests/test_async_orders.py b/tests/test_async_orders.py index 4f851092..57dae8a2 100644 --- a/tests/test_async_orders.py +++ b/tests/test_async_orders.py @@ -965,36 +965,39 @@ async def test_accepts_typed_order_entries(self, orders: AsyncOrdersResource) -> ] -class TestAsyncBatchCancelRoutesThroughDeleteWithBody: - """Regression for issue #47: async batch_cancel must route through the - shared ``AsyncResource._delete_with_body`` helper so any future - retry / error-mapping behavior added to the helper applies to the +class TestAsyncBatchCancelRoutesThroughDeleteWithBodyJson: + """Regression for issue #47 / #223: async batch_cancel must route through + the shared ``AsyncResource._delete_with_body_json`` bytes helper so any + future retry / error-mapping behavior added to the helper applies to the async path. Sync and async paths must stay symmetric. """ @respx.mock @pytest.mark.asyncio - async def test_batch_cancel_uses_delete_with_body_helper( - self, orders: AsyncOrdersResource + async def test_batch_cancel_uses_delete_with_body_json_helper( + self, orders: AsyncOrdersResource, ) -> None: respx.delete("https://test.kalshi.com/trade-api/v2/portfolio/orders/batched").mock( return_value=httpx.Response(200, json={"orders": []}) ) - # AsyncMock + wraps forwards every call to the real async method, - # so the respx mock above still resolves and the helper's - # status-code branching runs end-to-end; the spy only records. with patch.object( orders, - "_delete_with_body", - wraps=orders._delete_with_body, + "_delete_with_body_json", + wraps=orders._delete_with_body_json, new_callable=AsyncMock, ) as spy: await orders.batch_cancel(["ord-1", "ord-2"]) - spy.assert_called_once_with( - "/portfolio/orders/batched", - json={"orders": [{"order_id": "ord-1"}, {"order_id": "ord-2"}]}, - ) + spy.assert_called_once() + args, kwargs = spy.call_args + assert args == ("/portfolio/orders/batched",) + # Bytes path: caller passes pre-serialized JSON via ``content=``, + # NOT a dict via ``json=`` (P4.2). + assert "json" not in kwargs + assert isinstance(kwargs["content"], bytes) + assert json.loads(kwargs["content"]) == { + "orders": [{"order_id": "ord-1"}, {"order_id": "ord-2"}] + } class TestAmendWireShapeAsync: diff --git a/tests/test_auth.py b/tests/test_auth.py index 9b2d7165..68d78e50 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -8,6 +8,7 @@ import tempfile import pytest +from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa @@ -268,10 +269,41 @@ async def test_lazy_executor_creation(self, test_auth: KalshiAuth) -> None: @pytest.mark.asyncio async def test_close_is_idempotent(self, test_auth: KalshiAuth) -> None: + """Double-close (and triple-close) must not raise. The terminality + check lives in ``test_close_is_terminal_no_silent_respawn`` / + ``test_closed_auth_raises_on_sign_request_async``.""" test_auth.close() # no executor yet + test_auth.close() + test_auth.close() # triple-close OK + + @pytest.mark.asyncio + async def test_closed_auth_raises_on_sign_request_async( + self, test_auth: KalshiAuth, + ) -> None: + """P4.6: ``close()`` is terminal — a subsequent ``sign_request_async`` + raises ``RuntimeError`` instead of silently respawning the + ThreadPoolExecutor (which would defeat the lifecycle contract and + prevent clients from detecting use-after-close in tests).""" + await test_auth.sign_request_async("GET", "/x", timestamp_ms=1) + test_auth.close() + with pytest.raises(RuntimeError, match=r"KalshiAuth has been closed"): + await test_auth.sign_request_async("GET", "/y", timestamp_ms=2) + + @pytest.mark.asyncio + async def test_close_is_terminal_no_silent_respawn( + self, test_auth: KalshiAuth, + ) -> None: + """P4.6: after ``close()``, the executor stays None — the lazy-init + in ``_get_sign_executor`` never reinstates it. Pre-fix it would + have re-allocated a fresh ``ThreadPoolExecutor`` on the next + async sign, leaking resources past the lifecycle bound.""" await test_auth.sign_request_async("GET", "/x", timestamp_ms=1) + assert test_auth._sign_executor is not None test_auth.close() - test_auth.close() # double-close OK + assert test_auth._sign_executor is None + with pytest.raises(RuntimeError): + await test_auth.sign_request_async("GET", "/y", timestamp_ms=2) + assert test_auth._sign_executor is None # no silent respawn @pytest.mark.asyncio async def test_concurrent_signs_do_not_stall_event_loop( @@ -453,6 +485,47 @@ def test_from_path_env_var( assert auth.key_id == "path-key" os.unlink(f.name) + def test_from_env_pem_takes_precedence_over_path( + self, monkeypatch: pytest.MonkeyPatch, + ) -> None: + """P4.4: when BOTH ``KALSHI_PRIVATE_KEY`` (inline PEM) and + ``KALSHI_PRIVATE_KEY_PATH`` (file) are set, ``from_env`` must + load the inline PEM. Pins the existing precedence so a future + refactor that flips the order doesn't silently swap which key + signs production requests.""" + key_a = rsa.generate_private_key(public_exponent=65537, key_size=2048) + key_b = rsa.generate_private_key(public_exponent=65537, key_size=2048) + pem_a = key_a.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode() + pem_b = key_b.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f: + f.write(pem_b) + f.flush() + try: + monkeypatch.setenv("KALSHI_KEY_ID", "precedence-key") + monkeypatch.setenv("KALSHI_PRIVATE_KEY", pem_a) + monkeypatch.setenv("KALSHI_PRIVATE_KEY_PATH", f.name) + auth = KalshiAuth.from_env() + headers = auth.sign_request("GET", "/x", timestamp_ms=1) + sig = base64.b64decode(headers["KALSHI-ACCESS-SIGNATURE"]) + pss = padding.PSS( + mgf=padding.MGF1(hashes.SHA256()), + salt_length=padding.PSS.DIGEST_LENGTH, + ) + # Verifies with A; FAILS to verify with B (proves A signed). + key_a.public_key().verify(sig, b"1GET/x", pss, hashes.SHA256()) + with pytest.raises(InvalidSignature): + key_b.public_key().verify(sig, b"1GET/x", pss, hashes.SHA256()) + finally: + os.unlink(f.name) + class TestTryFromEnv: def test_returns_auth_when_env_vars_set( @@ -483,6 +556,45 @@ def test_returns_none_when_key_id_set_but_no_key( auth = KalshiAuth.try_from_env() assert auth is None + def test_try_from_env_pem_takes_precedence_over_path( + self, monkeypatch: pytest.MonkeyPatch, + ) -> None: + """P4.4: mirror of + ``TestFromEnv.test_from_env_pem_takes_precedence_over_path`` + for ``try_from_env``. Same precedence contract.""" + key_a = rsa.generate_private_key(public_exponent=65537, key_size=2048) + key_b = rsa.generate_private_key(public_exponent=65537, key_size=2048) + pem_a = key_a.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode() + pem_b = key_b.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f: + f.write(pem_b) + f.flush() + try: + monkeypatch.setenv("KALSHI_KEY_ID", "precedence-key") + monkeypatch.setenv("KALSHI_PRIVATE_KEY", pem_a) + monkeypatch.setenv("KALSHI_PRIVATE_KEY_PATH", f.name) + auth = KalshiAuth.try_from_env() + assert auth is not None + headers = auth.sign_request("GET", "/x", timestamp_ms=1) + sig = base64.b64decode(headers["KALSHI-ACCESS-SIGNATURE"]) + pss = padding.PSS( + mgf=padding.MGF1(hashes.SHA256()), + salt_length=padding.PSS.DIGEST_LENGTH, + ) + key_a.public_key().verify(sig, b"1GET/x", pss, hashes.SHA256()) + with pytest.raises(InvalidSignature): + key_b.public_key().verify(sig, b"1GET/x", pss, hashes.SHA256()) + finally: + os.unlink(f.name) + class TestPassphraseSupport: """#217: KalshiAuth loaders accept ``password=`` so callers don't have to diff --git a/tests/test_base_helpers.py b/tests/test_base_helpers.py index 7dd46b41..32c7707d 100644 --- a/tests/test_base_helpers.py +++ b/tests/test_base_helpers.py @@ -170,14 +170,40 @@ def test_repeated_cursor_raises( assert route.call_count == 2 @respx.mock - def test_multi_page_loop_raises( + def test_list_all_cursor_loop_detection_uses_last_cursor_only_o1( self, test_auth: KalshiAuth, test_config: KalshiConfig ) -> None: - """A → B → A revisit also trips detection.""" + """P4.1: cursor guard is O(1) — only the previous page's cursor is + compared. A non-adjacent revisit (A → B → A) must NOT trip the guard; + the realistic server-pagination-bug shape is the *immediate* replay + captured by ``test_repeated_cursor_raises``. The old set-based + guard would have raised here and consumed unbounded memory on a + well-behaved server that legitimately reused cursor tokens across + non-adjacent pages.""" responses = [ httpx.Response(200, json={"items": [{"id": "1"}], "cursor": "A"}), httpx.Response(200, json={"items": [{"id": "2"}], "cursor": "B"}), httpx.Response(200, json={"items": [{"id": "3"}], "cursor": "A"}), + httpx.Response(200, json={"items": [{"id": "4"}], "cursor": ""}), + ] + respx.get("https://test.kalshi.com/trade-api/v2/things").mock( + side_effect=responses + ) + resource = SyncResource(SyncTransport(test_auth, test_config)) + + collected = list(resource._list_all("/things", _Item, "items")) + assert [item.id for item in collected] == ["1", "2", "3", "4"] + + @respx.mock + def test_list_all_cursor_loop_detection_catches_adjacent_replay( + self, test_auth: KalshiAuth, test_config: KalshiConfig + ) -> None: + """P4.1: adjacent replay (A → A) still trips. Mirrors + ``test_repeated_cursor_raises`` with an explicit name pinned to + the O(1) regression boundary.""" + responses = [ + httpx.Response(200, json={"items": [{"id": "1"}], "cursor": "A"}), + httpx.Response(200, json={"items": [{"id": "2"}], "cursor": "A"}), ] respx.get("https://test.kalshi.com/trade-api/v2/things").mock( side_effect=responses diff --git a/tests/test_orders.py b/tests/test_orders.py index 2a0d5a1f..1b92f01f 100644 --- a/tests/test_orders.py +++ b/tests/test_orders.py @@ -890,15 +890,18 @@ def test_accepts_typed_order_entries(self, orders: OrdersResource) -> None: ] -class TestBatchCancelRoutesThroughDeleteWithBody: - """Regression for issue #47: sync batch_cancel must route through the - shared ``SyncResource._delete_with_body`` helper so retry / error-mapping - behavior added to the helper applies to the sync path. Symmetric with - ``test_async_orders.py::TestAsyncBatchCancelRoutesThroughDeleteWithBody``. +class TestBatchCancelRoutesThroughDeleteWithBodyJson: + """Regression for issue #47 / #223: sync batch_cancel must route through + the shared ``SyncResource._delete_with_body_json`` bytes helper so retry / + error-mapping behavior added to the helper applies to the sync path. + Symmetric with + ``test_async_orders.py::TestAsyncBatchCancelRoutesThroughDeleteWithBodyJson``. """ @respx.mock - def test_batch_cancel_uses_delete_with_body_helper(self, orders: OrdersResource) -> None: + def test_batch_cancel_uses_delete_with_body_json_helper( + self, orders: OrdersResource, + ) -> None: respx.delete("https://test.kalshi.com/trade-api/v2/portfolio/orders/batched").mock( return_value=httpx.Response(200, json={"orders": []}) ) @@ -907,14 +910,20 @@ def test_batch_cancel_uses_delete_with_body_helper(self, orders: OrdersResource) # so the respx mock above still resolves; the spy only records. with patch.object( orders, - "_delete_with_body", - wraps=orders._delete_with_body, + "_delete_with_body_json", + wraps=orders._delete_with_body_json, ) as spy: orders.batch_cancel(["ord-1", "ord-2"]) - spy.assert_called_once_with( - "/portfolio/orders/batched", - json={"orders": [{"order_id": "ord-1"}, {"order_id": "ord-2"}]}, - ) + spy.assert_called_once() + args, kwargs = spy.call_args + assert args == ("/portfolio/orders/batched",) + # Bytes path: caller passes pre-serialized JSON via ``content=``, + # NOT a dict via ``json=`` (P4.2). + assert "json" not in kwargs + assert isinstance(kwargs["content"], bytes) + assert json.loads(kwargs["content"]) == { + "orders": [{"order_id": "ord-1"}, {"order_id": "ord-2"}] + } @respx.mock def test_batch_cancel_returns_typed_response_with_reduced_by_fp( @@ -1209,6 +1218,77 @@ def test_wraps_orders_key(self, orders: OrdersResource) -> None: assert set(body.keys()) == {"orders"} + +class TestBatchCreateUsesBytesPath: + """P4.2: batch_create / batch_cancel serialize via model_dump_json directly + to bytes and hand them to httpx as ``content=`` (not ``json=``). This + skips one full dict-walk pass on large payloads where the serializer + cost dominates.""" + + @respx.mock + def test_batch_create_uses_bytes_path(self, orders: OrdersResource) -> None: + """The transport is called with ``content=bytes`` and the + Content-Type header is set explicitly (httpx doesn't infer it for + a raw ``content=`` body).""" + respx.post("https://test.kalshi.com/trade-api/v2/portfolio/orders/batched").mock( + return_value=httpx.Response(200, json={"orders": []}) + ) + + with patch.object( + orders._transport, "request", wraps=orders._transport.request, + ) as spy: + orders.batch_create( + [CreateOrderRequest(ticker="A", side="yes", action="buy")] + ) + spy.assert_called_once() + args, kwargs = spy.call_args + assert args == ("POST", "/portfolio/orders/batched") + assert "json" not in kwargs or kwargs.get("json") is None + assert isinstance(kwargs["content"], bytes) + assert kwargs["headers"]["Content-Type"] == "application/json" + + @respx.mock + def test_batch_cancel_uses_bytes_path(self, orders: OrdersResource) -> None: + """Mirror of ``test_batch_create_uses_bytes_path`` for batch_cancel + — DELETE-with-body bytes path.""" + respx.delete("https://test.kalshi.com/trade-api/v2/portfolio/orders/batched").mock( + return_value=httpx.Response(200, json={"orders": []}) + ) + + with patch.object( + orders._transport, "request", wraps=orders._transport.request, + ) as spy: + orders.batch_cancel(["ord-1"]) + spy.assert_called_once() + args, kwargs = spy.call_args + assert args == ("DELETE", "/portfolio/orders/batched") + assert "json" not in kwargs or kwargs.get("json") is None + assert isinstance(kwargs["content"], bytes) + assert kwargs["headers"]["Content-Type"] == "application/json" + + @respx.mock + def test_batch_create_bytes_path_preserves_decimal_precision( + self, orders: OrdersResource, + ) -> None: + """P4.2: the bytes path uses ``model_dump_json`` which round-trips + Decimals via the configured DollarDecimal serializer. Round-trip + ``Decimal("0.5600")`` through the wire and confirm the exact + string is preserved (no float-conversion drift).""" + route = respx.post( + "https://test.kalshi.com/trade-api/v2/portfolio/orders/batched" + ).mock(return_value=httpx.Response(200, json={"orders": []})) + + orders.batch_create([ + CreateOrderRequest( + ticker="A", side="yes", action="buy", + count=10, yes_price=Decimal("0.5600"), + ), + ]) + + raw = route.calls[0].request.content + body = json.loads(raw) + assert body["orders"][0]["yes_price_dollars"] == "0.5600" + # ── V2 event-market orders (spec v3.18.0) ───────────────────