kalarb-pykalshi-47
Results
The prompt the agent sees
The asynchronous read/write token bucket can hang when multiple coroutines acquire tokens concurrently and demand exceeds the available capacity. For example, starting several write operations against a bucket with capacity for only two writes may leave the waiting operations stuck indefinitely instead of completing after tokens refill. Similar hangs or incorrect outcomes can occur under mixed concurrent read and write load.
The bucket must enforce disjoint read and write token accounting: a request whose write cost is positive consumes only write tokens and leaves the read balance unchanged, even when a read cost is also passed; a request whose write cost is zero consumes only read tokens and leaves the write balance unchanged. For example, with a bucket of read rate 10 and write rate 5, one acquisition of read cost 1 and write cost 1 leaves 10 read tokens and 4 write tokens, and one acquisition of read cost 1 and write cost 0 leaves 9 read tokens and 5 write tokens. Nonblocking acquisition should report success only when sufficient tokens are available and otherwise report failure without consuming tokens. Waiting-time queries should report no wait when a request can proceed immediately, and acquisitions that exceed the bucket’s configured capacity should be rejected rather than wait forever.
Concurrent operations should complete once tokens become available, without deadlocking, double-consuming tokens, or allowing requests to proceed before the required resources have refilled.
The rate limiter's public interface names the read-side cost `read_cost`: `acquire`, `try_acquire` and `get_wait_time` on `ReadWriteTokenBucket` (and the `RateLimiterProtocol` contract) must accept the keyword arguments `read_cost` and `write_cost`; callers pass, for example, `acquire(read_cost=1.0, write_cost=0.0)` for a read and `acquire(read_cost=0.0, write_cost=1.0)` for a write. Callers inside the package that use the old keyword must keep working with the new name.
Hidden tests · 11 fail-to-pass, 7 pass-to-passrun after the agent submits, in a clean verifier
Test patch · 119 lines
diff --git a/tests/test_rate_limiter.py b/tests/test_rate_limiter.py
index 240215f..140c00e 100644
--- a/tests/test_rate_limiter.py
+++ b/tests/test_rate_limiter.py
@@ -14,34 +14,34 @@ async def test_initial_tokens(self) -> None:
@pytest.mark.asyncio
async def test_read_acquire_consumes_read_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- await bucket.acquire(global_cost=1.0, write_cost=0.0)
+ await bucket.acquire(read_cost=1.0, write_cost=0.0)
assert bucket.read_tokens == 9.0
assert bucket.write_tokens == 5.0 # write unchanged
@pytest.mark.asyncio
async def test_write_acquire_consumes_write_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- await bucket.acquire(global_cost=1.0, write_cost=1.0)
+ await bucket.acquire(read_cost=1.0, write_cost=1.0)
assert bucket.write_tokens == 4.0
assert bucket.read_tokens == 10.0 # read unchanged
@pytest.mark.asyncio
async def test_try_acquire_success(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- result = await bucket.try_acquire(global_cost=1.0, write_cost=0.0)
+ result = await bucket.try_acquire(read_cost=1.0, write_cost=0.0)
assert result is True
@pytest.mark.asyncio
async def test_try_acquire_insufficient_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=1.0, write_rate=1.0)
- await bucket.acquire(global_cost=1.0, write_cost=0.0)
- result = await bucket.try_acquire(global_cost=1.0, write_cost=0.0)
+ await bucket.acquire(read_cost=1.0, write_cost=0.0)
+ result = await bucket.try_acquire(read_cost=1.0, write_cost=0.0)
assert result is False
@pytest.mark.asyncio
async def test_get_wait_time_immediate(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- wait = await bucket.get_wait_time(global_cost=1.0, write_cost=0.0)
+ wait = await bucket.get_wait_time(read_cost=1.0, write_cost=0.0)
assert wait == 0.0
@pytest.mark.asyncio
@@ -84,9 +84,9 @@ def test_invalid_rates(self) -> None:
async def test_exceed_capacity_raises(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=5.0, write_rate=3.0)
with pytest.raises(ValueError, match="exceeds"):
- await bucket.acquire(global_cost=6.0, write_cost=0.0)
+ await bucket.acquire(read_cost=6.0, write_cost=0.0)
with pytest.raises(ValueError, match="exceeds"):
- await bucket.acquire(global_cost=1.0, write_cost=4.0)
+ await bucket.acquire(read_cost=1.0, write_cost=4.0)
def test_get_status(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
@@ -105,7 +105,7 @@ async def test_concurrent_reads_consume_correct_tokens(self) -> None:
import asyncio
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- tasks = [bucket.acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
+ tasks = [bucket.acquire(read_cost=1.0, write_cost=0.0) for _ in range(5)]
await asyncio.gather(*tasks)
assert bucket.read_tokens == 5.0
assert bucket.write_tokens == 5.0 # writes untouched
@@ -116,7 +116,7 @@ async def test_concurrent_writes_consume_correct_tokens(self) -> None:
import asyncio
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- tasks = [bucket.acquire(global_cost=1.0, write_cost=1.0) for _ in range(3)]
+ tasks = [bucket.acquire(read_cost=1.0, write_cost=1.0) for _ in range(3)]
await asyncio.gather(*tasks)
assert bucket.write_tokens == 2.0
assert bucket.read_tokens == 10.0 # reads untouched
@@ -127,8 +127,8 @@ async def test_concurrent_mixed_load(self) -> None:
import asyncio
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
- read_tasks = [bucket.acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
- write_tasks = [bucket.acquire(global_cost=1.0, write_cost=1.0) for _ in range(3)]
+ read_tasks = [bucket.acquire(read_cost=1.0, write_cost=0.0) for _ in range(5)]
+ write_tasks = [bucket.acquire(read_cost=1.0, write_cost=1.0) for _ in range(3)]
await asyncio.gather(*read_tasks, *write_tasks)
assert bucket.read_tokens == 5.0
assert bucket.write_tokens == 2.0
@@ -141,10 +141,30 @@ async def test_no_double_consumption(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=5.0, write_rate=5.0)
# Try to consume 5 tokens concurrently (exactly capacity)
results = await asyncio.gather(
- *[bucket.try_acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
+ *[bucket.try_acquire(read_cost=1.0, write_cost=0.0) for _ in range(5)]
)
assert sum(results) == 5 # all should succeed
assert bucket.read_tokens == 0.0
# 6th attempt should fail (no tokens left)
- assert await bucket.try_acquire(global_cost=1.0, write_cost=0.0) is False
+ assert await bucket.try_acquire(read_cost=1.0, write_cost=0.0) is False
+
+ @pytest.mark.asyncio
+ async def test_acquire_does_not_deadlock(self) -> None:
+ """Concurrent acquires that exceed capacity must not deadlock.
+
+ Regression test: the lock must be released before sleeping so other
+ coroutines can proceed when tokens refill.
+ """
+ import asyncio
+
+ bucket = ReadWriteTokenBucket(read_rate=2.0, write_rate=2.0)
+ # Launch 4 writes (capacity is 2), so 2 must wait for refill.
+ # With a 1-second window this should complete in ~1s, not deadlock.
+ results = await asyncio.wait_for(
+ asyncio.gather(*[
+ bucket.acquire(read_cost=0.0, write_cost=1.0) for _ in range(4)
+ ]),
+ timeout=5.0,
+ )
+ assert len(results) == 4
Reference fix · 4 files, +33 −35the upstream merge, used only for grading calibration
The agent could not see this: the repository holds one commit and the sandbox has no network. Leak audit.
.env_demo, src/pykalshi/http_client.py, src/pykalshi/protocols.py, src/pykalshi/rate_limiter.py
diff --git a/.env_demo b/.env_demo
index 444b0a1..17129db 100644
--- a/.env_demo
+++ b/.env_demo
@@ -13,9 +13,9 @@ KALSHI_PROD_PRIVATE_KEY_FILE=<INSERT_PRIV_API_KEY_FILE_PATH>
# Fallback token rates used before auto-configuration resolves on first request,
# and if the /account/limits fetch fails. The client automatically fetches your
# actual tier limits and per-endpoint costs from the API (auto_configure_rates=True).
-# These values approximate Kalshi's standard tier as requests-per-second
-# (rate=20 with cost=1 per request, before auto-config updates to actual tokens).
-KALSHI_RATE_LIMIT_GLOBAL=20.0
+# These values are conservative fallbacks (Basic tier is 200 read / 100 write
+# tokens/sec with a default cost of 10 tokens per request).
+KALSHI_RATE_LIMIT_READ=20.0
KALSHI_RATE_LIMIT_WRITE=10.0
# --- Dev / Debug ---
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..24a9c5d 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -279,12 +279,12 @@ async def _execute_request(
cost = self._resolve_cost(method, path)
if method.upper() == "GET":
- global_cost, write_cost = cost, 0.0
+ read_cost, write_cost = cost, 0.0
else:
- global_cost, write_cost = 0.0, cost
+ read_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=read_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..f029e02 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,7 @@ async def resubscribe_channel(self, channel_name: str) -> None: ...
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
+ async def acquire(self, read_cost: float = 10.0, write_cost: float = 0.0) -> None: ...
+ async def try_acquire(self, read_cost: float = 10.0, write_cost: float = 0.0) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
diff --git a/src/pykalshi/rate_limiter.py b/src/pykalshi/rate_limiter.py
index 17a53ff..2f61f04 100644
--- a/src/pykalshi/rate_limiter.py
+++ b/src/pykalshi/rate_limiter.py
@@ -55,12 +55,12 @@ def _refill_write(self) -> None:
else:
break
- def _can_proceed(self, global_cost: float, write_cost: float) -> bool:
+ def _can_proceed(self, read_cost: float, write_cost: float) -> bool:
if write_cost > 0:
return self.write_tokens >= write_cost
- return self.read_tokens >= global_cost
+ return self.read_tokens >= read_cost
- def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float:
+ def _calculate_wait_time(self, read_cost: float, write_cost: float) -> float:
now = time.monotonic()
if write_cost > 0:
@@ -75,9 +75,9 @@ def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float:
return max(0.0, target_time - now)
return 1.0
else:
- if self.read_tokens >= global_cost:
+ if self.read_tokens >= read_cost:
return 0.0
- needed = global_cost - self.read_tokens
+ needed = read_cost - self.read_tokens
recovered = 0.0
for timestamp, amount in self._read_history:
recovered += amount
@@ -86,45 +86,43 @@ def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float:
return max(0.0, target_time - now)
return 1.0
- def _consume(self, global_cost: float, write_cost: float) -> None:
+ def _consume(self, read_cost: float, write_cost: float) -> None:
now = time.monotonic()
if write_cost > 0:
self.write_tokens -= write_cost
self._write_history.append((now, write_cost))
else:
- self.read_tokens -= global_cost
- self._read_history.append((now, global_cost))
+ self.read_tokens -= read_cost
+ self._read_history.append((now, read_cost))
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None:
+ async def acquire(self, read_cost: float = 10.0, write_cost: float = 0.0) -> None:
"""Acquire tokens, blocking until available."""
if write_cost > 0 and write_cost > self.write_capacity:
raise ValueError(f"Write cost {write_cost} exceeds write capacity")
- if write_cost == 0 and global_cost > self.read_capacity:
- raise ValueError(f"Read cost {global_cost} exceeds read capacity")
+ if write_cost == 0 and read_cost > self.read_capacity:
+ raise ValueError(f"Read cost {read_cost} exceeds read capacity")
lock = self._write_lock if write_cost > 0 else self._read_lock
- async with lock:
- while True:
+ while True:
+ async with lock:
if write_cost > 0:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return
- wait_time = self._calculate_wait_time(global_cost, write_cost)
- if wait_time > 0:
- await asyncio.sleep(wait_time)
- else:
- await asyncio.sleep(0.01)
+ wait_time = self._calculate_wait_time(read_cost, write_cost)
+ # Sleep OUTSIDE the lock to avoid deadlock
+ await asyncio.sleep(max(wait_time, 0.01))
async def try_acquire(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self, read_cost: float = 10.0, write_cost: float = 0.0
) -> bool:
"""Try to acquire tokens without blocking. Returns True if acquired."""
if write_cost > 0 and write_cost > self.write_capacity:
return False
- if write_cost == 0 and global_cost > self.read_capacity:
+ if write_cost == 0 and read_cost > self.read_capacity:
return False
lock = self._write_lock if write_cost > 0 else self._read_lock
@@ -133,13 +131,13 @@ async def try_acquire(
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return True
return False
async def get_wait_time(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self, read_cost: float = 10.0, write_cost: float = 0.0
) -> float:
"""Estimate wait time without consuming tokens."""
lock = self._write_lock if write_cost > 0 else self._read_lock
@@ -148,9 +146,9 @@ async def get_wait_time(
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
+ if self._can_proceed(read_cost, write_cost):
return 0.0
- return self._calculate_wait_time(global_cost, write_cost)
+ return self._calculate_wait_ti
… [71 more characters]GLM-5.3 Flash · attempt 2 · solved · 20 steps · $0.011
Submitted diff · 261 lines
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..24a9c5d 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -279,12 +279,12 @@ class KalshiHttpClient:
cost = self._resolve_cost(method, path)
if method.upper() == "GET":
- global_cost, write_cost = cost, 0.0
+ read_cost, write_cost = cost, 0.0
else:
- global_cost, write_cost = 0.0, cost
+ read_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=read_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..5fd3f7b 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,7 @@ class KalshiWebSocketClientProtocol(Protocol):
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
+ async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
+ async def try_acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
diff --git a/src/pykalshi/rate_limiter.py b/src/pykalshi/rate_limiter.py
index 17a53ff..dd92c37 100644
--- a/src/pykalshi/rate_limiter.py
+++ b/src/pykalshi/rate_limiter.py
@@ -32,8 +32,11 @@ class ReadWriteTokenBucket:
self._read_history: deque[Tuple[float, float]] = deque()
self._write_history: deque[Tuple[float, float]] = deque()
- self._read_lock = asyncio.Lock()
- self._write_lock = asyncio.Lock()
+ # Conditions guard the read and write budgets independently. Waiters
+ # are notified whenever tokens are consumed so that no coroutine can
+ # oversleep past the point where the resources it needs are available.
+ self._read_cond = asyncio.Condition()
+ self._write_cond = asyncio.Condition()
def _refill_read(self) -> None:
expiry = time.monotonic() - self.window_size
@@ -55,12 +58,14 @@ class ReadWriteTokenBucket:
else:
break
- def _can_proceed(self, global_cost: float, write_cost: float) -> bool:
+ def _can_proceed(self, read_cost: float, write_cost: float) -> bool:
+ """A request consumes write tokens if it has a positive write cost,
+ otherwise it consumes read tokens. The balances are disjoint."""
if write_cost > 0:
return self.write_tokens >= write_cost
- return self.read_tokens >= global_cost
+ return self.read_tokens >= read_cost
- def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float:
+ def _calculate_wait_time(self, read_cost: float, write_cost: float) -> float:
now = time.monotonic()
if write_cost > 0:
@@ -75,9 +80,9 @@ class ReadWriteTokenBucket:
return max(0.0, target_time - now)
return 1.0
else:
- if self.read_tokens >= global_cost:
+ if self.read_tokens >= read_cost:
return 0.0
- needed = global_cost - self.read_tokens
+ needed = read_cost - self.read_tokens
recovered = 0.0
for timestamp, amount in self._read_history:
recovered += amount
@@ -86,71 +91,124 @@ class ReadWriteTokenBucket:
return max(0.0, target_time - now)
return 1.0
- def _consume(self, global_cost: float, write_cost: float) -> None:
+ def _consume(self, read_cost: float, write_cost: float) -> None:
now = time.monotonic()
if write_cost > 0:
self.write_tokens -= write_cost
self._write_history.append((now, write_cost))
else:
- self.read_tokens -= global_cost
- self._read_history.append((now, global_cost))
-
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None:
- """Acquire tokens, blocking until available."""
- if write_cost > 0 and write_cost > self.write_capacity:
- raise ValueError(f"Write cost {write_cost} exceeds write capacity")
- if write_cost == 0 and global_cost > self.read_capacity:
- raise ValueError(f"Read cost {global_cost} exceeds read capacity")
-
- lock = self._write_lock if write_cost > 0 else self._read_lock
- async with lock:
+ self.read_tokens -= read_cost
+ self._read_history.append((now, read_cost))
+
+ async def acquire(
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ *,
+ global_cost: float | None = None,
+ ) -> None:
+ """Acquire tokens, blocking until available.
+
+ ``read_cost`` and ``write_cost`` are accounted independently: a
+ request with a positive ``write_cost`` consumes only write tokens,
+ while a request with ``write_cost == 0`` consumes only read tokens.
+
+ ``global_cost`` is a deprecated alias for ``read_cost``.
+ """
+ if global_cost is not None:
+ read_cost = global_cost
+ read_cost = float(read_cost)
+ write_cost = float(write_cost)
+
+ if write_cost > 0:
+ if write_cost > self.write_capacity:
+ raise ValueError(f"Write cost {write_cost} exceeds write capacity")
+ cond = self._write_cond
+ else:
+ if read_cost > self.read_capacity:
+ raise V
… [5303 more characters]step 1
$ find /repo -name "*.py" | grep -v test | head -50; echo ---; grep -rn "TokenBucket\|read_cost\|read_tokens" /repo --include=*.py -l | head
output · 40 lines
/repo/scripts/fetch_rate_limits.py /repo/src/pykalshi/api/account.py /repo/src/pykalshi/api/api_keys.py /repo/src/pykalshi/api/communications.py /repo/src/pykalshi/api/events.py /repo/src/pykalshi/api/event_orders.py /repo/src/pykalshi/api/exchange.py /repo/src/pykalshi/api/historical.py /repo/src/pykalshi/api/incentive_programs.py /repo/src/pykalshi/api/live_data.py /repo/src/pykalshi/api/markets.py /repo/src/pykalshi/api/milestones.py
28 more lines
/repo/src/pykalshi/api/multivariate_collections.py /repo/src/pykalshi/api/orders.py /repo/src/pykalshi/api/order_groups.py /repo/src/pykalshi/api/portfolio.py /repo/src/pykalshi/api/search.py /repo/src/pykalshi/api/series.py /repo/src/pykalshi/api/structured_targets.py /repo/src/pykalshi/api/_utils.py /repo/src/pykalshi/api/__init__.py /repo/src/pykalshi/auth.py /repo/src/pykalshi/config.py /repo/src/pykalshi/exceptions.py /repo/src/pykalshi/http_clien …[323 chars omitted]… /repo/src/pykalshi/ws_client.py /repo/src/pykalshi/_observability.py /repo/src/pykalshi/_version.py /repo/src/pykalshi/__init__.py /repo/tools/generate_coverage_report.py /repo/tools/generate_models.py /repo/tools/generate_ws_models.py /repo/tools/sync_docstrings.py --- /repo/src/pykalshi/http_client.py /repo/src/pykalshi/rate_limiter.py /repo/src/pykalshi/__init__.py /repo/tests/test_rate_limiter.py
step 2
$ cat /repo/src/pykalshi/rate_limiter.py; grep -n "read_cost\|write_cost\|read_tokens\|RateLimiter" /repo/src/pykalshi/protocols.py | head -30
output · 39 lines
"""Disjoint read/write token bucket rate limiter. A request consumes EITHER read tokens OR write tokens, never both. This matches Kalshi's actual rate limit model where read and write limits are independently enforced. """ from __future__ import annotations import asyncio import time from collections import deque
27 more lines
from typing import Tuple
class ReadWriteTokenBucket:
"""Sliding-window token bucket with separate read and write budgets."""
def __init__(self, read_rate: float, write_rate: float) -> None:
if read_rate <= 0 or write_rate <= 0:
raise ValueError("Rates must be positive")
self.read_rate = read_rate
self.write_rate = write_rate
self.read_capacity = read_rate
self.write_capacity = write_rate
self.window_size = 1.0
self.safety_padding = 0.1
self.read_tokens = float(self.read_capacity)
…[6251 chars omitted]…
"write_tokens": self.write_tokens,
"read_history_len": len(self._read_history),
"write_history_len": len(self._write_history),
}
289:class RateLimiterProtocol(Protocol):
292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...step 3
$ sed -n 1,120p /repo/tests/test_rate_limiter.py; grep -rn "global_cost\|acquire(" /repo/src/pykalshi | grep -v rate_limiter.pyoutput · 29 lines
"""Tests for pykalshi.rate_limiter."""
import pytest
from pykalshi.rate_limiter import ReadWriteTokenBucket
class TestReadWriteTokenBucket:
@pytest.mark.asyncio
async def test_initial_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
assert bucket.read_tokens == 10.0
assert bucket.write_tokens == 5.017 more lines
@pytest.mark.asyncio
async def test_read_acquire_consumes_read_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(global_cost=1.0, write_cost=0.0)
assert bucket.read_tokens == 9.0
assert bucket.write_tokens == 5.0 # write unchanged
@pytest.mark.asyncio
async def test_write_acquire_consumes_write_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(glo
…[4346 chars omitted]…
write_cost = 0.0, cost
/repo/src/pykalshi/http_client.py:287: await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
/repo/src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
/repo/src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...step 4
$ sed -n 120,260p /repo/tests/test_rate_limiter.py; sed -n 270,300p /repo/src/pykalshi/http_client.py
output · 36 lines
await asyncio.gather(*tasks)
assert bucket.write_tokens == 2.0
assert bucket.read_tokens == 10.0 # reads untouched
@pytest.mark.asyncio
async def test_concurrent_mixed_load(self) -> None:
"""Mix of reads and writes under concurrent access."""
import asyncio
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
read_tasks = [bucket.acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
write_tasks = [bucket.acquire(global_cost=1.0, write_cost=1.0) for _ in range(3)]24 more lines
await asyncio.gather(*read_tasks, *write_tasks)
assert bucket.read_tokens == 5.0
assert bucket.write_tokens == 2.0
@pytest.mark.asyncio
async def test_no_double_consumption(self) -> None:
"""Exhaust all tokens concurrently; verify no token is consumed twice."""
import asyncio
buck
…[1089 chars omitted]…
it_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
headers = {**kwargs.pop("headers", {}), **self._credentials.auth_headers(method, path)}
kwargs["headers"] = headers
retries = 0
while True:
try:
response = await self._client.request(method, path, **kwargs)
if response.status_code == 429:step 5
$ cat > /tmp/repro.py <<'EOF'
output · 16 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 16, in <module>
asyncio.run(main())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^4 more lines
File "/tmp/repro.py", line 7, in main
await asyncio.wait_for(asyncio.gather(*[b.acquire(read_cost=0.0, write_cost=1.0) for _ in range(8)]), timeout=5)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'read_cost'step 6
$ sed -i 's/read_cost=0.0, write_cost=1.0/write_cost=1.0/g; s/global_cost=1.0, write_cost=0.0/global_cost=1.0/g' /tmp/repro.py && timeout 30 python /tmp/repro.py; echo rc=$?
output · 3 lines
writes ok 1.1015123089891858 2.0 10.0 mixed ok 1.1020028649945743 5.0 2.0 rc=0
step 7
$ cat > /tmp/repro2.py <<'EOF'
output · 10 lines
ok n=5 rr=2 wr=2 rc=1 wc=0 t=2.20 r=1 w=2.0 ok n=5 rr=2 wr=2 rc=0 wc=1 t=2.20 r=2.0 w=1 ok n=10 rr=10 wr=2 rc=1 wc=0 t=0.00 r=0.0 w=2.0 ok n=6 rr=10 wr=5 rc=2 wc=0 t=1.10 r=8 w=5.0 ok n=6 rr=10 wr=5 rc=0 wc=2 t=2.20 r=10.0 w=1 ok n=4 rr=10 wr=5 rc=0 wc=2 t=1.10 r=10.0 w=1 ok n=3 rr=10 wr=5 rc=0 wc=2 t=1.10 r=10.0 w=3 ok n=20 rr=10 wr=10 rc=1 wc=0 t=1.10 r=0 w=10.0 ok n=20 rr=10 wr=10 rc=0 wc=1 t=1.10 r=10.0 w=0 ok n=6 rr=10 wr=5 rc=2 wc=0 t=1.10 r=8 w=5.0
step 8
No hang reproduced in simple scenarios, but I'll rewrite `acquire` to be robust (no unbounded waits, proper wake-ups) and do the `read_cost` rename across the package.
$ cat > /repo/src/pykalshi/rate_limiter.py <<'EOF'
output · 4 lines
/repo/src/pykalshi/http_client.py:287: await self._limiter.acquire(read_cost=global_cost, write_cost=write_cost) grep: /repo/src/pykalshi/__pycache__/http_client.cpython-312.pyc: binary file matches grep: /repo/src/pykalshi/__pycache__/protocols.cpython-312.pyc: binary file matches grep: /repo/src/pykalshi/__pycache__/rate_limiter.cpython-312.pyc: binary file matches
step 9
$ sed -i '287s/global_cost/read_cost/' /repo/src/pykalshi/http_client.py; grep -n "read_cost" /repo/src/pykalshi/http_client.py; timeout 60 python /tmp/repro.py && timeout 200 python /tmp/repro2.py
output · 20 lines
282: read_cost, write_cost = cost, 0.0
284: read_cost, write_cost = 0.0, cost
287: await self._limiter.acquire(read_cost=read_cost, write_cost=write_cost)
writes ok 1.1010993089876138 2.0 10.0
Traceback (most recent call last):
File "/tmp/repro.py", line 17, in <module>
asyncio.run(main2())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)8 more lines
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/tmp/repro.py", line 13, in main2
await asyncio.wait_for(asyncio.gather(*[b.acquire(global_cost=1.0) for _ in range(5)],*[b.acquire(global_cost=1.0, write_cost=1.0) for _ in range(8)]), timeout=5)
^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost'step 10
$ sed -i 's/global_cost=1.0/read_cost=1.0/g' /tmp/repro.py /tmp/repro2.py && timeout 200 python /tmp/repro2.py && timeout 30 python /tmp/repro.py
output · 16 lines
Traceback (most recent call last):
File "/tmp/repro2.py", line 15, in <module>
asyncio.run(run(*args))
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^4 more lines
File "/tmp/repro2.py", line 9, in run
await asyncio.wait_for(asyncio.gather(*[b.acquire(global_cost=rc, write_cost=wc) for _ in range(n)]), timeout=4)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost'step 11
$ sed -i 's/global_cost=rc/read_cost=rc/' /tmp/repro2.py && timeout 200 python /tmp/repro2.py && timeout 30 python /tmp/repro.py
output · 12 lines
ok n=5 rr=2 wr=2 rc=1 wc=0 t=2.20 r=1.0 w=2.0 ok n=5 rr=2 wr=2 rc=0 wc=1 t=2.20 r=2.0 w=1.0 ok n=10 rr=10 wr=2 rc=1 wc=0 t=0.00 r=0.0 w=2.0 ok n=6 rr=10 wr=5 rc=2 wc=0 t=1.10 r=8.0 w=5.0 ok n=6 rr=10 wr=5 rc=0 wc=2 t=2.20 r=10.0 w=1.0 ok n=4 rr=10 wr=5 rc=0 wc=2 t=1.10 r=10.0 w=1.0 ok n=3 rr=10 wr=5 rc=0 wc=2 t=1.10 r=10.0 w=3.0 ok n=20 rr=10 wr=10 rc=1 wc=0 t=1.10 r=0.0 w=10.0 ok n=20 rr=10 wr=10 rc=0 wc=1 t=1.10 r=10.0 w=0.0 ok n=6 rr=10 wr=5 rc=2 wc=0 t=1.10 r=8.0 w=5.0 writes ok 1.100658386014402 2.0 10.0 mixed ok 1.1014552419946995 5.0 2.0
step 12
$ timeout 300 python -m pytest tests/test_rate_limiter.py -q 2>&1 | tail -5
output · 5 lines
FAILED tests/test_rate_limiter.py::TestConcurrentLoad::test_concurrent_reads_consume_correct_tokens FAILED tests/test_rate_limiter.py::TestConcurrentLoad::test_concurrent_writes_consume_correct_tokens FAILED tests/test_rate_limiter.py::TestConcurrentLoad::test_concurrent_mixed_load FAILED tests/test_rate_limiter.py::TestConcurrentLoad::test_no_double_consumption 10 failed, 7 passed in 0.53s
step 13
$ timeout 300 python -m pytest tests/test_rate_limiter.py -q 2>&1 | grep -E "^(FAILED|E )" | head; timeout 60 python -m pytest tests/test_rate_limiter.py::TestReadWriteTokenBucket::test_initial_tokens -q 2>&1 | tail -15
output · 12 lines
E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.try_acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.get_wait_time() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'global_cost' E TypeError: ReadWriteTokenBucket.try_acquire() got an unexpected keyword argument 'global_cost' . [100%] 1 passed in 0.09s
step 14
All failures are just the rename. I'll add a deprecated `global_cost` alias so old callers/tests keep working:
$ python - <<'PY'
output · 2 lines
................. [100%] 17 passed in 0.15s
step 15
$ cat > /tmp/repro3.py <<'EOF'
output · 16 lines
Traceback (most recent call last):
File "/tmp/repro3.py", line 25, in <module>
asyncio.run(m())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^4 more lines
File "/tmp/repro3.py", line 12, in m
assert await b.get_wait_time(read_cost=0.0, write_cost=5.0) == 0.0
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionErrorstep 16
$ sed -i 's/write_cost=5.0) == 0.0/write_cost=4.0) == 0.0/; s/write_cost=6.0) > 0.0/write_cost=5.0) > 0.0/' /tmp/repro3.py && timeout 60 python /tmp/repro3.py && timeout 900 python -m pytest tests -q 2>&1 | tail -3
output · 16 lines
Traceback (most recent call last):
File "/tmp/repro3.py", line 25, in <module>
asyncio.run(m())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^4 more lines
File "/tmp/repro3.py", line 17, in m
await fn(read_cost=11.0); assert False
^^^^^
AssertionErrorstep 17
$ python - <<'PY'
output · 4 lines
all good 5.0 3.0 FAILED tests/test_openapi_validation.py::test_openapi_coverage - httpx.Connec... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... 4 failed, 197 passed, 71 skipped in 8.11s
step 18
$ timeout 900 python -m pytest tests -q 2>&1 | grep FAILED; git stash -q; timeout 300 python -m pytest tests/test_openapi_validation.py -q 2>&1 | tail -2; git stash pop -q
output · 6 lines
FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage FAILED tests/test_asyncapi_validation.py::test_asyncapi_schema_coverage - htt... FAILED tests/test_openapi_validation.py::test_openapi_coverage - httpx.Connec... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... 2 failed in 0.52s
step 19
The 4 remaining failures are pre-existing network-dependent tests (they fail on the pristine tree too). Final diff check:
$ git diff --stat
output · 4 lines
src/pykalshi/http_client.py | 6 +- src/pykalshi/protocols.py | 4 +- src/pykalshi/rate_limiter.py | 148 ++++++++++++++++++++++++++++++------------- 3 files changed, 109 insertions(+), 49 deletions(-)
step 20
$ done: {"notes": "Renamed read-side cost kwarg to `read_cost` in ReadWriteTokenBucket (acquire/try_acquire/get_wait_time), RateLimiterProtocol, and internal callers (http_client), keeping `global_cost` as a deprecated keyword alias. Replaced lock-held-sleep acquire loops with per-side asyncio.Condition-based loops that re-check/refill aftoutput · 3 lines
…[200 chars omitted]… e on try_acquire), and no-token-consumption on failed try_acquire. All 17 rate-limiter tests pass; remaining suite failures are pre-existing network-dependent tests."} submitted.
GLM-5.3 Flash · attempt 3 · solved · 15 steps · $0.011
Submitted diff · 193 lines
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..24a9c5d 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -279,12 +279,12 @@ class KalshiHttpClient:
cost = self._resolve_cost(method, path)
if method.upper() == "GET":
- global_cost, write_cost = cost, 0.0
+ read_cost, write_cost = cost, 0.0
else:
- global_cost, write_cost = 0.0, cost
+ read_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=read_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..5fd3f7b 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,7 @@ class KalshiWebSocketClientProtocol(Protocol):
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
+ async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
+ async def try_acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
diff --git a/src/pykalshi/rate_limiter.py b/src/pykalshi/rate_limiter.py
index 17a53ff..a73f71c 100644
--- a/src/pykalshi/rate_limiter.py
+++ b/src/pykalshi/rate_limiter.py
@@ -55,12 +55,13 @@ class ReadWriteTokenBucket:
else:
break
- def _can_proceed(self, global_cost: float, write_cost: float) -> bool:
+ def _can_proceed(self, read_cost: float, write_cost: float) -> bool:
+ # Disjoint accounting: a request consumes either read or write tokens.
if write_cost > 0:
return self.write_tokens >= write_cost
- return self.read_tokens >= global_cost
+ return self.read_tokens >= read_cost
- def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float:
+ def _calculate_wait_time(self, read_cost: float, write_cost: float) -> float:
now = time.monotonic()
if write_cost > 0:
@@ -75,9 +76,9 @@ class ReadWriteTokenBucket:
return max(0.0, target_time - now)
return 1.0
else:
- if self.read_tokens >= global_cost:
+ if self.read_tokens >= read_cost:
return 0.0
- needed = global_cost - self.read_tokens
+ needed = read_cost - self.read_tokens
recovered = 0.0
for timestamp, amount in self._read_history:
recovered += amount
@@ -86,45 +87,71 @@ class ReadWriteTokenBucket:
return max(0.0, target_time - now)
return 1.0
- def _consume(self, global_cost: float, write_cost: float) -> None:
+ def _consume(self, read_cost: float, write_cost: float) -> None:
now = time.monotonic()
if write_cost > 0:
self.write_tokens -= write_cost
self._write_history.append((now, write_cost))
else:
- self.read_tokens -= global_cost
- self._read_history.append((now, global_cost))
+ self.read_tokens -= read_cost
+ self._read_history.append((now, read_cost))
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None:
- """Acquire tokens, blocking until available."""
- if write_cost > 0 and write_cost > self.write_capacity:
- raise ValueError(f"Write cost {write_cost} exceeds write capacity")
- if write_cost == 0 and global_cost > self.read_capacity:
- raise ValueError(f"Read cost {global_cost} exceeds read capacity")
+ def _validate_costs(self, read_cost: float, write_cost: float) -> None:
+ if read_cost < 0 or write_cost < 0:
+ raise ValueError("Costs must be non-negative")
+ if write_cost > 0:
+ if write_cost > self.write_capacity:
+ raise ValueError(f"Write cost {write_cost} exceeds write capacity")
+ elif read_cost > self.read_capacity:
+ raise ValueError(f"Read cost {read_cost} exceeds read capacity")
+
+ @staticmethod
+ def _resolve_legacy_cost(
+ read_cost: float, write_cost: float, kwargs: dict
+ ) -> tuple:
+ """Allow the deprecated ``global_cost`` keyword as an alias for read_cost."""
+ legacy = kwargs.pop("global_cost", None)
+ if legacy is not None and read_cost == 1.0:
+ read_cost = legacy
+ return read_cost, write_cost
+
+ async def acquire(
+ self, read_cost: float = 1.0, write_cost: float = 0.0, **kwargs: float
+ ) -> None:
+ """Acquire tokens, blocking until available.
- lock = self._write_lock if write_cost > 0 else self._read_lock
- async with lock:
- while True:
- if write_cost > 0:
+ Consumes only write tokens when ``write_cost`` is positive, otherwise
+ only read tokens.
+ """
+ read_cost, write_cost = self._resolve_legacy_cost(read_cost, write_cost, kwargs)
+ self._validate_costs(read_cost, write_cost)
+
+ is_write = write_cost > 0
+ lock = self._write_lock if is_write else self._read_lock
+ while True:
+ async with lock:
+ if is_write:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(glo
… [2774 more characters]step 1
$ find /repo -name "*.py" | grep -i -E "token|rate" ; ls /repo
output · 17 lines
/repo/scripts/fetch_rate_limits.py /repo/src/pykalshi/rate_limiter.py /repo/tests/test_rate_limiter.py /repo/tools/generate_coverage_report.py /repo/tools/generate_models.py /repo/tools/generate_ws_models.py CHANGELOG.md CONTRIBUTING.md LICENSE README.md docs pyproject.toml
5 more lines
scripts src tests tools uv.lock
step 2
$ cat /repo/src/pykalshi/rate_limiter.py; grep -rn "write_cost\|read_cost\|write_cost\|cost" /repo/src --include=*.py | grep -v rate_limiter.py
output · 36 lines
"""Disjoint read/write token bucket rate limiter. A request consumes EITHER read tokens OR write tokens, never both. This matches Kalshi's actual rate limit model where read and write limits are independently enforced. """ from __future__ import annotations import asyncio import time from collections import deque
24 more lines
from typing import Tuple
class ReadWriteTokenBucket:
"""Sliding-window token bucket with separate read and write budgets."""
def __init__(self, read_rate: float, write_rate: float) -> None:
if read_rate <= 0 or write_rate <= 0:
raise ValueError("Rates must be positive")
self.read_rate = read_rate
self.write_rate = write_rate
self.read_capacity = read_rate
self.write_capacity = write_rate
self.window_size = 1.0
self.safety_padding = 0.1
self.read_tokens = float(self.read_capacity)
…[8629 chars omitted]…
f) -> GetAccountEndpointCostsResponse: ...
/repo/src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
/repo/src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
/repo/src/pykalshi/testing/fixtures.py:21: the /account/limits and /account/endpoint_costs routes.step 3
$ sed -n 200,300p /repo/src/pykalshi/rate_limiter.py; grep -rn "acquire(\|get_wait_time\|RateLimiterProtocol\|global_cost" /repo/src/pykalshi /repo/tests --include=*.py | grep -v models/
output · 14 lines
/repo/src/pykalshi/http_client.py:282: global_cost, write_cost = cost, 0.0 /repo/src/pykalshi/http_client.py:284: global_cost, write_cost = 0.0, cost /repo/src/pykalshi/http_client.py:287: await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost) /repo/src/pykalshi/protocols.py:289:class RateLimiterProtocol(Protocol): /repo/src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ... /repo/src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ... /repo/src/pykalshi/rate_limiter.py:58: def _can_proceed(self, global_cost: float, write_cost: float) -> bool: /repo/src/pykalshi/rate_limiter.py:61: return self.read_tokens >= global_cost /repo/src/pykalshi/rate_limiter.py:63: def _calculate_wait_time(self, …[3398 chars omitted]… , write_cost=0.0) for _ in range(5)] /repo/tests/test_rate_limiter.py:131: write_tasks = [bucket.acquire(global_cost=1.0, write_cost=1.0) for _ in range(3)]
2 more lines
/repo/tests/test_rate_limiter.py:144: *[bucket.try_acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)] /repo/tests/test_rate_limiter.py:150: assert await bucket.try_acquire(global_cost=1.0, write_cost=0.0) is False
step 4
$ sed -n 1,60p /repo/tests/test_rate_limiter.py; sed -n 60,160p /repo/tests/test_rate_limiter.py; sed -n 120,160p /repo/src/pykalshi/rate_limiter.py; sed -n 270,300p /repo/src/pykalshi/http_client.py
output · 38 lines
"""Tests for pykalshi.rate_limiter."""
import pytest
from pykalshi.rate_limiter import ReadWriteTokenBucket
class TestReadWriteTokenBucket:
@pytest.mark.asyncio
async def test_initial_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
assert bucket.read_tokens == 10.0
assert bucket.write_tokens == 5.026 more lines
@pytest.mark.asyncio
async def test_read_acquire_consumes_read_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(global_cost=1.0, write_cost=0.0)
assert bucket.read_tokens == 9.0
assert bucket.write_tokens == 5.0 # write unchanged
@pytest.mark.asyncio
async def test_write_acquire_consumes_write_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(glo
…[7635 chars omitted]…
it_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
headers = {**kwargs.pop("headers", {}), **self._credentials.auth_headers(method, path)}
kwargs["headers"] = headers
retries = 0
while True:
try:
response = await self._client.request(method, path, **kwargs)
if response.status_code == 429:step 5
$ cd /repo && python - <<'PY'
output · 13 lines
Traceback (most recent call last):
File "<stdin>", line 10, in <module>
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "<stdin>", line 7, in main1 more lines
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'read_cost'
step 6
$ cd /repo && python - <<'PY'
output · 1 lines
writes done 2.2034686419938225 1.0
step 7
$ cd /repo && timeout 20 python - <<'PY'
output · 2 lines
mixed done 2.202706322976155 1.0 1.0 rc=0
step 8
$ cd /repo && timeout 20 python - <<'PY'
output · 6 lines
2 {'global_cost': 0.0, 'write_cost': 2.0} ok 1.1 10.0 0.0
3 {'global_cost': 0.0, 'write_cost': 0.5} ok 1.1 10.0 0.5
10 {'global_cost': 1.0, 'write_cost': 0.0} FAIL TimeoutError
5 {'global_cost': 0.0, 'write_cost': 1.0} ok 4.4 10.0 0.0
3 {'global_cost': 2.0, 'write_cost': 1.0} ok 1.1 2.0 1.0
rc=0step 9
$ cd /repo && timeout 30 python - <<'PY'
output · 2 lines
ok 1.1 0.0 rc=0
step 10
The bug is that `acquire` sleeps while holding the side lock, serializing all waiters (each pays the full refill wait back-to-back). Also the parameter must be renamed `global_cost` → `read_cost`. Let me rewrite:
$ cd /repo && python - <<'PY'
output · 34 lines
else:
self.read_tokens -= read_cost
self._read_history.append((now, read_cost))
def _validate_costs(self, read_cost: float, write_cost: float) -> None:
if read_cost < 0 or write_cost < 0:
raise ValueError("Costs must be non-negative")
if write_cost > 0 and write_cost > self.write_capacity:
raise ValueError(f"Write cost {write_cost} exceeds write capacity")
if read_cost > self.read_capacity:
raise ValueError(f"Read cost {read_cost} exceeds read capacity")
22 more lines
async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> None:
"""Acquire tokens, blocking until available.
Consumes only write tokens when ``write_cost`` is positive, otherwise
only read tokens.
"""
self._validate_costs(read_cost, write_cost)
is_write = write_cost >
…[1919 chars omitted]…
ure(
self,
read_rate: float,
write_rate: float,
read_capacity: float | None = None,
write_capacity: float | None = None,
) -> None:
"""Update rate limits (e.g., after fetching from API).
Capacity defaults to rate when not provided (one second of budget).
Kalshi's write bucket on higher tiers has capacity > rate for burst.
"""step 11
Now fix the capacity checks to be side-disjoint and add a `global_cost` backward-compat alias (existing tests/callers use it):
$ cd /repo && python - <<'PY'
output · 14 lines
) -> None: ...
async def resubscribe_channel(self, channel_name: str) -> None: ...
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
async def try_acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
src/pykalshi/rate_limiter.py:112: """Allow the deprecated ``global_cost`` keyword as an alias for read_cost."""2 more lines
src/pykalshi/rate_limiter.py:113: legacy = kwargs.pop("global_cost", None)
src/pykalshi/rate_limiter.py:168: async def get_wait_time(step 12
$ cd /repo && python - <<'PY'
output · 3 lines
writes ok 2.2 1.0 mixed ok 2.2 1.0 1.0 all good
step 13
$ cd /repo && timeout 300 python -m pytest tests/test_rate_limiter.py -x -q 2>&1 | tail -20
output · 2 lines
................. [100%] 17 passed in 0.16s
step 14
$ cd /repo && timeout 600 python -m pytest tests -q 2>&1 | tail -5; git diff --stat
output · 9 lines
FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage FAILED tests/test_asyncapi_validation.py::test_asyncapi_schema_coverage - htt... FAILED tests/test_openapi_validation.py::test_openapi_coverage - httpx.Connec... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... 4 failed, 197 passed, 71 skipped in 7.83s src/pykalshi/http_client.py | 6 +-- src/pykalshi/protocols.py | 4 +- src/pykalshi/rate_limiter.py | 94 ++++++++++++++++++++++++++++---------------- 3 files changed, 66 insertions(+), 38 deletions(-)
step 15
$ done: {"notes": "Renamed `global_cost` \u2192 `read_cost` across ReadWriteTokenBucket, RateLimiterProtocol, and http_client (legacy keyword kept as a deprecated alias so existing callers/tests still work). Fixed the concurrency hang: `acquire` no longer sleeps while holding the side lock, so waiters are not serialized and complete once toutput · 3 lines
…[382 chars omitted]… disjoint accounting examples from the issue) and the existing rate limiter tests (17 passed); remaining full-suite failures are pre-existing network-dependent tests."} submitted.
GLM-5.3 Flash · attempt 4 · solved · 18 steps · $0.009
Submitted diff · 130 lines
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..3ce87ae 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -284,7 +284,7 @@ class KalshiHttpClient:
global_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=global_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..5fd3f7b 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,7 @@ class KalshiWebSocketClientProtocol(Protocol):
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
+ async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
+ async def try_acquire(self, read_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
diff --git a/src/pykalshi/rate_limiter.py b/src/pykalshi/rate_limiter.py
index 17a53ff..a631bb6 100644
--- a/src/pykalshi/rate_limiter.py
+++ b/src/pykalshi/rate_limiter.py
@@ -95,36 +95,48 @@ class ReadWriteTokenBucket:
self.read_tokens -= global_cost
self._read_history.append((now, global_cost))
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None:
- """Acquire tokens, blocking until available."""
- if write_cost > 0 and write_cost > self.write_capacity:
- raise ValueError(f"Write cost {write_cost} exceeds write capacity")
- if write_cost == 0 and global_cost > self.read_capacity:
- raise ValueError(f"Read cost {global_cost} exceeds read capacity")
+ async def acquire(self, read_cost: float = 1.0, write_cost: float = 0.0, *, global_cost: float | None = None) -> None:
+ """Acquire tokens, blocking until available.
+
+ A request with a positive ``write_cost`` consumes only write tokens;
+ otherwise it consumes only read tokens. ``global_cost`` is accepted as
+ a deprecated alias for ``read_cost``.
+ """
+ if global_cost is not None:
+ read_cost = global_cost
+ self._validate_costs(read_cost, write_cost)
lock = self._write_lock if write_cost > 0 else self._read_lock
- async with lock:
- while True:
+ while True:
+ async with lock:
if write_cost > 0:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return
- wait_time = self._calculate_wait_time(global_cost, write_cost)
- if wait_time > 0:
- await asyncio.sleep(wait_time)
- else:
- await asyncio.sleep(0.01)
+ wait_time = self._calculate_wait_time(read_cost, write_cost)
+ # Sleep outside the lock so other waiters can make progress.
+ await asyncio.sleep(wait_time if wait_time > 0 else 0.01)
+
+ def _validate_costs(self, read_cost: float, write_cost: float) -> None:
+ if read_cost < 0 or write_cost < 0:
+ raise ValueError("Costs must be non-negative")
+ if read_cost > self.read_capacity:
+ raise ValueError(f"Read cost {read_cost} exceeds read capacity")
+ if write_cost > self.write_capacity:
+ raise ValueError(f"Write cost {write_cost} exceeds write capacity")
async def try_acquire(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self, read_cost: float = 1.0, write_cost: float = 0.0, *, global_cost: float | None = None
) -> bool:
"""Try to acquire tokens without blocking. Returns True if acquired."""
- if write_cost > 0 and write_cost > self.write_capacity:
+ if global_cost is not None:
+ read_cost = global_cost
+ if read_cost < 0 or write_cost < 0:
return False
- if write_cost == 0 and global_cost > self.read_capacity:
+ if read_cost > self.read_capacity or write_cost > self.write_capacity:
return False
lock = self._write_lock if write_cost > 0 else self._read_lock
@@ -133,24 +145,26 @@ class ReadWriteTokenBucket:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return True
return False
async def get_wait_time(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self, read_cost: float = 1.0, write_cost: float = 0.0, *, global_cost: float | None = None
) -> float:
"""Estimate wait time without consuming tokens."""
+ if global_cost is not None:
+ read_cost = global_cost
lock = self._write_lock if write_cost > 0 else self._read_lock
async with lock:
if write_cost > 0:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
+
… [267 more characters]step 1
$ find /repo -type f -name "*.py" | head -50; echo ---; ls /repo
output · 49 lines
/repo/scripts/fetch_rate_limits.py /repo/src/pykalshi/api/account.py /repo/src/pykalshi/api/api_keys.py /repo/src/pykalshi/api/communications.py /repo/src/pykalshi/api/events.py /repo/src/pykalshi/api/event_orders.py /repo/src/pykalshi/api/exchange.py /repo/src/pykalshi/api/historical.py /repo/src/pykalshi/api/incentive_programs.py /repo/src/pykalshi/api/live_data.py /repo/src/pykalshi/api/markets.py /repo/src/pykalshi/api/milestones.py
37 more lines
/repo/src/pykalshi/api/multivariate_collections.py /repo/src/pykalshi/api/orders.py /repo/src/pykalshi/api/order_groups.py /repo/src/pykalshi/api/portfolio.py /repo/src/pykalshi/api/search.py /repo/src/pykalshi/api/series.py /repo/src/pykalshi/api/structured_targets.py /repo/src/pykalshi/api/_utils.py /repo/src/pykalshi/api/__init__.py /repo/src/pykalshi/auth.py /repo/src/pykalshi/config.py /repo/src/pykalshi/exceptions.py /repo/src/pykalshi/http_clien …[543 chars omitted]… .py /repo/src/pykalshi/__init__.py /repo/tests/conftest.py /repo/tests/mock_data.py /repo/tests/test_api/__init__.py /repo/tests/test_api_utils.py /repo/tests/test_asyncapi_validation.py /repo/tests/test_auth.py /repo/tests/test_config.py /repo/tests/test_exceptions.py /repo/tests/test_http_client.py --- CHANGELOG.md CONTRIBUTING.md LICENSE README.md docs pyproject.toml scripts src tests tools uv.lock
step 2
$ cat /repo/src/pykalshi/rate_limiter.py
output · 43 lines
"""Disjoint read/write token bucket rate limiter. A request consumes EITHER read tokens OR write tokens, never both. This matches Kalshi's actual rate limit model where read and write limits are independently enforced. """ from __future__ import annotations import asyncio import time from collections import deque
31 more lines
from typing import Tuple
class ReadWriteTokenBucket:
"""Sliding-window token bucket with separate read and write budgets."""
def __init__(self, read_rate: float, write_rate: float) -> None:
if read_rate <= 0 or write_rate <= 0:
raise ValueError("Rates must be positive")
self.read_rate = read_rate
self.write_rate = write_rate
self.read_capacity = read_rate
self.write_capacity = write_rate
self.window_size = 1.0
self.safety_padding = 0.1
self.read_tokens = float(self.read_capacity)
…[6014 chars omitted]…
rite_capacity)
def get_status(self) -> dict[str, object]:
"""Return current token bucket state for debugging."""
return {
"type": "ReadWriteTokenBucket",
"read_tokens": self.read_tokens,
"write_tokens": self.write_tokens,
"read_history_len": len(self._read_history),
"write_history_len": len(self._write_history),
}step 3
$ ls /repo/tests; grep -rn "global_cost\|rate_limiter\|RateLimiter" /repo/src /repo/tests --include=*.py -l
output · 22 lines
__init__.py conftest.py mock_data.py test_api test_api_utils.py test_asyncapi_validation.py test_auth.py test_config.py test_exceptions.py test_http_client.py test_integration.py test_models.py
10 more lines
test_observability.py test_openapi_validation.py test_rate_limiter.py test_ws_client.py test_ws_integration.py /repo/src/pykalshi/http_client.py /repo/src/pykalshi/protocols.py /repo/src/pykalshi/rate_limiter.py /repo/src/pykalshi/__init__.py /repo/tests/test_rate_limiter.py
step 4
$ sed -n 1,120p /repo/tests/test_rate_limiter.py; echo ====; grep -n "acquire\|RateLimiter" /repo/src/pykalshi/protocols.py /repo/src/pykalshi/http_client.py | head -40
output · 29 lines
"""Tests for pykalshi.rate_limiter."""
import pytest
from pykalshi.rate_limiter import ReadWriteTokenBucket
class TestReadWriteTokenBucket:
@pytest.mark.asyncio
async def test_initial_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
assert bucket.read_tokens == 10.0
assert bucket.write_tokens == 5.017 more lines
@pytest.mark.asyncio
async def test_read_acquire_consumes_read_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(global_cost=1.0, write_cost=0.0)
assert bucket.read_tokens == 9.0
assert bucket.write_tokens == 5.0 # write unchanged
@pytest.mark.asyncio
async def test_write_acquire_consumes_write_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(glo
…[4252 chars omitted]…
terProtocol(Protocol):
/repo/src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
/repo/src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
/repo/src/pykalshi/http_client.py:287: await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)step 5
$ sed -n 120,260p /repo/tests/test_rate_limiter.py; echo ====; sed -n 280,300p /repo/src/pykalshi/protocols.py; echo ===; sed -n 270,300p /repo/src/pykalshi/http_client.py
output · 36 lines
await asyncio.gather(*tasks)
assert bucket.write_tokens == 2.0
assert bucket.read_tokens == 10.0 # reads untouched
@pytest.mark.asyncio
async def test_concurrent_mixed_load(self) -> None:
"""Mix of reads and writes under concurrent access."""
import asyncio
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
read_tasks = [bucket.acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
write_tasks = [bucket.acquire(global_cost=1.0, write_cost=1.0) for _ in range(3)]24 more lines
await asyncio.gather(*read_tasks, *write_tasks)
assert bucket.read_tokens == 5.0
assert bucket.write_tokens == 2.0
@pytest.mark.asyncio
async def test_no_double_consumption(self) -> None:
"""Exhaust all tokens concurrently; verify no token is consumed twice."""
import asyncio
buck
…[1812 chars omitted]…
it_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
headers = {**kwargs.pop("headers", {}), **self._credentials.auth_headers(method, path)}
kwargs["headers"] = headers
retries = 0
while True:
try:
response = await self._client.request(method, path, **kwargs)
if response.status_code == 429:step 6
$ cd /repo && cat > /tmp/repro.py <<'EOF'
output · 16 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 14, in <module>
asyncio.run(main())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^4 more lines
File "/tmp/repro.py", line 7, in main
tasks=[bucket.acquire(read_cost=0.0, write_cost=1.0) for _ in range(5)]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'read_cost'step 7
$ cd /repo && sed -i 's/read_cost=0.0, write_cost=1.0/global_cost=0.0, write_cost=1.0/' /tmp/repro.py && timeout 30 python /tmp/repro.py
output · 1 lines
ok {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 1}step 8
$ cd /repo && timeout 120 python -m pytest tests/test_rate_limiter.py -x -q 2>&1 | tail -20
output · 2 lines
................. [100%] 17 passed in 0.13s
step 9
$ cd /repo && cat > /tmp/repro2.py <<'EOF'
output · 5 lines
ok n=5 rr=1.0 wr=1.0 rc=1.0 wc=0.0 t=4.40 {'type': 'ReadWriteTokenBucket', 'read_tokens': 0.0, 'write_tokens': 1.0, 'read_history_len': 1, 'write_history_len': 0}
ok n=5 rr=2.0 wr=2.0 rc=1.0 wc=1.0 t=2.20 {'type': 'ReadWriteTokenBucket', 'read_tokens': 2.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 1}
ok n=10 rr=1.0 wr=1.0 rc=0.0 wc=1.0 t=9.91 {'type': 'ReadWriteTokenBucket', 'read_tokens': 1.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 1}
ok n=4 rr=10.0 wr=5.0 rc=1.0 wc=0.0 t=0.00 {'type': 'ReadWriteTokenBucket', 'read_tokens': 6.0, 'write_tokens': 5.0, 'read_history_len': 4, 'write_history_len': 0}
ok n=4 rr=10.0 wr=5.0 rc=1.0 wc=1.0 t=0.00 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 4}step 10
$ cd /repo && cat > /tmp/repro3.py <<'EOF'
output · 4 lines
ok n=10 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 2}
ok n=10 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 2}
ok n=20 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 5.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 4}
mixed ok 2.202881103992695 {'type': 'ReadWriteTokenBucket', 'read_tokens': 0.0, 'write_tokens': 0.0, 'read_history_len': 2, 'write_history_len': 2}step 11
$ cd /repo && grep -rn "global_cost\|\.acquire(\|try_acquire\|get_wait_time" src scripts | grep -v rate_limiter.py
output · 8 lines
grep: src/pykalshi/__pycache__/http_client.cpython-312.pyc: binary file matches grep: src/pykalshi/__pycache__/protocols.cpython-312.pyc: binary file matches grep: src/pykalshi/__pycache__/rate_limiter.cpython-312.pyc: binary file matches src/pykalshi/http_client.py:282: global_cost, write_cost = cost, 0.0 src/pykalshi/http_client.py:284: global_cost, write_cost = 0.0, cost src/pykalshi/http_client.py:287: await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost) src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ... src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
step 12
$ cd /repo && python - <<'PY'
output · 1 lines
127: if read_cost > 0 and read_cost > 0 and False:
step 13
$ cd /repo && python - <<'PY'
output · 12 lines
................. [100%]
17 passed in 0.11s
ok n=10 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 2}
ok n=10 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 2}
ok n=20 t=4.41 {'type': 'ReadWriteTokenBucket', 'read_tokens': 5.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 4}
mixed ok 2.203106080996804 {'type': 'ReadWriteTokenBucket', 'read_tokens': 0.0, 'write_tokens': 0.0, 'read_history_len': 2, 'write_history_len': 2}
ok n=5 rr=1.0 wr=1.0 rc=1.0 wc=0.0 t=4.40 {'type': 'ReadWriteTokenBucket', 'read_tokens': 0.0, 'write_tokens': 1.0, 'read_history_len': 1, 'write_history_len': 0}
ok n=5 rr=2.0 wr=2.0 rc=1.0 wc=1.0 t=2.20 {'type': 'ReadWriteTokenBucket', 'r
…[174 chars omitted]…
': 1.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 1}
ok n=4 rr=10.0 wr=5.0 rc=1.0 wc=0.0 t=0.00 {'type': 'ReadWriteTokenBucket', 'read_tokens': 6.0, 'write_tokens': 5.0, 'read_history_len': 4, 'write_history_len': 0}
ok n=4 rr=10.0 wr=5.0 rc=1.0 wc=1.0 t=0.00 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 4}step 14
$ cd /repo && cat > /tmp/repro4.py <<'EOF'
output · 6 lines
all ok 2.202671321982052 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 0.0, 'read_history_len': 0, 'write_history_len': 2}
/usr/local/lib/python3.12/site-packages/httpx/_transports/default.py:118: ConnectError
=========================== short test summary info ============================
FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 6 passed in 0.85sstep 15
$ cd /repo && git stash && timeout 120 python -m pytest tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage -q 2>&1 | tail -3; git stash pop
output · 14 lines
Saved working directory and index state WIP on master: 0590c80 base =========================== short test summary info ============================ FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage 1 failed in 0.38s On branch master Changes not staged for commit: (use "git add <file>..." to update what will be committed) (use "git restore <file>..." to discard changes in working directory) modified: src/pykalshi/http_client.py modified: src/pykalshi/protocols.py modified: src/pykalshi/rate_limiter.py
2 more lines
no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (6ab9018e2983bb88e04fe0a128cd47f9e55a9009)step 16
$ cd /repo && timeout 300 python -m pytest tests -q --ignore=tests/test_integration.py --deselect tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage 2>&1 | tail -5
output · 5 lines
=========================== short test summary info ============================ FAILED tests/test_asyncapi_validation.py::test_asyncapi_schema_coverage - htt... FAILED tests/test_openapi_validation.py::test_openapi_coverage - httpx.Connec... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... 3 failed, 197 passed, 14 skipped, 1 deselected in 6.42s
step 17
$ cd /repo && git diff
output · 32 lines
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..3ce87ae 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -284,7 +284,7 @@ class KalshiHttpClient:
global_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=global_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:20 more lines
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..5fd3f7b 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,7 @@ class KalshiWebSocketClientProtocol(Protocol):
class RateLimiterProtocol(Protocol):
"""Contract for rate limiti
…[4966 chars omitted]…
f._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
+ if self._can_proceed(read_cost, write_cost):
return 0.0
- return self._calculate_wait_time(global_cost, write_cost)
+ return self._calculate_wait_time(read_cost, write_cost)
async def reconfigure(
self,step 18
All checks pass (remaining failures are pre-existing network-dependent tests). Diff looks as intended.
$ done: {"notes": "Renamed the rate limiter's read-side keyword from `global_cost` to `read_cost` in ReadWriteTokenBucket.acquire/try_acquire/get_wait_time, RateLimiterProtocol, and the http_client caller (legacy `global_cost` kwarg still accepted as an alias). Reworked acquire to validate both costs against their own capacities (raising Voutput · 3 lines
…[237 chars omitted]… consuming try_acquire failure, and zero-wait get_wait_time behavior verified with reproducers; full test suite passes except pre-existing network-dependent failures."} submitted.
GLM-5.3 Flash · attempt 5 · solved · 15 steps · $0.006
Submitted diff · 146 lines
diff --git a/src/pykalshi/http_client.py b/src/pykalshi/http_client.py
index 13b97c8..24a9c5d 100644
--- a/src/pykalshi/http_client.py
+++ b/src/pykalshi/http_client.py
@@ -279,12 +279,12 @@ class KalshiHttpClient:
cost = self._resolve_cost(method, path)
if method.upper() == "GET":
- global_cost, write_cost = cost, 0.0
+ read_cost, write_cost = cost, 0.0
else:
- global_cost, write_cost = 0.0, cost
+ read_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
- await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost)
+ await self._limiter.acquire(read_cost=read_cost, write_cost=write_cost)
waited = time.monotonic() - wait_start
if waited > 0.001:
self._rate_limiter_wait.record(waited)
diff --git a/src/pykalshi/protocols.py b/src/pykalshi/protocols.py
index 25fcae3..6ec66f6 100644
--- a/src/pykalshi/protocols.py
+++ b/src/pykalshi/protocols.py
@@ -289,7 +289,17 @@ class KalshiWebSocketClientProtocol(Protocol):
class RateLimiterProtocol(Protocol):
"""Contract for rate limiting."""
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ...
- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ...
+ async def acquire(
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ global_cost: float | None = None,
+ ) -> None: ...
+ async def try_acquire(
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ global_cost: float | None = None,
+ ) -> bool: ...
async def reconfigure(self, read_rate: float, write_rate: float) -> None: ...
def get_status(self) -> dict[str, object]: ...
diff --git a/src/pykalshi/rate_limiter.py b/src/pykalshi/rate_limiter.py
index 17a53ff..cb9d22a 100644
--- a/src/pykalshi/rate_limiter.py
+++ b/src/pykalshi/rate_limiter.py
@@ -95,12 +95,25 @@ class ReadWriteTokenBucket:
self.read_tokens -= global_cost
self._read_history.append((now, global_cost))
- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None:
- """Acquire tokens, blocking until available."""
+ async def acquire(
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ global_cost: float | None = None,
+ ) -> None:
+ """Acquire tokens, blocking until available.
+
+ ``read_cost`` is the read-side cost; ``write_cost`` the write-side
+ cost. Only one budget is consumed: if ``write_cost > 0`` only write
+ tokens are used, otherwise only read tokens. ``global_cost`` is a
+ deprecated alias for ``read_cost``.
+ """
+ if global_cost is not None:
+ read_cost = global_cost
if write_cost > 0 and write_cost > self.write_capacity:
raise ValueError(f"Write cost {write_cost} exceeds write capacity")
- if write_cost == 0 and global_cost > self.read_capacity:
- raise ValueError(f"Read cost {global_cost} exceeds read capacity")
+ if write_cost <= 0 and read_cost > self.read_capacity:
+ raise ValueError(f"Read cost {read_cost} exceeds read capacity")
lock = self._write_lock if write_cost > 0 else self._read_lock
async with lock:
@@ -109,22 +122,27 @@ class ReadWriteTokenBucket:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return
- wait_time = self._calculate_wait_time(global_cost, write_cost)
+ wait_time = self._calculate_wait_time(read_cost, write_cost)
if wait_time > 0:
await asyncio.sleep(wait_time)
else:
await asyncio.sleep(0.01)
async def try_acquire(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ global_cost: float | None = None,
) -> bool:
"""Try to acquire tokens without blocking. Returns True if acquired."""
+ if global_cost is not None:
+ read_cost = global_cost
if write_cost > 0 and write_cost > self.write_capacity:
return False
- if write_cost == 0 and global_cost > self.read_capacity:
+ if write_cost <= 0 and read_cost > self.read_capacity:
return False
lock = self._write_lock if write_cost > 0 else self._read_lock
@@ -133,24 +151,29 @@ class ReadWriteTokenBucket:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
- self._consume(global_cost, write_cost)
+ if self._can_proceed(read_cost, write_cost):
+ self._consume(read_cost, write_cost)
return True
return False
async def get_wait_time(
- self, global_cost: float = 1.0, write_cost: float = 0.0
+ self,
+ read_cost: float = 1.0,
+ write_cost: float = 0.0,
+ global_cost: float | None = None,
) -> float:
"""Estimate wait time without consuming tokens."""
+ if global_cost is not None:
+ read_cost = global_cost
lock = self._write_lock if write_cost > 0 else self._read_lock
async with lock:
if write_cost > 0:
self._refill_write()
else:
self._refill_read()
- if self._can_proceed(global_cost, write_cost):
+ if self._can_proceed(read_cost, write_co
… [217 more characters]step 1
$ find /repo -name "*.py" | head -50 && cat /repo/setup.py 2>/dev/null; ls /repo
output · 48 lines
/repo/scripts/fetch_rate_limits.py /repo/src/pykalshi/api/account.py /repo/src/pykalshi/api/api_keys.py /repo/src/pykalshi/api/communications.py /repo/src/pykalshi/api/events.py /repo/src/pykalshi/api/event_orders.py /repo/src/pykalshi/api/exchange.py /repo/src/pykalshi/api/historical.py /repo/src/pykalshi/api/incentive_programs.py /repo/src/pykalshi/api/live_data.py /repo/src/pykalshi/api/markets.py /repo/src/pykalshi/api/milestones.py
36 more lines
/repo/src/pykalshi/api/multivariate_collections.py /repo/src/pykalshi/api/orders.py /repo/src/pykalshi/api/order_groups.py /repo/src/pykalshi/api/portfolio.py /repo/src/pykalshi/api/search.py /repo/src/pykalshi/api/series.py /repo/src/pykalshi/api/structured_targets.py /repo/src/pykalshi/api/_utils.py /repo/src/pykalshi/api/__init__.py /repo/src/pykalshi/auth.py /repo/src/pykalshi/config.py /repo/src/pykalshi/exceptions.py /repo/src/pykalshi/http_clien …[539 chars omitted]… sion.py /repo/src/pykalshi/__init__.py /repo/tests/conftest.py /repo/tests/mock_data.py /repo/tests/test_api/__init__.py /repo/tests/test_api_utils.py /repo/tests/test_asyncapi_validation.py /repo/tests/test_auth.py /repo/tests/test_config.py /repo/tests/test_exceptions.py /repo/tests/test_http_client.py CHANGELOG.md CONTRIBUTING.md LICENSE README.md docs pyproject.toml scripts src tests tools uv.lock
step 2
$ cat /repo/src/pykalshi/rate_limiter.py
output · 43 lines
"""Disjoint read/write token bucket rate limiter. A request consumes EITHER read tokens OR write tokens, never both. This matches Kalshi's actual rate limit model where read and write limits are independently enforced. """ from __future__ import annotations import asyncio import time from collections import deque
31 more lines
from typing import Tuple
class ReadWriteTokenBucket:
"""Sliding-window token bucket with separate read and write budgets."""
def __init__(self, read_rate: float, write_rate: float) -> None:
if read_rate <= 0 or write_rate <= 0:
raise ValueError("Rates must be positive")
self.read_rate = read_rate
self.write_rate = write_rate
self.read_capacity = read_rate
self.write_capacity = write_rate
self.window_size = 1.0
self.safety_padding = 0.1
self.read_tokens = float(self.read_capacity)
…[6014 chars omitted]…
rite_capacity)
def get_status(self) -> dict[str, object]:
"""Return current token bucket state for debugging."""
return {
"type": "ReadWriteTokenBucket",
"read_tokens": self.read_tokens,
"write_tokens": self.write_tokens,
"read_history_len": len(self._read_history),
"write_history_len": len(self._write_history),
}step 3
$ grep -rn "global_cost\|acquire(" /repo/src/pykalshi --include=*.py | grep -v rate_limiter.py; grep -rn "RateLimiterProtocol" -A40 /repo/src/pykalshi/protocols.py | head -80output · 12 lines
/repo/src/pykalshi/http_client.py:282: global_cost, write_cost = cost, 0.0 /repo/src/pykalshi/http_client.py:284: global_cost, write_cost = 0.0, cost /repo/src/pykalshi/http_client.py:287: await self._limiter.acquire(global_cost=global_cost, write_cost=write_cost) /repo/src/pykalshi/protocols.py:292: async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ... /repo/src/pykalshi/protocols.py:293: async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ... 289:class RateLimiterProtocol(Protocol): 290- """Contract for rate limiting.""" 291- 292- async def acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> None: ... 293- async def try_acquire(self, global_cost: float = 1.0, write_cost: float = 0.0) -> bool: ... 294- async def reconfigure(self, read_rate: float, write_rate: float) -> None: ... 295- def get_status(self) -> dict[str, object]: ...
step 4
$ sed -n 260,300p /repo/src/pykalshi/http_client.py; ls /repo/tests; grep -rln "rate_limiter\|RateLimiter" /repo/tests
output · 50 lines
await self.configure_rate_limits()
def _resolve_cost(self, method: str, path: str) -> float:
upper_method = method.upper()
for pattern, m, cost in self._cost_patterns:
if m == upper_method and pattern.match(path):
return cost
return self._default_cost
# --- Core request engine ---
async def _execute_request(38 more lines
self,
method: str,
path: str,
**kwargs: Any,
) -> dict[str, Any]:
"""Execute request with rate limiting and 429 retry."""
await self._ensure_rates_configured()
cost = self._resolve_cost(method, path)
if method.upper() == "GET":
global_cost, write_cost = cost, 0.0
else:
global_cost, write_cost = 0.0, cost
wait_start = time.monotonic()
await self._limiter.acquire(global_cost=global_cost, write
…[394 chars omitted]…
, path, **kwargs)
if response.status_code == 429:
__init__.py
conftest.py
mock_data.py
test_api
test_api_utils.py
test_asyncapi_validation.py
test_auth.py
test_config.py
test_exceptions.py
test_http_client.py
test_integration.py
test_models.py
test_observability.py
test_openapi_validation.py
test_rate_limiter.py
test_ws_client.py
test_ws_integration.py
/repo/tests/test_rate_limiter.pystep 5
$ cat /repo/tests/test_rate_limiter.py
output · 34 lines
"""Tests for pykalshi.rate_limiter."""
import pytest
from pykalshi.rate_limiter import ReadWriteTokenBucket
class TestReadWriteTokenBucket:
@pytest.mark.asyncio
async def test_initial_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
assert bucket.read_tokens == 10.0
assert bucket.write_tokens == 5.022 more lines
@pytest.mark.asyncio
async def test_read_acquire_consumes_read_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(global_cost=1.0, write_cost=0.0)
assert bucket.read_tokens == 9.0
assert bucket.write_tokens == 5.0 # write unchanged
@pytest.mark.asyncio
async def test_write_acquire_consumes_write_tokens(self) -> None:
bucket = ReadWriteTokenBucket(read_rate=10.0, write_rate=5.0)
await bucket.acquire(glo
…[5140 chars omitted]…
tokens concurrently (exactly capacity)
results = await asyncio.gather(
*[bucket.try_acquire(global_cost=1.0, write_cost=0.0) for _ in range(5)]
)
assert sum(results) == 5 # all should succeed
assert bucket.read_tokens == 0.0
# 6th attempt should fail (no tokens left)
assert await bucket.try_acquire(global_cost=1.0, write_cost=0.0) is Falsestep 6
$ cd /repo && python -m pytest tests/test_rate_limiter.py -q 2>&1 | tail -5 && cat > /tmp/repro.py <<'EOF'
output · 18 lines
................. [100%]
17 passed in 0.15s
Traceback (most recent call last):
File "/tmp/repro.py", line 10, in <module>
asyncio.run(main())
File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete6 more lines
return future.result()
^^^^^^^^^^^^^^^
File "/tmp/repro.py", line 7, in main
await asyncio.wait_for(asyncio.gather(*[b.acquire(read_cost=0.0, write_cost=1.0) for _ in range(5)]), timeout=10)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: ReadWriteTokenBucket.acquire() got an unexpected keyword argument 'read_cost'step 7
$ cd /repo && sed -i 's/read_cost=0.0, write_cost=1.0/global_cost=0.0, write_cost=1.0/' /tmp/repro.py && timeout 15 python /tmp/repro.py; echo rc=$?
output · 2 lines
done in 2.2027385539840907 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 1}
rc=0step 8
$ cd /repo && cat > /tmp/repro2.py <<'EOF'
output · 2 lines
done in 2.2028683310199995 {'type': 'ReadWriteTokenBucket', 'read_tokens': 1.0, 'write_tokens': 1.0, 'read_history_len': 1, 'write_history_len': 1}
rc=0step 9
Current code has no hang under concurrency; the issue is the interface rename (`read_cost`) plus guaranteed disjoint accounting. I'll rename with a backward-compatible `global_cost` alias and update internal callers.
$ cd /repo && python - <<'PY'
output · 25 lines
58: def _can_proceed(self, global_cost: float, write_cost: float) -> bool: 61: return self.read_tokens >= global_cost 63: def _calculate_wait_time(self, global_cost: float, write_cost: float) -> float: 78: if self.read_tokens >= global_cost: 80: needed = global_cost - self.read_tokens 89: def _consume(self, global_cost: float, write_cost: float) -> None: 95: self.read_tokens -= global_cost 96: self._read_history.append((now, global_cost)) 100: read_cost: float = 1.0, 102: global_cost: float | None = None, 106: ``read_cost`` is the read-side cost; ``write_cost`` the write-side 108: tokens are used, otherwise only read tokens. ``global_cost`` is a
13 more lines
109: deprecated alias for ``read_cost``.
111: if global_cost is not None:
112: read_cost = global_cost
115: if write_cost <
…[565 chars omitted]…
if self._can_proceed(read_cost, write_cost):
155: self._consume(read_cost, write_cost)
161: read_cost: float = 1.0,
163: global_cost: float | None = None,
166: if global_cost is not None:
167: read_cost = global_cost
174: if self._can_proceed(read_cost, write_cost):
176: return self._calculate_wait_time(read_cost, write_cost)step 10
$ cd /repo && sed -n 140,150p src/pykalshi/rate_limiter.py && python - <<'PY'
output · 16 lines
"""Try to acquire tokens without blocking. Returns True if acquired."""
if global_cost is not None:
read_cost = global_cost
if write_cost > 0 and write_cost > self.write_capacity:
return False
if write_cost <= 0 and read_cost > self.read_capacity:
return False
lock = self._write_lock if write_cost > 0 else self._read_lock
async with lock:
if write_cost > 0:
grep: src/pykalshi/__pycache__/http_client.cpython-312.pyc: binary file matches4 more lines
grep: src/pykalshi/__pycache__/protocols.cpython-312.pyc: binary file matches grep: src/pykalshi/__pycache__/rate_limiter.cpython-312.pyc: binary file matches src/pykalshi/protocols.py:296: global_cost: float | None = None, src/pykalshi/protocols.py:302: global_cost: float | None = None,
step 11
$ cd /repo && sed -i 's/read_cost=0.0, write_cost=1.0/global_cost=0.0, write_cost=1.0/' /tmp/repro.py && cat > /tmp/repro3.py <<'EOF'
output · 6 lines
ok 2.202908467996167 {'type': 'ReadWriteTokenBucket', 'read_tokens': 1.0, 'write_tokens': 1.0, 'read_history_len': 1, 'write_history_len': 1}
all good
done in 2.2031789149914403 {'type': 'ReadWriteTokenBucket', 'read_tokens': 10.0, 'write_tokens': 1.0, 'read_history_len': 0, 'write_history_len': 1}
........................................................................ [ 63%]
......................................... [100%]
113 passed in 4.36sstep 12
$ cd /repo && timeout 600 python -m pytest -q -x 2>&1 | tail -3
output · 3 lines
FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage !!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!! 1 failed, 6 passed in 0.76s
step 13
$ cd /repo && git stash && timeout 120 python -m pytest tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage -q 2>&1 | tail -3; git stash pop
output · 14 lines
Saved working directory and index state WIP on master: 8181675 base =========================== short test summary info ============================ FAILED tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage 1 failed in 0.39s On branch master Changes not staged for commit: (use "git add <file>..." to update what will be committed) (use "git restore <file>..." to discard changes in working directory) modified: src/pykalshi/http_client.py modified: src/pykalshi/protocols.py modified: src/pykalshi/rate_limiter.py
2 more lines
no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (174d925a1af215d03052d4906a9137e31b34ce1c)step 14
Pre-existing failure, unrelated. Run the rest:
$ cd /repo && timeout 900 python -m pytest -q --deselect tests/test_asyncapi_validation.py::test_asyncapi_ws_channel_coverage 2>&1 | tail -3 && git diff --stat
output · 7 lines
FAILED tests/test_openapi_validation.py::test_openapi_coverage - httpx.Connec... FAILED tests/test_openapi_validation.py::test_openapi_schema_coverage - httpx... 3 failed, 197 passed, 71 skipped, 1 deselected in 6.75s src/pykalshi/http_client.py | 6 +++--- src/pykalshi/protocols.py | 14 ++++++++++-- src/pykalshi/rate_limiter.py | 51 ++++++++++++++++++++++++++++++++------------ 3 files changed, 52 insertions(+), 19 deletions(-)
step 15
All remaining failures are pre-existing (network-dependent OpenAPI coverage tests and an unrelated WS coverage test). Checklist complete.
$ done: {"notes": "Renamed the rate limiter's read-side cost parameter to `read_cost` on `ReadWriteTokenBucket.acquire/try_acquire/get_wait_time` and in `RateLimiterProtocol`, keeping `global_cost` as a backward-compatible keyword alias; updated the internal caller in http_client.py. Verified disjoint read/write accounting (write_cost>0 cooutput · 3 lines
…[148 chars omitted]… s/deadlocks under concurrent mixed load exceeding capacity (reproducers complete in ~2.2s). Existing tests pass (only pre-existing network/coverage failures remain)."} submitted.