Morningstar
morningstar ¶
__all__ = ['MorningstarClient', 'MorningstarRecord'] module-attribute ¶
MorningstarClient(timeout: float = 30.0, seleniumwrapper: object | None = None, clean_nonUS_ticker: bool = True, test_wo_browser: bool = False) ¶
Morningstar HTTP / API client.
Source code in src/equities_classifier/clients/morningstar/client.py
def __init__(
self,
timeout: float = 30.0,
seleniumwrapper: object | None = None,
clean_nonUS_ticker: bool = True,
test_wo_browser: bool = False,
) -> None:
"""Initialize Morningstar client."""
if not test_wo_browser:
if seleniumwrapper is None:
options = uc.ChromeOptions()
# options.add_argument("--headless=new") # does not work with Morningstar because CloudFront-detected
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--disable-gpu")
options.add_argument("--window-size=1920,1080")
try:
self._client = uc.Chrome(options=options)
except Exception as e:
print("Failed to start Chrome")
raise e
try:
config = StabilizationConfig(
timeout=10, # Max wait time (seconds)
network_idle_threshold=5, # Max pending requests (allows background traffic)
strictness='normal', # 'strict' | 'normal' | 'relaxed'
debug_mode=False # Enable logging
)
self._client = stabilize(self._client, config=config)
except StabilizationTimeout as e:
print("Failed to stabilize Chrome with 'waitless'")
diagnostics = get_diagnostics(self._client)
print_report(diagnostics) # Print detailed report
raise e
else:
self._client = seleniumwrapper
else:
self._client = None
self._timeout = timeout
self._access_token: str | None = None
self._access_token_expires: datetime.date | None = None
self._clean_nonUS_ticker = clean_nonUS_ticker
__enter__() -> Self ¶
__exit__(*_: object) -> None ¶
close() -> None ¶
leaf_paths(data: Mapping[str, Any], path: tuple[str, ...] = (), exclude_leaves: Collection[str] = []) -> list[tuple[str, ...]] staticmethod ¶
Return all key paths from the root to every leaf.
Source code in src/equities_classifier/clients/morningstar/client.py
@staticmethod
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(MorningstarClient.leaf_paths(value, current, exclude_leaves))
else:
result.append(current)
return result
read_provider_base_data(source_identifiers: Sequence[SecurityIdentifier], raise_error: bool = False) -> list[MorningstarRecord] ¶
Read base data for one or more identifiers from Morningstar.
Source code in src/equities_classifier/clients/morningstar/client.py
def read_provider_base_data(
self,
source_identifiers: Sequence[SecurityIdentifier],
raise_error: bool = False
) -> list[MorningstarRecord]:
"""Read base data for one or more identifiers from Morningstar."""
records: list[MorningstarRecord] = []
for source_identifier in source_identifiers:
if source_identifier.type not in self._MORNINGSTAR_IDENTIFIER_TYPES.values():
ClientHelper.invalid_security_type(
DataSourceID.MORNINGSTAR,
source_identifier
)
continue
response_data = self._execute_search_request(source_identifier)
search_results = self._parse_search_results(
source_identifier,
response_data,
raise_error,
)
search_results = [
result for result in search_results
if (
source_identifier.type == SecurityIdentifierType.ISIN
and result.isin == source_identifier.value
)
or (
source_identifier.type == SecurityIdentifierType.TICKER
and result.ticker == source_identifier.value_cleaned
)
]
if search_results is not None:
self._parse_records(
source_identifier,
search_results,
records,
raise_error
)
return records
read_provider_profile_data(records: list[MorningstarRecord], raise_error: bool = False) -> list[MorningstarRecord] ¶
Read profile data from Morningstar.
Source code in src/equities_classifier/clients/morningstar/client.py
def read_provider_profile_data(
self,
records: list[MorningstarRecord],
raise_error: bool = False
) -> list[MorningstarRecord]:
"""Read profile data from Morningstar."""
for record in records:
profile_data = self._execute_profile_request(
security_id=record.company_id if record.company_id else record.performance_id[0]
)
record = self._parse_profile_to_record(profile_data, record, raise_error)
return records
supports_identifier_type(identifier_type: SecurityIdentifierType) -> bool classmethod ¶
Check if identifier type supported
MorningstarRecord(name: str | None = None, ticker: str | None = None, short_name: str | None = None, isin: str | None = None, company_id: str | None = None, business_description: str | None = None, universe: str | None = None, sector: str | None = None, industry: str | None = None, security_id: list[str] = list(), performance_id: list[str] = list(), ticker_exchange: list[str] = list(), exchange: list[str] = list(), exchange_name: list[str] = list(), exchange_country: list[str] = list(), exchange_country_name: list[str] = list(), *, identifiers: list[SecurityIdentifier] = list()) dataclass ¶
Internal representation of a Morningstar search result and classification record.