diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index 57b91211..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 @@ -93,9 +94,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: @@ -399,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/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/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/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_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 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.""" 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() 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