Skip to content

Clienthelper

clienthelper

client helper for all clients.

bloomberg_exchange_mapping = _get_bloomberg_exchange_mapping() module-attribute

ClientHelperErrorHandler

Helper methods for client processing.

inconsistent_provider_data(provider: DataSourceID, name: str, description: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when Provider record shows data inconsistency.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def inconsistent_provider_data(
    provider: DataSourceID,
    name: str,
    description: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when Provider record shows data inconsistency."""

    message = f"Provider record for '{name}' shows inconsistent data: {description}."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

invalid_security_type(provider: DataSourceID, identifier: SecurityIdentifier, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider is called with invalid security type.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def invalid_security_type(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider is called with invalid security type."""

    message = f"Identifier type {identifier.type} invalid or invalid for source '{provider}' (value was '{identifier.value}')."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

missing_provider_attribute(provider: DataSourceID, identifier: SecurityIdentifier, attribute: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider returns an unmapped JSON attribute.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def missing_provider_attribute(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    attribute: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider returns an unmapped JSON attribute."""

    message = f"Missing provider attribute '{attribute}' from source '{provider}' for {identifier.type} '{identifier.value}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

missing_record_attribute(provider: DataSourceID, attribute: str, value: Any, context: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the mapping points to a record attribute that does not exist.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def missing_record_attribute(
    provider: DataSourceID,
    attribute: str,
    value: Any,
    context: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the mapping points to a record attribute that does not exist."""

    message = f"Unknown/not handled provider attribute '{attribute}' with value '{value}' in response from '{provider}', not mapped in '{context}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

other_error_with_message(provider: DataSourceID, message: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider is called with invalid security type.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def other_error_with_message(
    provider: DataSourceID,
    message: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider is called with invalid security type."""

    logger_equities_classifier.error(message)

    if raise_object:
        raise raise_object(message)

primary_ticker_not_unique(provider: DataSourceID, isin: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when Provider record does not allow reliable identification of primary ticker.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def primary_ticker_not_unique(
    provider: DataSourceID,
    isin: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when Provider record does not allow reliable identification of primary ticker."""

    message = f"Provider record from '{provider}' shows more than one candidate for primary ticker for ISIN, '{isin}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

records_cleaned(provider: DataSourceID, clean_counter: int, reason: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when provider records where cleaned.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def records_cleaned(
    provider: DataSourceID,
    clean_counter: int,
    reason: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when provider records where cleaned."""

    message = f"{clean_counter} provider record from '{provider}' was/were cleaned/deleted: {reason}."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

search_result_counter_issue(provider: DataSourceID, identifier: SecurityIdentifier, count_provider: int, count_found: int, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider delivers inconsistent number of data records.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def search_result_counter_issue(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    count_provider: int,
    count_found: int,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider delivers inconsistent number of data records."""

    message = f"Search result from source '{provider}' for {identifier.type} '{identifier.value}' contains inconsistent counter value, expected {count_provider} vs {count_found} found."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

search_result_not_unique(provider: DataSourceID, identifier: SecurityIdentifier, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider cannot find unique dataset.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def search_result_not_unique(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider cannot find unique dataset."""

    message = f"Search result from source '{provider}' for {identifier.type} '{identifier.value}' is not unique. Processing first match."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

unknown_identifier_type(provider: DataSourceID, identifier_type: SecurityIdentifierType, name: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when expected identifier of an (expected) security type is not present.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def unknown_identifier_type(
    provider: DataSourceID,
    identifier_type: SecurityIdentifierType,
    name: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when expected identifier of  an (expected) security type is not present."""

    message = f"Unknown identifier type '{identifier_type}' for security '{name}') in data form '{provider}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

unknown_provider_attribute(provider: DataSourceID, identifier: SecurityIdentifier, attribute: str, value: str, context: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider returns an unmapped JSON attribute.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def unknown_provider_attribute(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    attribute: str,
    value: str,
    context: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider returns an unmapped JSON attribute."""

    message = f"Provider attribute {attribute}, value '{value}' from source '{provider}' for {identifier.type} '{identifier.value}' not mapped in '{context}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

unknown_provider_attributes(provider: DataSourceID, identifier: SecurityIdentifier, attributes: set[tuple[str, ...]], context: str, raise_object: type[ClientResponseError] | None = None) -> None staticmethod

Called when the provider returns unmapped JSON attributes.

Source code in src/equities_classifier/clients/clienthelper.py 
@staticmethod
def unknown_provider_attributes(
    provider: DataSourceID,
    identifier: SecurityIdentifier,
    attributes: set[tuple[str, ...]],
    context: str,
    raise_object: type[ClientResponseError] | None = None,
) -> None:
    """Called when the provider returns unmapped JSON attributes."""

    message = f"Provider attributes {attributes} from source '{provider}' for {identifier.type} '{identifier.value}' not mapped in '{context}'."
    logger_equities_classifier.warning(message)

    if raise_object:
        raise raise_object(message)

get_nested_value(data: dict[str, Any], path: tuple[str, ...]) -> Any

Return a nested value from a dictionary.

Source code in src/equities_classifier/clients/clienthelper.py 
def get_nested_value(data: dict[str, Any], path: tuple[str, ...], ) -> Any:
    """Return a nested value from a dictionary."""

    value: Any = data
    for key in path:
        if not isinstance(value, dict):
            return None
        value = value.get(key)
        if value is None:
            return None

    return value

get_primary_ticker(provider: DataSourceID, isin: str, name: str, ticker_for_check: str, list_ticker: list[str], list_mic_code: list[str], error_object: type[ClientResponseError] | None = None) -> str | None

Get primary ticker defined as ticker in home country / country of isin registration.

Source code in src/equities_classifier/clients/clienthelper.py 
def get_primary_ticker(
    provider: DataSourceID,
    isin: str,
    name: str,
    ticker_for_check: str,
    list_ticker: list[str],
    list_mic_code: list[str],
    error_object: type[ClientResponseError] | None = None,
) -> str | None:
    """Get primary ticker defined as ticker in home country / country of isin registration."""

    if list_ticker and len(list_ticker) > 0 and list_mic_code and len(list_mic_code) > 0:

        if len(list_ticker) != len(list_mic_code):
            ClientHelperErrorHandler.inconsistent_provider_data(
                provider,
                name,
                "Count of 'ticker_exchange' and 'mic_code' differ",
                error_object,
            )
            return None

        regulated_mics_for_country = _REGULATED_MICS_BY_COUNTRY.get(
            isin[:2],
            frozenset(),
        )
        tickers = [
            ticker
            for ticker, mic in zip(list_ticker, list_mic_code, strict=True, )
            if mic in regulated_mics_for_country
        ]
        counter_tickers = Counter(tickers)
        if len(counter_tickers) > 0:
            ticker_new = counter_tickers.most_common(1)[0][0]
            if ticker_for_check and ticker_for_check != ticker_new:
                ClientHelperErrorHandler.inconsistent_provider_data(
                    provider,
                    name,
                    f"Ticker '{ticker_for_check}' currently set does not match assumed primary ticker '{ticker_new}'",
                    error_object,
                )
            if (len(counter_tickers.most_common()) > 1 and
                counter_tickers.most_common(1)[0][1] == counter_tickers.most_common(2)[1][1]):
                ClientHelperErrorHandler.primary_ticker_not_unique(
                    DataSourceID.OPENFIGI,
                    isin,
                    error_object,
                )

            return ticker_new

    return None

get_us_ticker(list_ticker_exchange: list[str], list_exch_code: list[str]) -> str | None

Get US ticke r- assumption: ticker identical for all (regulated) US exchanges

Source code in src/equities_classifier/clients/clienthelper.py 
def get_us_ticker(
    list_ticker_exchange: list[str],
    list_exch_code: list[str],
) -> str | None:
    """Get US ticke r- assumption: ticker identical for all (regulated) US exchanges"""

    for ticker_exchange, exch_code in zip(list_ticker_exchange, list_exch_code, strict=True):
        country = next(
            (
                entry["country"]
                for entry in bloomberg_exchange_mapping
                if entry["bloomberg_exchange"] == exch_code
            ),
            None
        )
        if country in {"US", "USA"}:
            return ticker_exchange

    return None

leaf_paths(data: Mapping[str, Any], path: tuple[str, ...] = (), exclude_leaves: Collection[str] = []) -> list[tuple[str, ...]]

Return all key paths from the root to every leaf.

Source code in src/equities_classifier/clients/clienthelper.py 
def leaf_paths(
    data: Mapping[str, Any],
    path: tuple[str, ...] = (),
    exclude_leaves: Collection[str] = []
) -> list[tuple[str, ...]]:
    """Return all key paths from the root to every leaf."""

    result: list[tuple[str, ...]] = []

    for key, value in data.items():

        if key in exclude_leaves:
            continue

        current = (*path, key)
        if isinstance(value, Mapping):
            result.extend(leaf_paths(value, current, exclude_leaves))
        elif value:
            result.append(current)

    return result