From e6c57ddf751eebc1ef3240d90fe31b4944b489fe Mon Sep 17 00:00:00 2001 From: Jeff West Date: Thu, 21 May 2026 08:23:17 -0500 Subject: [PATCH 1/4] fix(auth): recheck _closed under lock in _get_sign_executor (#267) Pre-fix, sign_request_async and close() could race: thread A read _closed=False outside the lock, close() ran end-to-end (set _closed, shut down + nulled the executor), then thread A entered the locked double-checked init and constructed a brand-new ThreadPoolExecutor on a closed auth. The new pool was never shut down (close() already returned), defeating the documented terminality and leaking a 2-thread pool until interpreter shutdown. Move the _closed check inside the lock: keep a lock-free fast path for the already-constructed case, but on the slow path re-read _closed under the lock and raise without instantiating a new executor if close() has won the race. Regression test simulates the interleaving by wrapping the lock's __enter__ to call close() mid-acquire. --- kalshi/auth.py | 35 +++++++++++++++++++-------------- tests/test_auth.py | 48 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 14 deletions(-) diff --git a/kalshi/auth.py b/kalshi/auth.py index b5c64a2f..9f09741d 100644 --- a/kalshi/auth.py +++ b/kalshi/auth.py @@ -296,26 +296,33 @@ def sign_request( def _get_sign_executor(self) -> ThreadPoolExecutor: """Return the dedicated sign-offload executor, creating it on first use. - Thread-safe lazy init: the lock guards the assignment, and the - double-checked pattern keeps the fast path lock-free after first use. + Thread-safe lazy init: the lock guards both the ``_closed`` recheck + and the assignment, so a concurrent :meth:`close` cannot interleave + between "saw open" and "constructed pool" and leave a fresh executor + dangling on a closed auth (#267). 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: - self._sign_executor = ThreadPoolExecutor( - max_workers=2, - thread_name_prefix="kalshi-sign", - ) - return self._sign_executor + # Fast path: already constructed and not closed. ``_closed`` is only + # ever flipped True under the lock, so reading it lock-free here is + # safe — a stale False just falls through to the locked recheck. + executor = self._sign_executor + if executor is not None and not self._closed: + return executor + with self._sign_executor_lock: + if self._closed: + raise RuntimeError( + "KalshiAuth has been closed; create a new instance to sign further requests." + ) + if self._sign_executor is None: + self._sign_executor = ThreadPoolExecutor( + max_workers=2, + thread_name_prefix="kalshi-sign", + ) + return self._sign_executor async def sign_request_async( self, method: str, path: str, timestamp_ms: int | None = None diff --git a/tests/test_auth.py b/tests/test_auth.py index 54dd52d6..520acf32 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -326,6 +326,54 @@ async def test_close_is_terminal_no_silent_respawn( 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_close_during_locked_init_does_not_spawn_executor( + self, + test_auth: KalshiAuth, + ) -> None: + """#267 item 1: simulate close() racing with the locked branch of + ``_get_sign_executor``. Pre-fix, thread A saw ``_closed=False`` outside + the lock, ``close()`` then ran to completion, and thread A still + entered the locked init and spun up a fresh ``ThreadPoolExecutor`` on + a closed auth — leaking the pool past the documented lifecycle bound. + Post-fix the recheck under the lock raises before construction. + """ + # Force the fast path to fall through (no cached executor), then + # interpose ``close()`` between the fast-path check and the locked + # construction by wrapping the lock's ``__enter__``. + assert test_auth._sign_executor is None + original_lock = test_auth._sign_executor_lock + + class _RacyLock: + def __init__(self, inner: object, auth: KalshiAuth) -> None: + self._inner = inner + self._auth = auth + self._fired = False + + def __enter__(self) -> object: + # Race: close() runs after the lock-free check observed + # _closed=False but before we hold the lock. Restore the + # real lock first so close()'s own ``with`` uses it (and + # so we only fire the race once). + if not self._fired: + self._fired = True + self._auth._sign_executor_lock = self._inner # type: ignore[assignment] + self._auth.close() + return self._inner.__enter__() # type: ignore[attr-defined] + + def __exit__(self, *exc: object) -> None: + self._inner.__exit__(*exc) # type: ignore[attr-defined] + + test_auth._sign_executor_lock = _RacyLock(original_lock, test_auth) # type: ignore[assignment] + try: + with pytest.raises(RuntimeError, match=r"KalshiAuth has been closed"): + test_auth._get_sign_executor() + finally: + test_auth._sign_executor_lock = original_lock + assert test_auth._sign_executor is None, ( + "raced close() must not leave a fresh ThreadPoolExecutor dangling" + ) + @pytest.mark.asyncio async def test_concurrent_signs_do_not_stall_event_loop(self, test_auth: KalshiAuth) -> None: """Microbench (#178). Run a real ``asyncio.sleep(0.01)`` ticker From 3939722545489a2c0ecfdec69e1ee5dccb686bbc Mon Sep 17 00:00:00 2001 From: Jeff West Date: Thu, 21 May 2026 08:24:09 -0500 Subject: [PATCH 2/4] fix(transport): clamp negative Retry-After delta to 0.0 for date-form parity (#267) Pre-fix, Retry-After: -5 returned retry_after=None (falls back to computed exponential backoff) while the semantically equivalent ``Retry-After: `` returned retry_after=0.0 (retry immediately). Tests that pinned one shape silently disagreed with the other, and clock-skewed proxies emitting past timestamps got inconsistent behavior depending on which form they used. Pick the more permissive policy and clamp negative delta-seconds to 0.0 \u2014 matching the existing HTTP-date branch. Non-finite (NaN/inf) still falls back to None since those would survive the transport caller's min() cap and either crash time.sleep or schedule a near-infinite delay. Regression test asserts both forms produce identical retry_after. --- kalshi/_base_client.py | 12 ++++++++++-- tests/test_client.py | 27 ++++++++++++++++++++++++--- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index 57b91211..a162a038 100644 --- a/kalshi/_base_client.py +++ b/kalshi/_base_client.py @@ -93,9 +93,17 @@ def _map_error(response: httpx.Response) -> KalshiError: if retry_after: try: retry_after_val = float(retry_after) - # Reject non-finite/negative: NaN crashes time.sleep, negative busy-loops. - if retry_after_val < 0 or not math.isfinite(retry_after_val): + # Reject non-finite (NaN/inf) which would survive the + # transport caller's min() cap. Negative deltas — a server + # signalling "retry now" or a clock-skewed proxy emitting a + # past timestamp — clamp to 0.0 so both Retry-After forms + # behave the same (#267): a negative delta-seconds value and + # a past HTTP-date below now both produce ``retry_after=0.0`` + # rather than diverging into None vs 0.0. + if not math.isfinite(retry_after_val): retry_after_val = None + elif retry_after_val < 0: + retry_after_val = 0.0 except ValueError: # RFC 7231 §7.1.3: Retry-After may also be an HTTP-date. try: diff --git a/tests/test_client.py b/tests/test_client.py index b19d7fed..aa177211 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -134,12 +134,33 @@ def test_validation_error_with_details(self) -> None: # transport never calls time.sleep with a non-positive or non-finite # delay (busy-loop or ValueError crash respectively). - def test_429_retry_after_negative_falls_back(self) -> None: - """Negative Retry-After is rejected so backoff isn't bypassed.""" + def test_429_retry_after_negative_clamps_to_zero(self) -> None: + """#267 item 2: negative Retry-After delta-seconds clamps to 0.0 so it + matches the HTTP-date branch's past-date behavior. ``time.sleep(0.0)`` + is safe; the busy-loop concern (#96) only applies to unbounded + negative loops, not a single 0-second retry.""" resp = httpx.Response(429, json={"message": "slow"}, headers={"Retry-After": "-1"}) err = _map_error(resp) assert isinstance(err, KalshiRateLimitError) - assert err.retry_after is None + assert err.retry_after == 0.0 + + def test_429_retry_after_negative_and_past_date_agree(self) -> None: + """#267 item 2: ``Retry-After: -5`` and a past HTTP-date produce the + same ``retry_after`` value. Pre-fix the former returned None (computed + backoff) while the latter clamped to 0.0 — silently divergent.""" + neg = httpx.Response( + 429, json={"message": "slow"}, headers={"Retry-After": "-5"} + ) + past = httpx.Response( + 429, + json={"message": "slow"}, + headers={"Retry-After": "Wed, 21 Oct 1999 07:28:00 GMT"}, + ) + neg_err = _map_error(neg) + past_err = _map_error(past) + assert isinstance(neg_err, KalshiRateLimitError) + assert isinstance(past_err, KalshiRateLimitError) + assert neg_err.retry_after == past_err.retry_after == 0.0 def test_429_retry_after_nan_falls_back(self) -> None: """NaN Retry-After is rejected so time.sleep never raises ValueError.""" From 226bcebc487dab6cc9a40d16dbda8d6df16cb8c0 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Thu, 21 May 2026 08:26:00 -0500 Subject: [PATCH 3/4] perf(ws,transport): drop MessageQueue._size; hoist asyncio import (#271) Two small per-message hot-path savings: - MessageQueue._size: an O(1) int counter incremented/decremented on every put / popleft purely so qsize() could exclude the single terminal sentinel. Two extra int ops per message on the WS recv hot path with no functional gain \u2014 derive qsize from len(self._buffer) minus a sentinel adjustment when the queue is closed. The deque's maxlen=maxsize+1 was already the real memory ceiling. - import asyncio inside AsyncTransport.request: a sys.modules lookup on every async request. The module is already imported at top level by KalshiAuth.sign_request_async (and many others); move the import to the file header. Updated the deque-maxlen regression to assert the cap via sustained overflow rather than simulated counter drift (the counter no longer exists). --- kalshi/_base_client.py | 3 +-- kalshi/ws/backpressure.py | 28 ++++++++++++++-------------- tests/ws/test_backpressure.py | 20 ++++++-------------- 3 files changed, 21 insertions(+), 30 deletions(-) diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index a162a038..e3be8274 100644 --- a/kalshi/_base_client.py +++ b/kalshi/_base_client.py @@ -6,6 +6,7 @@ from __future__ import annotations +import asyncio import email.utils import logging import math @@ -407,8 +408,6 @@ async def request( See :meth:`SyncTransport.request` for ``extra_headers`` semantics. """ - import asyncio - if json is not None and content is not None: raise TypeError("request() accepts `json=` or `content=`, not both.") diff --git a/kalshi/ws/backpressure.py b/kalshi/ws/backpressure.py index bf29d85a..5f21ef9f 100644 --- a/kalshi/ws/backpressure.py +++ b/kalshi/ws/backpressure.py @@ -66,17 +66,13 @@ def __init__( # e.client_id, ...)`` without parsing the message string. self._channel = channel self._client_id = client_id - # `maxlen=maxsize+1` is a hard memory ceiling enforced by deque itself, - # independent of `_size`. If the counter ever drifts (a put path that - # forgets to increment, an exception between append and increment, etc.) - # the buffer still cannot grow without bound. +1 because put_sentinel - # appends without going through the size-check overflow path. (#173) + # `maxlen=maxsize+1` is a hard memory ceiling enforced by deque + # itself. Even if the size check is bypassed for any reason, the buffer + # still cannot grow without bound. +1 because put_sentinel appends + # without going through the size-check overflow path. (#173) self._buffer: collections.deque[T | object] = collections.deque(maxlen=maxsize + 1) self._event = asyncio.Event() self._closed = False - # O(1) size tracking: counts real items only (excludes sentinel). - # Kept in sync with _buffer on every put/popleft path. - self._size = 0 async def put(self, item: T) -> None: """Add an item to the queue, applying overflow strategy if full.""" @@ -86,7 +82,6 @@ async def put(self, item: T) -> None: if len(self._buffer) >= self._maxsize: if self._overflow is OverflowStrategy.DROP_OLDEST: self._buffer.popleft() - self._size -= 1 elif self._overflow is OverflowStrategy.ERROR: raise KalshiBackpressureError( f"Message queue full ({self._maxsize} items). " @@ -98,7 +93,6 @@ async def put(self, item: T) -> None: ) self._buffer.append(item) - self._size += 1 self._event.set() async def put_sentinel(self) -> None: @@ -144,12 +138,19 @@ async def get(self) -> T: raise StopAsyncIteration if isinstance(item, _ErrorSentinel): raise item.exc - self._size -= 1 return item # type: ignore[return-value] def qsize(self) -> int: - """Number of items currently in the queue (excludes sentinel). O(1).""" - return self._size + """Number of items currently in the queue (excludes sentinel). O(1). + + Derived from ``len(self._buffer)`` minus one when a terminal sentinel + has been appended (``put_sentinel`` / ``put_error`` both flip + ``_closed`` and push exactly one sentinel-shaped item). + """ + n = len(self._buffer) + if self._closed and n: + return n - 1 + return n def __aiter__(self) -> AsyncIterator[T]: return self @@ -164,5 +165,4 @@ async def __anext__(self) -> T: raise StopAsyncIteration if isinstance(item, _ErrorSentinel): raise item.exc - self._size -= 1 return item # type: ignore[return-value] diff --git a/tests/ws/test_backpressure.py b/tests/ws/test_backpressure.py index 362c04b7..dec8c0aa 100644 --- a/tests/ws/test_backpressure.py +++ b/tests/ws/test_backpressure.py @@ -118,27 +118,19 @@ async def test_qsize_stays_accurate_across_put_get_drop_and_sentinel(self) -> No assert await q.get() == 4 assert q.qsize() == 0 - async def test_deque_maxlen_caps_memory_even_if_counter_drifts(self) -> None: + async def test_deque_maxlen_caps_memory_under_sustained_overflow(self) -> None: """Regression for #173: the underlying deque carries `maxlen=maxsize+1` - as a defense-in-depth ceiling. Even if the manual `_size` counter - drifts (a hypothetical bug that fails to track an eviction), the deque - cannot grow without bound — `maxlen` enforces the cap at the C level. - - Simulates counter drift by zeroing `_size` mid-stream so the overflow - check (`len(self._buffer) >= self._maxsize`) still triggers via the - real buffer length, but a counter-based bound would not. Then asserts - memory stays bounded regardless. + as a defense-in-depth ceiling. Even under sustained overflow far past + the configured ``maxsize``, the buffer cannot grow without bound — + ``maxlen`` enforces the cap at the C level. """ q: MessageQueue[int] = MessageQueue(maxsize=5, overflow=OverflowStrategy.DROP_OLDEST) - for i in range(5): - await q.put(i) - # Inject counter drift. From here `_size` lies about the queue's - # occupancy; the only thing protecting memory is the deque's `maxlen`. - q._size = 0 for i in range(1_000): await q.put(i) # Hard ceiling: maxsize + 1 (the +1 covers the put_sentinel append). assert len(q._buffer) <= q._maxsize + 1 + # qsize() reflects the real (capped) occupancy. + assert q.qsize() == q._maxsize @pytest.mark.asyncio From 1e97725453ac090b4b233beb2952ff23e7f61adf Mon Sep 17 00:00:00 2001 From: Jeff West Date: Thu, 21 May 2026 08:29:45 -0500 Subject: [PATCH 4/4] test(transport,bench): http2 wiring test + iterator/dataframe benches (#271) #271 items 3 + 4 \u2014 testing-coverage polish. tests/test_http2.py: verify KalshiConfig(http2=True) propagates to the underlying httpx.Client / httpx.AsyncClient's connection pool, so a regression that silently drops the kwarg surfaces here instead of landing without any failing test. Guarded by skipif when h2 isn't installed in the env (httpx rejects http2=True without it). scripts/bench_orderbook_iterator.py: drives _OrderbookIterator.__anext__ end-to-end through a synthetic stream that mimics the recv loop (apply_delta_inplace + yield). Existing bench_orderbook_delta only measures the in-place mutation; this one captures the mgr.get + Orderbook materialization that real async for book in ws.orderbook() consumers pay. scripts/bench_page_to_dataframe.py: builds a 1000-item Page[Market] and times to_dataframe() / to_polars(). Catches regressions in the column-oriented build path (#264) and confirms the DollarDecimal contract (#225/#190) doesn't quietly fall back to per-row model_dump. Closes #267 Closes #271 --- scripts/bench_orderbook_iterator.py | 120 ++++++++++++++++++++++++++++ scripts/bench_page_to_dataframe.py | 116 +++++++++++++++++++++++++++ tests/test_http2.py | 83 +++++++++++++++++++ 3 files changed, 319 insertions(+) create mode 100644 scripts/bench_orderbook_iterator.py create mode 100644 scripts/bench_page_to_dataframe.py create mode 100644 tests/test_http2.py diff --git a/scripts/bench_orderbook_iterator.py b/scripts/bench_orderbook_iterator.py new file mode 100644 index 00000000..b48d3458 --- /dev/null +++ b/scripts/bench_orderbook_iterator.py @@ -0,0 +1,120 @@ +"""Microbench for :class:`_OrderbookIterator.__anext__` end-to-end (#271 item 4). + +Existing ``bench_orderbook_delta.py`` measures the in-place mutation cost +(``_apply_delta_inplace``) but stops short of the materialization that +``async for book in ws.orderbook(ticker)`` consumers actually pay: +``mgr.get(ticker)`` re-sorts the inner price levels and validates an +:class:`Orderbook` model on every update. + +This bench drives a synthetic stream through the iterator so we measure +the full hot path a strategy sees: stream.__anext__ + mgr.get + +Orderbook validation. + +Usage:: + + uv run python scripts/bench_orderbook_iterator.py [--depth D] [--updates N] +""" + +from __future__ import annotations + +import argparse +import asyncio +import gc +import time +from collections.abc import AsyncIterator +from decimal import Decimal + +from kalshi.ws.client import _OrderbookIterator +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] + ), + ) + + +async def _drive(depth: int, updates: int) -> tuple[float, int]: + ticker = "BENCH-ITER" + mgr = OrderbookManager() + mgr._apply_snapshot_inplace(_build_snapshot(ticker, depth)) + + deltas = [ + _build_delta( + ticker, + seq=i + 2, + price=Decimal("0." + f"{(i % depth) + 1:04d}"), + delta=Decimal("1.00") if i % 2 == 0 else Decimal("-1.00"), + side="yes" if i % 2 == 0 else "no", + ) + for i in range(updates) + ] + + async def stream() -> AsyncIterator[OrderbookDeltaMessage]: + # Simulates the recv loop: apply delta to the manager, then yield. + # The iterator under test reads `mgr.get(ticker)` after each yield. + for d in deltas: + mgr._apply_delta_inplace(d) + yield d + + it = _OrderbookIterator(stream(), mgr, ticker) + gc.collect() + t0 = time.perf_counter() + consumed = 0 + async for _book in it: + consumed += 1 + elapsed = time.perf_counter() - t0 + return elapsed, consumed + + +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=1_000) + args = p.parse_args() + + elapsed, consumed = asyncio.run(_drive(args.depth, args.updates)) + rate = consumed / elapsed if elapsed > 0 else float("inf") + print(f"depth={args.depth} updates={args.updates} consumed={consumed}") + print(f" elapsed={elapsed * 1000:.1f} ms rate={rate:,.0f} updates/sec") + + +if __name__ == "__main__": + main() diff --git a/scripts/bench_page_to_dataframe.py b/scripts/bench_page_to_dataframe.py new file mode 100644 index 00000000..9b347ad8 --- /dev/null +++ b/scripts/bench_page_to_dataframe.py @@ -0,0 +1,116 @@ +"""Microbench for :meth:`Page.to_dataframe` / :meth:`Page.to_polars` (#271 item 4). + +A 1000-item Page of :class:`Market` instances is built once, then handed +to pandas and polars in turn. Catches regressions in the column-oriented +build path (#264) and confirms the Decimal contract (#225/#190) doesn't +silently fall back to per-row ``model_dump``. + +Usage:: + + uv run python scripts/bench_page_to_dataframe.py [--rows N] +""" + +from __future__ import annotations + +import argparse +import gc +import importlib.util +import time +from typing import Any + +from kalshi.models.common import Page +from kalshi.models.markets import Market + + +def _market_dict(ticker: str) -> dict[str, Any]: + """Spec-shaped wire dict; mirrors ``tests/_model_fixtures.market_dict`` so + bench scripts don't pull a tests-only helper.""" + return { + "ticker": ticker, + "event_ticker": "EVT-A", + "market_type": "binary", + "yes_sub_title": "Yes", + "no_sub_title": "No", + "status": "open", + "yes_bid_dollars": "0.5000", + "yes_ask_dollars": "0.5100", + "no_bid_dollars": "0.4900", + "no_ask_dollars": "0.5000", + "last_price_dollars": "0.5000", + "previous_yes_bid_dollars": "0.5000", + "previous_yes_ask_dollars": "0.5100", + "previous_price_dollars": "0.5000", + "notional_value_dollars": "1.0000", + "liquidity_dollars": "100.0000", + "yes_bid_size_fp": "0.00", + "yes_ask_size_fp": "0.00", + "volume_fp": "0.00", + "volume_24h_fp": "0.00", + "open_interest_fp": "0.00", + "created_time": "2026-01-01T00:00:00Z", + "updated_time": "2026-01-01T00:00:00Z", + "open_time": "2026-01-01T00:00:00Z", + "close_time": "2026-12-31T23:59:59Z", + "latest_expiration_time": "2026-12-31T23:59:59Z", + "settlement_timer_seconds": 0, + "result": "", + "can_close_early": False, + "fractional_trading_enabled": False, + "expiration_value": "", + "rules_primary": "", + "rules_secondary": "", + "price_level_structure": "binary", + "price_ranges": [], + } + + +def _build_page(rows: int) -> Page[Market]: + items = [Market.model_validate(_market_dict(f"BENCH-{i:05d}")) for i in range(rows)] + return Page[Market](items=items, cursor=None) + + +def _time_call(fn: Any, repeats: int) -> float: + gc.collect() + t0 = time.perf_counter() + for _ in range(repeats): + fn() + return time.perf_counter() - t0 + + +def main() -> None: + p = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + p.add_argument("--rows", type=int, default=1_000) + p.add_argument("--repeats", type=int, default=10) + args = p.parse_args() + + page = _build_page(args.rows) + total_conversions = args.rows * args.repeats + + pandas_available = importlib.util.find_spec("pandas") is not None + polars_available = importlib.util.find_spec("polars") is not None + + if pandas_available: + elapsed = _time_call(page.to_dataframe, args.repeats) + rate = total_conversions / elapsed if elapsed > 0 else float("inf") + print( + f"pandas: rows={args.rows} repeats={args.repeats} " + f"elapsed={elapsed * 1000:.1f} ms rate={rate:,.0f} rows/sec" + ) + else: + print("pandas: not installed (skip — install kalshi-sdk[pandas])") + + if polars_available: + elapsed = _time_call(page.to_polars, args.repeats) + rate = total_conversions / elapsed if elapsed > 0 else float("inf") + print( + f"polars: rows={args.rows} repeats={args.repeats} " + f"elapsed={elapsed * 1000:.1f} ms rate={rate:,.0f} rows/sec" + ) + else: + print("polars: not installed (skip — install kalshi-sdk[polars])") + + +if __name__ == "__main__": + main() diff --git a/tests/test_http2.py b/tests/test_http2.py new file mode 100644 index 00000000..50f942ec --- /dev/null +++ b/tests/test_http2.py @@ -0,0 +1,83 @@ +"""Tests that ``KalshiConfig(http2=True)`` wires through to httpx (#271 item 3). + +The advertised HTTP/2 multiplexing benefit (#233) only kicks in if the +underlying ``httpx.Client`` / ``httpx.AsyncClient`` actually has +``http2=True`` set, which in turn requires the ``h2`` package to be +importable. Without this check, a regression that silently disables h2 +(e.g. dropping the ``http2`` kwarg from the transport's ``client_kwargs``) +would land with no test failing. + +We avoid hitting the network: just construct each client and inspect +the httpx-internal flag. +""" + +from __future__ import annotations + +import importlib.util + +import pytest +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 + +_h2_missing = importlib.util.find_spec("h2") is None + + +def _make_auth() -> KalshiAuth: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return KalshiAuth(key_id="test", private_key=key) + + +@pytest.mark.skipif(_h2_missing, reason="h2 package not installed; httpx rejects http2=True") +class TestHttp2Wiring: + """``http2=True`` on ``KalshiConfig`` must propagate to the httpx client.""" + + def test_sync_client_enables_http2(self) -> None: + config = KalshiConfig(http2=True) + client = KalshiClient(auth=_make_auth(), config=config) + try: + # httpx exposes the negotiated h2 flag on the private transport + # config; ``http2`` is also reflected on the public client's + # ``_transport`` chain. We read the transport directly since + # that is what carries the kwarg. + httpx_client = client._transport._client + # ``_transport`` is httpx's default ``HTTPTransport`` which stores + # the http2 flag on its ``_pool``. Cheaper to just confirm the + # kwarg made it onto the high-level client. + assert httpx_client._transport is not None + # Probe via the same path httpx uses internally: every default + # transport built with http2=True sets ``http2`` on its pool. + pool = httpx_client._transport._pool # type: ignore[attr-defined] + assert pool._http2 is True, ( + "KalshiConfig(http2=True) did not propagate to httpx.Client's pool" + ) + finally: + client.close() + + @pytest.mark.asyncio + async def test_async_client_enables_http2(self) -> None: + config = KalshiConfig(http2=True) + client = AsyncKalshiClient(auth=_make_auth(), config=config) + try: + httpx_client = client._transport._client + pool = httpx_client._transport._pool # type: ignore[attr-defined] + assert pool._http2 is True, ( + "KalshiConfig(http2=True) did not propagate to httpx.AsyncClient's pool" + ) + finally: + await client.close() + + def test_default_config_does_not_enable_http2(self) -> None: + """``http2`` defaults to False so the test above is meaningful.""" + config = KalshiConfig() + assert config.http2 is False + client = KalshiClient(auth=_make_auth(), config=config) + try: + httpx_client = client._transport._client + pool = httpx_client._transport._pool # type: ignore[attr-defined] + assert pool._http2 is False + finally: + client.close()