[BugFix] openbb-fred: Fix 429 Error From FRED API & Implement Global Rate Limiter For Provider (#7527)

* fix 429 error from fred api due to change in rate limit calculation

* mypy

* black

* pylint

* pin aiohttp in devtools so vcrpy doesn't break

openbb_platform/providers/fred/openbb_fred/models/economic_calendar.py

@@ -56,7 +56,7 @@ class FredEconomicCalendarFetcher(
         from io import StringIO
         from math import ceil
 
-        from openbb_core.provider.utils.helpers import amake_request
+        from openbb_fred.utils.rate_limiter import fred_get
         from pandas import (
             DataFrame,
             concat,
@@ -137,13 +137,8 @@ class FredEconomicCalendarFetcher(
         ) -> DataFrame:
             """Fetch all pages of the FRED releases calendar for a single date or date range."""
             end_date = end_date or start_date
-            base_url = (
-                "https://fred.stlouisfed.org/releases/calendar"
-                f"?po=1&ptic=0&vs={start_date}&ve={end_date}&rid={rid}"
-            )
-            first_page = await amake_request(
-                base_url, response_callback=response_callback
-            )
+            base_url = f"https://fred.stlouisfed.org/releases/calendar?po=1&ptic=0&vs={start_date}&ve={end_date}&rid={rid}"
+            first_page = await fred_get(base_url, response_callback=response_callback)
             n_results = first_page.get("ptic", 0)  # type: ignore
 
             if n_results == 0:
@@ -161,7 +156,7 @@ class FredEconomicCalendarFetcher(
             ]
 
             async def _fetch_page(url: str):
-                return await amake_request(url, response_callback=response_callback)
+                return await fred_get(url, response_callback=response_callback)
 
             responses = await asyncio.gather(*[_fetch_page(u) for u in remaining_urls])
 
@@ -222,7 +217,8 @@ class FredEconomicCalendarFetcher(
         try:
             all_results = await fetch_all_pages(start_date, end_date, rid)
         except Exception as e:
-            raise OpenBBError(e) from e
+            message = str(e) or f"FRED request failed ({type(e).__name__})."
+            raise OpenBBError(message) from e
 
         if all_results.empty:
             raise OpenBBError(

openbb_platform/providers/fred/openbb_fred/models/high_quality_market.py

@@ -76,7 +76,7 @@ class FredHighQualityMarketCorporateBondFetcher(
         # pylint: disable=import-outside-toplevel
         import asyncio  # noqa
         from dateutil import parser  # noqa
-        from openbb_core.provider.utils.helpers import amake_request  # noqa
+        from openbb_fred.utils.rate_limiter import fred_get  # noqa
 
         api_key = credentials.get("fred_api_key") if credentials else ""
 
@@ -99,7 +99,7 @@ class FredHighQualityMarketCorporateBondFetcher(
 
         async def get_one(URL):
             """Get the observations for a single date."""
-            data = await amake_request(URL)
+            data = await fred_get(URL)
             if data:
                 elements = dict(data.get("elements", {}).items())  # type: ignore
                 for k, v in elements.items():  # pylint: disable=W0612

openbb_platform/providers/fred/openbb_fred/models/manufacturing_outlook_ny.py

@@ -489,7 +489,8 @@ class FredManufacturingOutlookNYFetcher(
                 credentials,
             )
         except Exception as e:
-            raise OpenBBError(e) from e
+            message = str(e) or f"FRED request failed ({type(e).__name__})."
+            raise OpenBBError(message) from e
 
         return {
             "metadata": response.metadata,

openbb_platform/providers/fred/openbb_fred/models/non_farm_payrolls.py

@@ -278,7 +278,7 @@ class FredNonFarmPayrollsFetcher(
         """Extract data."""
         # pylint: disable=import-outside-toplevel
         import asyncio  # noqa
-        from openbb_core.provider.utils.helpers import amake_request
+        from openbb_fred.utils.rate_limiter import fred_get
         from numpy import nan
         from pandas import DataFrame, to_datetime
 
@@ -304,7 +304,7 @@ class FredNonFarmPayrollsFetcher(
 
         async def get_one(URL):
             """Get the observations for a single date."""
-            response = await amake_request(URL)
+            response = await fred_get(URL)
             data = [v for v in response.get("elements", {}).values() if v.get("observation_value") != "."]  # type: ignore
             if data:
                 df = (

openbb_platform/providers/fred/openbb_fred/models/personal_consumption_expenditures.py

@@ -122,7 +122,7 @@ class FredPersonalConsumptionExpendituresFetcher(
         """Extract data."""
         # pylint: disable=import-outside-toplevel
         import asyncio  # noqa
-        from openbb_core.provider.utils.helpers import amake_request
+        from openbb_fred.utils.rate_limiter import fred_get
         from numpy import nan
         from pandas import DataFrame, to_datetime
 
@@ -148,7 +148,7 @@ class FredPersonalConsumptionExpendituresFetcher(
 
         async def get_one(URL):
             """Get the observations for a single date."""
-            response = await amake_request(URL)
+            response = await fred_get(URL)
             data = [
                 v
                 for v in response.get("elements", {}).values()  # type: ignore

openbb_platform/providers/fred/openbb_fred/models/regional.py

@@ -219,10 +219,8 @@ class FredRegionalDataFetcher(
     ) -> dict:
         """Extract the raw data."""
         # pylint: disable=import-outside-toplevel
-        from openbb_core.provider.utils.helpers import (
-            amake_request,
-            get_querystring,
-        )
+        from openbb_core.provider.utils.helpers import get_querystring
+        from openbb_fred.utils.rate_limiter import fred_get
 
         api_key = credentials.get("fred_api_key") if credentials else ""
         season = query.season.upper()
@@ -254,7 +252,7 @@ class FredRegionalDataFetcher(
                 )
                 + f"&file_type=json&api_key={api_key}"
             )
-        return await amake_request(url)  # type: ignore
+        return await fred_get(url)  # type: ignore
 
     @staticmethod
     def transform_data(

openbb_platform/providers/fred/openbb_fred/models/release_table.py

@@ -77,7 +77,7 @@ class FredReleaseTableFetcher(
         """Extract data."""
         # pylint: disable=import-outside-toplevel
         import asyncio  # noqa
-        from openbb_core.provider.utils.helpers import amake_request
+        from openbb_fred.utils.rate_limiter import fred_get
         from openbb_fred.models.search import FredSearchFetcher
         from numpy import nan
         from pandas import DataFrame, to_datetime
@@ -132,7 +132,7 @@ class FredReleaseTableFetcher(
 
         async def get_one(URL):
             """Get the observations for a single date."""
-            response = await amake_request(URL)
+            response = await fred_get(URL)
 
             # If the response has no elements we return empty and try the next URL.
             # If all URLs return empty, it will raise in `transform_data`.

openbb_platform/providers/fred/openbb_fred/models/search.py

@@ -187,10 +187,8 @@ class FredSearchFetcher(
         """Extract the raw data."""
         # pylint: disable=import-outside-toplevel
         import asyncio  # noqa
-        from openbb_core.provider.utils.helpers import (
-            amake_request,
-            get_querystring,
-        )
+        from openbb_core.provider.utils.helpers import get_querystring
+        from openbb_fred.utils.rate_limiter import fred_get
 
         api_key = credentials.get("fred_api_key") if credentials else ""
 
@@ -201,7 +199,7 @@ class FredSearchFetcher(
                 """Get data for one series."""
                 data: dict = {}
                 url = f"https://api.stlouisfed.org/geofred/series/group?series_id={_id}&api_key={api_key}&file_type=json"
-                response = await amake_request(url)
+                response = await fred_get(url)
                 data = response.get("series_group")  # type: ignore
                 if data:
                     data.update({"series_id": _id})
@@ -215,7 +213,7 @@ class FredSearchFetcher(
 
         if query.search_type == "release" and query.release_id is None:
             url = f"https://api.stlouisfed.org/fred/releases?api_key={api_key}&file_type=json"
-            response = await amake_request(url)
+            response = await fred_get(url)
             results = response.get("releases")  # type: ignore
             if results:
                 return results
@@ -238,7 +236,7 @@ class FredSearchFetcher(
 
         querystring = get_querystring(query.model_dump(), exclude).replace(" ", "%20")
         url = url + querystring + f"&file_type=json&api_key={api_key}"
-        response = await amake_request(url)
+        response = await fred_get(url)
 
         if isinstance(response, dict) and "error_code" in response:
             raise OpenBBError(

openbb_platform/providers/fred/openbb_fred/models/series.py

@@ -117,12 +117,10 @@ class FredSeriesFetcher(
     ) -> list[dict]:
         """Extract data."""
         # pylint: disable=import-outside-toplevel
-        from openbb_core.provider.utils.helpers import (
-            ClientResponse,
-            ClientSession,
-            amake_requests,
-            get_querystring,
-        )
+        import asyncio
+
+        from openbb_core.provider.utils.helpers import get_querystring
+        from openbb_fred.utils.rate_limiter import fred_get
         from pandas import DataFrame
 
         api_key = credentials.get("fred_api_key") if credentials else ""
@@ -133,21 +131,15 @@ class FredSeriesFetcher(
         querystring = get_querystring(query.model_dump(), ["series_id"])
         series_ids = query.symbol.split(",") if "," in query.symbol else [query.symbol]
 
-        urls = [
-            f"{base_url}?series_id={series_id}&{querystring}&file_type=json&api_key={api_key}"
-            for series_id in series_ids
-        ]
-
-        async def callback(response: ClientResponse, session: ClientSession) -> dict:
-            observations_response = await response.json()
-            series_id = response.url.query.get("series_id")
-
-            metadata_response = await session.get_json(
-                f"{metadata_url}?series_id={series_id}&file_type=json&api_key={api_key}",
-                timeout=5,
+        async def fetch_one(series_id: str) -> dict:
+            obs_url = f"{base_url}?series_id={series_id}&{querystring}&file_type=json&api_key={api_key}"
+            meta_url = (
+                f"{metadata_url}?series_id={series_id}&file_type=json&api_key={api_key}"
             )
 
-            # seriess is not a typo, it's the actual key in the response
+            observations_response = await fred_get(obs_url, timeout=5, **kwargs)
+            metadata_response = await fred_get(meta_url, timeout=5, **kwargs)
+
             _metadata = (
                 metadata_response.get("seriess", [{}])[0]
                 if isinstance(metadata_response, dict)
@@ -158,6 +150,7 @@ class FredSeriesFetcher(
                 if isinstance(observations_response, dict)
                 else []
             ) or []
+
             try:
                 for d in observations:
                     d.pop("realtime_start")
@@ -186,12 +179,20 @@ class FredSeriesFetcher(
             }
 
         try:
-            results = await amake_requests(
-                urls, response_callback=callback, timeout=5, **kwargs
-            )
+            results: list[dict] = []
+            for result in await asyncio.gather(
+                *[fetch_one(sid) for sid in series_ids], return_exceptions=True
+            ):
+                if isinstance(result, Exception):
+                    raise result
+                if result:
+                    results.append(result)  # type: ignore
             return results
+        except OpenBBError:
+            raise
         except Exception as e:
-            raise OpenBBError(e) from e
+            message = str(e) or f"FRED request failed ({type(e).__name__})."
+            raise OpenBBError(message) from e
 
     @staticmethod
     def transform_data(

openbb_platform/providers/fred/openbb_fred/models/tips_yields.py

@@ -158,7 +158,8 @@ class FredTipsYieldsFetcher(
             ids_df = await get_tips_series()
             ids = ids_df.series_id.to_list()
         except Exception as e:
-            raise OpenBBError(e) from e
+            message = str(e) or f"FRED request failed ({type(e).__name__})."
+            raise OpenBBError(message) from e
 
         # If we are looking for a specific tenor, the request will be smaller.
         if query.maturity:
@@ -197,7 +198,8 @@ class FredTipsYieldsFetcher(
             df = DataFrame([d.model_dump() for d in res.result])  # type: ignore
             meta: dict = res.metadata or {}  # type: ignore
         except Exception as e:
-            raise OpenBBError(e) from e
+            message = str(e) or f"FRED request failed ({type(e).__name__})."
+            raise OpenBBError(message) from e
 
         for k, v in title_map.items():
             if k in meta:

openbb_platform/providers/fred/openbb_fred/utils/rate_limiter.py

@@ -0,0 +1,315 @@
+"""Process-wide FRED API rate limiter, in-memory cache, and request helper.
+
+Every FRED API request in this provider should be routed through `fred_get`.
+The helper provides:
+
+* A global minimum-interval throttle (default 600 ms ≈ 100 req/min), well
+  below FRED's 120 req/min ceiling.
+* Auto-retry on HTTP 429 with `Retry-After` / exponential backoff, so
+  transient throttling does not surface as an error.
+* An in-memory TTL LRU cache so that repeated identical requests (typical
+  when a user refreshes or re-opens widgets) reuse the prior response.
+* Single-flight de-duplication: concurrent callers asking for the same URL
+  share a single in-flight request instead of fanning out duplicates.
+"""
+
+# pylint: disable=import-outside-toplevel
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import os
+import re
+import threading
+import time
+from collections import OrderedDict
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from openbb_core.app.model.abstract.error import OpenBBError
+
+
+def _float_env(name: str, default: float) -> float:
+    raw = os.environ.get(name)
+    if not raw:
+        return default
+    try:
+        return float(raw)
+    except ValueError:
+        return default
+
+
+def _int_env(name: str, default: int) -> int:
+    raw = os.environ.get(name)
+    if not raw:
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        return default
+
+
+MIN_INTERVAL_SECONDS = _float_env("OPENBB_FRED_MIN_INTERVAL", 0.6)
+MAX_RETRIES = _int_env("OPENBB_FRED_MAX_RETRIES", 4)
+DEFAULT_BACKOFF_SECONDS = _float_env("OPENBB_FRED_BACKOFF", 2.0)
+MAX_BACKOFF_SECONDS = _float_env("OPENBB_FRED_MAX_BACKOFF", 30.0)
+CACHE_TTL_SECONDS = _float_env("OPENBB_FRED_CACHE_TTL", 300.0)
+CACHE_MAX_ENTRIES = _int_env("OPENBB_FRED_CACHE_SIZE", 512)
+
+_RATE_LIMIT_MESSAGE = (
+    "FRED API rate limit exceeded (HTTP 429). Please wait a moment and try again."
+)
+
+_throttle_lock = threading.Lock()
+_last_request_at: float = 0.0
+
+_cache_lock = threading.Lock()
+_cache: OrderedDict[str, tuple[float, Any]] = OrderedDict()
+
+_inflight_lock = threading.Lock()
+_inflight: dict[tuple[int, str], asyncio.Future] = {}
+
+_API_KEY_RE = re.compile(r"([?&])api_key=[^&]*")
+
+
+class _RateLimitedResponse(Exception):
+    """Internal sentinel raised when a 429 is observed on a FRED response."""
+
+    def __init__(self, retry_after: float | None = None) -> None:
+        super().__init__(_RATE_LIMIT_MESSAGE)
+        self.retry_after = retry_after
+
+
+def _cache_key(url: str) -> str:
+    """Build a cache key from a URL, stripping the API key."""
+    return _API_KEY_RE.sub(r"\1api_key=__redacted__", url)
+
+
+def _cache_get(key: str) -> Any:
+    """Return a cached value if present and not expired, else None.
+
+    Returns a deep copy so downstream mutation of dicts / lists does not
+    corrupt subsequent cache hits.
+    """
+    if CACHE_TTL_SECONDS <= 0 or CACHE_MAX_ENTRIES <= 0:
+        return None
+    with _cache_lock:
+        entry = _cache.get(key)
+        if entry is None:
+            return None
+        stored_at, value = entry
+        if time.monotonic() - stored_at > CACHE_TTL_SECONDS:
+            _cache.pop(key, None)
+            return None
+        _cache.move_to_end(key)
+        cached_value = value
+    return copy.deepcopy(cached_value)
+
+
+def _cache_set(key: str, value: Any) -> None:
+    """Insert a deep-copied value into the cache and evict oldest entries when full."""
+    if CACHE_TTL_SECONDS <= 0 or CACHE_MAX_ENTRIES <= 0:
+        return
+    snapshot = copy.deepcopy(value)
+    with _cache_lock:
+        _cache[key] = (time.monotonic(), snapshot)
+        _cache.move_to_end(key)
+        while len(_cache) > CACHE_MAX_ENTRIES:
+            _cache.popitem(last=False)
+
+
+def cache_clear() -> None:
+    """Clear the in-memory FRED response cache (useful for tests)."""
+    with _cache_lock:
+        _cache.clear()
+
+
+async def acquire() -> None:
+    """Enforce a minimum gap between successive FRED API requests."""
+    global _last_request_at  # noqa: PLW0603  # pylint: disable=global-statement
+    while True:
+        with _throttle_lock:
+            now = time.monotonic()
+            wait_for = _last_request_at + MIN_INTERVAL_SECONDS - now
+            if wait_for <= 0:
+                _last_request_at = now
+                return
+        await asyncio.sleep(wait_for)
+
+
+def _parse_retry_after(value: str | None) -> float | None:
+    if not value:
+        return None
+    try:
+        return max(float(value), 0.0)
+    except (TypeError, ValueError):
+        return None
+
+
+def _check_payload_rate_limit(payload: Any) -> None:
+    if not isinstance(payload, dict):
+        return
+    code = payload.get("error_code") or payload.get("status_code")
+    message = payload.get("error_message") or payload.get("message") or ""
+    if str(code) == "429" or "too many requests" in str(message).lower():
+        raise _RateLimitedResponse()
+
+
+async def _fetch(
+    url: str,
+    response_callback: Callable[[Any, Any], Awaitable[dict | list[dict]]] | None,
+    retries: int,
+    **kwargs: Any,
+) -> Any:
+    """Perform a single throttled fetch with retry-on-429."""
+    from openbb_core.provider.utils.helpers import amake_request
+
+    async def _wrapped_callback(response, session):
+        status = getattr(response, "status", None)
+        if status == 429:
+            retry_after = _parse_retry_after(
+                response.headers.get("Retry-After")
+                if hasattr(response, "headers")
+                else None
+            )
+            raise _RateLimitedResponse(retry_after)
+        if response_callback is not None:
+            return await response_callback(response, session)
+        payload = await response.json()
+        _check_payload_rate_limit(payload)
+        return payload
+
+    attempt = 0
+    while True:
+        await acquire()
+        try:
+            return await amake_request(
+                url, response_callback=_wrapped_callback, **kwargs
+            )
+        except _RateLimitedResponse as rl:
+            attempt += 1
+            if attempt > retries:
+                raise OpenBBError(ValueError(_RATE_LIMIT_MESSAGE)) from rl
+            backoff = (
+                rl.retry_after
+                if rl.retry_after is not None
+                else min(
+                    DEFAULT_BACKOFF_SECONDS * (2 ** (attempt - 1)),
+                    MAX_BACKOFF_SECONDS,
+                )
+            )
+            await asyncio.sleep(max(backoff, MIN_INTERVAL_SECONDS))
+
+
+def _is_cacheable(value: Any) -> bool:
+    """Don't cache empty / falsy payloads — they're usually transient errors."""
+    if value is None:
+        return False
+    return not (isinstance(value, (list, dict, str, bytes)) and len(value) == 0)
+
+
+async def fred_get(
+    url: str,
+    *,
+    response_callback: Callable[[Any, Any], Awaitable[dict | list[dict]]] | None = None,
+    max_retries: int | None = None,
+    use_cache: bool = True,
+    **kwargs: Any,
+) -> Any:
+    """Throttled, retrying, cached GET wrapper around `amake_request` for FRED.
+
+    Concurrent calls for the same URL share a single in-flight request so
+    duplicates collapse into one upstream API call.
+    """
+    retries = MAX_RETRIES if max_retries is None else max_retries
+    cache_key = _cache_key(url) if use_cache else None
+
+    if cache_key is not None:
+        cached = _cache_get(cache_key)
+        if cached is not None:
+            return cached
+
+    inflight_key: tuple[int, str] | None = None
+    future: asyncio.Future | None = None
+    owner = False
+    if cache_key is not None:
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        if loop is not None:
+            inflight_key = (id(loop), cache_key)
+            with _inflight_lock:
+                existing = _inflight.get(inflight_key)
+                if existing is not None:
+                    future = existing
+                else:
+                    future = loop.create_future()
+                    _inflight[inflight_key] = future
+                    owner = True
+
+    if future is not None and not owner:
+        shared = await future
+        return copy.deepcopy(shared)
+
+    try:
+        result = await _fetch(
+            url,
+            response_callback=response_callback,
+            retries=retries,
+            **kwargs,
+        )
+    except BaseException as exc:
+        if future is not None and owner and not future.done():
+            future.set_exception(exc)
+        raise
+    finally:
+        if inflight_key is not None and owner:
+            with _inflight_lock:
+                _inflight.pop(inflight_key, None)
+
+    if cache_key is not None and _is_cacheable(result):
+        _cache_set(cache_key, result)
+
+    if future is not None and owner and not future.done():
+        future.set_result(result)
+
+    return result
+
+
+async def fred_get_many(
+    urls: list[str],
+    *,
+    response_callback: Callable[[Any, Any], Awaitable[dict | list[dict]]] | None = None,
+    return_exceptions: bool = False,
+    use_cache: bool = True,
+    **kwargs: Any,
+) -> list[Any]:
+    """Issue many FRED requests under the global rate limit and cache."""
+    tasks = [
+        fred_get(
+            url,
+            response_callback=response_callback,
+            use_cache=use_cache,
+            **kwargs,
+        )
+        for url in urls
+    ]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    cleaned: list[Any] = []
+    first_error: Exception | None = None
+    for result in results:
+        if isinstance(result, Exception):
+            if return_exceptions:
+                cleaned.append(result)
+            elif first_error is None:
+                first_error = result
+            continue
+        cleaned.append(result)
+
+    if first_error is not None and not cleaned:
+        raise first_error
+
+    return cleaned

openbb_platform/providers/fred/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "openbb-fred"
-version = "1.6.1"
+version = "1.6.2"
 description = "FRED extension for OpenBB"
 authors = ["OpenBB Team <hello@openbb.co>"]
 license = "AGPL-3.0-only"