From 0d35dff5947878e6f77189481626f74c1157d2fe Mon Sep 17 00:00:00 2001 From: MarketDataDev02 Date: Fri, 2 Oct 2026 15:42:32 -0300 Subject: [PATCH 1/4] feat(client): cap the requests in flight at 50 per client SDK requirements section 12 asks for one client-level pool of 50 requests, shared by every endpoint, that hands a freed slot to the next request at once. stocks.candles and options.quotes each built their own 50-worker executor per call, so two calls running at the same time put 100 requests on the wire and the API answered the excess with 429. MarketDataClient now owns a BoundedSemaphore(50) that _make_request holds around the HTTP exchange only. Every request of the client draws from it, the fan-outs keep their executors, and a retry sleeping through its backoff holds no slot. A request that finds none waits for one. --- CHANGELOG.md | 1 + README.md | 6 +- docs/options.md | 4 +- docs/stocks.md | 8 +- src/marketdata/client.py | 56 ++++- src/tests/test_request_pool.py | 386 +++++++++++++++++++++++++++++++++ 6 files changed, 451 insertions(+), 10 deletions(-) create mode 100644 src/tests/test_request_pool.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a08b89..6d5a190 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `stocks.quotes()` now requests `stocks/quotes/?symbols=...` instead of the deprecated `stocks/bulkquotes/`; the method, its parameters and its output are unchanged (#74) - **BREAKING**: `UserRateLimits` speaks in API credits, as the product does (SDK requirements §8.1): `requests_limit` → `credit_limit`, `requests_remaining` → `credits_remaining`, `requests_consumed` → `credits_consumed`, `requests_reset` → `reset_time`. The old names are gone, with no aliases; its string form now reads `Credits used X/Y, remaining: Z, reset at: ` (#48) - The pre-flight credit check reads a private, thread-safe tracker fed by every response that carries credit headers, error answers included; out-of-order responses (an older reset window, a higher remaining in the same window) are ignored, and every response logs its credits at DEBUG. An envelope with a `credit_limit` of zero is not recorded: the API answers a missing or unknown token with demo data and a `0/0` envelope from another window, which would otherwise tell the check this account has no credits. `client.utilities.user()` is asked for precisely to learn the balance, so its answer replaces the state rather than being weighed against it (#49) +- A client sends at most 50 requests at a time, the limit the API enforces with a `429` (SDK requirements §12). Every request of the client takes one of 50 slots for the length of its HTTP exchange, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot instead of being sent and refused. Before, `stocks.candles()` and `options.quotes()` each ran 50 workers of their own, so two calls at the same time put 100 requests on the wire. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter, and it belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread (#65) ### Added diff --git a/README.md b/README.md index 9af93ee..1161c35 100644 --- a/README.md +++ b/README.md @@ -883,8 +883,10 @@ The SDK uses Pydantic for input validation: The SDK uses concurrent requests for efficient data fetching in specific scenarios: -- **`stocks.candles()`**: For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long, non-overlapping chunks and fetched concurrently (up to 50 concurrent requests by default) -- **`options.quotes()`**: Multiple option symbols are fetched concurrently (up to 50 concurrent requests by default) +- **`stocks.candles()`**: For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long, non-overlapping chunks and fetched concurrently +- **`options.quotes()`**: Multiple option symbols are fetched concurrently + +A client sends at most 50 requests at a time, the limit the API enforces with a `429`. Every request takes one of the client's 50 slots for as long as its HTTP exchange lasts, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot to free instead of being sent and refused. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter. It belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread. When concurrent requests are used, responses are automatically merged into a single result. A chunk or symbol with no data is left out of the merge; if every part has no data the call returns an empty result, and if the API answered with nothing usable at all a `MarketdataHttpError` is raised. diff --git a/docs/options.md b/docs/options.md index c16f1eb..7f82e7b 100644 --- a/docs/options.md +++ b/docs/options.md @@ -571,7 +571,7 @@ if csv_file: ### `quotes()` -Fetches options quotes for one or more option symbols. This method includes API status checking, automatic retry logic, and supports concurrent requests for multiple symbols (up to 50 concurrent requests by default). Each symbol's request retries on its own, so a failed symbol never re-sends the healthy ones. A symbol whose response body cannot be decoded raises `ParseError`, as in every other method, on every output format; in CSV output the merged file keeps the header the API sent (the requested `columns`, the human-readable names) and every row of every symbol. On the other formats the symbols are merged on the model's columns among those the API sent, in the order it sent them (the request order under `columns=`), and a symbol whose answer lacks one of them, or carries one that is not a list as long as its others, raises `ParseError` as well, since merging it would put the next symbol's values on its row. +Fetches options quotes for one or more option symbols. This method includes API status checking, automatic retry logic, and supports concurrent requests for multiple symbols (within the client's 50 request slots). Each symbol's request retries on its own, so a failed symbol never re-sends the healthy ones. A symbol whose response body cannot be decoded raises `ParseError`, as in every other method, on every output format; in CSV output the merged file keeps the header the API sent (the requested `columns`, the human-readable names) and every row of every symbol. On the other formats the symbols are merged on the model's columns among those the API sent, in the order it sent them (the request order under `columns=`), and a symbol whose answer lacks one of them, or carries one that is not a list as long as its others, raises `ParseError` as well, since merging it would put the next symbol's values on its row. > **Note:** The `symbols` parameter can be passed as the first positional argument or as a keyword argument. All other parameters must be keyword-only. @@ -608,7 +608,7 @@ Fetches options quotes for one or more option symbols. This method includes API - If `output_format=OutputFormat.CSV`: A string containing the filename where CSV data was written (merged from all requested symbols) - Raises a `BaseMarketdataException` subclass if an error occurs (rate limits, validation errors, request failures, no valid responses received, etc.); see the [README](../README.md#error-handling) -> **Note:** The method returns the requested output, an empty result when the API has no data for every symbol (`404` with `s: "no_data"`, see the [README](../README.md#no-data-is-not-an-error)), or raises. The method uses concurrent requests (up to 50 concurrent requests by default) to fetch quotes for multiple symbols efficiently. +> **Note:** The method returns the requested output, an empty result when the API has no data for every symbol (`404` with `s: "no_data"`, see the [README](../README.md#no-data-is-not-an-error)), or raises. The method uses concurrent requests, within the client's 50 request slots (see the [README](../README.md#concurrent-requests)), to fetch quotes for multiple symbols efficiently. #### Examples diff --git a/docs/stocks.md b/docs/stocks.md index 69a1f38..55ea84f 100644 --- a/docs/stocks.md +++ b/docs/stocks.md @@ -316,7 +316,7 @@ print(df) ### `candles()` -Fetches stock candles (OHLCV data) for a symbol with support for various timeframes and date ranges. This method includes API status checking and automatic retry logic. For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long chunks and fetched concurrently using up to 50 concurrent requests. Each chunk's request retries on its own, so a failed chunk never re-sends the others. +Fetches stock candles (OHLCV data) for a symbol with support for various timeframes and date ranges. This method includes API status checking and automatic retry logic. For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long chunks and fetched concurrently, within the client's 50 request slots (see the [README](../README.md#concurrent-requests)). Each chunk's request retries on its own, so a failed chunk never re-sends the others. > **Note:** The `symbol` parameter can be passed as the first positional argument or as a keyword argument. All other parameters must be keyword-only. @@ -327,8 +327,8 @@ Fetches stock candles (OHLCV data) for a symbol with support for various timefra - Numeric with unit: `"1"`, `"15M"`, `"1H"`, `"1D"`, `"1W"`, `"1M"`, `"1Y"` - Unit only: `"M"`, `"H"`, `"D"`, `"W"`, `"M"`, `"Y"` - Descriptive: `"minutely"`, `"hourly"`, `"daily"`, `"weekly"`, `"monthly"`, `"yearly"` -- `from_date` (datetime.date, optional): The start date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly) using up to 50 concurrent requests. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. -- `to_date` (datetime.date, optional): The end date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly) using up to 50 concurrent requests. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. +- `from_date` (datetime.date, optional): The start date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly), within the client's 50 request slots. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. +- `to_date` (datetime.date, optional): The end date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly), within the client's 50 request slots. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. - `countback` (int, optional): The number of candles to fetch (alternative to date range) - `extended` (bool, optional): Whether to fetch extended candles (pre-market and after-hours data) - `adjust_splits` (bool, optional): Whether to adjust for stock splits. Uses API alias `adjustsplits`. @@ -364,7 +364,7 @@ Fetches stock candles (OHLCV data) for a symbol with support for various timefra When both `from_date` and `to_date` are provided, the method behavior depends on the resolution: -- **For intraday resolutions (minutely/hourly)**: The date range is automatically split into year-long chunks, fetched concurrently using a thread pool executor (up to 50 concurrent requests), and merged into a single response. Chunks never share a calendar day, so no candle is requested twice. This allows efficient fetching of large historical date ranges without manual pagination. +- **For intraday resolutions (minutely/hourly)**: The date range is automatically split into year-long chunks, fetched concurrently using a thread pool executor within the client's 50 request slots, and merged into a single response. Chunks never share a calendar day, so no candle is requested twice. This allows efficient fetching of large historical date ranges without manual pagination. - **For non-intraday resolutions (daily/weekly/monthly/yearly)**: The entire date range is fetched in a single request without splitting. diff --git a/src/marketdata/client.py b/src/marketdata/client.py index 5145c45..a5a4f1d 100644 --- a/src/marketdata/client.py +++ b/src/marketdata/client.py @@ -1,3 +1,4 @@ +import threading import time from importlib.metadata import version from logging import DEBUG, INFO, Logger @@ -21,6 +22,7 @@ from marketdata.input_types.base import UserUniversalAPIParams from marketdata.internal_settings import ( HEADER_REQUEST_ID, + MAX_CONCURRENT_REQUESTS, MAX_CREDIT_WINDOW_SECONDS, MAX_RETRY_ATTEMPTS, NO_TOKEN_VALUE, @@ -53,6 +55,18 @@ def __init__( logger: Logger = None, max_retries: int = MAX_RETRY_ATTEMPTS, ): + """Build the client and, when it has a token, seed the credit tracker. + + Args: + token: The API token. Falls back to the ``MARKETDATA_TOKEN`` + setting; with neither, the client starts in demo mode. + logger: The logger to write to. The SDK's own when omitted. + max_retries: Retries after the first attempt of each request. + + Raises: + ValueError: ``max_retries`` is negative. + BaseMarketdataException: The start-up ``/user/`` request failed. + """ if max_retries < 0: raise ValueError("max_retries must be >= 0") self.token = token or settings.marketdata_token @@ -75,7 +89,12 @@ def __init__( self.client = self._get_client() self.default_params = UserUniversalAPIParams() - # The private credit tracker behind the pre-flight check (#49); the + # One budget for every request of this client, whichever endpoint or + # fan-out sends it. It exists before `_setup_rate_limits`, whose + # `/user/` request already takes a slot. + self._request_slots = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS) + + # The private credit tracker behind the pre-flight check; the # request-scoped numbers travel with each result (marketdata.get_meta). self._rate_limits = RateLimitTracker() self._setup_rate_limits() @@ -298,6 +317,38 @@ def _make_request( response_log_level: int = INFO, **kwargs, ) -> Response: + """Send one HTTP request and map its outcome to the SDK's exceptions. + + The request holds one of the client's request slots only for the HTTP + exchange itself: it waits for a free slot after the pre-flight check, + and a retry sleeping through its backoff holds none. + + Args: + method: The HTTP method. + url: The path, relative to the base URL. + check_rate_limits: Refuse the request when the account is known to + have no credits left. + part_of_result: Record the response's metadata in the caller's + scope. + include_api_version: Prefix the path with the API version. + authoritative_credits: Let this response replace the tracked credit + state instead of being weighed against it. + response_log_level: The level of the line logged for the response. + **kwargs: Passed to ``httpx.Client.request``. + + Returns: + The response, including a 404 without ``errmsg``, which is the + API's "no data" answer. + + Raises: + RateLimitError: The pre-flight check refused the request, or the + API answered 429. + NetworkError: There was no usable answer (a connection failure, a + timeout, a protocol or proxy error). + ParseError: The body does not match its ``Content-Encoding``. + MarketdataHttpError: The API answered with an error status; the + subclass names it. + """ if self.token is NO_TOKEN_VALUE: check_rate_limits = False @@ -311,7 +362,8 @@ def _make_request( # No `timeout=` here: the client carries the fixed one (§10), # so there is one place to read it and no argument a resource # could use to give itself a longer bound. - response = self.client.request(method, url, **kwargs) + with self._request_slots: + response = self.client.request(method, url, **kwargs) except DecodingError as exc: # The API answered but the body does not match its Content-Encoding # (an intercepting proxy): the answer is unusable, not missing. diff --git a/src/tests/test_request_pool.py b/src/tests/test_request_pool.py new file mode 100644 index 0000000..c527406 --- /dev/null +++ b/src/tests/test_request_pool.py @@ -0,0 +1,386 @@ +"""One request pool per client (SDK requirements §12): at most 50 requests in +flight, shared by every endpoint and every fan-out, and a freed slot goes to the +next request at once.""" + +import threading +import time + +import httpx +import pytest + +from marketdata.api_status import API_STATUS_DATA +from marketdata.client import MarketDataClient +from marketdata.exceptions import NetworkError, ServerError +from marketdata.input_types.base import OutputFormat + +PRICES_URL = "https://api.marketdata.app/v1/stocks/prices/" +FAN_OUT_URL = r".*/v1/(options/quotes|stocks/candles)/.*" +POOL_SIZE = 50 +CALLERS = 120 +ENOUGH = 1000 +WAIT = 10.0 +HOLD = 20.0 +FINISH = 15.0 + + +class Gate: + """Holds every request inside the mocked transport and counts them.""" + + def __init__(self, answer): + """Build a gate that lets nothing through until told to. + + Args: + answer: Builds the response of a request once the gate lets it go. + """ + self._answer = answer + self._changed = threading.Condition() + self._permits = threading.Semaphore(0) + self.current = 0 + self.started = 0 + self.peak = 0 + + def handle(self, request): + """Count a request in, hold it until it is let go, and count it out. + + Args: + request: The request the client sent. + + Returns: + The response built by ``answer``. + """ + with self._changed: + self.current += 1 + self.started += 1 + self.peak = max(self.peak, self.current) + self._changed.notify_all() + try: + self._permits.acquire(timeout=HOLD) + finally: + with self._changed: + self.current -= 1 + self._changed.notify_all() + return self._answer(request) + + def let_go(self, count): + """Let held requests go. + + Args: + count: How many requests to let go. + """ + self._permits.release(count) + + def wait_for(self, condition): + """Wait until a condition on the gate holds. + + Args: + condition: Takes the gate and says whether the wait is over. + + Returns: + Whether the condition held before ``WAIT`` seconds went by. + """ + with self._changed: + return self._changed.wait_for(lambda: condition(self), WAIT) + + def settle(self, quiet=0.3): + """Wait until the number of held requests stops changing. + + Args: + quiet: How long the number must stay the same, in seconds. + + Returns: + The number of held requests once it has been steady that long. + """ + deadline = time.monotonic() + WAIT + with self._changed: + while time.monotonic() < deadline: + before = self.current + if not self._changed.wait_for(lambda: self.current != before, quiet): + break + return self.current + + +class Callers: + """Runs calls in daemon threads and keeps what each returned or raised. + + The threads are daemons so that a call stuck waiting for a slot cannot keep + the test process alive. + """ + + def __init__(self, calls): + """Start one thread per call. + + Args: + calls: The callables to run, each in a thread of its own. + """ + self.outcomes = [None] * len(calls) + self._threads = [ + threading.Thread(target=self._run, args=(index, call), daemon=True) + for index, call in enumerate(calls) + ] + for thread in self._threads: + thread.start() + + def _run(self, index, call): + """Run one call and keep its outcome. + + Args: + index: Where the outcome goes in ``outcomes``. + call: The callable to run. + """ + try: + self.outcomes[index] = call() + except BaseException as exc: + self.outcomes[index] = exc + + def finish(self): + """Wait for every call to end. + + Returns: + What each call returned or raised, in the order they were given. + + Raises: + AssertionError: A call is still running after ``FINISH`` seconds. + """ + deadline = time.monotonic() + FINISH + for thread in self._threads: + thread.join(max(0.0, deadline - time.monotonic())) + stuck = sum(thread.is_alive() for thread in self._threads) + assert not stuck, f"{stuck} calls still waiting after {FINISH} seconds" + return self.outcomes + + +def raised(outcomes): + """Pick the exceptions out of what a set of calls produced. + + Args: + outcomes: What each call returned or raised. + + Returns: + The outcomes that are exceptions. + """ + return [outcome for outcome in outcomes if isinstance(outcome, BaseException)] + + +def free_slots(client): + """Count the request slots of a client that are free right now. + + Takes every free slot without waiting and gives them all back, so a leaked + slot shows as a smaller number instead of a call that never returns. + + Args: + client: The client whose pool is read. + + Returns: + The number of free slots. + """ + taken = 0 + while client._request_slots.acquire(blocking=False): + taken += 1 + for _ in range(taken): + client._request_slots.release() + return taken + + +def service_unavailable(request): + """Answer a request with a 503. + + Args: + request: The request the client sent. + + Returns: + A 503 response. + """ + return httpx.Response(503, json={}) + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + """Skip the retry backoff.""" + monkeypatch.setattr("time.sleep", lambda *_: None) + + +@pytest.fixture(autouse=True) +def _synchronous_status_refresh(monkeypatch): + """Refresh ``/status/`` inline, so no helper thread holds a slot while a test + counts them.""" + monkeypatch.setattr( + API_STATUS_DATA, "_trigger_async_refresh", lambda c: API_STATUS_DATA.refresh(c) + ) + + +def test_120_callers_never_put_more_than_50_requests_in_flight( + load_json, respx_mock, client +): + """Callers beyond the 50th wait for a slot instead of reaching the API.""" + body = load_json("stocks_prices_response_200") + gate = Gate(lambda request: httpx.Response(200, json=body)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(CALLERS) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert gate.started == CALLERS + assert not raised(outcomes) + + +def test_two_fan_outs_started_together_share_the_pool(load_json, respx_mock, client): + """A ``stocks.candles`` and an ``options.quotes`` running at once draw from + one budget of 50 requests.""" + quotes = load_json("options_quotes_response_200") + candles = load_json("stocks_candles_response_200") + + def answer(request): + """Answer with the body of the endpoint the request is for. + + Args: + request: The request the client sent. + + Returns: + A 200 response. + """ + body = candles if "/stocks/candles/" in request.url.path else quotes + return httpx.Response(200, json=body) + + gate = Gate(answer) + respx_mock.get(url__regex=FAN_OUT_URL).mock(side_effect=gate.handle) + symbols = [f"AAPL250117C{150000 + 500 * i:08d}" for i in range(60)] + start = threading.Barrier(2, timeout=WAIT) + + def quote_symbols(): + """Quote 60 symbols once both fan-outs are ready.""" + start.wait() + return client.options.quotes(symbols, output_format=OutputFormat.JSON) + + def fetch_candles(): + """Fetch 60 years of hourly candles, one request per year, once both + fan-outs are ready.""" + start.wait() + return client.stocks.candles( + "AAPL", + resolution="H", + from_date="1966-01-01", + to_date="2026-01-01", + output_format=OutputFormat.JSON, + ) + + callers = Callers([quote_symbols, fetch_candles]) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert not raised(outcomes) + + +def test_a_freed_slot_starts_the_next_request_without_waiting_for_the_others( + load_json, respx_mock, client +): + """With 50 requests in flight and one waiting, one completion starts the + waiting one while the other 49 are still running.""" + body = load_json("stocks_prices_response_200") + gate = Gate(lambda request: httpx.Response(200, json=body)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE + 1) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + assert gate.settle() == POOL_SIZE + gate.let_go(1) + + started = gate.wait_for(lambda g: g.started == POOL_SIZE + 1) + in_flight = gate.current + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert started, "the waiting request did not start when a slot freed" + assert in_flight == POOL_SIZE + assert not raised(outcomes) + + +@pytest.mark.parametrize( + ("transport", "ends_in"), + [ + (httpx.ConnectTimeout("slow"), NetworkError), + (httpx.ConnectError("refused"), NetworkError), + (RuntimeError("unexpected"), RuntimeError), + (service_unavailable, ServerError), + ], + ids=["timeout", "connection-error", "unexpected-exception", "5xx-after-retries"], +) +def test_a_failed_request_gives_its_slot_back(transport, ends_in, respx_mock, client): + """Whatever ends a request, its slot is free before the next one starts. + + Args: + transport: What the mocked transport does: an exception to raise or a + function that builds the response. + ends_in: The exception the call ends in. + """ + respx_mock.get(PRICES_URL).mock(side_effect=transport) + + for _ in range(POOL_SIZE + 10): + with pytest.raises(ends_in): + client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + assert free_slots(client) == POOL_SIZE + + +def test_a_retry_holds_no_slot_while_it_backs_off( + load_json, respx_mock, client, monkeypatch +): + """Between two attempts of one call, all 50 slots are free.""" + body = load_json("stocks_prices_response_200") + free_during_backoff = [] + monkeypatch.setattr( + "time.sleep", lambda *_: free_during_backoff.append(free_slots(client)) + ) + route = respx_mock.get(PRICES_URL).mock( + side_effect=[httpx.Response(503, json={}), httpx.Response(200, json=body)] + ) + + client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + + assert route.call_count == 2 + assert free_during_backoff == [POOL_SIZE] + + +def test_each_client_has_its_own_pool(load_json, respx_mock, client): + """A client with all 50 slots taken leaves another client's slots free.""" + body = load_json("stocks_prices_response_200") + gate = Gate(lambda request: httpx.Response(200, json=body)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + other = MarketDataClient(token="test") + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + taken = free_slots(client) + free_on_other = free_slots(other) + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert taken == 0 + assert free_on_other == POOL_SIZE + assert not raised(outcomes) From f3ccfdc77b65a82f0403bb6257b0025ecf19ad58 Mon Sep 17 00:00:00 2001 From: MarketDataDev02 Date: Fri, 2 Oct 2026 16:17:40 -0300 Subject: [PATCH 2/4] test(client): check the request pool through saturated public calls The failure, backoff and second-client cases read the private slot count of the client. They now saturate the pool with public calls instead: the calls waiting behind requests that fail still run, fifty retries asleep in their backoff leave room for another call, and a second client is not held up by a client whose slots are all taken. --- src/tests/test_request_pool.py | 175 ++++++++++++++++++++++++--------- 1 file changed, 126 insertions(+), 49 deletions(-) diff --git a/src/tests/test_request_pool.py b/src/tests/test_request_pool.py index c527406..6f0413f 100644 --- a/src/tests/test_request_pool.py +++ b/src/tests/test_request_pool.py @@ -2,6 +2,7 @@ flight, shared by every endpoint and every fan-out, and a freed slot goes to the next request at once.""" +import collections import threading import time @@ -14,6 +15,7 @@ from marketdata.input_types.base import OutputFormat PRICES_URL = "https://api.marketdata.app/v1/stocks/prices/" +MARKETS_STATUS_URL = "https://api.marketdata.app/v1/markets/status/" FAN_OUT_URL = r".*/v1/(options/quotes|stocks/candles)/.*" POOL_SIZE = 50 CALLERS = 120 @@ -161,24 +163,40 @@ def raised(outcomes): return [outcome for outcome in outcomes if isinstance(outcome, BaseException)] -def free_slots(client): - """Count the request slots of a client that are free right now. +def connect_timeout(request): + """Fail as a request that never connects does. - Takes every free slot without waiting and gives them all back, so a leaked - slot shows as a smaller number instead of a call that never returns. + Args: + request: The request the client sent. + + Raises: + httpx.ConnectTimeout: Always. + """ + raise httpx.ConnectTimeout("slow") + + +def connection_refused(request): + """Fail as a request the server refuses does. Args: - client: The client whose pool is read. + request: The request the client sent. - Returns: - The number of free slots. + Raises: + httpx.ConnectError: Always. """ - taken = 0 - while client._request_slots.acquire(blocking=False): - taken += 1 - for _ in range(taken): - client._request_slots.release() - return taken + raise httpx.ConnectError("refused") + + +def unexpected_failure(request): + """Fail with an exception the client does not know. + + Args: + request: The request the client sent. + + Raises: + RuntimeError: Always. + """ + raise RuntimeError("unexpected") def service_unavailable(request): @@ -318,69 +336,128 @@ def test_a_freed_slot_starts_the_next_request_without_waiting_for_the_others( @pytest.mark.parametrize( - ("transport", "ends_in"), + ("fails", "ends_in"), [ - (httpx.ConnectTimeout("slow"), NetworkError), - (httpx.ConnectError("refused"), NetworkError), - (RuntimeError("unexpected"), RuntimeError), + (connect_timeout, NetworkError), + (connection_refused, NetworkError), + (unexpected_failure, RuntimeError), (service_unavailable, ServerError), ], ids=["timeout", "connection-error", "unexpected-exception", "5xx-after-retries"], ) -def test_a_failed_request_gives_its_slot_back(transport, ends_in, respx_mock, client): - """Whatever ends a request, its slot is free before the next one starts. +def test_calls_waiting_for_a_slot_go_on_when_the_requests_ahead_fail( + fails, ends_in, respx_mock, client +): + """With the pool full of requests that fail, the 10 calls waiting behind them + still run and fail in turn. Args: - transport: What the mocked transport does: an exception to raise or a - function that builds the response. - ends_in: The exception the call ends in. + fails: What the mocked transport does with each request once the gate + lets it go. + ends_in: The exception every call ends in. """ - respx_mock.get(PRICES_URL).mock(side_effect=transport) + gate = Gate(fails) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE + 10) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() - for _ in range(POOL_SIZE + 10): - with pytest.raises(ends_in): - client.stocks.prices("AAPL", output_format=OutputFormat.JSON) - assert free_slots(client) == POOL_SIZE + assert settled == POOL_SIZE + assert all(isinstance(outcome, ends_in) for outcome in outcomes) -def test_a_retry_holds_no_slot_while_it_backs_off( +def test_retries_asleep_in_their_backoff_do_not_hold_up_another_call( load_json, respx_mock, client, monkeypatch ): - """Between two attempts of one call, all 50 slots are free.""" + """With 50 calls asleep between two attempts, one more call still goes + through.""" body = load_json("stocks_prices_response_200") - free_during_backoff = [] - monkeypatch.setattr( - "time.sleep", lambda *_: free_during_backoff.append(free_slots(client)) - ) - route = respx_mock.get(PRICES_URL).mock( - side_effect=[httpx.Response(503, json={}), httpx.Response(200, json=body)] - ) + attempts = collections.Counter() + attempts_lock = threading.Lock() + asleep = threading.Semaphore(0) + wake_up = threading.Event() - client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + def answer(request): + """Refuse the first request of each ``SYM`` symbol with a 503. - assert route.call_count == 2 - assert free_during_backoff == [POOL_SIZE] + Args: + request: The request the client sent. + Returns: + A 503 for the first request of a ``SYM`` symbol, a 200 otherwise. + """ + symbol = request.url.params["symbols"] + with attempts_lock: + attempts[symbol] += 1 + first = attempts[symbol] == 1 + if first and symbol.startswith("SYM"): + return httpx.Response(503, json={}) + return httpx.Response(200, json=body) -def test_each_client_has_its_own_pool(load_json, respx_mock, client): - """A client with all 50 slots taken leaves another client's slots free.""" - body = load_json("stocks_prices_response_200") - gate = Gate(lambda request: httpx.Response(200, json=body)) + def sleep_until_woken(seconds): + """Stand in for the backoff: sleep until the test wakes the call up. + + Args: + seconds: How long the retry asked to wait. Ignored. + """ + asleep.release() + wake_up.wait(HOLD) + + monkeypatch.setattr("time.sleep", sleep_until_woken) + respx_mock.get(PRICES_URL).mock(side_effect=answer) + + retrying = Callers( + [ + lambda symbol=f"SYM{index}": client.stocks.prices( + symbol, output_format=OutputFormat.JSON + ) + for index in range(POOL_SIZE) + ] + ) + for _ in range(POOL_SIZE): + assert asleep.acquire(timeout=WAIT), "a retry never reached its backoff" + other = Callers( + [lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON)] + ).finish() + wake_up.set() + outcomes = retrying.finish() + + assert not raised(other) + assert not raised(outcomes) + + +def test_a_client_with_every_slot_taken_does_not_hold_up_another_client( + load_json, respx_mock, client +): + """A client whose 50 slots are all taken leaves a second client's call free + to go through.""" + prices = load_json("stocks_prices_response_200") + markets_status = load_json("markets_status_response_200") + gate = Gate(lambda request: httpx.Response(200, json=prices)) respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + respx_mock.get(MARKETS_STATUS_URL).respond(json=markets_status, status_code=200) other = MarketDataClient(token="test") - callers = Callers( + saturating = Callers( [ lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) for _ in range(POOL_SIZE) ] ) assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" - taken = free_slots(client) - free_on_other = free_slots(other) + on_other = Callers( + [lambda: other.markets.status(output_format=OutputFormat.JSON)] + ).finish() gate.let_go(ENOUGH) - outcomes = callers.finish() + outcomes = saturating.finish() - assert taken == 0 - assert free_on_other == POOL_SIZE + assert not raised(on_other) assert not raised(outcomes) From 7d10cd4a349c29b080a9b072d144a451ba5b9539 Mon Sep 17 00:00:00 2001 From: MarketDataDev02 Date: Tue, 13 Oct 2026 13:40:37 -0300 Subject: [PATCH 3/4] docs(client): say the 50 request limit is the one the API documents The live runs of the pool did not see the API refuse requests above 50 in flight, so README and CHANGELOG state the limit the API documents instead of saying it enforces it with a 429. --- CHANGELOG.md | 2 +- README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d219f33..0125ba5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,7 +38,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `stocks.quotes()` now requests `stocks/quotes/?symbols=...` instead of the deprecated `stocks/bulkquotes/`; the method, its parameters and its output are unchanged (#74) - **BREAKING**: `UserRateLimits` speaks in API credits, as the product does (SDK requirements §8.1): `requests_limit` → `credit_limit`, `requests_remaining` → `credits_remaining`, `requests_consumed` → `credits_consumed`, `requests_reset` → `reset_time`. The old names are gone, with no aliases; its string form now reads `Credits used X/Y, remaining: Z, reset at: ` (#48) - The pre-flight credit check reads a private, thread-safe tracker fed by every response that carries credit headers, error answers included; out-of-order responses (an older reset window, a higher remaining in the same window) are ignored, and every response logs its credits at DEBUG. An envelope with a `credit_limit` of zero is not recorded: the API answers a missing or unknown token with demo data and a `0/0` envelope from another window, which would otherwise tell the check this account has no credits. `client.utilities.user()` is asked for precisely to learn the balance, so its answer replaces the state rather than being weighed against it (#49) -- A client sends at most 50 requests at a time, the limit the API enforces with a `429` (SDK requirements §12). Every request of the client takes one of 50 slots for the length of its HTTP exchange, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot instead of being sent and refused. Before, `stocks.candles()` and `options.quotes()` each ran 50 workers of their own, so two calls at the same time put 100 requests on the wire. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter, and it belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread (#65) +- A client sends at most 50 requests at a time, the concurrent-request limit the API documents (it states that requests above it get a `429`, SDK requirements §12). Every request of the client takes one of 50 slots for the length of its HTTP exchange, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot instead of being sent. Before, `stocks.candles()` and `options.quotes()` each ran 50 workers of their own, so two calls at the same time put 100 requests on the wire. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter, and it belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread (#65) ### Added diff --git a/README.md b/README.md index fe91467..29d7294 100644 --- a/README.md +++ b/README.md @@ -908,7 +908,7 @@ The SDK uses concurrent requests for efficient data fetching in specific scenari - **`stocks.candles()`**: For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long, non-overlapping chunks and fetched concurrently - **`options.quotes()`**: Multiple option symbols are fetched concurrently -A client sends at most 50 requests at a time, the limit the API enforces with a `429`. Every request takes one of the client's 50 slots for as long as its HTTP exchange lasts, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot to free instead of being sent and refused. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter. It belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread. +A client sends at most 50 requests at a time, the concurrent-request limit the API documents (it states that requests above it get a `429`). Every request takes one of the client's 50 slots for as long as its HTTP exchange lasts, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot to free instead of being sent. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter. It belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread. When concurrent requests are used, responses are automatically merged into a single result. A chunk or symbol with no data is left out of the merge; if every part has no data the call returns an empty result, and if the API answered with nothing usable at all a `MarketdataHttpError` is raised. From f6c6878e115e9eefb94861a1117991e7f8f499e7 Mon Sep 17 00:00:00 2001 From: MarketDataDev02 Date: Tue, 13 Oct 2026 14:18:33 -0300 Subject: [PATCH 4/4] test(client): run the pool scenario on the four output formats The pool sits before any output-format code, so the format cannot change how many requests are in flight. A scenario per format checks it through the public client: with the pool full, ten more calls wait and every call is served on INTERNAL, JSON, CSV and DATAFRAME. --- src/tests/test_request_pool.py | 72 ++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/src/tests/test_request_pool.py b/src/tests/test_request_pool.py index 6f0413f..fcb32e2 100644 --- a/src/tests/test_request_pool.py +++ b/src/tests/test_request_pool.py @@ -16,6 +16,9 @@ PRICES_URL = "https://api.marketdata.app/v1/stocks/prices/" MARKETS_STATUS_URL = "https://api.marketdata.app/v1/markets/status/" +PRICES_CSV = ( + "symbol,mid,change,changepct,updated\nAAPL,280.02,-0.68,-0.0024,1764950546\n" +) FAN_OUT_URL = r".*/v1/(options/quotes|stocks/candles)/.*" POOL_SIZE = 50 CALLERS = 120 @@ -335,6 +338,75 @@ def test_a_freed_slot_starts_the_next_request_without_waiting_for_the_others( assert not raised(outcomes) +@pytest.mark.parametrize( + "output_format", + [ + OutputFormat.INTERNAL, + OutputFormat.JSON, + OutputFormat.CSV, + OutputFormat.DATAFRAME, + ], + ids=["internal", "json", "csv", "dataframe"], +) +def test_the_output_format_does_not_change_what_the_pool_allows( + output_format, load_json, respx_mock, client +): + """Calls asking for any of the four output formats draw from the same 50 + slots: with the pool full, the 10 beyond it wait and every call is served. + + Args: + output_format: The output format every call asks for. + """ + body = load_json("stocks_prices_response_200") + + def answer(request): + """Answer with a body in the format the request asked for. + + Args: + request: The request the client sent. + + Returns: + A 200 response, CSV when the request asked for CSV and JSON + otherwise. + """ + if request.url.params.get("format") == "csv": + return httpx.Response(200, text=PRICES_CSV) + return httpx.Response(200, json=body) + + def call(index): + """Make one ``stocks.prices`` call in the output format of the test. + + Args: + index: Which call of the burst this is. It names the CSV file, which + is created exclusively, so no two calls share one. + + Returns: + What the call returned. + """ + extra = ( + {"filename": f"prices-{index}.csv"} if output_format.name == "CSV" else {} + ) + return client.stocks.prices("AAPL", output_format=output_format, **extra) + + gate = Gate(answer) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [lambda index=index: call(index) for index in range(POOL_SIZE + 10)] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert gate.started == POOL_SIZE + 10 + assert not raised(outcomes) + + @pytest.mark.parametrize( ("fails", "ends_in"), [