Move fred provider

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)

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"]

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

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],
+)

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]]