mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 21:12:16 +00:00
869cfed125
* merge branch develop with main * bump desktop version * fix the merge * sync Cargo.lock to desktop 2.0.0 * fix cli conftest.py --------- Co-authored-by: deeleeramone <>
335 lines
8.0 KiB
Python
335 lines
8.0 KiB
Python
"""Typed access to the FRED API, through the throttled and cached transport."""
|
|
|
|
from typing import Any
|
|
|
|
ROOT_URL = "https://api.stlouisfed.org/fred"
|
|
GEO_ROOT_URL = "https://api.stlouisfed.org/geofred"
|
|
|
|
|
|
def build_url(
|
|
path: str, api_key: str | None, *, root: str = ROOT_URL, **params: Any
|
|
) -> str:
|
|
"""Build a FRED request URL.
|
|
|
|
Parameters
|
|
----------
|
|
path : str
|
|
The endpoint path, relative to the API root.
|
|
api_key : str or None
|
|
The FRED API key.
|
|
root : str
|
|
The API root the path is read from.
|
|
**params : Any
|
|
Query parameters, with None values dropped.
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
The signed request URL.
|
|
"""
|
|
from urllib.parse import urlencode
|
|
|
|
query = {k: v for k, v in params.items() if v is not None}
|
|
query["api_key"] = api_key or ""
|
|
query["file_type"] = "json"
|
|
|
|
return f"{root}/{path.lstrip('/')}?{urlencode(query, doseq=True)}"
|
|
|
|
|
|
def headers(api_key: str | None = None) -> dict:
|
|
"""Return the headers a FRED request carries.
|
|
|
|
Parameters
|
|
----------
|
|
api_key : str or None
|
|
A key to send as a bearer token, for the endpoints that read one from
|
|
the header rather than the query string.
|
|
|
|
Returns
|
|
-------
|
|
dict
|
|
The request headers.
|
|
"""
|
|
from aiohttp.http import SERVER_SOFTWARE
|
|
|
|
sent = {
|
|
"Accept": "application/json",
|
|
"Accept-Encoding": "gzip, deflate",
|
|
"Connection": "keep-alive",
|
|
"User-Agent": SERVER_SOFTWARE,
|
|
}
|
|
|
|
if api_key is not None:
|
|
sent["Authorization"] = f"Bearer {api_key or ''}"
|
|
|
|
return sent
|
|
|
|
|
|
def _as_date(value: Any) -> Any:
|
|
"""Return a date as ``YYYY-MM-DD``, and anything else unchanged."""
|
|
return value.strftime("%Y-%m-%d") if hasattr(value, "strftime") else value or None
|
|
|
|
|
|
def observations_url(
|
|
series_id: str,
|
|
api_key: str | None,
|
|
start_date=None,
|
|
end_date=None,
|
|
**params: Any,
|
|
) -> str:
|
|
"""Build the observations URL for one FRED series.
|
|
|
|
Parameters
|
|
----------
|
|
series_id : str
|
|
The FRED series id, such as 'IORB'.
|
|
api_key : str or None
|
|
The FRED API key.
|
|
start_date : date or None
|
|
The earliest observation to return.
|
|
end_date : date or None
|
|
The latest observation to return.
|
|
**params : Any
|
|
Any further parameters the endpoint accepts.
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
The signed request URL.
|
|
"""
|
|
params.pop("preferences", None)
|
|
params.pop("use_cache", None)
|
|
|
|
return build_url(
|
|
"series/observations",
|
|
api_key,
|
|
series_id=series_id,
|
|
observation_start=_as_date(start_date),
|
|
observation_end=_as_date(end_date),
|
|
**params,
|
|
)
|
|
|
|
|
|
def observation_dates(value: Any) -> list:
|
|
"""Return the period-start dates a release table is asked for.
|
|
|
|
Parameters
|
|
----------
|
|
value : Any
|
|
The ``date`` query parameter, in any shape it accepts.
|
|
|
|
Returns
|
|
-------
|
|
list
|
|
One entry per requested period, or ``[None]`` for the latest.
|
|
"""
|
|
from openbb_fred.utils.query import join_dates
|
|
|
|
joined = join_dates(value)
|
|
|
|
if not joined:
|
|
return [None]
|
|
|
|
return sorted({f"{entry[:-2]}01" for entry in joined.split(",")})
|
|
|
|
|
|
def release_tables_url(
|
|
release_id: Any,
|
|
element_id: Any,
|
|
api_key: str | None,
|
|
observation_date: str | None = None,
|
|
) -> str:
|
|
"""Build the release tables URL for one element of a release.
|
|
|
|
Parameters
|
|
----------
|
|
release_id : Any
|
|
The FRED release id.
|
|
element_id : Any
|
|
The element within the release, or None for the whole release.
|
|
api_key : str or None
|
|
The FRED API key.
|
|
observation_date : str or None
|
|
The period to read, or None for the latest.
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
The signed request URL.
|
|
"""
|
|
return build_url(
|
|
"release/tables",
|
|
api_key,
|
|
release_id=release_id,
|
|
element_id=element_id,
|
|
observation_date=observation_date,
|
|
include_observation_values="true",
|
|
)
|
|
|
|
|
|
def published_value(value: Any) -> float | None:
|
|
"""Return one observation as a number.
|
|
|
|
Parameters
|
|
----------
|
|
value : Any
|
|
The observation as FRED wrote it.
|
|
|
|
Returns
|
|
-------
|
|
float or None
|
|
The value, or None on a date FRED published nothing for.
|
|
"""
|
|
if value in (".", "", None):
|
|
return None
|
|
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _observations(payload: Any, series_id: str) -> list[dict]:
|
|
"""Return the observations carried by one response.
|
|
|
|
Parameters
|
|
----------
|
|
payload : Any
|
|
The response the endpoint returned.
|
|
series_id : str
|
|
The series the response was asked for.
|
|
|
|
Returns
|
|
-------
|
|
list[dict]
|
|
One entry per observation.
|
|
|
|
Raises
|
|
------
|
|
OpenBBError
|
|
If FRED reports an error, or the series has no observations.
|
|
"""
|
|
from openbb_core.app.model.abstract.error import OpenBBError
|
|
|
|
if not payload:
|
|
raise OpenBBError(f"No data exists for series id: {series_id}")
|
|
|
|
if isinstance(payload, dict) and payload.get("error_message"):
|
|
raise OpenBBError(payload["error_message"])
|
|
|
|
return payload.get("observations", [])
|
|
|
|
|
|
async def get_observations(
|
|
series_id: str,
|
|
api_key: str | None,
|
|
start_date=None,
|
|
end_date=None,
|
|
use_cache: bool = True,
|
|
**params: Any,
|
|
) -> list[dict]:
|
|
"""Get the observations for one FRED series.
|
|
|
|
Parameters
|
|
----------
|
|
series_id : str
|
|
The FRED series id, such as 'IORB'.
|
|
api_key : str or None
|
|
The FRED API key.
|
|
start_date : date or None
|
|
The earliest observation to return.
|
|
end_date : date or None
|
|
The latest observation to return.
|
|
use_cache : bool
|
|
Whether to read and write the response cache.
|
|
**params : Any
|
|
Any further parameters the endpoint accepts.
|
|
|
|
Returns
|
|
-------
|
|
list[dict]
|
|
One entry per observation.
|
|
|
|
Raises
|
|
------
|
|
OpenBBError
|
|
If FRED reports an error, or the series has no observations.
|
|
"""
|
|
from openbb_fred.utils.rate_limiter import fred_get
|
|
|
|
url = observations_url(series_id, api_key, start_date, end_date, **params)
|
|
|
|
return _observations(await fred_get(url, use_cache=use_cache), series_id)
|
|
|
|
|
|
async def get_observations_many(
|
|
series_ids: "list[str]",
|
|
api_key: str | None,
|
|
start_date=None,
|
|
end_date=None,
|
|
use_cache: bool = True,
|
|
**params: Any,
|
|
) -> "list[list[dict]]":
|
|
"""Get the observations for many FRED series, in the order asked for.
|
|
|
|
Parameters
|
|
----------
|
|
series_ids : list[str]
|
|
The FRED series ids to read.
|
|
api_key : str or None
|
|
The FRED API key.
|
|
start_date : date or None
|
|
The earliest observation to return.
|
|
end_date : date or None
|
|
The latest observation to return.
|
|
use_cache : bool
|
|
Whether to read and write the response cache.
|
|
**params : Any
|
|
Any further parameters the endpoint accepts.
|
|
|
|
Returns
|
|
-------
|
|
list[list[dict]]
|
|
The observations for each series, aligned with ``series_ids``.
|
|
|
|
Raises
|
|
------
|
|
OpenBBError
|
|
If FRED reports an error, or a series has no observations.
|
|
"""
|
|
from openbb_fred.utils.rate_limiter import fred_get_many
|
|
|
|
urls = [
|
|
observations_url(series_id, api_key, start_date, end_date, **params)
|
|
for series_id in series_ids
|
|
]
|
|
payloads = await fred_get_many(urls, use_cache=use_cache)
|
|
|
|
return [
|
|
_observations(payload, series_id)
|
|
for payload, series_id in zip(payloads, series_ids)
|
|
]
|
|
|
|
|
|
def unwrap_series(
|
|
response: Any,
|
|
) -> "tuple[list, dict]":
|
|
"""Split a series fetch into its rows and its metadata.
|
|
|
|
Parameters
|
|
----------
|
|
response : Any
|
|
An ``AnnotatedResult``, or the rows on their own.
|
|
|
|
Returns
|
|
-------
|
|
tuple[list, dict]
|
|
The rows, and the metadata keyed by series id.
|
|
"""
|
|
result = getattr(response, "result", None)
|
|
|
|
if result is None:
|
|
return list(response or []), {}
|
|
|
|
return list(result or []), getattr(response, "metadata", None) or {}
|