Morningstar
morningstar ¶
__all__ = ['MorningstarClient', 'MorningstarRecord'] module-attribute ¶
MorningstarClient(timeout: float = 30.0, clean_nonUS_ticker: bool = False, 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,
clean_nonUS_ticker: bool = False,
test_wo_browser: bool = False,
) -> None:
"""Initialize Morningstar client."""
self._client: uc.Chrome | None
if not test_wo_browser:
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 = 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 ¶
check_and_set_primary_ticker(records: list[MorningstarRecord], set_ticker: bool = True, raise_error: bool = False) -> None staticmethod ¶
Check and optionally set the primary ticker for Morningstar records.
Source code in src/equities_classifier/clients/morningstar/client.py
@staticmethod
def check_and_set_primary_ticker(
records: list[MorningstarRecord],
set_ticker: bool = True,
raise_error: bool = False,
) -> None:
"""Check and optionally set the primary ticker for Morningstar records."""
for record in records:
if not record.has_identifier(SecurityIdentifierType.ISIN):
ClientHelperErrorHandler.unknown_identifier_type(
DataSourceID.OPENFIGI,
SecurityIdentifierType.ISIN,
record.name,
MorningstarResponseError if raise_error else None,
)
continue
isin = record.identifier(SecurityIdentifierType.ISIN).value
ticker_new = get_primary_ticker(
DataSourceID.OPENFIGI,
isin,
record.name,
record.ticker,
record.ticker_exchange,
record.exchange,
MorningstarResponseError,
)
if ticker_new and set_ticker:
record.ticker = ticker_new
record.identifiers.replace(SecurityIdentifier(SecurityIdentifierType.TICKER, ticker_new))
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 not self.supports_identifier_type(source_identifier.type):
ClientHelperErrorHandler.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) -> None ¶
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
) -> None:
"""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]
)
self._parse_profile_to_record(profile_data, record, raise_error)
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: SecurityIdentifierList = SecurityIdentifierList()) dataclass ¶
Internal representation of a Morningstar search result and classification record.