Cache parsed-and-filtered get() results per session (mitigates OOM on repeated non-templated bulk reads)

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 |
 

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",

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,16 +425,65 @@ 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,
             date_column=spec.date_column,
             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,

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)