"""A rank-persistence / stability screen over `analyst_consensus_history`
(issue #39).
`discover_symbols()`'s analyst-scan source, and `trader scan-universe`
directly, already populate this append-only table on every run. A single
point-in-time reading — today's `strong_buy` count, or today's
`recommendation_mean` — says nothing about whether that reading is durable.
This ranks every symbol present in a given historical fetch (`fetched_at`)
against its peers *from that same fetch*, then reports the mean and standard
deviation of each symbol's rank across every fetch it appeared in. A low mean
rank and a low rank standard deviation together identify a name that is
*consistently* well-regarded, not merely well-regarded today — closing the
gap `docs/ideas.md`'s "Analyst consensus as a non-price input" entry flagged:
the delta is stored but was "never computed or surfaced anywhere."
Computed entirely over data already fetched — no new provider, no live
lookup. `ConsensusHistoryPoint` is a small, duck-typed row shape rather than
the SQLAlchemy `AnalystConsensusHistory` model, so this module — like the
rest of `discovery/` — never imports `persistence/`.
"""
from collections import defaultdict
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
__all__ = [
"METRIC_RECOMMENDATION_MEAN",
"METRIC_STRONG_BUY",
"ConsensusHistoryPoint",
"RankStability",
"compute_rank_stability",
]
#: Rank by `strong_buy` count, highest first (rank 1 = most strong-buys).
METRIC_STRONG_BUY = "strong_buy"
#: Rank by `recommendation_mean`, lowest first (Yahoo's 1=strong-buy ..
#: 5=strong-sell scale, so a low mean is the favourable end).
METRIC_RECOMMENDATION_MEAN = "recommendation_mean"
_SUPPORTED_METRICS = frozenset({METRIC_STRONG_BUY, METRIC_RECOMMENDATION_MEAN})
#: Below this many appearances, a standard deviation is undefined.
_MIN_APPEARANCES_FOR_STDDEV = 2
[docs]
@dataclass(frozen=True, slots=True)
class ConsensusHistoryPoint:
"""One `analyst_consensus_history` row, reduced to what ranking needs."""
symbol: str
fetched_at: datetime
strong_buy: int
recommendation_mean: float | None
[docs]
@dataclass(frozen=True, slots=True)
class RankStability:
"""One symbol's rank mean/stddev across every fetch it appeared in.
`stddev_rank` is `None` when the symbol appeared in exactly one fetch —
a standard deviation is undefined from a single point, and reporting it
as `0` would read as "proven stable" for a name that has simply never
been re-scanned. The same "unknown is not zero" discipline this codebase
already applies to missing analyst coverage and the fill-poll settle
guard.
"""
symbol: str
mean_rank: Decimal
stddev_rank: Decimal | None
appearances: int
[docs]
def compute_rank_stability(
points: Sequence[ConsensusHistoryPoint], *, metric: str = METRIC_STRONG_BUY
) -> list[RankStability]:
"""Rank every symbol within each `fetched_at` snapshot, then reduce each
symbol's ranks across snapshots to a mean and standard deviation.
Rank 1 is "most favoured" for either metric. A symbol missing the ranked
metric in a given snapshot (a `None` `recommendation_mean`, most often)
is excluded from *that* snapshot's ranking only — it neither claims a
rank nor drags one down, the same "absence is not zero" shape as
everything else this predicate touches.
Sorted with the most consistently well-regarded symbol first: ascending
mean rank, then ascending stddev rank, with a `None` stddev (a
single-appearance symbol — unproven consistency, not proven stability)
sorted after every symbol with a measured one.
Raises `ValueError` for an unsupported `metric` — a typo here must not
silently rank by the wrong column.
"""
if metric not in _SUPPORTED_METRICS:
raise ValueError(
f"Unsupported metric {metric!r}; expected one of {sorted(_SUPPORTED_METRICS)}"
)
by_snapshot: dict[datetime, list[ConsensusHistoryPoint]] = defaultdict(list)
for point in points:
by_snapshot[point.fetched_at].append(point)
ranks_by_symbol: dict[str, list[int]] = defaultdict(list)
for snapshot_points in by_snapshot.values():
for symbol, rank in _rank_snapshot(snapshot_points, metric):
ranks_by_symbol[symbol].append(rank)
results = [_stability_for(symbol, ranks) for symbol, ranks in ranks_by_symbol.items()]
results.sort(
key=lambda r: (
r.mean_rank,
r.stddev_rank if r.stddev_rank is not None else Decimal("Infinity"),
)
)
return results
def _rank_snapshot(
points: Sequence[ConsensusHistoryPoint], metric: str
) -> list[tuple[str, int]]:
"""One snapshot's `(symbol, rank)` pairs, best first, rank 1 = best.
Ties (and, for `recommendation_mean`, the exclusion of a `None` value)
are broken/handled deterministically by symbol so the same input always
produces the same ranking, never an artifact of dict/insertion order.
"""
if metric == METRIC_STRONG_BUY:
eligible: list[tuple[str, float]] = [(p.symbol, p.strong_buy) for p in points]
eligible.sort(key=lambda pair: (-pair[1], pair[0]))
else:
eligible = [
(p.symbol, p.recommendation_mean)
for p in points
if p.recommendation_mean is not None
]
eligible.sort(key=lambda pair: (pair[1], pair[0]))
return [(symbol, index + 1) for index, (symbol, _value) in enumerate(eligible)]
def _stability_for(symbol: str, ranks: Sequence[int]) -> RankStability:
count = len(ranks)
mean_rank = sum((Decimal(r) for r in ranks), Decimal(0)) / Decimal(count)
if count < _MIN_APPEARANCES_FOR_STDDEV:
stddev_rank = None
else:
variance = sum(
((Decimal(r) - mean_rank) ** 2 for r in ranks), Decimal(0)
) / Decimal(count - 1)
stddev_rank = variance.sqrt()
return RankStability(
symbol=symbol, mean_rank=mean_rank, stddev_rank=stddev_rank, appearances=count
)