feat: Implement in-memory TTL cache for SEC EDGAR API responses and integrate with services

Author: areed1192

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **edgar/cache.py**: In-memory TTL cache for SEC EDGAR API responses.
+  - `TTLCache` class with `get()`, `set()`, `invalidate()`, `clear()`, `__len__()`, `__repr__()`.
+  - Uses `time.monotonic()` for expiration immune to wall-clock adjustments.
+  - Module-level TTL constants: `TTL_TICKERS` (24h), `TTL_TAXONOMY` (24h), `TTL_SUBMISSIONS` (1h).
+- **tests/test_cache.py**: 29 unit tests for `TTLCache` (get/set, expiration, invalidate, clear, len/repr, constants) and cache integration with `Tickers`, `Submissions`, and `Xbrl` services.
 - **edgar/\_\_init\_\_.py**: Top-level convenience functions for reduced boilerplate.
   - `edgar.company("AAPL")` — create a `Company` without instantiating `EdgarClient`.
   - `edgar.get_filings("AAPL", form="10-K")` — fetch filings in one call.
@@ -37,6 +42,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **edgar/client.py**: `EdgarClient` now accepts `rate_limit` parameter: `EdgarClient(user_agent="...", rate_limit=5)`.
+  - Defaults to 10 (SEC's maximum). Validates range 1–10.
+  - Passed through to `EdgarSession` for sliding-window enforcement.
+- **edgar/client.py**: `EdgarClient` now accepts `cache` parameter (`bool`, default `True`).
+  - `cache=True` creates a shared `TTLCache` passed to `EdgarSession`.
+  - `cache=False` disables caching; all requests hit the network.
+- **edgar/session.py**: `EdgarSession` accepts optional `cache` parameter storing a `TTLCache` instance.
+- **edgar/tickers.py**: `Tickers._load()` checks/stores data in the TTL cache (`TTL_TICKERS`).
+- **edgar/submissions.py**: `Submissions.get_submissions()` checks/stores responses in the TTL cache (`TTL_SUBMISSIONS`).
+- **edgar/xbrl.py**: `Xbrl.company_facts()` checks/stores responses in the TTL cache (`TTL_TAXONOMY`).
+- **tests/test_rate_limiter.py**: 8 new tests for configurable rate limit (custom values, boundary validation, client passthrough). Total: 17 tests.
 - **xbrl.py**: `company_concepts()` and `frames()` now accept an optional `taxonomy` parameter (default `"us-gaap"`). Previously hardcoded to `us-gaap`, now supports `"ifrs-full"`, `"dei"`, or any other taxonomy.
 - **edgar/tickers.py**: New `Tickers` service for ticker/CIK/company name resolution via `sec.gov/files/company_tickers.json`.
   - `resolve_ticker("AAPL")` → zero-padded CIK string (`"0000320193"`).

edgar/cache.py

@@ -0,0 +1,58 @@
+"""In-memory TTL cache for SEC EDGAR API responses."""
+
+from __future__ import annotations
+
+import time
+
+
+# Default TTLs in seconds.
+TTL_TICKERS = 86400       # 24 hours — ticker data changes ~quarterly
+TTL_TAXONOMY = 86400      # 24 hours — static reference data
+TTL_SUBMISSIONS = 3600    # 1 hour — changes on new filings
+
+
+class TTLCache:
+    """Simple in-memory cache with per-key time-to-live expiration.
+
+    Uses ``time.monotonic()`` so expiration is immune to wall-clock
+    adjustments.
+    """
+
+    def __init__(self) -> None:
+        self._store: dict[str, tuple[object, float]] = {}
+
+    def get(self, key: str) -> object | None:
+        """Return the cached value for *key*, or ``None`` if missing/expired."""
+
+        entry = self._store.get(key)
+        if entry is None:
+            return None
+        value, expires_at = entry
+        if time.monotonic() >= expires_at:
+            del self._store[key]
+            return None
+        return value
+
+    def set(self, key: str, value: object, ttl: float) -> None:
+        """Store *value* under *key* with a TTL of *ttl* seconds."""
+
+        self._store[key] = (value, time.monotonic() + ttl)
+
+    def invalidate(self, key: str) -> None:
+        """Remove a single key from the cache."""
+
+        self._store.pop(key, None)
+
+    def clear(self) -> None:
+        """Remove all entries from the cache."""
+
+        self._store.clear()
+
+    def __len__(self) -> int:
+        """Return the count of non-expired entries."""
+
+        now = time.monotonic()
+        return sum(1 for _, (_, exp) in self._store.items() if now < exp)
+
+    def __repr__(self) -> str:
+        return f"<TTLCache entries={len(self)}>"

edgar/client.py

@@ -1,5 +1,6 @@
 """Main entry-point client for the SEC EDGAR API."""
 
+from edgar.cache import TTLCache
 from edgar.xbrl import Xbrl
 from edgar.series import Series
 from edgar.search import Search
@@ -26,15 +27,36 @@ class EdgarClient:
     instantiate the different endpoints.
     """
 
-    def __init__(self, user_agent: str) -> None:
+    def __init__(self, user_agent: str, rate_limit: int = 10, cache: bool = True) -> None:
         """Initializes the `EdgarClient`.
 
+        ### Parameters
+        ----
+        user_agent : str
+            SEC EDGAR requires a User-Agent header in the format
+            ``"Company/Name email@example.com"``.
+
+        rate_limit : int (optional, Default=10)
+            Maximum requests per second. SEC allows 10 req/s.
+            Set lower to be more conservative.
+
+        cache : bool (optional, Default=True)
+            Enable in-memory TTL caching for ticker resolution (24h),
+            submission metadata (1h), and taxonomy data (24h).
+            Set ``False`` to always fetch fresh data from SEC.
+
         ### Usage
         ----
             >>> edgar_client = EdgarClient(user_agent="Your Name your-email@example.com")
+            >>> edgar_client = EdgarClient(user_agent="...", rate_limit=5)
+            >>> edgar_client = EdgarClient(user_agent="...", cache=False)
         """
 
-        self.edgar_session = EdgarSession(client=self, user_agent=user_agent)
+        self._ttl_cache = TTLCache() if cache else None
+        self.edgar_session = EdgarSession(
+            client=self, user_agent=user_agent, rate_limit=rate_limit,
+            cache=self._ttl_cache,
+        )
         self._services: dict = {}
 
     def __repr__(self) -> str:

edgar/session.py

@@ -33,27 +33,50 @@ class EdgarSession:
     handles all the requests made to EDGAR.
     """
 
-    def __init__(self, client: EdgarClient, user_agent: str) -> None:
+    def __init__(
+        self,
+        client: EdgarClient,
+        user_agent: str,
+        rate_limit: int = MAX_REQUESTS_PER_SECOND,
+        cache: object | None = None,
+    ) -> None:
         """Initializes the `EdgarSession` client.
 
         ### Parameters
         ----
-        client (str): The `edgar.EdgarClient` Python Client.
+        client : EdgarClient
+            The `edgar.EdgarClient` Python Client.
+
+        user_agent : str
+            SEC EDGAR User-Agent header value.
+
+        rate_limit : int (optional, Default=MAX_REQUESTS_PER_SECOND)
+            Maximum requests per second. Must be between 1 and 10.
+
+        cache : TTLCache | None (optional, Default=None)
+            Shared TTL cache instance. ``None`` disables caching.
 
         ### Usage
         ----
             >>> edgar_client = EdgarClient(user_agent="Your Name your-email@example.com")
             >>> edgar_session = EdgarSession(client=edgar_client, user_agent="your_user_agent")
         """
 
+        if not 1 <= rate_limit <= MAX_REQUESTS_PER_SECOND:
+            raise ValueError(
+                f"rate_limit must be between 1 and {MAX_REQUESTS_PER_SECOND}, "
+                f"got {rate_limit}"
+            )
+
         self.client: EdgarClient = client
         self.resource = "https://www.sec.gov"
         self.api_resource = "https://data.sec.gov"
         self.user_agent = user_agent
+        self.cache = cache
 
         # Sliding-window rate limiter: track timestamps of recent requests.
         self._request_times: deque[float] = deque()
-        self._rate_limit = MAX_REQUESTS_PER_SECOND
+        self._rate_limit = rate_limit
 
         # Create a single reusable session with connection pooling.
         self.http_session = requests.Session()

edgar/submissions.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from edgar.cache import TTL_SUBMISSIONS
 from edgar.session import EdgarSession
 
 
@@ -69,11 +70,23 @@ def get_submissions(self, cik: str) -> dict | None:
             num_of_zeros = 10 - len(cik)
             cik = num_of_zeros*"0" + cik
 
+        # Check TTL cache.
+        cache = self.edgar_session.cache
+        cache_key = f"submissions:{cik}"
+        if cache is not None:
+            cached = cache.get(cache_key)
+            if cached is not None:
+                return cached
+
         # Grab the Data.
         response = self.edgar_session.make_request(
             method='get',
             endpoint=f'/submissions/CIK{cik}.json',
             use_api=True
         )
 
+        # Store in TTL cache.
+        if cache is not None and response is not None:
+            cache.set(cache_key, response, TTL_SUBMISSIONS)
+
         return response

edgar/tickers.py

@@ -5,6 +5,7 @@
 import logging
 from typing import TYPE_CHECKING
 
+from edgar.cache import TTL_TICKERS
 from edgar.exceptions import EdgarRequestError
 
 if TYPE_CHECKING:
@@ -13,6 +14,7 @@
 logger = logging.getLogger(__name__)
 
 TICKERS_ENDPOINT = "/files/company_tickers.json"
+_CACHE_KEY = "tickers"
 
 
 class Tickers:
@@ -30,6 +32,14 @@ def _load(self) -> None:
         if self._data is not None:
             return
 
+        # Check TTL cache for previously fetched data.
+        cache = self._session.cache
+        if cache is not None:
+            cached = cache.get(_CACHE_KEY)
+            if cached is not None:
+                self._data, self._ticker_to_cik, self._cik_to_entries = cached
+                return
+
         raw = self._session.make_request(
             method="GET",
             endpoint=TICKERS_ENDPOINT,
@@ -53,6 +63,14 @@ def _load(self) -> None:
 
             self._cik_to_entries.setdefault(cik, []).append(entry)
 
+        # Store in TTL cache for reuse across service re-instantiations.
+        if cache is not None:
+            cache.set(
+                _CACHE_KEY,
+                (self._data, self._ticker_to_cik, self._cik_to_entries),
+                TTL_TICKERS,
+            )
+
     def resolve_ticker(self, ticker: str) -> str:
         """Resolves a ticker symbol to a zero-padded CIK string.
 

edgar/xbrl.py

@@ -4,6 +4,8 @@
 
 from enum import Enum
 from typing import Union
+
+from edgar.cache import TTL_TAXONOMY
 from edgar.models import Facts
 from edgar.session import EdgarSession
 
@@ -133,6 +135,14 @@ def company_facts(self, cik: str) -> dict | None:
             num_of_zeros = 10 - len(cik)
             cik = num_of_zeros*"0" + cik
 
+        # Check TTL cache.
+        cache = self.edgar_session.cache
+        cache_key = f"company_facts:{cik}"
+        if cache is not None:
+            cached = cache.get(cache_key)
+            if cached is not None:
+                return cached
+
         endpoint = f'/api/xbrl/companyfacts/CIK{cik}.json'
 
         # Grab the Data.
@@ -142,6 +152,10 @@ def company_facts(self, cik: str) -> dict | None:
             use_api=True
         )
 
+        # Store in TTL cache.
+        if cache is not None and response is not None:
+            cache.set(cache_key, response, TTL_TAXONOMY)
+
         return response
 
     def get_facts(self, cik: str) -> object: