Skip to content

Client

client

Client for SeekingAlpha.

SeekingAlphaClient(timeout: float = 30.0, test_wo_browser: bool = False)

HTTP client for SeekingAlpha.

Source code in src/equities_classifier/clients/seekingalpha/client.py
def __init__(
    self,
    timeout: float = 30.0,
    test_wo_browser: bool = False,
) -> None:
    """Initialize SeekingAlpha 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")
        # preferences (from utils-seleniumxp)
        prefsdict = {
            "profile.default_content_setting_values.notifications": 2,
            "profile.managed_default_content_settings.images": 2,
            "profile.managed_default_content_settings.stylesheets": 2,
            "profile.managed_default_content_settings.cookies": 1,
            "profile.managed_default_content_settings.javascript": 1,
            "profile.managed_default_content_settings.plugins": 1,
            "profile.managed_default_content_settings.popups": 2,
            "profile.managed_default_content_settings.geolocation": 2,
            "profile.managed_default_content_settings.media_stream": 2
        }
        options.add_experimental_option("prefs", prefsdict)
        try:
            self._client = uc.Chrome(options=options)
        except Exception as e:
            print("Failed to start Chrome")
            raise e
        try:
            config = StabilizationConfig(
                timeout=25,  # Max wait time (seconds)
                network_idle_threshold=5,  # Max pending requests (allows background traffic)
                animation_detection=False,  # Track CSS animations (non-blocking in normal mode)
                layout_stability=True,  # Wait for element positions to stabilized, default True in 'strict' mode
                strictness='strict',  # '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

__enter__() -> SeekingAlphaClient

Source code in src/equities_classifier/clients/seekingalpha/client.py
def __enter__(self) -> "SeekingAlphaClient":
    return self

__exit__(*_: object) -> None

Source code in src/equities_classifier/clients/seekingalpha/client.py
def __exit__(self, *_: object) -> None:
    self._close()

read_provider_profile_data(source_identifiers: Sequence[SecurityIdentifier], raise_error: bool = False) -> list[SeekingAlphaRecord]

Read profile data from SeekingAlpha.

Source code in src/equities_classifier/clients/seekingalpha/client.py
def read_provider_profile_data(
    self,
    source_identifiers: Sequence[SecurityIdentifier],
    raise_error: bool = False,
) -> list[SeekingAlphaRecord]:
    """Read profile data from SeekingAlpha."""

    records: list[SeekingAlphaRecord] = []

    for source_identifier in source_identifiers:

        if not self.supports_identifier_type(source_identifier.type):
            ClientHelperErrorHandler.invalid_security_type(DataSourceID.SEEKINGALPHA, source_identifier)
            continue

        response = self._execute_request(source_identifier, raise_error)
        if response is None:
            continue

        record = self._parse_record(source_identifier, response, raise_error)
        # check for duplicates
        if record:
            duplicate = False
            for existing_record in records:
                if record.name == existing_record.name:
                    duplicate = True
                    for identifier in record.identifiers:
                        if not existing_record.has_identifier(identifier.type):
                            existing_record.identifiers.append(identifier)
            if not duplicate:
                records.append(record)

    return records

supports_identifier_type(identifier_type: SecurityIdentifierType) -> bool classmethod

Return whether the provider supports an identifier type.

Source code in src/equities_classifier/clients/seekingalpha/client.py
@classmethod
def supports_identifier_type(cls, identifier_type: SecurityIdentifierType) -> bool:
    """Return whether the provider supports an identifier type."""
    return identifier_type in cls._SEEKINGALPHA_IDENTIFIER_TYPES.values()

SeekingAlphaResponseError

SeekingAlpha returned an error response.