From 5ae66d3c2e672f2e61a85cb22c0aed9b15af10a1 Mon Sep 17 00:00:00 2001 From: Theodore Aptekarev Date: Fri, 30 Jun 2023 11:54:33 +0200 Subject: [PATCH] Move fred provider --- .../builtin_providers/fred/cpi.py | 76 ++++++++++++++++ .../builtin_providers/fred/fred_base.py | 61 +++++++++++++ .../builtin_providers/fred/fred_helpers.py | 86 +++++++++++++++++++ .../builtin_providers/fred/provider.py | 16 ++++ .../openbb_provider/model/data/cpi.py | 60 +++++++++++++ 5 files changed, 299 insertions(+) create mode 100644 openbb_providers/builtin_providers/fred/cpi.py create mode 100644 openbb_providers/builtin_providers/fred/fred_base.py create mode 100644 openbb_providers/builtin_providers/fred/fred_helpers.py create mode 100644 openbb_providers/builtin_providers/fred/provider.py create mode 100644 openbb_providers/openbb_provider/model/data/cpi.py diff --git a/openbb_providers/builtin_providers/fred/cpi.py b/openbb_providers/builtin_providers/fred/cpi.py new file mode 100644 index 00000000000..5c17fe79f50 --- /dev/null +++ b/openbb_providers/builtin_providers/fred/cpi.py @@ -0,0 +1,76 @@ +"""FRED CPI Fetcher.""" + +# IMPORT STANDARD +from typing import Dict, List, Optional + +from builtin_providers.fred.fred_base import Fred +from builtin_providers.fred.fred_helpers import all_cpi_options + +# IMPORT THIRD-PARTY +# IMPORT INTERNAL +from openbb_provider.model.data.cpi import CPIData, CPIDataPoint, CPIQueryParams +from openbb_provider.provider.abstract.fetcher import Fetcher +from openbb_provider.provider.provider_helpers import data_transformer + + +class FREDCPIQueryParams(CPIQueryParams): + """CPI query. + When other provders are added, this will probably need less strict types + + Parameter + --------- + countries: List[CPI_COUNTRIES] + The country or countries you want to see. + units: List[CPI_UNITS] + The units you want to see, can be "growth_previous", "growth_same" or "index_2015". + frequency: List[CPI_FREQUENCY] + The frequency you want to see, either "annual", monthly" or "quarterly". + harmonized: bool + Whether you wish to obtain harmonized data. + start_date: Optional[date] + Start date, formatted YYYY-MM-DD + end_date: Optional[date] + End date, formatted YYYY-MM-DD + """ + + __name__ = "FREDCPIQueryParams" + + +class FREDCPIData(CPIData): + __name__ = "FREDCPIData" + + +class FREDCPIFetcher(Fetcher[CPIQueryParams, CPIData, FREDCPIQueryParams, FREDCPIData]): + @staticmethod + def transform_query( + query: CPIQueryParams, extra_params: Optional[Dict] = None + ) -> FREDCPIQueryParams: + return FREDCPIQueryParams( + **query.dict(), **extra_params if extra_params else {} + ) + + @staticmethod + def extract_data(query: FREDCPIQueryParams, api_key: str) -> FREDCPIData: + all_options = all_cpi_options(query.harmonized) + + step_1 = [x for x in all_options if x["country"] in query.countries] + step_2 = [x for x in step_1 if x["units"] == query.units] + step_3 = [x for x in step_2 if x["frequency"] == query.frequency] + + if not step_3: + FREDCPIData(data={}) + + series_dict = {} + + fred = Fred(api_key) + for item in step_3: + loc = f"{item['country']}-{item['frequency']}-{item['units']}" + temp = fred.get_series(item["series_id"], query.start_date, query.end_date) + clean_temp = [CPIDataPoint(**x) for x in temp] + series_dict[loc] = clean_temp + + return FREDCPIData(data=series_dict) + + @staticmethod + def transform_data(data: List[FREDCPIData]) -> List[CPIData]: + return data_transformer(data, CPIData) diff --git a/openbb_providers/builtin_providers/fred/fred_base.py b/openbb_providers/builtin_providers/fred/fred_base.py new file mode 100644 index 00000000000..dc3557ba280 --- /dev/null +++ b/openbb_providers/builtin_providers/fred/fred_base.py @@ -0,0 +1,61 @@ +# IMPORT STANDARD +from datetime import date +from typing import Optional +from urllib.parse import urlencode + +# IMPORT THIRD-PARTY +import requests + +# IMPORT INTERNAL + +root_url = "https://api.stlouisfed.org/fred" + + +class Fred: + def __init__(self, api_key: str): + self.api_key = api_key + + def __fetch_data(self, url: str): + full_url = f"{url}&api_key={self.api_key}&file_type=json" + response = requests.get(full_url, timeout=10) + return response.json() + + def get_series( + self, + series_id: str, + start_date: Optional[date] = None, + end_date: Optional[date] = None, + **kwargs, + ) -> dict: + """ + Get data for a Fred series id. This fetches the latest known data. + Code copied from: https://github.com/mortada/fredapi/blob/master/fredapi/fred.py + + Parameters + ---------- + series_id : str + Fred series id such as 'CPIAUCSL' + start_date : date + earliest observation date + end_date : date + latest observation date + kwargs : additional parameters + Any additional parameters supported by FRED. You can see the full list here: + https://api.stlouisfed.org/docs/fred/series_observations.html + + Returns + ------- + data : Series + a Series where each index is the observation date and the value is the data for the Fred series + """ + url = f"{root_url}/series/observations?series_id={series_id}" + if start_date: + url += "&observation_start=" + start_date.strftime("%Y-%m-%d") + if end_date: + url += "&observation_end=" + end_date.strftime("%Y-%m-%d") + if kwargs.keys(): + url += "&" + urlencode(kwargs) + root = self.__fetch_data(url) + if root is None: + raise ValueError("No data exists for series id: " + series_id) + return root["observations"] diff --git a/openbb_providers/builtin_providers/fred/fred_helpers.py b/openbb_providers/builtin_providers/fred/fred_helpers.py new file mode 100644 index 00000000000..139ed24543d --- /dev/null +++ b/openbb_providers/builtin_providers/fred/fred_helpers.py @@ -0,0 +1,86 @@ +# IMPORT STANDARD +import csv +import pathlib +from typing import List, Literal + +# IMPORT THIRD-PARTY +# IMPORT INTERNAL + +harmonized_cpi_path = pathlib.Path(__file__).parent / "datasets" / "harmonized_cpi.csv" +cpi_path = pathlib.Path(__file__).parent / "datasets" / "cpi.csv" + + +CPI_COUNTRIES = Literal[ + "australia", + "austria", + "belgium", + "brazil", + "bulgaria", + "canada", + "chile", + "china", + "croatia", + "cyprus", + "czech_republic", + "denmark", + "estonia", + "euro_area", + "finland", + "france", + "germany", + "greece", + "hungary", + "iceland", + "india", + "indonesia", + "ireland", + "israel", + "italy", + "japan", + "korea", + "latvia", + "lithuania", + "luxembourg", + "malta", + "mexico", + "netherlands", + "new_zealand", + "norway", + "poland", + "portugal", + "romania", + "russian_federation", + "slovak_republic", + "slovakia", + "slovenia", + "south_africa", + "spain", + "sweden", + "switzerland", + "turkey", + "united_kingdom", + "united_states", +] + +CPI_UNITS = Literal["growth_previous", "growth_same", "index_2015"] + +CPI_FREQUENCY = Literal["monthly", "quarterly", "annual"] + + +def all_cpi_options(harmonized: bool = False) -> List[dict]: + data = [] + + the_path = harmonized_cpi_path if harmonized else cpi_path + with open(the_path, encoding="utf-8") as csv_file_handler: + csv_reader = csv.DictReader(csv_file_handler) + for rows in csv_reader: + row = {key.lstrip("\ufeff"): value for key, value in rows.items()} + data.append(row) + return data + + +def get_cpi_options(harmonized: bool = False) -> List[dict]: + series = all_cpi_options(harmonized) + for item in series: + item.pop("series_id") + return series diff --git a/openbb_providers/builtin_providers/fred/provider.py b/openbb_providers/builtin_providers/fred/provider.py new file mode 100644 index 00000000000..07692a0ef86 --- /dev/null +++ b/openbb_providers/builtin_providers/fred/provider.py @@ -0,0 +1,16 @@ +"""FRED provider module.""" + +# IMPORT STANDARD +# IMPORT THIRD-PARTY +# IMPORT INTERNAL +from builtin_providers.fred.cpi import FREDCPIFetcher +from openbb_provider.provider.abstract.provider import Provider + +# mypy: disable-error-code="list-item" + +# ignoring because I dont know how to type the string properly +fred_provider = Provider( + name="fred", # type: ignore + description="Provider for FRED.", + fetcher_list=[FREDCPIFetcher], +) diff --git a/openbb_providers/openbb_provider/model/data/cpi.py b/openbb_providers/openbb_provider/model/data/cpi.py new file mode 100644 index 00000000000..d28ec88e3f2 --- /dev/null +++ b/openbb_providers/openbb_provider/model/data/cpi.py @@ -0,0 +1,60 @@ +# IMPORT STANDARD +from datetime import date as dateType +from typing import Dict, List, Optional + +# IMPORT THIRD-PARTY +from pydantic import BaseModel, validator + +from builtin_providers.fred.fred_helpers import CPI_COUNTRIES, CPI_FREQUENCY, CPI_UNITS + +# IMPORT INTERNAL +from openbb_provider.model.abstract.data import Data, QueryParams + + +class CPIQueryParams(QueryParams): + """CPI query. + When other provders are added, this will probably need less strict types + + Parameter + --------- + countries: List[CPI_COUNTRIES] + The country or countries you want to see. + units: List[CPI_UNITS] + The units you want to see, can be "growth_previous", "growth_same" or "index_2015". + frequency: List[CPI_FREQUENCY] + The frequency you want to see, either "annual", monthly" or "quarterly". + harmonized: bool + Whether you wish to obtain harmonized data. + start_date: Optional[date] + Start date, formatted YYYY-MM-DD + end_date: Optional[date] + End date, formatted YYYY-MM-DD + """ + + __name__ = "CPIQueryParams" + + countries: List[CPI_COUNTRIES] + units: CPI_UNITS = "growth_same" + frequency: CPI_FREQUENCY = "monthly" + harmonized: bool = False + start_date: Optional[dateType] + end_date: Optional[dateType] + + +class CPIDataPoint(BaseModel): + realtime_start: dateType + realtime_end: dateType + date: dateType + value: float + + @validator("value", pre=True) + def value_validate(cls, v: str): # pylint: disable=E0213 + if v == ".": + return 0.0 + return float(v) + + +class CPIData(Data): + __name__ = "FREDCPIData" + # I don't love this, but the keys are dynamic based on input so I don't know another way + data: Dict[str, List[CPIDataPoint]]