jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 12:40:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Enhancement] Add IMF Financial Soundness Indicators Set (#6960)

Browse Source 
* add IMF Financial Soundness Indicators Set

* finish the APIEx

* fix docstring

* typo

* lint

* update readme

* fix test..?

* add Eastern Caribbean Currency Union to country map

* mypy

* more linting

* more linting

* improve title consistency between IRFCL and FSI

* clear commented out code
This commit is contained in:
Danglewood
2024-12-05 08:15:45 -08:00
committed by GitHub
parent 320a66f710
commit e787e13d3d
14 changed files with 23540 additions and 706 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
openbb_platform/extensions/economy/integration/test_economy_api.py
View File

@@ -638,6 +638,7 @@ def test_economy_indicators(params, headers):
"params",
[
({"provider": "econdb", "use_cache": False}),
({"provider": "imf", "query": "balance sheet;households;debt"}),
],
)
@pytest.mark.integration

1
openbb_platform/extensions/economy/integration/test_economy_python.py
View File

@@ -580,6 +580,7 @@ def test_economy_country_profile(params, obb):
"params",
[
({"provider": "econdb", "use_cache": False}),
({"provider": "imf", "query": "balance sheet;households;debt"}),
],
)
@pytest.mark.integration

18
openbb_platform/extensions/economy/openbb_economy/economy_router.py
View File

