polish(auth+transport): sign-executor race, Retry-After parity, MessageQueue, http2 test, benches

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.")
 

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

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]

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()

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()