"""Analyst consensus for a symbol.
The cheapest real signal available about whether the market's professional
readers like a name: 0.05-0.23s per symbol measured on 2026-08-03, against a
400-day bar fetch plus a backtest for the alternative. That is why discovery
consults this before doing anything expensive. (Issue #27, 2026-08-19, added
a second yfinance surface — `.info`, for the mean rating and price targets —
alongside `.recommendations`, so this call now costs two round trips, not
one; not re-measured, which is why `trader scan-universe` rate-limits itself
sequentially rather than assuming this is still as cheap as before.)
Absent coverage is `None`, never zeros. ETFs have no analysts at all — SPY
returns nothing — and an all-zero opinion would answer "is it net negative?"
with a confident no for entirely the wrong reason.
"""
import math
from collections.abc import Callable
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import Protocol, runtime_checkable
from trader.errors import MarketDataError
__all__ = ["AnalystOpinion", "AnalystProvider", "YFinanceAnalystProvider"]
#: Weight applied to each count when `recommendationMean` is absent from
#: yfinance's `.info` — the standard 1 (strong buy) .. 5 (strong sell) scale,
#: matching how Yahoo itself derives the mean it usually reports directly.
_MEAN_WEIGHTS = (1, 2, 3, 4, 5)
[docs]
@dataclass(frozen=True, slots=True)
class AnalystOpinion:
"""How many analysts hold each view, for the current period.
The four fields below `strong_sell` are issue #27 additions and default
to "unknown" (`None`/`False`) so every pre-existing call site
(`AnalystOpinion(strong_buy=.., buy=.., ...)`) keeps working unchanged.
"""
strong_buy: int
buy: int
hold: int
sell: int
strong_sell: int
#: 1 (strong buy) .. 5 (strong sell), Yahoo's own consensus scale. A
#: rating, not money — `float`, not `Decimal`.
recommendation_mean: float | None = None
#: `True` when yfinance omitted `recommendationMean` and this was
#: computed from the five counts instead — never silently indistinguishable
#: from a Yahoo-reported value.
recommendation_mean_is_derived: bool = False
#: These two are money — a dollar price target — so `Decimal`.
target_mean_price: Decimal | None = None
target_high_price: Decimal | None = None
@property
def is_net_negative(self) -> bool:
"""Whether the bears strictly outnumber the bulls.
`hold` is deliberately excluded: a wall of holds is indecision, not
disapproval, and treating it as negative would reject every quiet
large-cap. Strictly greater, so a tie passes — discovery's job is to
remove the clearly disliked, not to arbitrate close calls the model is
better placed to read.
"""
return (self.sell + self.strong_sell) > (self.strong_buy + self.buy)
[docs]
@runtime_checkable
class AnalystProvider(Protocol):
"""Current analyst consensus, or `None` when nobody covers the symbol."""
[docs]
def get_opinion(self, symbol: str) -> AnalystOpinion | None:
"""Raises `MarketDataError` if the lookup itself fails."""
...
def _default_ticker_factory(symbol: str) -> object:
import yfinance
return yfinance.Ticker(symbol)
[docs]
class YFinanceAnalystProvider:
"""Analyst opinions from yfinance. No pandas type escapes.
Caches a successful answer per symbol for this instance's lifetime
(issue #100) — "no coverage" for an ETF is a structural fact, not
something that changes cycle to cycle, so re-deriving it fresh from
yfinance every ~15 minutes, forever, was pure waste (and, for SPY
specifically, a guaranteed 404 every time — issue #98's reordering
below removed one of two redundant calls but, corrected by #100, cannot
remove the other: `.recommendations` alone already triggers it).
Deliberately no TTL: a symbol that gains real coverage later stays
cached as "none" until the process restarts, an accepted tradeoff — see
issue #100. Never cached on an exception, so a transient fetch failure
is retried fresh next call rather than permanently misremembered.
"""
def __init__(self, ticker_factory: Callable[[str], object] | None = None) -> None:
self._ticker_factory = ticker_factory or _default_ticker_factory
self._cache: dict[str, AnalystOpinion | None] = {}
[docs]
def get_opinion(self, symbol: str) -> AnalystOpinion | None:
ticker = symbol.strip().upper()
if ticker in self._cache:
return self._cache[ticker]
opinion = self._fetch_opinion(ticker)
self._cache[ticker] = opinion
return opinion
def _fetch_opinion(self, ticker: str) -> AnalystOpinion | None:
try:
instrument = self._ticker_factory(ticker)
frame = instrument.recommendations
except Exception as exc: # noqa: BLE001 - yfinance raises many types
raise MarketDataError(
f"Failed to fetch analyst opinions for {ticker}: {exc}"
) from exc
if frame is None or len(frame) == 0:
return None
# Row 0 is the current period; later rows are previous months.
row = frame.iloc[0]
try:
strong_buy = int(row["strongBuy"])
buy = int(row["buy"])
hold = int(row["hold"])
sell = int(row["sell"])
strong_sell = int(row["strongSell"])
except (KeyError, TypeError, ValueError) as exc:
raise MarketDataError(
f"yfinance returned an unusable recommendations row for {ticker}: {exc}"
) from exc
# `.info` is a second, separate yfinance surface from `.recommendations`,
# fetched only now that we know there is real coverage. An ETF's empty
# `.recommendations` frame already returned above without ever using
# `.info` — fetching it unconditionally cost every ETF a second
# request on top of the first (issue #98). The class docstring above
# covers the rest: caching the whole answer (issue #100) is what
# actually stops the repeat cost, since `.recommendations` alone
# already 404s for an ETF before this line is ever reached. Still
# `getattr(..., None)`: a test double or a future minimal fake with
# no `.info` attribute at all must not turn "no mean/target data"
# into a fetch failure for the five counts that already fetched
# successfully.
try:
info = getattr(instrument, "info", None)
except Exception as exc: # noqa: BLE001 - yfinance raises many types
raise MarketDataError(
f"Failed to fetch analyst opinions for {ticker}: {exc}"
) from exc
# Can be missing keys (or absent entirely) without that being a fetch
# failure — only an exception from either surface above raises.
info = info or {}
recommendation_mean = _as_float(info.get("recommendationMean"))
is_derived = False
if recommendation_mean is None:
recommendation_mean = _derive_recommendation_mean(
strong_buy, buy, hold, sell, strong_sell
)
is_derived = recommendation_mean is not None
return AnalystOpinion(
strong_buy=strong_buy,
buy=buy,
hold=hold,
sell=sell,
strong_sell=strong_sell,
recommendation_mean=recommendation_mean,
recommendation_mean_is_derived=is_derived,
target_mean_price=_as_decimal(info.get("targetMeanPrice")),
target_high_price=_as_decimal(info.get("targetHighPrice")),
)
def _as_float(value: object) -> float | None:
"""`value` as a finite `float`, or `None` for anything else.
Missing/malformed is degraded to "unknown" here, not raised: a mangled
`recommendationMean` must not cost the five counts that were fetched
successfully.
"""
if value is None:
return None
try:
result = float(value)
except (TypeError, ValueError):
return None
return result if math.isfinite(result) else None
def _as_decimal(value: object) -> Decimal | None:
"""`value` as a finite `Decimal`, or `None`.
`Decimal(str(value))`, not `Decimal(value)`: the same guard
`brokers/alpaca.py`'s `_to_decimal` uses, so a float from yfinance is
converted through its own printed representation rather than its binary
one.
"""
if value is None:
return None
try:
parsed = Decimal(str(value))
except (InvalidOperation, TypeError, ValueError):
return None
return parsed if parsed.is_finite() else None
def _derive_recommendation_mean(
strong_buy: int, buy: int, hold: int, sell: int, strong_sell: int
) -> float | None:
"""The weighted mean of the five counts, on Yahoo's 1..5 scale.
`None` when nobody has an opinion at all (`total == 0`) — the same "no
coverage" claim `AnalystProvider.get_opinion` returning `None` already
makes, never a fabricated 0.0.
"""
counts = (strong_buy, buy, hold, sell, strong_sell)
total = sum(counts)
if total == 0:
return None
return (
sum(weight * count for weight, count in zip(_MEAN_WEIGHTS, counts, strict=True))
/ total
)