Bases: AioHttpClient
Federal Reserve Economic Data API client
Fetch economic data from FRED_.
.. _FRED: https://fred.stlouisfed.org/
url
class-attribute
instance-attribute
url = 'https://api.stlouisfed.org/fred'
key
class-attribute
instance-attribute
key = field(default_factory=lambda: get('FRED_API_KEY', ''))
freq
Bases: StrEnum
Fred historical frequencies
d
class-attribute
instance-attribute
w
class-attribute
instance-attribute
bw
class-attribute
instance-attribute
m
class-attribute
instance-attribute
q
class-attribute
instance-attribute
sa
class-attribute
instance-attribute
a
class-attribute
instance-attribute
categiories
async
Get categories
Source code in quantflow/data/fred.py
| async def categiories(self, **kw: Any) -> dict:
"""Get categories"""
return await self.get_path("category", **kw)
|
subcategories
async
Get subcategories of a given category
Source code in quantflow/data/fred.py
| async def subcategories(self, **kw: Any) -> dict:
"""Get subcategories of a given category"""
return await self.get_path("category/children", **kw)
|
series
async
Get series of a given category
Source code in quantflow/data/fred.py
| async def series(self, **kw: Any) -> dict:
"""Get series of a given category"""
return await self.get_path("category/series", **kw)
|
serie_data
async
serie_data(*, to_date=False, **kw)
Get series data frame
Source code in quantflow/data/fred.py
| async def serie_data(self, *, to_date: bool = False, **kw: Any) -> pd.DataFrame:
"""Get series data frame"""
data = await self.get_path("series/observations", **kw)
df = pd.DataFrame(data["observations"])
df["value"] = pd.to_numeric(df["value"])
if to_date and "date" in df.columns:
df["date"] = pd.to_datetime(df["date"])
return df
|
get_path
async
Source code in quantflow/data/fred.py
| async def get_path(self, path: str, **kw: Any) -> dict:
result = await self.get(f"{self.url}/{path}", **self.params(**kw))
return cast(dict, result)
|
params
params(params=None, **kw)
Source code in quantflow/data/fred.py
| def params(self, params: dict | None = None, **kw: Any) -> dict:
params = params.copy() if params is not None else {}
params.update(api_key=self.key, file_type="json")
return {"params": params, **kw}
|