diff --git a/README.md b/README.md index 2f1e086..bcf4ef5 100644 --- a/README.md +++ b/README.md @@ -245,6 +245,18 @@ client.download("binance_spot_1h") # just make sure it is on dis client.download("fx_daily", symbol="EURUSD") ``` +A cache hit above still means `get()` re-reads and re-filters that file from +disk into a fresh DataFrame — for a dataset with no `Template.YEAR` +partitioning (a single export spanning its whole date range, e.g. +`liquid_universe_prices`), that's the *entire* export, every call, regardless +of how narrow `start`/`end` are. A **separate**, smaller in-memory LRU +(`RWPYTOOLS_FRAME_CACHE_MAX_ENTRIES`, default 8) returns the identical +already-parsed-and-filtered result instead when a call repeats an earlier one +exactly — useful for retrying after a partial failure, or for a caller that +iterates windows itself to bound its own memory. It's keyed by the export's +own on-disk identity plus the request shape, so it can never return stale or +mismatched data, and `frame_cache_max_entries=0` disables it. + ## Errors ```python @@ -292,6 +304,7 @@ Every setting takes a constructor argument or an environment variable | `RWPYTOOLS_RATE_LIMIT_WAIT` | 20 | Seconds per retry after a 429 without `Retry-After` | | `RWPYTOOLS_MAX_RETRIES` | 3 | Retry attempts | | `RWPYTOOLS_DOWNLOAD_CONCURRENCY` | 8 | Parallel downloads / live requests | +| `RWPYTOOLS_FRAME_CACHE_MAX_ENTRIES` | 8 | In-memory LRU size for parsed-and-filtered `get()` results per client (0 disables) | | `RWPYTOOLS_REQUEST_TIMEOUT` | 30 | Umbrella HTTP timeout (per-phase overrides exist) | | `RWPYTOOLS_NO_PROMPT` | — | Set to disable the interactive key prompt | diff --git a/src/rwpytools/config.py b/src/rwpytools/config.py index 2906605..a2364e9 100644 --- a/src/rwpytools/config.py +++ b/src/rwpytools/config.py @@ -46,6 +46,7 @@ _ENV_LIVE_DIFF_MAX_DAYS = "RWPYTOOLS_LIVE_DIFF_MAX_DAYS" _ENV_LIVE_ONLY_WINDOW_DAYS = "RWPYTOOLS_LIVE_ONLY_WINDOW_DAYS" _ENV_RATE_LIMIT_WAIT = "RWPYTOOLS_RATE_LIMIT_WAIT" +_ENV_FRAME_CACHE_MAX_ENTRIES = "RWPYTOOLS_FRAME_CACHE_MAX_ENTRIES" DEFAULT_API_BASE_URL = "https://api.robotwealth.com" DEFAULT_CACHE_MAX_BYTES = 10 * 1024 * 1024 * 1024 # 10 GiB on-disk cache cap. @@ -75,6 +76,18 @@ #: Seconds to wait before retrying a 429 that carries no ``Retry-After``. #: rw-api's limits are per minute and it does not send the header. DEFAULT_RATE_LIMIT_WAIT = 20.0 +#: How many parsed-and-filtered ``AsyncSession._read_object`` results to keep +#: in memory per ``Client``/``AsyncSession``, LRU-evicted. Session.get() calls +#: formats.read_frame() fresh on every call -- for a bulk dataset without +#: Template.YEAR (a single large export covering its whole date range, e.g. +#: liquid_universe_prices), that means re-reading and re-filtering the same +#: on-disk export from scratch for every distinct (or repeated) window a +#: caller asks for, even seconds apart in the same process. A caller who +#: repeats an identical (dataset, cell, window, symbols) request -- a retry +#: after a partial failure, or a caller iterating windows itself to bound its +#: own memory -- gets that exact result back without re-reading the export. +#: 0 disables the cache entirely. +DEFAULT_FRAME_CACHE_MAX_ENTRIES = 8 def _mask_api_key(api_key: str) -> str: @@ -126,6 +139,7 @@ class ClientConfig: live_diff_max_days: int = DEFAULT_LIVE_DIFF_MAX_DAYS live_only_window_days: int = DEFAULT_LIVE_ONLY_WINDOW_DAYS rate_limit_wait: float = DEFAULT_RATE_LIMIT_WAIT + frame_cache_max_entries: int = DEFAULT_FRAME_CACHE_MAX_ENTRIES user_agent: str = field(default_factory=lambda: _default_user_agent()) def __post_init__(self) -> None: @@ -151,6 +165,8 @@ def __post_init__(self) -> None: raise ConfigError("live_diff_max_days / live_only_window_days must be >= 0") if self.rate_limit_wait < 0: raise ConfigError("rate_limit_wait must be >= 0") + if self.frame_cache_max_entries < 0: + raise ConfigError("frame_cache_max_entries must be >= 0 (0 disables the cache)") for phase, value in ( ("connect_timeout", self.connect_timeout), ("read_timeout", self.read_timeout), @@ -202,6 +218,7 @@ def from_env( live_diff_max_days: int | None = None, live_only_window_days: int | None = None, rate_limit_wait: float | None = None, + frame_cache_max_entries: int | None = None, ) -> ClientConfig: resolved_key = api_key or os.environ.get(_ENV_API_KEY) or "" if not resolved_key: @@ -269,6 +286,11 @@ def from_env( rate_limit_wait=_float_env( _ENV_RATE_LIMIT_WAIT, rate_limit_wait, DEFAULT_RATE_LIMIT_WAIT ), + frame_cache_max_entries=_int_env( + _ENV_FRAME_CACHE_MAX_ENTRIES, + frame_cache_max_entries, + DEFAULT_FRAME_CACHE_MAX_ENTRIES, + ), ) @@ -329,6 +351,7 @@ def _bool_env(env_var: str, override: bool | None, default: bool) -> bool: "DEFAULT_CACHE_MAX_BYTES", "DEFAULT_CONNECT_TIMEOUT", "DEFAULT_DOWNLOAD_CONCURRENCY", + "DEFAULT_FRAME_CACHE_MAX_ENTRIES", "DEFAULT_LIVE_DIFF_MAX_DAYS", "DEFAULT_LIVE_ONLY_WINDOW_DAYS", "DEFAULT_MAX_CONNECTIONS", diff --git a/src/rwpytools/session.py b/src/rwpytools/session.py index 4beec32..fe1b5b4 100644 --- a/src/rwpytools/session.py +++ b/src/rwpytools/session.py @@ -23,6 +23,7 @@ import dataclasses import datetime as _dt import logging +from collections import OrderedDict from collections.abc import Callable, Sequence from typing import Any @@ -54,6 +55,15 @@ #: itself is behind, and refreshing again would change nothing. _NO_REFRESH = 10**6 +#: ``(path, size, date_column, start, end, symbol_column, symbols)`` -- +#: identifies one ``_read_object`` result. ``(path, size)`` stands in for the +#: on-disk export's identity: it's the same pair ``BulkClient.fetch_one`` +#: already re-validates a cache entry against (a changed size means a +#: changed export), so a stale entry can never outlive the file it read. +_FrameCacheKey = tuple[ + str, int, str | None, _dt.date | None, _dt.date | None, str | None, tuple[str, ...] | None +] + class AsyncSession: """The auto-routing engine shared by ``Client`` and ``AsyncClient``.""" @@ -70,6 +80,13 @@ def __init__( self._bulk = bulk self._live = live self._today = today or _dt.date.today + # LRU of _read_object's own parsed-and-filtered results, keyed by + # _FrameCacheKey. See _read_object's own docstring for why this + # exists. Plain dict + move_to_end is safe under asyncio's + # single-threaded interleaving: a lookup/store never spans an + # ``await``, so two concurrent calls racing on the same key can at + # worst both miss and both compute (redundant, not incorrect). + self._frame_cache: OrderedDict[_FrameCacheKey, pd.DataFrame] = OrderedDict() # ---- planning ----------------------------------------------------------- @@ -408,7 +425,50 @@ async def _read_object( symbols: list[str], per_symbol_files: bool, ) -> pd.DataFrame: - return await asyncio.to_thread( + """Read one cached export, filtered to ``window``/``symbols``. + + For a bulk dataset without ``Template.YEAR`` (a single large export + spanning its whole date range, e.g. ``liquid_universe_prices``), + every call here re-reads and re-filters the *entire* on-disk export + via ``formats.read_frame`` -- there's no server-side or file-level + way to narrow what gets read for a non-templated export. A caller + that repeats an identical request (retrying after a partial + failure, or a caller that iterates windows itself specifically to + bound its own memory -- e.g. one calendar year at a time) pays that + full-export cost again on every repeat, in the same process, for a + result that hasn't changed. ``self._frame_cache`` (an LRU, size set + by ``ClientConfig.frame_cache_max_entries``) returns the identical + result instead of re-reading. Keyed by the export's own on-disk + identity (``path``, ``size``) plus the request shape, so a refreshed + export (new size) or a genuinely different request always misses. + """ + + effective_symbols = None if per_symbol_files else (symbols or None) + cache_key: _FrameCacheKey | None = None + if self._config.frame_cache_max_entries > 0: + cache_key = ( + str(result.path), + result.size, + spec.date_column, + window.start, + window.end, + spec.symbol_column, + tuple(sorted(effective_symbols)) if effective_symbols else None, + ) + cached = self._frame_cache.get(cache_key) + if cached is not None: + self._frame_cache.move_to_end(cache_key) + # Never hand back the cached object itself -- a caller that + # mutates its result in place (session.py's own _conform/ + # _merge do, on the *live* half, not this bulk half, but a + # third-party caller of Client.get() might) must not corrupt + # what every future cache hit returns. A shallow copy is + # enough: it's a new DataFrame wrapper over the same + # (immutable-in-practice) underlying column arrays, so this + # costs far less than re-reading the export. + return cached.copy(deep=False) + + frame = await asyncio.to_thread( formats.read_frame, result.path, object_name=result.object_name, @@ -416,8 +476,14 @@ async def _read_object( start=window.start, end=window.end, symbol_column=spec.symbol_column, - symbols=None if per_symbol_files else (symbols or None), + symbols=effective_symbols, ) + if cache_key is not None: + self._frame_cache[cache_key] = frame + self._frame_cache.move_to_end(cache_key) + while len(self._frame_cache) > self._config.frame_cache_max_entries: + self._frame_cache.popitem(last=False) + return frame async def _read_live( self, diff --git a/tests/test_session.py b/tests/test_session.py index 809bb96..239adc9 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -183,6 +183,90 @@ def test_cached_export_is_used_without_any_api_call( assert len(again) == 6 +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_repeated_identical_request_reads_the_export_only_once( + respx_mock: respx.MockRouter, client: Client, monkeypatch: pytest.MonkeyPatch +) -> None: + """A caller repeating an identical (dataset, window, symbols) request -- + retrying after a partial failure, or iterating windows itself to bound + its own memory -- must not pay to re-read and re-filter the same + on-disk export twice in one process. This is a separate cost from the + signed-URL/API call `test_cached_export_is_used_without_any_api_call` + already covers: that test shows the *network* call isn't repeated; this + one shows the *parse* isn't either.""" + import rwpytools.formats as formats_module + + _mock_vix(respx_mock) + real_read_frame = formats_module.read_frame + calls: list[Any] = [] + + def _counting_read_frame(*args: Any, **kwargs: Any) -> pd.DataFrame: + calls.append((args, kwargs)) + return real_read_frame(*args, **kwargs) + + monkeypatch.setattr(formats_module, "read_frame", _counting_read_frame) + + first = client.get("vix", source="bulk") + second = client.get("vix", source="bulk") + + assert len(calls) == 1 + pd.testing.assert_frame_equal(first.df, second.df) + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_frame_cache_discriminates_by_request_shape( + respx_mock: respx.MockRouter, client: Client, monkeypatch: pytest.MonkeyPatch +) -> None: + """A genuinely different request (different symbols here) must still + read/filter fresh -- the cache must never return another request's + result.""" + import rwpytools.formats as formats_module + + _mock_vix(respx_mock) + real_read_frame = formats_module.read_frame + calls: list[Any] = [] + + def _counting_read_frame(*args: Any, **kwargs: Any) -> pd.DataFrame: + calls.append(kwargs) + return real_read_frame(*args, **kwargs) + + monkeypatch.setattr(formats_module, "read_frame", _counting_read_frame) + + vix_only = client.get("vix", source="bulk", symbols=["VIX"]) + vix3m_only = client.get("vix", source="bulk", symbols=["VIX3M"]) + + assert len(calls) == 2 + assert set(vix_only.df["ticker"]) == {"VIX"} + assert set(vix3m_only.df["ticker"]) == {"VIX3M"} + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_frame_cache_max_entries_zero_disables_it( + respx_mock: respx.MockRouter, config: ClientConfig, monkeypatch: pytest.MonkeyPatch +) -> None: + import rwpytools.formats as formats_module + + _mock_vix(respx_mock) + real_read_frame = formats_module.read_frame + calls: list[Any] = [] + + def _counting_read_frame(*args: Any, **kwargs: Any) -> pd.DataFrame: + calls.append(kwargs) + return real_read_frame(*args, **kwargs) + + monkeypatch.setattr(formats_module, "read_frame", _counting_read_frame) + + disabled = Client(config=config.with_overrides(frame_cache_max_entries=0)) + disabled._stack.session._today = lambda: TODAY + try: + disabled.get("vix", source="bulk") + disabled.get("vix", source="bulk") + finally: + disabled.close() + + assert len(calls) == 2 # no caching at all: read fresh every time, as before this feature + + @respx.mock(base_url=FAKE_API, assert_all_called=False) def test_source_bulk_never_calls_live(respx_mock: respx.MockRouter, client: Client) -> None: _files, live = _mock_vix(respx_mock)