Source code for trader.marketdata.index_provider

"""Index membership, scraped from Wikipedia (issue #27).

Never raises past `get_members`: a scrape failure degrades to `[]`, the same
shape a discovery source degrades to when `themes: []` — `trader
scan-universe` still records whatever `analyst_consensus_history` already has,
and `discover_symbols`'s analyst-scan candidate source simply has nothing new
to check that cycle.
"""

import logging
import re
from collections.abc import Callable
from typing import ClassVar

from bs4 import BeautifulSoup

__all__ = ["WikipediaIndexProvider", "validate_symbol"]

_logger = logging.getLogger("trader.marketdata.index_provider")

_SP500_URL = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
#: Wikipedia blocks the default urllib/requests user agent on some tables.
_USER_AGENT = "claude-alpaca-trader/1.0 (+index membership scan; issue #27)"

#: 1-8 chars, uppercase letters/digits plus `.`/`-` only.
_MIN_SYMBOL_LENGTH = 1
_MAX_SYMBOL_LENGTH = 8
_ALLOWED_CHARS = re.compile(r"^[A-Z0-9.\-]+$")

#: Which header text names the symbol column. Wikipedia has used more than
#: one wording for this over time.
_SYMBOL_HEADERS = frozenset({"symbol", "ticker symbol", "ticker"})




def validate_symbol(raw: str) -> str | None:
    """A cleaned ticker symbol, or `None` if `raw` is not one.

    Wikipedia table scraping mixes years, footnote markers, and company names
    into the symbol column (a confirmed pattern, per `docs/ideas.md`'s
    "validate ticker symbols at the boundary" entry) — this is the one
    validator at the scrape boundary that rejects those, the same discipline
    `listing_gate` applies to a warrant suffix: logged, countable, never a
    crash reaching `discover_symbols`.
    """
    candidate = raw.strip().upper()
    if not _MIN_SYMBOL_LENGTH <= len(candidate) <= _MAX_SYMBOL_LENGTH:
        return None
    if not _ALLOWED_CHARS.fullmatch(candidate):
        return None
    if not any(char.isalpha() for char in candidate):
        return None
    if candidate[0] in ".-" or candidate[-1] in ".-":
        return None
    return candidate



def _default_html_source(url: str) -> str:
    import requests

    response = requests.get(url, headers={"User-Agent": _USER_AGENT}, timeout=30)
    response.raise_for_status()
    return response.text


def _symbol_column(table) -> list[str]:
    """Raw cell text from whichever column's header names the symbol."""
    rows = table.find_all("tr")
    if not rows:
        return []
    header_cells = rows[0].find_all(["th", "td"])
    header_text = [cell.get_text(strip=True).lower() for cell in header_cells]
    try:
        column = next(i for i, text in enumerate(header_text) if text in _SYMBOL_HEADERS)
    except StopIteration:
        return []

    raw_symbols: list[str] = []
    for row in rows[1:]:
        cells = row.find_all(["td", "th"])
        if column >= len(cells):
            continue
        raw_symbols.append(cells[column].get_text(strip=True))
    return raw_symbols




class WikipediaIndexProvider:
    """Index constituents scraped from Wikipedia. No `bs4` type escapes."""

    #: v1 supports only the S&P 500 — the spec's own Open Question #4 defers
    #: which other indices to a later session, and this is an operator-invoked
    #: tool, not something `config/watchlist.yaml` drives.
    _SUPPORTED: ClassVar[dict[str, str]] = {"sp500": _SP500_URL}

    def __init__(self, html_source: Callable[[str], str] | None = None) -> None:
        self._html_source = html_source or _default_html_source



    def get_members(self, index_name: str) -> list[str]:
        """Cleaned, deduplicated symbols for `index_name`, or `[]` on failure.

        Raises `ValueError` only for an unsupported `index_name` — a caller
        programming error, not a runtime condition to degrade from. Every
        other failure (network, markup change, an empty page) is caught and
        logged; `trader scan-universe` still has a table to record.
        """
        url = self._SUPPORTED.get(index_name)
        if url is None:
            raise ValueError(
                f"unsupported index {index_name!r}; supported: {sorted(self._SUPPORTED)}"
            )

        try:
            html = self._html_source(url)
            soup = BeautifulSoup(html, "html.parser")
            table = soup.find("table", {"id": "constituents"}) or soup.find(
                "table", class_="wikitable"
            )
            raw_symbols = _symbol_column(table) if table is not None else []
        except Exception:  # noqa: BLE001 - a scrape outage must not raise
            _logger.warning(
                "index_provider: scrape failed for %r", index_name, exc_info=True
            )
            return []

        members: list[str] = []
        seen: set[str] = set()
        for raw in raw_symbols:
            cleaned = validate_symbol(raw)
            if cleaned is None:
                _logger.info("index_provider: rejected invalid symbol %r", raw)
                continue
            if cleaned in seen:
                continue
            seen.add(cleaned)
            members.append(cleaned)
        return members