@@ -434,6 +434,24 @@ async def available_indicators(
"frequency": "annual",
},
),
APIEx(
description=(
"When the provider is 'imf', additional presets return the core Financial Soundness Indicators."
"\n 'fsi_core' - Core FSIs"
"\n 'fsi_encouraged_set' - Encouraged Set of FSIs,"
"\n 'fsi_core_underlying' - Underlying data for the Core FSIs."
"\n 'fsi_other' - Additional/Other FSIs that are not in the Core or Encouraged Set."
"\n 'fsi_all' - all FSI data for a single country."
),
parameters={
"provider": "imf",
"symbol": "fsi_encouraged_set",
"country": "us,fr,gb",
"start_date": "2022-01-01",
"end_date": "2023-12-31",
"frequency": "annual",
},
),
],
)
async def indicators(

53
openbb_platform/openbb/assets/reference.json vendored
View File

@@ -7679,7 +7679,16 @@
"choices": null
}
],
"imf": []
"imf": [
{
"name": "query",
"type": "Union[str, List[str]]",
"description": "The query string to search through the available indicators. Use semicolons to separate multiple terms. Multiple items allowed for provider(s): imf.",
"default": null,
"optional": true,
"choices": null
}
]
},
"returns": {
"OBBject": [
@@ -7828,6 +7837,14 @@
}
],
"imf": [
{
"name": "dataset",
"type": "str",
"description": "The IMF dataset associated with the symbol.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "table",
"type": "str",
@@ -7846,11 +7863,27 @@
},
{
"name": "order",
"type": "int",
"type": "Union[int, float]",
"description": "Order of the data, relative to the table.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "children",
"type": "str",
"description": "The symbol of the child data, if any.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "unit",
"type": "str",
"description": "The unit of the data.",
"default": null,
"optional": true,
"choices": null
}
]
},
@@ -7862,7 +7895,7 @@
"message": null
},
"description": "Get economic indicators by country and indicator.",
"examples": "\nExamples\n--------\n\n```python\nfrom openbb import obb\nobb.economy.indicators(provider='econdb', symbol=PCOCO)\n# Enter the country as the full name, or iso code. Use `available_indicators()` to get a list of supported indicators from EconDB.\nobb.economy.indicators(symbol=CPI, country='united_states,jp', provider='econdb')\n# Use the `main` symbol to get the group of main indicators for a country.\nobb.economy.indicators(provider='econdb', symbol=main, country='eu')\n# When the provider is 'imf', the absence of a symbol will default to 'irfcl_top_lines'. Use 'IRFCL' to get all the data from the set of indicators.\nobb.economy.indicators(provider='imf')\n# When the provider is 'imf', complete tables are returned by using a 'preset'. Refer to the function's docstring for descriptions of each preset. When no country is supplied, the data is returned for all countries.\nobb.economy.indicators(provider='imf', symbol=gold_reserves)\n# When the provider is 'imf', multiple countries and symbols can be supplied. Enter countries as a two-letter ISO country code, or the country name in lower_snake_case.\nobb.economy.indicators(provider='imf', symbol=RAFA_USD,RAPFA_USD,RAFA_RAPFA_RO, country='us,china,jp,4f,gb', start_date='2010-01-01', end_date='2020-12-31', frequency=annual)\n```\n\n",
"examples": "\nExamples\n--------\n\n```python\nfrom openbb import obb\nobb.economy.indicators(provider='econdb', symbol=PCOCO)\n# Enter the country as the full name, or iso code. Use `available_indicators()` to get a list of supported indicators from EconDB.\nobb.economy.indicators(symbol=CPI, country='united_states,jp', provider='econdb')\n# Use the `main` symbol to get the group of main indicators for a country.\nobb.economy.indicators(provider='econdb', symbol=main, country='eu')\n# When the provider is 'imf', the absence of a symbol will default to 'irfcl_top_lines'. Use 'IRFCL' to get all the data from the set of indicators.\nobb.economy.indicators(provider='imf')\n# When the provider is 'imf', complete tables are returned by using a 'preset'. Refer to the function's docstring for descriptions of each preset. When no country is supplied, the data is returned for all countries.\nobb.economy.indicators(provider='imf', symbol=gold_reserves)\n# When the provider is 'imf', multiple countries and symbols can be supplied. Enter countries as a two-letter ISO country code, or the country name in lower_snake_case.\nobb.economy.indicators(provider='imf', symbol=RAFA_USD,RAPFA_USD,RAFA_RAPFA_RO, country='us,china,jp,4f,gb', start_date='2010-01-01', end_date='2020-12-31', frequency=annual)\n# When the provider is 'imf', additional presets return the core Financial Soundness Indicators.\n 'fsi_core' - Core FSIs\n 'fsi_encouraged_set' - Encouraged Set of FSIs,\n 'fsi_core_underlying' - Underlying data for the Core FSIs.\n 'fsi_other' - Additional/Other FSIs that are not in the Core or Encouraged Set.\n 'fsi_all' - all FSI data for a single country.\nobb.economy.indicators(provider='imf', symbol=fsi_encouraged_set, country='us,fr,gb', start_date='2022-01-01', end_date='2023-12-31', frequency=annual)\n```\n\n",
"parameters": {
"standard": [
{
@@ -8145,7 +8178,7 @@
{
"name": "symbol",
"type": "Union[str, List[str]]",
"description": "Symbol to get data for. Use `available_indicators()` to get the list of available symbols. Use 'IRFCL' to get all the data from the set of indicators. Complete tables are available only by single country, and are keyed as described below. The default is 'irfcl_top_lines'. Available presets not listed in `available_indicators()` are:\n\n 'IRFCL': All the data from the set of indicators. Not compatible with multiple countries.\n 'irfcl_top_lines': The default, top line items from the IRFCL data. Compatible with multiple countries.\n 'reserve_assets_and_other_fx_assets': Table I of the IRFCL data. Not compatible with multiple countries.\n 'predetermined_drains_on_fx_assets': Table II of the IRFCL data. Not compatible with multiple countries.\n 'contingent_drains_fx_assets': Table III of the IRFCL data. Not compatible with multiple countries.\n 'memorandum_items': The memorandum items table of the IRFCL data. Not compatible with multiple countries.\n 'gold_reserves': Gold reserves as value in USD and Fine Troy Ounces. Compatible with multiple countries.\n 'derivative_assets': Net derivative assets as value in USD. Compatible with multipile countries. Multiple items allowed for provider(s): imf.",
"description": "Symbol to get data for. Use `available_indicators()` to get the list of available symbols. Use 'IRFCL' to get all the data from International Reserves & Foreign Currency Liquidity indicators. Use 'core_fsi' to get the core Financial Soundness Indicators. Use 'core_fsi_underlying' to include underlying data for the core Financial Soundness Indicators. Complete tables are available only by single country, and are keyed as described below. The default is 'irfcl_top_lines'. Available presets not listed in `available_indicators()` are:\n\n 'IRFCL': All the data from the set of indicators. Not compatible with multiple countries.\n 'irfcl_top_lines': The default, top line items from the IRFCL data. Compatible with multiple countries.\n 'reserve_assets_and_other_fx_assets': Table I of the IRFCL data. Not compatible with multiple countries.\n 'predetermined_drains_on_fx_assets': Table II of the IRFCL data. Not compatible with multiple countries.\n 'contingent_drains_fx_assets': Table III of the IRFCL data. Not compatible with multiple countries.\n 'memorandum_items': The memorandum items table of the IRFCL data. Not compatible with multiple countries.\n 'gold_reserves': Gold reserves as value in USD and Fine Troy Ounces. Compatible with multiple countries.\n 'derivative_assets': Net derivative assets as value in USD. Compatible with multipile countries.\n 'fsi_core': The core Financial Soundness Indicators. Compatible with multiple countries.\n 'fsi_core_underlying': The core FSIs underlying series data. Not compatible with country='all'.\n 'fsi_encouraged_set': The encouraged set of Financial Soundness Indicators. Not compatible with country='all'.\n 'fsi_other': The other Financial Soundness Indicators. Not compatible with country='all'.\n 'fsi_balance_sheets': Data categorized as Balance Sheets and Income Statements. Not compatible with country='all'.\n 'fsi_all': All the Financial Soundness Indicators. Not compatible with multiple countries. Multiple items allowed for provider(s): imf.",
"default": "irfcl_top_lines",
"optional": true,
"choices": null
@@ -8153,7 +8186,7 @@
{
"name": "frequency",
"type": "Literal['annual', 'quarter', 'month']",
"description": "Frequency of the data.",
"description": "Frequency of the data, default is 'quarter'.",
"default": "quarter",
"optional": true,
"choices": [
@@ -8238,6 +8271,14 @@
],
"econdb": [],
"imf": [
{
"name": "unit",
"type": "str",
"description": "The unit of the value.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "scale",
"type": "str",
@@ -8264,7 +8305,7 @@
},
{
"name": "order",
"type": "int",
"type": "Union[int, float]",
"description": "Order of the data, relative to the table.",
"default": null,
"optional": true,

206
openbb_platform/openbb/package/economy.py
View File

@@ -70,6 +70,8 @@ class ROUTER_economy(Container):
The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: econdb, imf.
use_cache : bool
Whether to use cache or not, by default is True The cache of indicator symbols will persist for one week. (provider: econdb)
query : Optional[str]
The query string to search through the available indicators. Use semicolons to separate multiple terms. Multiple comma separated items allowed. (provider: imf)
Returns
-------
@@ -115,12 +117,18 @@ class ROUTER_economy(Container):
The last date of the data. (provider: econdb)
last_insert_timestamp : Optional[datetime]
The time of the last update. Data is typically reported with a lag. (provider: econdb)
dataset : Optional[str]
The IMF dataset associated with the symbol. (provider: imf)
table : Optional[str]
The name of the table associated with the symbol. (provider: imf)
level : Optional[int]
The indentation level of the data, relative to the table and symbol_root (provider: imf)
order : Optional[int]
order : Optional[Union[int, float]]
Order of the data, relative to the table. (provider: imf)
children : Optional[str]
The symbol of the child data, if any. (provider: imf)
unit : Optional[str]
The unit of the data. (provider: imf)
Examples
--------
@@ -140,6 +148,9 @@ class ROUTER_economy(Container):
},
standard_params={},
extra_params=kwargs,
info={
"query": {"imf": {"multiple_items_allowed": True, "choices": None}}
},
)
)
@@ -2946,99 +2957,114 @@ class ROUTER_economy(Container):
) -> OBBject:
"""Get economic indicators by country and indicator.
Parameters
----------
country : Union[str, None, List[Optional[str]]]
The country to get data. The country represented by the indicator, if available. Multiple comma separated items allowed for provider(s): econdb, imf.
start_date : Union[date, None, str]
Start date of the data, in YYYY-MM-DD format.
end_date : Union[date, None, str]
End date of the data, in YYYY-MM-DD format.
provider : Optional[Literal['econdb', 'imf']]
The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: econdb, imf.
symbol : Optional[str]
Symbol to get data for. The base symbol for the indicator (e.g. GDP, CPI, etc.). Use `available_indicators()` to get a list of available symbols. Multiple comma separated items allowed. (provider: econdb);
Symbol to get data for. Use `available_indicators()` to get the list of available symbols. Use 'IRFCL' to get all the data from the set of indicators. Complete tables are available only by single country, and are keyed as described below. The default is 'irfcl_top_lines'. Available presets not listed in `available_indicators()` are:
'IRFCL': All the data from the set of indicators. Not compatible with multiple countries.
'irfcl_top_lines': The default, top line items from the IRFCL data. Compatible with multiple countries.
'reserve_assets_and_other_fx_assets': Table I of the IRFCL data. Not compatible with multiple countries.
'predetermined_drains_on_fx_assets': Table II of the IRFCL data. Not compatible with multiple countries.
'contingent_drains_fx_assets': Table III of the IRFCL data. Not compatible with multiple countries.
'memorandum_items': The memorandum items table of the IRFCL data. Not compatible with multiple countries.
'gold_reserves': Gold reserves as value in USD and Fine Troy Ounces. Compatible with multiple countries.
'derivative_assets': Net derivative assets as value in USD. Compatible with multipile countries.
Multiple comma separated items allowed. (provider: imf)
transform : Optional[Literal['toya', 'tpop', 'tusd', 'tpgp']]
The transformation to apply to the data, default is None.
tpop: Change from previous period
toya: Change from one year ago
tusd: Values as US dollars
tpgp: Values as a percent of GDP
Only 'tpop' and 'toya' are applicable to all indicators. Applying transformations across multiple indicators/countries may produce unexpected results.
This is because not all indicators are compatible with all transformations, and the original units and scale differ between entities.
`tusd` should only be used where values are currencies. (provider: econdb)
frequency : Literal['annual', 'quarter', 'month']
The frequency of the data, default is 'quarter'. Only valid when 'symbol' is 'main'. (provider: econdb);
Frequency of the data. (provider: imf)
use_cache : bool
If True, the request will be cached for one day. Using cache is recommended to avoid needlessly requesting the same data. (provider: econdb)
Returns
-------
OBBject
results : List[EconomicIndicators]
Serializable results.
Parameters
----------
country : Union[str, None, List[Optional[str]]]
The country to get data. The country represented by the indicator, if available. Multiple comma separated items allowed for provider(s): econdb, imf.
start_date : Union[date, None, str]
Start date of the data, in YYYY-MM-DD format.
end_date : Union[date, None, str]
End date of the data, in YYYY-MM-DD format.
provider : Optional[Literal['econdb', 'imf']]
Provider name.
warnings : Optional[List[Warning_]]
List of warnings.
chart : Optional[Chart]
Chart object.
extra : Dict[str, Any]
Extra info.
The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: econdb, imf.
symbol : Optional[str]
Symbol to get data for. The base symbol for the indicator (e.g. GDP, CPI, etc.). Use `available_indicators()` to get a list of available symbols. Multiple comma separated items allowed. (provider: econdb);
Symbol to get data for. Use `available_indicators()` to get the list of available symbols. Use 'IRFCL' to get all the data from International Reserves & Foreign Currency Liquidity indicators. Use 'core_fsi' to get the core Financial Soundness Indicators. Use 'core_fsi_underlying' to include underlying data for the core Financial Soundness Indicators. Complete tables are available only by single country, and are keyed as described below. The default is 'irfcl_top_lines'. Available presets not listed in `available_indicators()` are:
EconomicIndicators
------------------
date : date
The date of the data.
symbol_root : Optional[str]
The root symbol for the indicator (e.g. GDP).
symbol : Optional[str]
Symbol representing the entity requested in the data.
country : Optional[str]
The country represented by the data.
value : Optional[Union[int, float]]
'IRFCL': All the data from the set of indicators. Not compatible with multiple countries.
'irfcl_top_lines': The default, top line items from the IRFCL data. Compatible with multiple countries.
'reserve_assets_and_other_fx_assets': Table I of the IRFCL data. Not compatible with multiple countries.
'predetermined_drains_on_fx_assets': Table II of the IRFCL data. Not compatible with multiple countries.
'contingent_drains_fx_assets': Table III of the IRFCL data. Not compatible with multiple countries.
'memorandum_items': The memorandum items table of the IRFCL data. Not compatible with multiple countries.
'gold_reserves': Gold reserves as value in USD and Fine Troy Ounces. Compatible with multiple countries.
'derivative_assets': Net derivative assets as value in USD. Compatible with multipile countries.
'fsi_core': The core Financial Soundness Indicators. Compatible with multiple countries.
'fsi_core_underlying': The core FSIs underlying series data. Not compatible with country='all'.
'fsi_encouraged_set': The encouraged set of Financial Soundness Indicators. Not compatible with country='all'.
'fsi_other': The other Financial Soundness Indicators. Not compatible with country='all'.
'fsi_balance_sheets': Data categorized as Balance Sheets and Income Statements. Not compatible with country='all'.
'fsi_all': All the Financial Soundness Indicators. Not compatible with multiple countries.
Multiple comma separated items allowed. (provider: imf)
transform : Optional[Literal['toya', 'tpop', 'tusd', 'tpgp']]
The transformation to apply to the data, default is None.
scale : Optional[str]
The scale of the value. (provider: imf)
table : Optional[str]
The name of the table associated with the symbol. (provider: imf)
level : Optional[int]
The indentation level of the data, relative to the table and symbol_root (provider: imf)
order : Optional[int]
Order of the data, relative to the table. (provider: imf)
reference_sector : Optional[str]
The reference sector for the data. (provider: imf)
title : Optional[str]
The title of the series associated with the symbol. (provider: imf)
tpop: Change from previous period
toya: Change from one year ago
tusd: Values as US dollars
tpgp: Values as a percent of GDP
Examples
--------
>>> from openbb import obb
>>> obb.economy.indicators(provider='econdb', symbol='PCOCO')
>>> # Enter the country as the full name, or iso code. Use `available_indicators()` to get a list of supported indicators from EconDB.
>>> obb.economy.indicators(symbol='CPI', country='united_states,jp', provider='econdb')
>>> # Use the `main` symbol to get the group of main indicators for a country.
>>> obb.economy.indicators(provider='econdb', symbol='main', country='eu')
>>> # When the provider is 'imf', the absence of a symbol will default to 'irfcl_top_lines'. Use 'IRFCL' to get all the data from the set of indicators.
>>> obb.economy.indicators(provider='imf')
>>> # When the provider is 'imf', complete tables are returned by using a 'preset'. Refer to the function's docstring for descriptions of each preset. When no country is supplied, the data is returned for all countries.
>>> obb.economy.indicators(provider='imf', symbol='gold_reserves')
>>> # When the provider is 'imf', multiple countries and symbols can be supplied. Enter countries as a two-letter ISO country code, or the country name in lower_snake_case.
>>> obb.economy.indicators(provider='imf', symbol='RAFA_USD,RAPFA_USD,RAFA_RAPFA_RO', country='us,china,jp,4f,gb', start_date='2010-01-01', end_date='2020-12-31', frequency='annual')
Only 'tpop' and 'toya' are applicable to all indicators. Applying transformations across multiple indicators/countries may produce unexpected results.
This is because not all indicators are compatible with all transformations, and the original units and scale differ between entities.
`tusd` should only be used where values are currencies. (provider: econdb)
frequency : Literal['annual', 'quarter', 'month']
The frequency of the data, default is 'quarter'. Only valid when 'symbol' is 'main'. (provider: econdb);
Frequency of the data, default is 'quarter'. (provider: imf)
use_cache : bool
If True, the request will be cached for one day. Using cache is recommended to avoid needlessly requesting the same data. (provider: econdb)
Returns
-------
OBBject
results : List[EconomicIndicators]
Serializable results.
provider : Optional[Literal['econdb', 'imf']]
Provider name.
warnings : Optional[List[Warning_]]
List of warnings.
chart : Optional[Chart]
Chart object.
extra : Dict[str, Any]
Extra info.
EconomicIndicators
------------------
date : date
The date of the data.
symbol_root : Optional[str]
The root symbol for the indicator (e.g. GDP).
symbol : Optional[str]
Symbol representing the entity requested in the data.
country : Optional[str]
The country represented by the data.
value : Optional[Union[int, float]]
unit : Optional[str]
The unit of the value. (provider: imf)
scale : Optional[str]
The scale of the value. (provider: imf)
table : Optional[str]
The name of the table associated with the symbol. (provider: imf)
level : Optional[int]
The indentation level of the data, relative to the table and symbol_root (provider: imf)
order : Optional[Union[int, float]]
Order of the data, relative to the table. (provider: imf)
reference_sector : Optional[str]
The reference sector for the data. (provider: imf)
title : Optional[str]
The title of the series associated with the symbol. (provider: imf)
Examples
--------
>>> from openbb import obb
>>> obb.economy.indicators(provider='econdb', symbol='PCOCO')
>>> # Enter the country as the full name, or iso code. Use `available_indicators()` to get a list of supported indicators from EconDB.
>>> obb.economy.indicators(symbol='CPI', country='united_states,jp', provider='econdb')
>>> # Use the `main` symbol to get the group of main indicators for a country.
>>> obb.economy.indicators(provider='econdb', symbol='main', country='eu')
>>> # When the provider is 'imf', the absence of a symbol will default to 'irfcl_top_lines'. Use 'IRFCL' to get all the data from the set of indicators.
>>> obb.economy.indicators(provider='imf')
>>> # When the provider is 'imf', complete tables are returned by using a 'preset'. Refer to the function's docstring for descriptions of each preset. When no country is supplied, the data is returned for all countries.
>>> obb.economy.indicators(provider='imf', symbol='gold_reserves')
>>> # When the provider is 'imf', multiple countries and symbols can be supplied. Enter countries as a two-letter ISO country code, or the country name in lower_snake_case.
>>> obb.economy.indicators(provider='imf', symbol='RAFA_USD,RAPFA_USD,RAFA_RAPFA_RO', country='us,china,jp,4f,gb', start_date='2010-01-01', end_date='2020-12-31', frequency='annual')
>>> # When the provider is 'imf', additional presets return the core Financial Soundness Indicators.
'fsi_core' - Core FSIs
'fsi_encouraged_set' - Encouraged Set of FSIs,
'fsi_core_underlying' - Underlying data for the Core FSIs.
'fsi_other' - Additional/Other FSIs that are not in the Core or Encouraged Set.
'fsi_all' - all FSI data for a single country.
>>> obb.economy.indicators(provider='imf', symbol='fsi_encouraged_set', country='us,fr,gb', start_date='2022-01-01', end_date='2023-12-31', frequency='annual')
""" # noqa: E501
return self._run(

2
openbb_platform/providers/imf/README.md vendored
View File

@@ -21,10 +21,12 @@ No authorization is required to use, but IP addresses are bound by the limitatio
- Databases:
- International Reserves and Foreign Currency Liquidity
- Direction of Trade Statistics
- Financial Soundness Indicators
Coverage:
- All IRFCL tables.
- Individual, or multiple, time series from single or multiple countries.
- Core and Encouraged Set tables, plus all individual underlying series.
### Endpoints

1
openbb_platform/providers/imf/openbb_imf/assets/imf_country_map.json vendored
View File

@@ -36,6 +36,7 @@
"KH": "Cambodia",
"CM": "Cameroon",
"CA": "Canada",
"5Y": "Eastern Caribbean Currency Union",
"CF": "Central African Republic",
"TD": "Chad",
"CL": "Chile",

23340
openbb_platform/providers/imf/openbb_imf/assets/imf_symbols.json vendored
View File

File diff suppressed because it is too large Load Diff

77
openbb_platform/providers/imf/openbb_imf/models/available_indicators.py
View File

@@ -2,7 +2,7 @@
# pylint: disable=unused-argument
from typing import Any, Dict, List, Optional
from typing import Any, Optional, Union
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -16,6 +16,14 @@ from pydantic import Field
class ImfAvailableIndicatorsQueryParams(AvailableIndicesQueryParams):
"""IMF Available Indicators Query Parameters."""
__json_schema_extra__ = {"query": {"multiple_items_allowed": True}}
query: Optional[str] = Field(
default=None,
description="The query string to search through the available indicators."
+ " Use semicolons to separate multiple terms.",
)
class ImfAvailableIndicatorsData(AvailableIndicatorsData):
"""IMF Available Indicators Data."""
@@ -24,7 +32,10 @@ class ImfAvailableIndicatorsData(AvailableIndicatorsData):
"symbol_root": "parent",
"description": "title",
}
dataset: Optional[str] = Field(
default=None,
description="The IMF dataset associated with the symbol.",
)
table: Optional[str] = Field(
default=None,
description="The name of the table associated with the symbol.",
@@ -33,49 +44,83 @@ class ImfAvailableIndicatorsData(AvailableIndicatorsData):
default=None,
description="The indentation level of the data, relative to the table and symbol_root",
)
order: Optional[int] = Field(
order: Optional[Union[int, float]] = Field(
default=None,
description="Order of the data, relative to the table.",
)
children: Optional[str] = Field(
default=None,
description="The symbol of the child data, if any.",
)
unit: Optional[str] = Field(
default=None,
description="The unit of the data.",
)
class ImfAvailableIndicatorsFetcher(
Fetcher[ImfAvailableIndicatorsQueryParams, List[ImfAvailableIndicatorsData]]
Fetcher[ImfAvailableIndicatorsQueryParams, list[ImfAvailableIndicatorsData]]
):
"""IMF Available Indicators Fetcher."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> ImfAvailableIndicatorsQueryParams:
def transform_query(params: dict[str, Any]) -> ImfAvailableIndicatorsQueryParams:
"""Transform the query."""
return ImfAvailableIndicatorsQueryParams(**params)
@staticmethod
def extract_data(
query: ImfAvailableIndicatorsQueryParams,
credentials: Optional[Dict[str, Any]] = None,
credentials: Optional[dict[str, Any]] = None,
**kwargs: Any,
) -> List[Dict]:
) -> list[dict]:
"""Fetch the data."""
# pylint: disable=import-outside-toplevel
from openbb_imf.utils.irfcl_helpers import load_irfcl_symbols
from numpy import nan
from openbb_core.provider.utils.errors import EmptyDataError
from openbb_imf.utils.constants import load_symbols
from pandas import DataFrame, Series
try:
irfcl_symbols = load_irfcl_symbols()
all_symbols = load_symbols("all")
except OpenBBError as e:
raise OpenBBError(f"Failed to load IMF IRFCL symbols: {e}") from e
raise OpenBBError(f"Failed to load IMF symbols static file: {e}") from e
records: List = []
for key, value in irfcl_symbols.items():
value["symbol"] = key
records.append(value)
terms = [term.strip() for term in query.query.split(";")] if query.query else []
df = (
DataFrame(all_symbols)
.T.reset_index()
.rename(columns={"index": "symbol"})
.replace({nan: None})
)
if not terms:
records = df.to_dict(orient="records")
else:
combined_mask = Series([True] * len(df))
for term in terms:
mask = df.apply(
lambda row, term=term: row.astype(str).str.contains(
term, case=False, regex=True, na=False
)
).any(axis=1)
combined_mask &= mask
matches = df[combined_mask]
if matches.empty:
raise EmptyDataError("No results found for the provided query.")
records = matches.to_dict(orient="records")
return records
@staticmethod
def transform_data(
query: ImfAvailableIndicatorsQueryParams,
data: List[Dict],
data: list[dict],
**kwargs: Any,
) -> List[ImfAvailableIndicatorsData]:
) -> list[ImfAvailableIndicatorsData]:
"""Transform the data."""
return [ImfAvailableIndicatorsData.model_validate(d) for d in data]

140
openbb_platform/providers/imf/openbb_imf/models/economic_indicators.py
View File

@@ -3,7 +3,7 @@
# pylint: disable=unused-argument
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional
from typing import Any, Dict, List, Literal, Optional, Union
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -13,14 +13,17 @@ from openbb_core.provider.standard_models.economic_indicators import (
)
from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS
from openbb_core.provider.utils.errors import EmptyDataError
from pydantic import Field
from openbb_imf.utils.constants import IRFCL_PRESET, IRFCL_TABLES
from openbb_imf.utils.constants import (
FSI_PRESETS,
IRFCL_PRESET,
IRFCL_TABLES,
load_symbols,
)
from openbb_imf.utils.irfcl_helpers import (
load_country_to_code_map,
load_irfcl_symbols,
validate_countries,
)
from pydantic import Field, field_validator
class ImfEconomicIndicatorsQueryParams(EconomicIndicatorsQueryParams):
@@ -42,7 +45,9 @@ class ImfEconomicIndicatorsQueryParams(EconomicIndicatorsQueryParams):
default="irfcl_top_lines",
description=QUERY_DESCRIPTIONS.get("symbol", "")
+ " Use `available_indicators()` to get the list of available symbols."
+ " Use 'IRFCL' to get all the data from the set of indicators."
+ " Use 'IRFCL' to get all the data from International Reserves & Foreign Currency Liquidity indicators."
+ " Use 'core_fsi' to get the core Financial Soundness Indicators."
+ " Use 'core_fsi_underlying' to include underlying data for the core Financial Soundness Indicators."
+ " Complete tables are available only by single country, and are keyed as described below."
+ " The default is 'irfcl_top_lines'. Available presets not listed in `available_indicators()` are:"
+ """\n
@@ -54,13 +59,34 @@ class ImfEconomicIndicatorsQueryParams(EconomicIndicatorsQueryParams):
'memorandum_items': The memorandum items table of the IRFCL data. Not compatible with multiple countries.
'gold_reserves': Gold reserves as value in USD and Fine Troy Ounces. Compatible with multiple countries.
'derivative_assets': Net derivative assets as value in USD. Compatible with multipile countries.
'fsi_core': The core Financial Soundness Indicators. Compatible with multiple countries.
'fsi_core_underlying': The core FSIs underlying series data. Not compatible with country='all'.
'fsi_encouraged_set': The encouraged set of Financial Soundness Indicators. Not compatible with country='all'.
'fsi_other': The other Financial Soundness Indicators. Not compatible with country='all'.
'fsi_balance_sheets': Data categorized as Balance Sheets and Income Statements. Not compatible with country='all'.
'fsi_all': All the Financial Soundness Indicators. Not compatible with multiple countries.
""",
)
frequency: Literal["annual", "quarter", "month"] = Field(
default="quarter",
description="Frequency of the data.",
description="Frequency of the data, default is 'quarter'.",
)
@field_validator("symbol", mode="before", check_fields=False)
@classmethod
def _count_presets(cls, v):
"""Validate the symbol."""
if not v:
return v
presets = list(IRFCL_PRESET) + FSI_PRESETS
n_preset = 0
symbols = v.split(",")
for symbol in symbols:
n_preset += 1 if symbol in presets else 0
if n_preset > 1:
raise ValueError("only one preset symbol can be used at a time.")
return v
class ImfEconomicIndicatorsData(EconomicIndicatorsData):
"""IMF Economic Indicators Data."""
@@ -69,6 +95,10 @@ class ImfEconomicIndicatorsData(EconomicIndicatorsData):
"symbol_root": "parent",
}
unit: Optional[str] = Field(
default=None,
description="The unit of the value.",
)
scale: Optional[str] = Field(
default=None,
description="The scale of the value.",
@@ -81,7 +111,7 @@ class ImfEconomicIndicatorsData(EconomicIndicatorsData):
default=None,
description="The indentation level of the data, relative to the table and symbol_root",
)
order: Optional[int] = Field(
order: Optional[Union[int, float]] = Field(
default=None,
description="Order of the data, relative to the table.",
)
@@ -106,10 +136,24 @@ class ImfEconomicIndicatorsFetcher(
symbols = params.get("symbol", "")
countries = params.get("country")
now = datetime.now().date()
symbols = "IRFCL" if ("all" in symbols or "IRFCL" in symbols) else symbols
if symbols == "IRFCL" and not (countries or countries == "all"):
symbols = (
"IRFCL"
if (("all" in symbols or "IRFCL" in symbols) and "fsi_all" not in symbols)
else symbols if symbols else "irfcl_top_lines"
)
incompatible = (
"fsi_other" in symbols
or "fsi_encouraged_set" in symbols
or "fsi_all" in symbols
or "fsi_core_underlying" in symbols
or "fsi_balance_sheets" in symbols
)
if (symbols == "IRFCL" or incompatible) and not (
countries or countries == "all"
):
raise OpenBBError(
f"The selected symbol(s), {params.get('symbol')}, requires a single country for the 'country' parameter."
f"The selected symbol(s), {params.get('symbol')}, is not compatible with the all-countries group."
" Please provide country names or two-letter ISO country codes."
)
if countries:
@@ -145,15 +189,69 @@ class ImfEconomicIndicatorsFetcher(
) -> List[Dict]:
"""Extract the data."""
# pylint: disable = import-outside-toplevel
from openbb_imf.utils.fsi_helpers import _get_fsi_data # noqa
from openbb_imf.utils.irfcl_helpers import _get_irfcl_data
from warnings import warn
fsi_symbols = load_symbols("FSI")
irfcl_symbols = load_symbols("IRFCL")
symbols = query.symbol.split(",")
new_symbols_irfcl: Union[list, str] = []
new_symbols_fsi: Union[list, str] = []
for symbol in symbols:
if symbol in list(IRFCL_PRESET) + ["all", "IRFCL"]:
new_symbols_irfcl = symbol
elif symbol in FSI_PRESETS:
new_symbols_fsi = symbol
elif symbol.upper() in fsi_symbols:
new_symbols_fsi.append(symbol.upper()) # type: ignore
elif symbol.upper() in irfcl_symbols:
new_symbols_irfcl.append(symbol.upper()) # type: ignore
if not new_symbols_irfcl and not new_symbols_fsi:
raise OpenBBError(
f"No valid symbols found -> {query.symbol} -> "
"Use 'available_indicators(provider='imf')' to get the list of available symbols."
)
results: list = []
exceptions: list = []
try:
res = await _get_irfcl_data(**query.model_dump(exclude_none=True))
except OpenBBError as e:
raise e from e
if not res:
try:
if new_symbols_irfcl:
_kwargs = query.model_dump(exclude_none=True)
_kwargs["symbol"] = new_symbols_irfcl
results.extend(await _get_irfcl_data(**_kwargs))
except (EmptyDataError, OpenBBError) as e:
if new_symbols_fsi:
exceptions.append(
f"IRFCL dataset error -> {e.__class__.__name__}: {e}"
)
else:
raise
if new_symbols_fsi:
try:
_kwargs = query.model_dump(exclude_none=True)
_kwargs["symbol"] = new_symbols_fsi
results.extend(await _get_fsi_data(**_kwargs))
except (EmptyDataError, OpenBBError) as e:
if new_symbols_irfcl and len(results) > 0:
exceptions.append(
f"FSI dataset error -> {e.__class__.__name__}: {e}"
)
elif not new_symbols_irfcl:
raise
except OpenBBError as exc:
raise exc from exc
if not results:
raise EmptyDataError("No results returned for the query.")
return res
if results and exceptions:
msgs = "\n".join(exceptions)
warn("An error occurred while fetching the data -> " + msgs)
return results
@staticmethod
def transform_data(
@@ -163,12 +261,15 @@ class ImfEconomicIndicatorsFetcher(
) -> List[ImfEconomicIndicatorsData]:
"""Transform the data."""
# pylint: disable = import-outside-toplevel
from numpy import nan
from pandas import Categorical, DataFrame
if not data:
raise EmptyDataError("The data is empty.")
all_symbols = load_irfcl_symbols()
all_symbols = {
**load_symbols("all"),
}
df = DataFrame(data)
if df.empty:
@@ -190,8 +291,9 @@ class ImfEconomicIndicatorsFetcher(
ordered=True,
)
df = df.sort_values(
by=["date", "symbol", "parent", "value"],
by=["date", "parent", "symbol", "value"],
ascending=[True, True, True, False],
).reset_index(drop=True)
records = df.replace({nan: None}).to_dict(orient="records")
return [ImfEconomicIndicatorsData(**d) for d in df.to_dict(orient="records")]
return [ImfEconomicIndicatorsData.model_validate(r) for r in records]

32
openbb_platform/providers/imf/openbb_imf/utils/constants.py
View File

@@ -3,6 +3,15 @@
# pylint: disable=line-too-long
# flake8: noqa: E501
FSI_PRESETS = [
"fsi_core",
"fsi_core_underlying",
"fsi_other",
"fsi_encouraged_set",
"fsi_balance_sheets",
"fsi_all",
]
IRFCL_HEADLINE = "RAF_USD,RAFA_USD,RAFAFX_USD,RAOFA_USD,RAPFA_USD,RAFAIMF_USD,RAFASDR_USD,RAFAGOLD_USD,RACFA_USD,RAMDCD_USD,RAMFIFC_USD,RAMSR_USD"
RESERVE_ASSETS_AND_OTHER_FX_ASSETS = "RAF_USD,RAFA_USD,RAFAFX_USD,RAFAFXS_USD,RAFAFXSI_USD,RAFAFXCD_USD,RAFAFXCDN_USD,RAFAFXCDBI_USD,RAFAFXCDBIA_USD,RAFAFXCDBO_USD,RAFAFXCDBOA_USD,RAFAIMF_USD,RAFASDR_USD,RAFAGOLD_USD,RAFAGOLDV_OZT,RAFAO_USD,RAFAOF_USD,RAFAOL_USD,RAFAOO_USD,RAOFA_USD,RAOFAS_USD,RAOFAD_USD,RAOFAL_USD,RAOFAF_USD,RAOFAG_USD"
@@ -91,3 +100,26 @@ REF_SECTOR_MAP = {
"1C_AS": "All Sectors",
"AllSectorsIncludingAllSectors": "All Sectors Including All Sectors",
}
def load_symbols(dataset: str) -> dict:
"""Load IMF symbol list."""
# pylint: disable=import-outside-toplevel
import json # noqa
from json.decoder import JSONDecodeError
from pathlib import Path
from openbb_core.app.model.abstract.error import OpenBBError
try:
symbols_file = Path(__file__).parents[1].joinpath("assets", "imf_symbols.json")
with symbols_file.open(encoding="utf-8") as file:
symbols = json.load(file)
except (FileNotFoundError, JSONDecodeError) as e:
raise OpenBBError(
f"Failed to load IMF symbols from the static file: {e}"
) from e
if dataset == "all":
return symbols
return {k: v for k, v in symbols.items() if v["dataset"] == dataset}

247
openbb_platform/providers/imf/openbb_imf/utils/fsi_helpers.py Normal file
View File

@@ -0,0 +1,247 @@
"""IMF FSI Data Set Helpers."""
from typing import Optional
def load_fsi_symbols() -> dict:
"""Load IMF FSI symbols."""
# pylint: disable=import-outside-toplevel
from openbb_imf.utils.constants import load_symbols
return load_symbols("FSI")
def validate_symbols(symbols) -> str:
"""Validate the IMF FSI symbols."""
# pylint: disable=import-outside-toplevel
from warnings import warn # noqa
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_imf.utils.constants import FSI_PRESETS
fsi_symbols = load_fsi_symbols()
if isinstance(symbols, str):
symbols = symbols.split(",")
elif not isinstance(symbols, list):
raise OpenBBError("Invalid symbols list.")
new_symbols: list = []
for symbol in symbols:
if symbol in FSI_PRESETS:
return symbol
if symbol.upper() not in fsi_symbols:
warn(f"Unsupported IMF FSI symbol: {symbol}")
new_symbols.append(symbol.upper())
return "+".join(new_symbols) if len(new_symbols) > 1 else new_symbols[0]
# pylint: disable=too-many-branches,too-many-statements,too-many-locals
async def _get_fsi_data(**kwargs) -> list[dict]: # noqa:PLR0912
"""Get IMF FSI data.
This function is not intended to be called directly,
but through the `ImfEconomicIndicatorsFetcher` class.
"""
# pylint: disable=import-outside-toplevel
from datetime import date # noqa
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_imf.utils.helpers import get_data
from openbb_imf.utils.irfcl_helpers import load_country_to_code_map
from openbb_imf.utils import constants
from pandas import to_datetime
from pandas.tseries import offsets
countries = kwargs.get("country", "")
countries = (
"" if countries == "all" else countries.replace(",", "+") if countries else ""
)
frequency = constants.FREQUENCY_DICT.get(kwargs.get("frequency", "quarter"), "Q")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
all_symbols = load_fsi_symbols()
core_only = [k for k, v in all_symbols.items() if v.get("table") == "fsi_core"]
encouraged_set = [
k for k, v in all_symbols.items() if v.get("table") == "fsi_encouraged_set"
]
indicator = kwargs.get("symbol")
indicators = (
"+".join(core_only if indicator == "fsi_core" else encouraged_set)
if indicator in ["fsi_core", "fsi_encouraged_set"]
else (
""
if indicator
in ["fsi_other", "fsi_all", "fsi_core_underlying", "fsi_balance_sheets"]
else validate_symbols(indicator)
)
)
if not indicators and not countries:
raise OpenBBError(
"All countries not supported for this group of indicators. Please supply a country."
)
if not start_date and not end_date and not countries:
start_date = (
date.today().replace(year=date.today().year - 1).strftime("%Y-%m-%d")
)
if start_date and not end_date:
end_date = date.today().strftime("%Y-%m-%d")
# Adjust the dates to the date relative to frequency.
# The API does not accept arbitrary dates, so we need to adjust them.
if start_date:
start_date = to_datetime(start_date)
if frequency == "Q":
start_date = offsets.QuarterBegin(startingMonth=1).rollback(start_date)
elif frequency == "A":
start_date = offsets.YearBegin().rollback(start_date)
else:
start_date = offsets.MonthBegin().rollback(start_date)
start_date = start_date.strftime("%Y-%m-%d")
if end_date:
end_date = to_datetime(end_date)
if frequency == "Q":
end_date = offsets.QuarterEnd().rollforward(end_date)
elif frequency == "A":
end_date = offsets.YearEnd().rollforward(end_date)
else:
end_date = offsets.MonthEnd().rollforward(end_date)
end_date = end_date.strftime("%Y-%m-%d")
date_range = (
f"?startPeriod={start_date}&endPeriod={end_date}"
if start_date and end_date
else ""
)
base_url = "http://dataservices.imf.org/REST/SDMX_JSON.svc/"
key = f"CompactData/FSI/{frequency}.{countries}.{indicators}"
url = f"{base_url}{key}{date_range}"
series = await get_data(url)
data: list = []
all_symbols = load_fsi_symbols()
if indicator in ["fsi_core", "fsi_encouraged_set"]:
all_symbols = {
k: v for k, v in all_symbols.items() if v.get("table") == indicator
}
elif indicator == "fsi_core_underlying":
all_symbols = {
k: v
for k, v in all_symbols.items()
if "core set" in v.get("title", "").lower() and v.get("unit") != "Percent"
}
elif indicator == "fsi_balance_sheets":
all_symbols = {
k: v
for k, v in all_symbols.items()
if "balance sheets" in v.get("title", "").lower()
}
elif indicator == "fsi_other":
all_symbols = {
k: v
for k, v in all_symbols.items()
if "Additional FSIs" in v.get("title", "")
}
country_map_dict = {
v: k.replace("_", " ").title().replace("Ecb", "ECB")
for k, v in load_country_to_code_map().items()
}
# Iterate over the series to extract observations and map the metadata.
for s in series:
if "Obs" not in s:
continue
meta = {
k.replace("@", "").lower(): (
constants.UNIT_MULTIPLIERS_MAP.get(str(v), v)
if k == "@UNIT_MULT"
else v
)
for k, v in s.items()
if k != "Obs"
}
_symbol = meta.get("indicator")
_parent: Optional[str] = None
_order: Optional[str] = None
_level: Optional[str] = None
_table: Optional[str] = None
_title: Optional[str] = None
_unit: Optional[str] = None
if _symbol not in all_symbols:
continue
_table = all_symbols.get(_symbol, {}).get("table", "")
if _table and indicator == "fsi_core_underlying":
_table = _table.replace("fsi_other", "fsi_core_underlying") # type: ignore
_parent = all_symbols.get(_symbol, {}).get("parent", "")
_order = all_symbols.get(_symbol, {}).get("order", "")
_level = all_symbols.get(_symbol, {}).get("level", "")
_title = all_symbols.get(_symbol, {}).get("title", "")
if _title:
_title = " - ".join(_title.split(", ")[1:-1]).replace(
"Financial Soundness Indicators - ", ""
)
_unit = all_symbols.get(_symbol, {}).get("unit", "")
_data = s.pop("Obs", [])
if isinstance(_data, dict):
_data = [_data]
for d in _data:
_date = d.pop("@TIME_PERIOD", None)
val = d.pop("@OBS_VALUE", None)
_ = d.pop("@OBS_STATUS", None)
if not val:
continue
val = float(val)
if _unit and _unit.lower() == "percent":
val = val / 100
if _date:
offset = (
offsets.QuarterEnd
if "Q" in _date
else offsets.YearEnd if len(str(_date)) == 4 else offsets.MonthEnd
)
_date = to_datetime(_date)
_date = _date + offset(0)
_date = _date.strftime("%Y-%m-%d")
vals = {
k: v
for k, v in {
"date": _date,
"table": _table,
"symbol": _symbol,
"parent": _parent if _parent else _symbol,
"level": _level,
"order": _order,
"country": country_map_dict.get(
meta.get("ref_area"), meta.get("ref_area")
),
"reference_sector": None,
"title": _title,
"value": val,
"unit": (
_unit.upper() if _unit and _unit in ["usd", "eur"] else _unit
),
"scale": (
"Basis Points"
if _unit and _unit.lower() == "percent"
else meta.get("unit_mult")
),
}.items()
if v
}
if vals.get("value") and vals.get("date"):
d.update(vals)
data.append(d)
if not data:
raise OpenBBError(f"No data found for, '{indicator}', in, '{countries}'.")
return data

34
openbb_platform/providers/imf/openbb_imf/utils/helpers.py Normal file
View File

@@ -0,0 +1,34 @@
"""IMF Helper Utilities."""
async def get_data(url: str) -> list[dict]:
"""Get data from the IMF API."""
# pylint: disable=import-outside-toplevel
from aiohttp.client_exceptions import ContentTypeError # noqa
from json.decoder import JSONDecodeError
from openbb_core.provider.utils.helpers import amake_request
from openbb_core.app.model.abstract.error import OpenBBError
try:
response = await amake_request(url, timeout=20)
except (JSONDecodeError, ContentTypeError) as e:
raise OpenBBError(
"Error fetching data; This might be rate-limiting. Try again later."
) from e
if "ErrorDetails" in response:
raise OpenBBError(
f"{response['ErrorDetails'].get('Code')} -> {response['ErrorDetails'].get('Message')}" # type: ignore
)
series = response.get("CompactData", {}).get("DataSet", {}).pop("Series", {}) # type: ignore
if not series:
raise OpenBBError(f"No time series data found -> {response}")
# If there is only one series, they ruturn a dict instead of a list.
if series and isinstance(series, dict):
series = [series]
return series

94
openbb_platform/providers/imf/openbb_imf/utils/irfcl_helpers.py
View File

@@ -1,27 +1,17 @@
"""IMF IRFCL Data Set Helpers."""
from typing import Dict, List, Optional
from typing import Optional
def load_irfcl_symbols() -> Dict:
def load_irfcl_symbols() -> dict:
"""Load IMF IRFCL symbols."""
# pylint: disable=import-outside-toplevel
import json # noqa
from json.decoder import JSONDecodeError
from pathlib import Path
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_imf.utils.constants import load_symbols
try:
symbols_file = Path(__file__).parents[1].joinpath("assets", "imf_symbols.json")
with symbols_file.open(encoding="utf-8") as file:
symbols = json.load(file)
except (FileNotFoundError, JSONDecodeError) as e:
raise OpenBBError(f"Failed to load IMF IRFCL symbols: {e}") from e
return {k: v for k, v in symbols.items() if v["dataset"] == "IRFCL"}
return load_symbols("IRFCL")
def load_country_map() -> Dict:
def load_country_map() -> dict:
"""Load IMF IRFCL country map."""
# pylint: disable=import-outside-toplevel
import json # noqa
@@ -47,7 +37,7 @@ def load_country_map() -> Dict:
}
def load_country_to_code_map() -> Dict:
def load_country_to_code_map() -> dict:
"""Load a map of lowercase country name to 2-letter ISO symbol."""
return {
(
@@ -76,7 +66,7 @@ def validate_countries(countries) -> str:
elif not isinstance(countries, list):
raise OpenBBError("Invalid countries list.")
new_countries: List = []
new_countries: list = []
if "all" in countries or "ALL" in countries:
return "all"
@@ -118,7 +108,7 @@ def validate_symbols(symbols) -> str:
if "IRFCL" in symbols or "all" in symbols:
return "all"
new_symbols: List = []
new_symbols: list = []
for symbol in symbols:
if symbol in IRFCL_PRESET:
@@ -132,22 +122,18 @@ def validate_symbols(symbols) -> str:
# We use this as a helper to allow future expansion of the supported IMF indicators.
# Each database has its own nuances with URL construction and schemas.
# In the future, we will check the supported symbols map and direct the call
# to the appropriate dtatabase's function.
# pylint: disable=too-many-branches,too-many-statements,too-many-locals
async def _get_irfcl_data(**kwargs) -> List[Dict]:
async def _get_irfcl_data(**kwargs) -> list[dict]:
"""Get IMF IRFCL data.
This function is not intended to be called directly,
but through the `ImfEconomicIndicatorsFetcher` class.
"""
# pylint: disable=import-outside-toplevel
from aiohttp.client_exceptions import ContentTypeError # noqa
from json.decoder import JSONDecodeError
from openbb_core.provider.utils.helpers import amake_request
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_imf.utils import constants
from openbb_imf.utils.helpers import get_data
from pandas import to_datetime
from pandas.tseries import offsets
@@ -167,8 +153,6 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
if start_date:
start_date = to_datetime(start_date)
if frequency == "Q":
# offset = offsets.QuarterBegin(startingMonth=1)
# start_date = start_date + offset
start_date = offsets.QuarterBegin(startingMonth=1).rollback(start_date)
elif frequency == "A":
start_date = offsets.YearBegin().rollback(start_date)
@@ -187,10 +171,10 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
end_date = end_date.strftime("%Y-%m-%d")
indicator = kwargs.get("symbol")
indicator = validate_symbols(indicator) if indicator else ""
indicator = "" if indicator == "all" else indicator
indicators = validate_symbols(indicator) if indicator else ""
indicators = "" if indicators == "all" else indicators
if not indicator and not countries:
if not indicators and not countries:
raise OpenBBError("Country is required when returning the complete dataset.")
date_range = (
@@ -199,31 +183,12 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
else ""
)
base_url = "http://dataservices.imf.org/REST/SDMX_JSON.svc/"
key = f"CompactData/IRFCL/{frequency}.{countries}.{indicator}.{sector}"
key = f"CompactData/IRFCL/{frequency}.{countries}.{indicators}.{sector}"
url = f"{base_url}{key}{date_range}"
try:
response = await amake_request(url, timeout=20)
except (JSONDecodeError, ContentTypeError) as e:
raise OpenBBError(
"Error fetching data; This might be rate-limiting. Try again later."
) from e
series = await get_data(url)
if "ErrorDetails" in response:
raise OpenBBError(
f"{response['ErrorDetails'].get('Code')} -> {response['ErrorDetails'].get('Message')}" # type: ignore
)
series = response.get("CompactData", {}).get("DataSet", {}).pop("Series", {}) # type: ignore
if not series:
raise OpenBBError(f"No time series data found -> {response}")
# If there is only one series, they ruturn a dict instead of a list.
if series and isinstance(series, dict):
series = [series]
data: List = []
data: list = []
all_symbols = load_irfcl_symbols()
country_map_dict = {
v: k.replace("_", " ").title().replace("Ecb", "ECB")
@@ -248,13 +213,20 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
_level: Optional[str] = None
_table: Optional[str] = None
_title: Optional[str] = None
_unit: Optional[str] = None
if _symbol in all_symbols:
_table = all_symbols.get(_symbol, {}).get("table")
_parent = all_symbols.get(_symbol, {}).get("parent", "")
_order = all_symbols.get(_symbol, {}).get("order", "")
_level = all_symbols.get(_symbol, {}).get("level", "")
_title = all_symbols.get(_symbol, {}).get("title", "")
if _symbol not in all_symbols:
continue
_table = all_symbols.get(_symbol, {}).get("table")
_parent = all_symbols.get(_symbol, {}).get("parent", "")
_order = all_symbols.get(_symbol, {}).get("order", "")
_level = all_symbols.get(_symbol, {}).get("level", "")
_title = all_symbols.get(_symbol, {}).get("title", "").replace(", ", " - ")
_unit = all_symbols.get(_symbol, {}).get("unit", "")
if _title:
_title = " - ".join(_title.split(", ")[:-1])
_data = s.pop("Obs", [])
@@ -264,6 +236,7 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
for d in _data:
_date = d.pop("@TIME_PERIOD", None)
val = d.pop("@OBS_VALUE", None)
_ = d.pop("@OBS_STATUS", None)
if not val:
continue
@@ -294,6 +267,7 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
),
"title": _title,
"value": float(val) if val else None,
"unit": _unit,
"scale": meta.get("unit_mult"),
}.items()
if v
@@ -301,11 +275,9 @@ async def _get_irfcl_data(**kwargs) -> List[Dict]:
if vals.get("value") and vals.get("date"):
d.update(vals)
if _data:
data.extend([d for d in _data if d])
data.append(d)
if not data:
raise OpenBBError("No data found.")
raise OpenBBError(f"No data found for '{indicator}' in '{countries}'.")
return data