summaryrefslogtreecommitdiffstats
path: root/openbb_platform/providers/fred/openbb_fred/models/cpi.py
diff options
context:
space:
mode:
Diffstat (limited to 'openbb_platform/providers/fred/openbb_fred/models/cpi.py')
-rw-r--r--openbb_platform/providers/fred/openbb_fred/models/cpi.py81
1 files changed, 0 insertions, 81 deletions
diff --git a/openbb_platform/providers/fred/openbb_fred/models/cpi.py b/openbb_platform/providers/fred/openbb_fred/models/cpi.py
deleted file mode 100644
index a7c55bb9b78..00000000000
--- a/openbb_platform/providers/fred/openbb_fred/models/cpi.py
+++ /dev/null
@@ -1,81 +0,0 @@
-"""FRED Consumer Price Index Model."""
-
-from typing import Any, Dict, List, Optional
-
-from openbb_core.provider.abstract.fetcher import Fetcher
-from openbb_core.provider.standard_models.cpi import (
- ConsumerPriceIndexData,
- ConsumerPriceIndexQueryParams,
-)
-from openbb_fred.utils.fred_base import Fred
-from openbb_fred.utils.fred_helpers import all_cpi_options
-
-
-class FREDConsumerPriceIndexQueryParams(ConsumerPriceIndexQueryParams):
- """FRED Consumer Price Index Query."""
-
- __json_schema_extra__ = {"country": {"multiple_items_allowed": True}}
-
-
-class FREDConsumerPriceIndexData(ConsumerPriceIndexData):
- """FRED Consumer Price Index Data."""
-
-
-class FREDConsumerPriceIndexFetcher(
- Fetcher[FREDConsumerPriceIndexQueryParams, List[FREDConsumerPriceIndexData]]
-):
- """Transform the query, extract and transform the data from the FRED endpoints."""
-
- @staticmethod
- def transform_query(params: Dict[str, Any]) -> FREDConsumerPriceIndexQueryParams:
- """Transform query."""
- return FREDConsumerPriceIndexQueryParams(**params)
-
- @staticmethod
- def extract_data(
- query: FREDConsumerPriceIndexQueryParams,
- credentials: Optional[Dict[str, str]],
- **kwargs: Any,
- ) -> Dict:
- """Extract data."""
- api_key = credentials.get("fred_api_key") if credentials else ""
-
- all_options = all_cpi_options(query.harmonized)
-
- step_1 = [x for x in all_options if x["country"] in query.country]
- 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]
-
- series_dict = {}
- fred = Fred(api_key)
- for item in step_3:
- loc = f"{item['country']}"
- temp = fred.get_series(
- item["series_id"], query.start_date, query.end_date, **kwargs
- )
- temp = [{"date": item["date"], "value": item["value"]} for item in temp]
- series_dict[loc] = [item for item in temp if item["value"] != "."]
-
- return series_dict
-
- @staticmethod
- def transform_data(
- query: FREDConsumerPriceIndexQueryParams, data: Dict, **kwargs: Any
- ) -> List[FREDConsumerPriceIndexData]:
- """Transform data."""
- transformed_data = {}
-
- # Iterate over the series_dict
- for country, data_list in data.items():
- for item in data_list:
- # If the date is not in the dictionary, add it
- if item["date"] not in transformed_data:
- transformed_data[item["date"]] = {"date": item["date"]}
- # Update the dictionary with the country's value data
- transformed_data[item["date"]].update({country: item["value"]})
-
- # Convert the dictionary to a list of dictionaries
- return [
- FREDConsumerPriceIndexData.model_validate(item)
- for item in list(transformed_data.values())
- ]