"""Search by phrase, returning the symbols the news is talking about.
`NewsProvider` goes symbol -> news. This goes the other way, which is what
discovery needs: given "nuclear energy", which tickers are in the story?
Two sources inside one response, and both matter. `quotes` are symbols whose
name or description matches the phrase. `news[].relatedTickers` are symbols
Yahoo tagged onto an article — noisier (an article about a small nuclear
company came back tagged NVDA and BAC on 2026-08-03) but the only way a
symbol that does not contain the phrase in its name is ever surfaced.
"""
from collections.abc import Callable
from dataclasses import dataclass
from typing import Protocol, runtime_checkable
from trader.errors import MarketDataError
__all__ = ["NewsSearchProvider", "SearchHit", "YFinanceSearchProvider"]
#: A direct name match, stronger evidence than a tag on someone's article.
ORIGIN_QUOTE = "quote"
#: Yahoo tagged this symbol onto a news item matching the phrase.
ORIGIN_RELATED = "related_ticker"
[docs]
@dataclass(frozen=True, slots=True)
class SearchHit:
"""One symbol a phrase surfaced, and why."""
symbol: str
theme: str
origin: str
headline: str | None = None
url: str | None = None
[docs]
@runtime_checkable
class NewsSearchProvider(Protocol):
"""Symbols mentioned in news matching a phrase."""
[docs]
def search(self, phrase: str, limit: int = 5) -> list[SearchHit]:
"""Raises `MarketDataError` if the search fails. `[]` is normal."""
...
def _default_search_factory(phrase: str, limit: int) -> object:
import yfinance
return yfinance.Search(phrase, max_results=limit, news_count=limit)
def _select_with_reserved_shares(
hits: dict[str, SearchHit], limit: int
) -> list[SearchHit]:
"""Truncate to `limit`, guaranteeing both origins a share rather than
letting whichever was inserted first fill every slot.
A plain `[:limit]` on the merged, insertion-ordered dict above starves
`relatedTickers`: quotes are always inserted first, so a query returning
`limit` or more quotes (measured: `Search("nuclear energy")` at the
default `limit=5` returns five) leaves zero slots for a tag Yahoo put on
a news item, and `relatedTickers` is the *only* way a symbol that does
not contain the search phrase in its own name is ever surfaced — the
slice's motivating case. Reserving roughly half of `limit` for each
origin (the larger half to quotes, the stronger signal, when `limit` is
odd) guarantees a surviving related-ticker hit is never fully crowded
out, while any capacity a thin origin leaves unused is backfilled from
the other — quotes first — so a real, wide result never comes back
shorter than `limit` just because one origin was reserved more room than
it had candidates for.
"""
quote_order = [symbol for symbol, hit in hits.items() if hit.origin == ORIGIN_QUOTE]
related_order = [
symbol for symbol, hit in hits.items() if hit.origin == ORIGIN_RELATED
]
quote_quota = limit - limit // 2 # ceil(limit / 2)
related_quota = limit // 2
selected: list[str] = []
seen: set[str] = set()
def take(symbols: list[str]) -> None:
for symbol in symbols:
if len(selected) >= limit:
return
if symbol not in seen:
seen.add(symbol)
selected.append(symbol)
take(quote_order[:quote_quota])
take(related_order[:related_quota])
# Backfill: whichever origin was reserved more room than it had
# candidates for leaves capacity unused above. Quotes first, matching
# the same "stronger signal" precedence used everywhere else in this
# method.
take(quote_order[quote_quota:])
take(related_order[related_quota:])
return [hits[symbol] for symbol in selected]
[docs]
class YFinanceSearchProvider:
"""Phrase search via `yf.Search`. No yfinance type escapes."""
def __init__(self, search_factory: Callable[..., object] | None = None) -> None:
self._search_factory = search_factory or _default_search_factory
[docs]
def search(self, phrase: str, limit: int = 5) -> list[SearchHit]:
try:
result = self._search_factory(phrase, limit)
quotes = list(getattr(result, "quotes", []) or [])
news = list(getattr(result, "news", []) or [])
except Exception as exc: # noqa: BLE001 - yfinance raises many types
raise MarketDataError(f"Search for {phrase!r} failed: {exc}") from exc
hits: dict[str, SearchHit] = {}
# Quotes first, so a direct match is not later overwritten by a
# weaker related-ticker tag for the same symbol.
for quote in quotes:
symbol = str(quote.get("symbol") or "").strip().upper()
if symbol and symbol not in hits:
hits[symbol] = SearchHit(symbol=symbol, theme=phrase, origin=ORIGIN_QUOTE)
for item in news:
related = item.get("relatedTickers") or []
for raw in related:
symbol = str(raw or "").strip().upper()
if symbol and symbol not in hits:
hits[symbol] = SearchHit(
symbol=symbol,
theme=phrase,
origin=ORIGIN_RELATED,
headline=item.get("title"),
url=item.get("link"),
)
return _select_with_reserved_shares(hits, limit)