Openfigi
openfigi ¶
__all__ = ['OpenFIGIClient', 'OpenFIGIRecord'] module-attribute ¶
OpenFIGIClient(api_key: str | None = None, timeout: float = 30.0) ¶
Client for the OpenFIGI mapping REST API.
Source code in src/equities_classifier/clients/openfigi/client.py
def __init__(self, api_key: str | None = None, timeout: float = 30.0) -> None:
"""init the HTTP client."""
headers: dict[str, str] = {
"Content-Type": "application/json"
}
self._api_key = api_key
if api_key:
headers["X-OPENFIGI-APIKEY"] = api_key
self._limits = self._AUTHENTICATED_LIMITS
else:
self._limits = self._ANONYMOUS_LIMITS
self._rate_limiter = RateLimiter(requests_per_minute=self._limits.requests_per_minute)
self._client = httpx.Client(
headers=headers,
timeout=timeout,
)
__enter__() -> Self ¶
__exit__(*_: object) -> None ¶
close() -> None ¶
read_provider_base_data(source_identifiers: Sequence[SecurityIdentifier], raise_error: bool = True) -> list[OpenFIGIRecord] ¶
Read base date for one or more identifiers from OpenFIGI.
Source code in src/equities_classifier/clients/openfigi/client.py
def read_provider_base_data(
self,
source_identifiers: Sequence[SecurityIdentifier],
raise_error: bool = True
) -> list[OpenFIGIRecord]:
"""Read base date for one or more identifiers from OpenFIGI."""
results: list[OpenFIGIRecord] = []
batches = self._create_batches(source_identifiers)
for batch in batches:
response_data = self._execute_request(batch)
for source_identifier, item in zip(batch, response_data, strict=True):
# parsing accroSs batches for comprehension of data for duplicate source identifiers
# (i.e. ticker and ISIN)
self._parse_records(
item=item,
source_identifier=source_identifier,
records=results,
raise_error=raise_error
)
return results
remove_records_without_share_class_figi(records: list[OpenFIGIRecord]) -> list[OpenFIGIRecord] staticmethod ¶
Remove records without share_class_FIGI if a ISIN-matching record exists.
Source code in src/equities_classifier/clients/openfigi/client.py
@staticmethod
def remove_records_without_share_class_figi(
records: list[OpenFIGIRecord],
) -> list[OpenFIGIRecord]:
"""Remove records without share_class_FIGI if a ISIN-matching record exists."""
identifiers_with_share_class_figi: set[tuple[SecurityIdentifierType, str]] = {
(identifier.type, str(identifier.value_cleaned))
for record in records
if record.share_class_figi is not None
for identifier in record.identifiers
if identifier.type == SecurityIdentifierType.ISIN
}
return [
record
for record in records
if (
record.share_class_figi is not None
or not any(
(identifier.type, identifier.value_cleaned)
in identifiers_with_share_class_figi
for identifier in record.identifiers
if identifier.type == SecurityIdentifierType.ISIN
)
)
]
OpenFIGIRecord(name: str | None = None, ticker: str | None = None, *, identifiers: list[SecurityIdentifier] = list(), figi: list[str] = list(), composite_figi: list[str] = list(), share_class_figi: str | None = None, security_description: list[str] = list(), security_type: str | None = None, security_type2: str | None = None, market_sector: str | None = None, exch_code: list[str] = list(), mic_code: list[str] = list(), currency: str | None = None, state_code: str | None = None) dataclass ¶
Internal representation of a single OpenFIGI mapping result.