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."""
self._client: httpx.Client | None
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 ¶
check_and_set_primary_ticker(records: list[OpenFIGIRecord], set_ticker: bool = True, raise_error: bool = False) -> None staticmethod ¶
Check and optionally set the primary ticker for OpenFIGI records.
Source code in src/equities_classifier/clients/openfigi/client.py
@staticmethod
def check_and_set_primary_ticker(
records: list[OpenFIGIRecord],
set_ticker: bool = True,
raise_error: bool = False,
) -> None:
"""Check and optionally set the primary ticker for OpenFIGI records."""
for record in records:
if record.has_identifier(SecurityIdentifierType.ISIN):
isin = record.identifier(SecurityIdentifierType.ISIN).value
ticker_new1 = get_primary_ticker(
DataSourceID.OPENFIGI,
isin,
record.name,
record.ticker,
record.ticker_mic,
record.mic_code,
OpenFIGIResponseError if raise_error else None,
)
if ticker_new1:
ticker_new1 = ticker_new1.replace("/", "-")
mics_from_exchange = [
next(
(
entry["operating_mic"]
for entry in bloomberg_exchange_mapping
if entry["bloomberg_exchange"] == exchange
),
None
)
for exchange in record.exch_code
]
ticker_new2 = get_primary_ticker(
DataSourceID.OPENFIGI,
isin,
record.name,
record.ticker,
record.ticker_exchange,
mics_from_exchange,
OpenFIGIResponseError if raise_error else None,
)
if ticker_new2:
ticker_new2 = ticker_new2.replace("/", "-")
if ticker_new1 and ticker_new2 and ticker_new1 != ticker_new2:
ClientHelperErrorHandler.inconsistent_provider_data(
DataSourceID.OPENFIGI,
record.name,
"primary ticker from Bloomberg exchanges and mic-codes differ",
OpenFIGIResponseError if raise_error else None,
)
if ticker_new1 and set_ticker:
record.ticker = ticker_new1
record.identifiers.replace(SecurityIdentifier(SecurityIdentifierType.TICKER, ticker_new1))
elif ticker_new2 and set_ticker:
record.ticker = ticker_new2
record.identifiers.replace(SecurityIdentifier(SecurityIdentifierType.TICKER, ticker_new2))
elif record.has_identifier(SecurityIdentifierType.TICKER):
if record.security_type == record.security_type2 == "Common Stock":
record.ticker = record.identifier(SecurityIdentifierType.TICKER).value
check_and_set_us_ticker(records: list[OpenFIGIRecord], set_ticker: bool = True, raise_error: bool = False) -> None staticmethod ¶
Check and optionally set the primary ticker for OpenFIGI records.
Source code in src/equities_classifier/clients/openfigi/client.py
@staticmethod
def check_and_set_us_ticker(
records: list[OpenFIGIRecord],
set_ticker: bool = True,
raise_error: bool = False,
) -> None:
"""Check and optionally set the primary ticker for OpenFIGI records."""
for record in records:
ticker_us = get_us_ticker(record.ticker_exchange, record.exch_code)
if ticker_us:
ticker_us = ticker_us.replace("/", "-")
record.identifiers.append(SecurityIdentifier(SecurityIdentifierType.TICKER_US, ticker_us))
read_provider_base_data(source_identifiers: Sequence[SecurityIdentifier], raise_error: bool = False) -> 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 = False
) -> 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 across 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
)
# repeat for tickers with share_class_figi to get US tickers
repeat_ticker_with_share_class_figi: list[SecurityIdentifier] = list({
result.identifier(SecurityIdentifierType.SHARE_CLASS_FIGI)
for result in results
if (
result.identifier(SecurityIdentifierType.SHARE_CLASS_FIGI) is not None and
not result.has_identifier(SecurityIdentifierType.ISIN) and
result.has_identifier(SecurityIdentifierType.TICKER) and
result.identifier(SecurityIdentifierType.TICKER).value_cleaned !=
get_us_ticker(result.ticker_exchange, result.exch_code)
)
})
if len(repeat_ticker_with_share_class_figi) == 0:
return results
batches = self._create_batches(repeat_ticker_with_share_class_figi)
for batch in batches:
response_data = self._execute_request(batch)
for source_identifier, item in zip(batch, response_data, strict=True):
# parsing across 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_1(records: list[OpenFIGIRecord]) -> list[OpenFIGIRecord] staticmethod ¶
Remove records without share_class_FIGI if no ISIN-matching record exists.
Source code in src/equities_classifier/clients/openfigi/client.py
@staticmethod
def remove_records_without_share_class_figi_1(
records: list[OpenFIGIRecord],
) -> list[OpenFIGIRecord]:
"""Remove records without share_class_FIGI if no ISIN-matching record exists."""
count_before = len(records)
identifiers_with_share_class_figi: set[SecurityIdentifier] = {
SecurityIdentifier(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
}
records = [
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
)
)
]
count_after = len(records)
if count_before - count_after > 0:
ClientHelperErrorHandler.records_cleaned(
DataSourceID.OPENFIGI,
count_before - count_after,
"records without shareClassFIGI and no matching to shareClassFIGI via ISIN"
)
return records
remove_records_without_share_class_figi_2(records: list[OpenFIGIRecord]) -> list[OpenFIGIRecord] staticmethod ¶
Remove records without share_class_FIGI and not Common Stock.
Source code in src/equities_classifier/clients/openfigi/client.py
@staticmethod
def remove_records_without_share_class_figi_2(
records: list[OpenFIGIRecord],
) -> list[OpenFIGIRecord]:
"""Remove records without share_class_FIGI and not Common Stock."""
count_before = len(records)
records = [
record
for record in records
if record.share_class_figi and
(record.security_type == "Common Stock" or record.security_type2 == "Common Stock")
]
count_after = len(records)
if count_before - count_after > 0:
ClientHelperErrorHandler.records_cleaned(
DataSourceID.OPENFIGI,
count_before - count_after,
"records without missing shareClassFIGI and no ISIN matching (i. e. non share fiancial instruments)"
)
return records
OpenFIGIRecord(name: str | None = None, ticker: str | None = None, *, identifiers: SecurityIdentifierList = SecurityIdentifierList(), 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, ticker_exchange: list[str] = list(), ticker_mic: list[str] = list(), 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.