From b61c3e26f30cc8938213ebdfb2446784a822b4c7 Mon Sep 17 00:00:00 2001 From: alexpipipi Date: Wed, 1 Jul 2026 12:56:55 -0400 Subject: [PATCH] feat: add Credit & Sovereign Risk, Sanctions, and Interest Rates APIs Add three new API families covering 14 GET endpoints: - Credit & Sovereign Risk (/api/credit-risk): sovereign risk-premium, credit-ratings, cds-spreads, default-spreads; corporate cmdi, hqm-yields; cds-market aggregates - Sanctions (/api/sanctions): entities, vessels, programs, sources - Interest Rates (/api/rates, /api/spreads): reference-rates, policy-rates, funding-stress (no pagination) credit-risk and rates/spreads use filter[...] deep-object params; sanctions uses BARE query params (q=, program=, type=, active=, imo=, flag=, vessel_type=, source=, country=) per prometheus-web + live prod. Pagination is page[offset]/page[limit] throughout (except funding-stress: none). Filter/param values are URL-encoded via a shared helper so multi-word values do not corrupt the query string. Sanctions validates source->ofac and type/active enums. Registered in APIs/__init__.py, exposed via APIClient facade, covered by pytest. --- eodhd/APIs/CreditSovereignRiskAPI.py | 244 +++++++++++++++++++++++ eodhd/APIs/InterestRatesAPI.py | 134 +++++++++++++ eodhd/APIs/SanctionsAPI.py | 217 +++++++++++++++++++++ eodhd/APIs/__init__.py | 3 + eodhd/apiclient.py | 282 +++++++++++++++++++++++++++ tests/test_credit_sovereign_risk.py | 140 +++++++++++++ tests/test_interest_rates.py | 104 ++++++++++ tests/test_sanctions.py | 190 ++++++++++++++++++ 8 files changed, 1314 insertions(+) create mode 100644 eodhd/APIs/CreditSovereignRiskAPI.py create mode 100644 eodhd/APIs/InterestRatesAPI.py create mode 100644 eodhd/APIs/SanctionsAPI.py create mode 100644 tests/test_credit_sovereign_risk.py create mode 100644 tests/test_interest_rates.py create mode 100644 tests/test_sanctions.py diff --git a/eodhd/APIs/CreditSovereignRiskAPI.py b/eodhd/APIs/CreditSovereignRiskAPI.py new file mode 100644 index 0000000..cfc9994 --- /dev/null +++ b/eodhd/APIs/CreditSovereignRiskAPI.py @@ -0,0 +1,244 @@ +# APIs/CreditSovereignRiskAPI.py + +from urllib.parse import quote + +from .BaseAPI import BaseAPI + + +class CreditSovereignRiskAPI(BaseAPI): + """ + Wrapper for Credit & Sovereign Risk endpoints: + + GET /api/credit-risk/sovereign/risk-premium + GET /api/credit-risk/sovereign/credit-ratings + GET /api/credit-risk/sovereign/cds-spreads + GET /api/credit-risk/sovereign/default-spreads + GET /api/credit-risk/corporate/cmdi + GET /api/credit-risk/corporate/hqm-yields + GET /api/credit-risk/cds-market/aggregates + + Notes: + - All endpoints return a JSON envelope { data, meta, links }. + - Filtering uses filter[...] deep-object params. + - Pagination uses page[offset] and page[limit]. + """ + + @staticmethod + def _filter(name: str, value) -> str: + """Build a URL-encoded &filter[name]=value fragment (empty if value is None).""" + if value is None: + return "" + return f"&filter[{name}]={quote(str(value), safe='')}" + + @staticmethod + def _pagination(page_offset: int = None, page_limit: int = None) -> str: + query_string = "" + if page_offset is not None: + page_offset = int(page_offset) + if page_offset < 0: + raise ValueError("page_offset must be >= 0.") + query_string += f"&page[offset]={page_offset}" + if page_limit is not None: + page_limit = int(page_limit) + if page_limit < 1: + raise ValueError("page_limit must be >= 1.") + query_string += f"&page[limit]={page_limit}" + return query_string + + def get_sovereign_risk_premium( + self, + api_token: str, + country: str = None, + region: str = None, + as_of: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/sovereign/risk-premium + + Fields: country_iso3, country_name, as_of_date, moodys_rating, + adj_default_spread, country_risk_premium, equity_risk_premium, + corporate_tax_rate, sovereign_cds (nullable), source. + """ + query_string = "" + query_string += self._filter("country", country) + query_string += self._filter("region", region) + query_string += self._filter("as_of", as_of) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="sovereign/risk-premium", + querystring=query_string, + ) + + def get_sovereign_credit_ratings( + self, + api_token: str, + country: str = None, + as_of: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/sovereign/credit-ratings + + Fields: country_iso3, country_name, as_of_date, moodys_rating, + sp_rating, fitch_rating, source. + """ + query_string = "" + query_string += self._filter("country", country) + query_string += self._filter("as_of", as_of) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="sovereign/credit-ratings", + querystring=query_string, + ) + + def get_sovereign_cds_spreads( + self, + api_token: str, + country: str = None, + as_of: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/sovereign/cds-spreads + + Fields: country_iso3, country_name, as_of_date, moodys_rating, + cds_spread (nullable), cds_spread_net_of_switzerland (nullable), source. + """ + query_string = "" + query_string += self._filter("country", country) + query_string += self._filter("as_of", as_of) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="sovereign/cds-spreads", + querystring=query_string, + ) + + def get_sovereign_default_spreads( + self, + api_token: str, + rating: str = None, + as_of: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/sovereign/default-spreads + + Fields: rating, as_of_date, default_spread, source. + """ + query_string = "" + query_string += self._filter("rating", rating) + query_string += self._filter("as_of", as_of) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="sovereign/default-spreads", + querystring=query_string, + ) + + def get_corporate_cmdi( + self, + api_token: str, + from_date: str = None, + to_date: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/corporate/cmdi + + Filters: filter[from], filter[to]. + Fields: as_of_date, market_cmdi, ig_cmdi, hy_cmdi, source. + """ + query_string = "" + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="corporate/cmdi", + querystring=query_string, + ) + + def get_corporate_hqm_yields( + self, + api_token: str, + tenor: str = None, + type: str = None, + from_date: str = None, + to_date: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/corporate/hqm-yields + + Filters: filter[tenor], filter[type] (par|spot), filter[from], filter[to]. + Fields: series_id, tenor_years, yield_type, as_of_date, yield_value, source. + """ + if type is not None and str(type).lower() not in ("par", "spot"): + raise ValueError("type must be 'par' or 'spot'.") + + query_string = "" + query_string += self._filter("tenor", tenor) + if type is not None: + query_string += self._filter("type", str(type).lower()) + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="corporate/hqm-yields", + querystring=query_string, + ) + + def get_cds_market_aggregates( + self, + api_token: str, + metric: str = None, + dimension: str = None, + from_date: str = None, + to_date: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/credit-risk/cds-market/aggregates + + Filters: filter[metric] (gross_notional), filter[dimension] (grade|cleared_status), + filter[from], filter[to]. + Fields: as_of_date, release_date, metric, breakdown_dimension, + breakdown_value, region, usd_notional_mn, source. + """ + query_string = "" + query_string += self._filter("metric", metric) + query_string += self._filter("dimension", dimension) + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="credit-risk", + uri="cds-market/aggregates", + querystring=query_string, + ) diff --git a/eodhd/APIs/InterestRatesAPI.py b/eodhd/APIs/InterestRatesAPI.py new file mode 100644 index 0000000..d7b3510 --- /dev/null +++ b/eodhd/APIs/InterestRatesAPI.py @@ -0,0 +1,134 @@ +# APIs/InterestRatesAPI.py + +from urllib.parse import quote + +from .BaseAPI import BaseAPI + + +class InterestRatesAPI(BaseAPI): + """ + Wrapper for Interest Rates & Spreads endpoints: + + GET /api/rates/reference-rates + GET /api/rates/policy-rates + GET /api/spreads/funding-stress + + Notes: + - All endpoints return a JSON envelope { data, meta, links }. + - Filtering uses filter[...] deep-object params. + - rates/* endpoints paginate with page[offset]/page[limit]. + - spreads/funding-stress does NOT support pagination. + """ + + @staticmethod + def _filter(name: str, value) -> str: + """Build a URL-encoded &filter[name]=value fragment (empty if value is None).""" + if value is None: + return "" + return f"&filter[{name}]={quote(str(value), safe='')}" + + @staticmethod + def _pagination(page_offset: int = None, page_limit: int = None) -> str: + query_string = "" + if page_offset is not None: + page_offset = int(page_offset) + if page_offset < 0: + raise ValueError("page_offset must be >= 0.") + query_string += f"&page[offset]={page_offset}" + if page_limit is not None: + page_limit = int(page_limit) + if page_limit < 1: + raise ValueError("page_limit must be >= 1.") + query_string += f"&page[limit]={page_limit}" + return query_string + + def get_reference_rates( + self, + api_token: str, + code: str = None, + currency: str = None, + from_date: str = None, + to_date: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/rates/reference-rates + + Filters: filter[code], filter[currency] (USD|GBP|EUR), filter[from], filter[to]. + Fields: date, code, currency, rate_type, rate, source, source_series_id. + """ + if currency is not None and str(currency).upper() not in ("USD", "GBP", "EUR"): + raise ValueError("currency must be one of 'USD', 'GBP', 'EUR'.") + + query_string = "" + query_string += self._filter("code", code) + if currency is not None: + query_string += self._filter("currency", str(currency).upper()) + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="rates", + uri="reference-rates", + querystring=query_string, + ) + + def get_policy_rates( + self, + api_token: str, + code: str = None, + country: str = None, + central_bank: str = None, + from_date: str = None, + to_date: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/rates/policy-rates + + Filters: filter[code], filter[country], filter[central_bank], filter[from], filter[to]. + Fields: date, code, country, central_bank, rate, source, source_series_id. + """ + query_string = "" + query_string += self._filter("code", code) + query_string += self._filter("country", country) + query_string += self._filter("central_bank", central_bank) + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="rates", + uri="policy-rates", + querystring=query_string, + ) + + def get_funding_stress( + self, + api_token: str, + code: str = None, + from_date: str = None, + to_date: str = None, + ): + """ + GET /api/spreads/funding-stress + + Filters: filter[code], filter[from], filter[to]. NO pagination. + Fields: date, code, value_bps, formula, leg_a, leg_b, leg_a_rate, leg_b_rate. + """ + query_string = "" + query_string += self._filter("code", code) + query_string += self._filter("from", from_date) + query_string += self._filter("to", to_date) + + return self._rest_get_method( + api_key=api_token, + endpoint="spreads", + uri="funding-stress", + querystring=query_string, + ) diff --git a/eodhd/APIs/SanctionsAPI.py b/eodhd/APIs/SanctionsAPI.py new file mode 100644 index 0000000..f47511c --- /dev/null +++ b/eodhd/APIs/SanctionsAPI.py @@ -0,0 +1,217 @@ +# APIs/SanctionsAPI.py + +from urllib.parse import quote + +from .BaseAPI import BaseAPI + + +class SanctionsAPI(BaseAPI): + """ + Wrapper for Sanctions endpoints: + + GET /api/sanctions/entities + GET /api/sanctions/vessels + GET /api/sanctions/programs + GET /api/sanctions/sources + + Notes: + - All endpoints return a JSON envelope { data, meta, links }. + - Filtering uses BARE query params (e.g. program=, q=, type=), NOT filter[...]. + (This differs from other EODHD endpoints; verified against prometheus-web + request classes + live production.) + - Pagination uses page[offset] and page[limit]. + """ + + _SOURCE_VALUES = ("ofac",) + _ENTITY_TYPE_VALUES = ("individual", "entity", "vessel", "aircraft") + + @staticmethod + def _param(name: str, value) -> str: + """Build a URL-encoded &name=value fragment. + + Returns "" for None or blank/whitespace-only values (so an empty string + is treated as "omit the param", not "send &name="). Values are + URL-encoded; the key is a bare query param (not filter[...]).""" + if value is None: + return "" + text = str(value).strip() + if text == "": + return "" + return f"&{name}={quote(text, safe='')}" + + @staticmethod + def _normalize_active(active): + """Normalise the `active` filter to the API's "true"/"false" scalar. + + Accepts Python bools, or the strings "true"/"false" (any case). Anything + else raises ValueError. Returns None if `active` is None.""" + if active is None: + return None + if isinstance(active, bool): + return "true" if active else "false" + text = str(active).strip().lower() + if text not in ("true", "false"): + raise ValueError("active must be a bool or one of 'true'/'false'.") + return text + + @staticmethod + def _pagination(page_offset: int = None, page_limit: int = None) -> str: + query_string = "" + if page_offset is not None: + page_offset = int(page_offset) + if page_offset < 0: + raise ValueError("page_offset must be >= 0.") + query_string += f"&page[offset]={page_offset}" + if page_limit is not None: + page_limit = int(page_limit) + if page_limit < 1: + raise ValueError("page_limit must be >= 1.") + query_string += f"&page[limit]={page_limit}" + return query_string + + def get_entities( + self, + api_token: str, + q: str = None, + program: str = None, + country: str = None, + source: str = None, + entity_type: str = None, + active: bool = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/sanctions/entities + + Query params (bare, not filter[...]): + q - free-text search (min 2 chars) + program - sanctions program + country - country + source - data source; currently only "ofac" + type - entity type: "individual" | "entity" | "vessel" | "aircraft" + (Python arg name: entity_type) + active - "true" | "false" (Python bool accepted) + page[offset], page[limit] - pagination + + Fields: source, source_uid, entity_type, name, programs[], country, + remarks, listed_date, is_active, aliases[], identifiers[]. + """ + if q is not None and len(str(q).strip()) < 2: + raise ValueError("q (search) must be at least 2 characters.") + if source is not None and str(source).strip().lower() not in self._SOURCE_VALUES: + raise ValueError(f"source must be one of {self._SOURCE_VALUES}.") + if entity_type is not None and str(entity_type).strip().lower() not in self._ENTITY_TYPE_VALUES: + raise ValueError(f"type must be one of {self._ENTITY_TYPE_VALUES}.") + active = self._normalize_active(active) + + query_string = "" + query_string += self._param("q", q) + query_string += self._param("program", program) + query_string += self._param("country", country) + if source is not None: + query_string += self._param("source", str(source).strip().lower()) + if entity_type is not None: + query_string += self._param("type", str(entity_type).strip().lower()) + if active is not None: + query_string += self._param("active", active) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="sanctions", + uri="entities", + querystring=query_string, + ) + + def get_vessels( + self, + api_token: str, + q: str = None, + imo: str = None, + flag: str = None, + vessel_type: str = None, + program: str = None, + source: str = None, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/sanctions/vessels + + Query params (bare, not filter[...]): + q - free-text search (min 2 chars) + imo - IMO number + flag - flag + vessel_type - vessel type + program - sanctions program + source - data source; currently only "ofac" + page[offset], page[limit] - pagination + + Fields: call_sign, vessel_type, flag, tonnage, gross_tonnage, owner, + imo_number, mmsi, entity_source_uid, entity_name, source, programs[], + country, is_active. + """ + if q is not None and len(str(q).strip()) < 2: + raise ValueError("q (search) must be at least 2 characters.") + if source is not None and str(source).strip().lower() not in self._SOURCE_VALUES: + raise ValueError(f"source must be one of {self._SOURCE_VALUES}.") + + query_string = "" + query_string += self._param("q", q) + query_string += self._param("imo", imo) + query_string += self._param("flag", flag) + query_string += self._param("vessel_type", vessel_type) + query_string += self._param("program", program) + if source is not None: + query_string += self._param("source", str(source).strip().lower()) + query_string += self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="sanctions", + uri="vessels", + querystring=query_string, + ) + + def get_programs( + self, + api_token: str, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/sanctions/programs + + No filter params (pagination only). + Fields: program, count. + """ + query_string = self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="sanctions", + uri="programs", + querystring=query_string, + ) + + def get_sources( + self, + api_token: str, + page_offset: int = None, + page_limit: int = None, + ): + """ + GET /api/sanctions/sources + + No filter params (pagination only). + Fields: name. + """ + query_string = self._pagination(page_offset, page_limit) + + return self._rest_get_method( + api_key=api_token, + endpoint="sanctions", + uri="sources", + querystring=query_string, + ) diff --git a/eodhd/APIs/__init__.py b/eodhd/APIs/__init__.py index 0798a7e..647bc8d 100644 --- a/eodhd/APIs/__init__.py +++ b/eodhd/APIs/__init__.py @@ -34,6 +34,9 @@ from .BulkFundamentalsAPI import BulkFundamentalsAPI from .TreasuryAPI import TreasuryAPI from .ExchangeDetailsV2API import ExchangeDetailsV2API +from .CreditSovereignRiskAPI import CreditSovereignRiskAPI +from .SanctionsAPI import SanctionsAPI +from .InterestRatesAPI import InterestRatesAPI #Marketplace endpoints from .MPIndexComponentsAPI import MPIndexComponentsAPI diff --git a/eodhd/apiclient.py b/eodhd/apiclient.py index b380002..ea868f1 100644 --- a/eodhd/apiclient.py +++ b/eodhd/apiclient.py @@ -51,6 +51,9 @@ from eodhd.APIs import BulkFundamentalsAPI from eodhd.APIs import TreasuryAPI from eodhd.APIs import ExchangeDetailsV2API +from eodhd.APIs import CreditSovereignRiskAPI +from eodhd.APIs import SanctionsAPI +from eodhd.APIs import InterestRatesAPI #Marketplace endpoints from eodhd.APIs import MPIndexComponentsAPI @@ -1644,6 +1647,285 @@ def mp_unicornbay_logo(self, symbol): api_call = MPUnicornbayExtrasAPI(session=self._session, timeout=self._timeout) return api_call.get_logo(api_token=self._api_key, symbol=symbol) + # ------------------------------------------------------------------ + # Credit & Sovereign Risk API (/api/credit-risk) + # ------------------------------------------------------------------ + def get_sovereign_risk_premium(self, country=None, region=None, as_of=None, + page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: sovereign risk premium + Endpoint: GET /api/credit-risk/sovereign/risk-premium + + Args: + country [OPTIONAL] - filter[country], e.g. "USA" or "US" + region [OPTIONAL] - filter[region] + as_of [OPTIONAL] - filter[as_of], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_sovereign_risk_premium( + api_token=self._api_key, country=country, region=region, as_of=as_of, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_sovereign_credit_ratings(self, country=None, as_of=None, + page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: sovereign credit ratings + Endpoint: GET /api/credit-risk/sovereign/credit-ratings + + Args: + country [OPTIONAL] - filter[country] + as_of [OPTIONAL] - filter[as_of], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_sovereign_credit_ratings( + api_token=self._api_key, country=country, as_of=as_of, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_sovereign_cds_spreads(self, country=None, as_of=None, + page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: sovereign CDS spreads + Endpoint: GET /api/credit-risk/sovereign/cds-spreads + + Args: + country [OPTIONAL] - filter[country] + as_of [OPTIONAL] - filter[as_of], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_sovereign_cds_spreads( + api_token=self._api_key, country=country, as_of=as_of, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_sovereign_default_spreads(self, rating=None, as_of=None, + page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: sovereign default spreads + Endpoint: GET /api/credit-risk/sovereign/default-spreads + + Args: + rating [OPTIONAL] - filter[rating] + as_of [OPTIONAL] - filter[as_of], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_sovereign_default_spreads( + api_token=self._api_key, rating=rating, as_of=as_of, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_corporate_cmdi(self, from_date=None, to_date=None, + page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: corporate CMDI (credit market distress index) + Endpoint: GET /api/credit-risk/corporate/cmdi + + Args: + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_corporate_cmdi( + api_token=self._api_key, from_date=from_date, to_date=to_date, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_corporate_hqm_yields(self, tenor=None, type=None, from_date=None, + to_date=None, page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: corporate HQM yields + Endpoint: GET /api/credit-risk/corporate/hqm-yields + + Args: + tenor [OPTIONAL] - filter[tenor] + type [OPTIONAL] - filter[type], "par" or "spot" + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_corporate_hqm_yields( + api_token=self._api_key, tenor=tenor, type=type, from_date=from_date, + to_date=to_date, page_offset=page_offset, page_limit=page_limit, + ) + + def get_cds_market_aggregates(self, metric=None, dimension=None, from_date=None, + to_date=None, page_offset=None, page_limit=None): + """ + Credit & Sovereign Risk: CDS market aggregates + Endpoint: GET /api/credit-risk/cds-market/aggregates + + Args: + metric [OPTIONAL] - filter[metric], e.g. "gross_notional" + dimension [OPTIONAL] - filter[dimension], "grade" or "cleared_status" + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = CreditSovereignRiskAPI(session=self._session, timeout=self._timeout) + return api_call.get_cds_market_aggregates( + api_token=self._api_key, metric=metric, dimension=dimension, + from_date=from_date, to_date=to_date, + page_offset=page_offset, page_limit=page_limit, + ) + + # ------------------------------------------------------------------ + # Sanctions API (/api/sanctions) + # ------------------------------------------------------------------ + def get_sanctions_entities(self, q=None, program=None, country=None, source=None, + entity_type=None, active=None, + page_offset=None, page_limit=None): + """ + Sanctions: sanctioned entities + Endpoint: GET /api/sanctions/entities + + Note: sanctions endpoints use bare query params, NOT filter[...]. + + Args: + q [OPTIONAL] - free-text search (min 2 chars) + program [OPTIONAL] - program + country [OPTIONAL] - country + source [OPTIONAL] - source (currently only "ofac") + entity_type [OPTIONAL] - type: "individual"|"entity"|"vessel"|"aircraft" + active [OPTIONAL] - "true"/"false" (bool accepted) + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = SanctionsAPI(session=self._session, timeout=self._timeout) + return api_call.get_entities( + api_token=self._api_key, q=q, program=program, country=country, + source=source, entity_type=entity_type, active=active, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_sanctions_vessels(self, q=None, imo=None, flag=None, vessel_type=None, + program=None, source=None, + page_offset=None, page_limit=None): + """ + Sanctions: sanctioned vessels + Endpoint: GET /api/sanctions/vessels + + Note: sanctions endpoints use bare query params, NOT filter[...]. + + Args: + q [OPTIONAL] - free-text search (min 2 chars) + imo [OPTIONAL] - IMO number + flag [OPTIONAL] - flag + vessel_type [OPTIONAL] - vessel type + program [OPTIONAL] - program + source [OPTIONAL] - source (currently only "ofac") + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = SanctionsAPI(session=self._session, timeout=self._timeout) + return api_call.get_vessels( + api_token=self._api_key, q=q, imo=imo, flag=flag, vessel_type=vessel_type, + program=program, source=source, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_sanctions_programs(self, page_offset=None, page_limit=None): + """ + Sanctions: list of sanctions programs + Endpoint: GET /api/sanctions/programs + + Returns: dict envelope { data, meta, links }; data items have program, count. + """ + api_call = SanctionsAPI(session=self._session, timeout=self._timeout) + return api_call.get_programs( + api_token=self._api_key, page_offset=page_offset, page_limit=page_limit, + ) + + def get_sanctions_sources(self, page_offset=None, page_limit=None): + """ + Sanctions: list of sanctions sources + Endpoint: GET /api/sanctions/sources + + Returns: dict envelope { data, meta, links }; data items have name. + """ + api_call = SanctionsAPI(session=self._session, timeout=self._timeout) + return api_call.get_sources( + api_token=self._api_key, page_offset=page_offset, page_limit=page_limit, + ) + + # ------------------------------------------------------------------ + # Interest Rates & Spreads API (/api/rates, /api/spreads) + # ------------------------------------------------------------------ + def get_reference_rates(self, code=None, currency=None, from_date=None, to_date=None, + page_offset=None, page_limit=None): + """ + Interest Rates: reference rates + Endpoint: GET /api/rates/reference-rates + + Args: + code [OPTIONAL] - filter[code] + currency [OPTIONAL] - filter[currency], "USD", "GBP" or "EUR" + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = InterestRatesAPI(session=self._session, timeout=self._timeout) + return api_call.get_reference_rates( + api_token=self._api_key, code=code, currency=currency, + from_date=from_date, to_date=to_date, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_policy_rates(self, code=None, country=None, central_bank=None, + from_date=None, to_date=None, + page_offset=None, page_limit=None): + """ + Interest Rates: central bank policy rates + Endpoint: GET /api/rates/policy-rates + + Args: + code [OPTIONAL] - filter[code] + country [OPTIONAL] - filter[country] + central_bank [OPTIONAL] - filter[central_bank] + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + page_offset / page_limit [OPTIONAL] - pagination + Returns: dict envelope { data, meta, links } + """ + api_call = InterestRatesAPI(session=self._session, timeout=self._timeout) + return api_call.get_policy_rates( + api_token=self._api_key, code=code, country=country, + central_bank=central_bank, from_date=from_date, to_date=to_date, + page_offset=page_offset, page_limit=page_limit, + ) + + def get_funding_stress(self, code=None, from_date=None, to_date=None): + """ + Spreads: funding stress spreads + Endpoint: GET /api/spreads/funding-stress + + Note: this endpoint does NOT support pagination. + + Args: + code [OPTIONAL] - filter[code] + from_date [OPTIONAL] - filter[from], YYYY-MM-DD + to_date [OPTIONAL] - filter[to], YYYY-MM-DD + Returns: dict envelope { data, meta, links } + """ + api_call = InterestRatesAPI(session=self._session, timeout=self._timeout) + return api_call.get_funding_stress( + api_token=self._api_key, code=code, from_date=from_date, to_date=to_date, + ) + class ScannerClient: """Scanner class""" diff --git a/tests/test_credit_sovereign_risk.py b/tests/test_credit_sovereign_risk.py new file mode 100644 index 0000000..183ec52 --- /dev/null +++ b/tests/test_credit_sovereign_risk.py @@ -0,0 +1,140 @@ +"""Tests for CreditSovereignRiskAPI.""" + +import pytest +from unittest.mock import MagicMock + +from eodhd.APIs.CreditSovereignRiskAPI import CreditSovereignRiskAPI + + +@pytest.fixture +def mock_session(): + return MagicMock() + + +def _make_api(session): + return CreditSovereignRiskAPI(session=session) + + +def _mock_response(session, data=None): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = data if data is not None else {"data": [], "meta": {}, "links": {}} + session.get.return_value = resp + + +def test_risk_premium_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sovereign_risk_premium(api_token="test1234567890123456") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/sovereign/risk-premium" in call_url + assert "api_token=test1234567890123456" in call_url + + +def test_risk_premium_filters(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sovereign_risk_premium( + api_token="test1234567890123456", country="USA", region="North America", + as_of="2024-01-01", page_offset=0, page_limit=50, + ) + + call_url = mock_session.get.call_args[0][0] + assert "filter[country]=USA" in call_url + # filter values are URL-encoded (spaces -> %20) + assert "filter[region]=North%20America" in call_url + assert "filter[as_of]=2024-01-01" in call_url + assert "page[offset]=0" in call_url + assert "page[limit]=50" in call_url + + +def test_credit_ratings_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sovereign_credit_ratings(api_token="test1234567890123456", country="DEU") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/sovereign/credit-ratings" in call_url + assert "filter[country]=DEU" in call_url + + +def test_cds_spreads_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sovereign_cds_spreads(api_token="test1234567890123456", as_of="2024-06-01") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/sovereign/cds-spreads" in call_url + assert "filter[as_of]=2024-06-01" in call_url + + +def test_default_spreads_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sovereign_default_spreads(api_token="test1234567890123456", rating="Aaa") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/sovereign/default-spreads" in call_url + assert "filter[rating]=Aaa" in call_url + + +def test_corporate_cmdi_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_corporate_cmdi(api_token="test1234567890123456", from_date="2024-01-01", to_date="2024-12-31") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/corporate/cmdi" in call_url + assert "filter[from]=2024-01-01" in call_url + assert "filter[to]=2024-12-31" in call_url + + +def test_hqm_yields_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_corporate_hqm_yields(api_token="test1234567890123456", tenor="10", type="par") + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/corporate/hqm-yields" in call_url + assert "filter[tenor]=10" in call_url + assert "filter[type]=par" in call_url + + +def test_hqm_yields_invalid_type(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_corporate_hqm_yields(api_token="test1234567890123456", type="bad") + + +def test_cds_market_aggregates_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_cds_market_aggregates( + api_token="test1234567890123456", metric="gross_notional", dimension="grade", + ) + + call_url = mock_session.get.call_args[0][0] + assert "/credit-risk/cds-market/aggregates" in call_url + assert "filter[metric]=gross_notional" in call_url + assert "filter[dimension]=grade" in call_url + + +def test_invalid_pagination(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_sovereign_risk_premium(api_token="test1234567890123456", page_offset=-1) + with pytest.raises(ValueError): + api.get_sovereign_risk_premium(api_token="test1234567890123456", page_limit=0) + + +def test_filter_values_are_url_encoded(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + # a value with a space and an ampersand must not break the query string + api.get_sovereign_risk_premium(api_token="test1234567890123456", region="A & B") + + call_url = mock_session.get.call_args[0][0] + assert "filter[region]=A%20%26%20B" in call_url + # the raw ampersand must not appear inside the filter value + assert "filter[region]=A & B" not in call_url diff --git a/tests/test_interest_rates.py b/tests/test_interest_rates.py new file mode 100644 index 0000000..1be36ce --- /dev/null +++ b/tests/test_interest_rates.py @@ -0,0 +1,104 @@ +"""Tests for InterestRatesAPI.""" + +import pytest +from unittest.mock import MagicMock + +from eodhd.APIs.InterestRatesAPI import InterestRatesAPI + + +@pytest.fixture +def mock_session(): + return MagicMock() + + +def _make_api(session): + return InterestRatesAPI(session=session) + + +def _mock_response(session, data=None): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = data if data is not None else {"data": [], "meta": {}, "links": {}} + session.get.return_value = resp + + +def test_reference_rates_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_reference_rates(api_token="test1234567890123456", code="SOFR", currency="USD") + + call_url = mock_session.get.call_args[0][0] + assert "/rates/reference-rates" in call_url + assert "filter[code]=SOFR" in call_url + assert "filter[currency]=USD" in call_url + + +def test_reference_rates_pagination(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_reference_rates( + api_token="test1234567890123456", from_date="2024-01-01", to_date="2024-06-01", + page_offset=10, page_limit=100, + ) + + call_url = mock_session.get.call_args[0][0] + assert "filter[from]=2024-01-01" in call_url + assert "filter[to]=2024-06-01" in call_url + assert "page[offset]=10" in call_url + assert "page[limit]=100" in call_url + + +def test_reference_rates_invalid_currency(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_reference_rates(api_token="test1234567890123456", currency="JPY") + + +def test_policy_rates_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_policy_rates( + api_token="test1234567890123456", code="FEDFUNDS", country="USA", + central_bank="Federal Reserve", + ) + + call_url = mock_session.get.call_args[0][0] + assert "/rates/policy-rates" in call_url + assert "filter[code]=FEDFUNDS" in call_url + assert "filter[country]=USA" in call_url + # filter values are URL-encoded (spaces -> %20) + assert "filter[central_bank]=Federal%20Reserve" in call_url + + +def test_funding_stress_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_funding_stress( + api_token="test1234567890123456", code="SOFR_OIS", from_date="2024-01-01", + to_date="2024-06-01", + ) + + call_url = mock_session.get.call_args[0][0] + assert "/spreads/funding-stress" in call_url + assert "filter[code]=SOFR_OIS" in call_url + assert "filter[from]=2024-01-01" in call_url + assert "filter[to]=2024-06-01" in call_url + + +def test_funding_stress_no_pagination_kwargs(mock_session): + # funding-stress has no pagination params; calling with only filters works + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_funding_stress(api_token="test1234567890123456", code="TED") + + call_url = mock_session.get.call_args[0][0] + assert "page[offset]" not in call_url + assert "page[limit]" not in call_url + + +def test_invalid_pagination(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_reference_rates(api_token="test1234567890123456", page_offset=-1) + with pytest.raises(ValueError): + api.get_policy_rates(api_token="test1234567890123456", page_limit=0) diff --git a/tests/test_sanctions.py b/tests/test_sanctions.py new file mode 100644 index 0000000..093acc2 --- /dev/null +++ b/tests/test_sanctions.py @@ -0,0 +1,190 @@ +"""Tests for SanctionsAPI. + +Note: sanctions endpoints use BARE query params (program=, q=, type=, active=, +imo=, flag=, vessel_type=, source=, country=), NOT filter[...]. Pagination still +uses page[offset]/page[limit]. +""" + +import pytest +from unittest.mock import MagicMock + +from eodhd.APIs.SanctionsAPI import SanctionsAPI + + +@pytest.fixture +def mock_session(): + return MagicMock() + + +def _make_api(session): + return SanctionsAPI(session=session) + + +def _mock_response(session, data=None): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = data if data is not None else {"data": [], "meta": {}, "links": {}} + session.get.return_value = resp + + +def test_entities_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_entities(api_token="test1234567890123456") + + call_url = mock_session.get.call_args[0][0] + assert "/sanctions/entities" in call_url + assert "api_token=test1234567890123456" in call_url + + +def test_entities_bare_params(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_entities( + api_token="test1234567890123456", q="Acme", program="OFAC-SDN", + country="RU", source="ofac", entity_type="individual", active=True, + page_offset=0, page_limit=25, + ) + + call_url = mock_session.get.call_args[0][0] + # bare keys, not filter[...] + assert "&q=Acme" in call_url + assert "&program=OFAC-SDN" in call_url + assert "&country=RU" in call_url + assert "&source=ofac" in call_url + assert "&type=individual" in call_url + assert "&active=true" in call_url + assert "page[offset]=0" in call_url + assert "page[limit]=25" in call_url + # must NOT use filter[...] style + assert "filter[" not in call_url + + +def test_entities_active_false(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_entities(api_token="test1234567890123456", active=False) + + call_url = mock_session.get.call_args[0][0] + assert "&active=false" in call_url + + +def test_entities_q_too_short(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", q="a") + + +def test_entities_invalid_source(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", source="eu") + + +def test_entities_invalid_type(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", entity_type="ship") + + +def test_entities_value_url_encoded(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_entities(api_token="test1234567890123456", program="A & B") + + call_url = mock_session.get.call_args[0][0] + assert "&program=A%20%26%20B" in call_url + assert "&program=A & B" not in call_url + + +def test_vessels_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_vessels( + api_token="test1234567890123456", imo="9074729", flag="RU", + vessel_type="cargo", q="tanker", program="OFAC-SDN", source="ofac", + ) + + call_url = mock_session.get.call_args[0][0] + assert "/sanctions/vessels" in call_url + assert "&imo=9074729" in call_url + assert "&flag=RU" in call_url + assert "&vessel_type=cargo" in call_url + assert "&q=tanker" in call_url + assert "&program=OFAC-SDN" in call_url + assert "&source=ofac" in call_url + assert "filter[" not in call_url + + +def test_vessels_q_too_short(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_vessels(api_token="test1234567890123456", q="x") + + +def test_vessels_invalid_source(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_vessels(api_token="test1234567890123456", source="un") + + +def test_programs_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_programs(api_token="test1234567890123456", page_limit=100) + + call_url = mock_session.get.call_args[0][0] + assert "/sanctions/programs" in call_url + assert "page[limit]=100" in call_url + + +def test_sources_url(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_sources(api_token="test1234567890123456") + + call_url = mock_session.get.call_args[0][0] + assert "/sanctions/sources" in call_url + + +def test_invalid_pagination(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", page_offset=-1) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", page_limit=0) + + +def test_active_string_accepted(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + api.get_entities(api_token="test1234567890123456", active="TRUE") + + call_url = mock_session.get.call_args[0][0] + assert "&active=true" in call_url + + +def test_active_invalid_value(mock_session): + api = _make_api(mock_session) + with pytest.raises(ValueError): + api.get_entities(api_token="test1234567890123456", active="yes") + + +def test_source_whitespace_tolerated(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + # surrounding whitespace should not cause a spurious ValueError + api.get_entities(api_token="test1234567890123456", source="ofac ") + + call_url = mock_session.get.call_args[0][0] + assert "&source=ofac" in call_url + + +def test_blank_value_omitted(mock_session): + _mock_response(mock_session) + api = _make_api(mock_session) + # empty/whitespace-only values are omitted, not sent as &program= + api.get_entities(api_token="test1234567890123456", program=" ") + + call_url = mock_session.get.call_args[0][0] + assert "&program=" not in call_url