"""Repository for `analyst_consensus_history` (issue #27).
Append-only writes from `trader scan-universe`; reads answer "what is the
newest-enough opinion for each of these symbols", which is the only question
`discover_symbols`'s analyst-scan candidate source and `scan-universe`'s own
freshness skip ever ask.
"""
from collections.abc import Collection
from dataclasses import dataclass
from datetime import datetime, timedelta
from sqlalchemy import func, select
from sqlalchemy.orm import Session, sessionmaker
from trader.marketdata.analysts import AnalystOpinion
from trader.persistence.models import AnalystConsensusHistory
__all__ = ["AnalystConsensusRepository", "ConsensusHistoryRow"]
[docs]
@dataclass(frozen=True, slots=True)
class ConsensusHistoryRow:
"""One `analyst_consensus_history` row, reduced to what a rank-over-time
read needs (issue #39) — deliberately not the full `AnalystOpinion`
shape, since `history()` can return hundreds of rows and every extra
field would be carried for nothing a caller here uses.
"""
symbol: str
fetched_at: datetime
strong_buy: int
recommendation_mean: float | None
[docs]
class AnalystConsensusRepository:
"""Writes and reads `analyst_consensus_history`."""
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
[docs]
def append(
self,
symbol: str,
opinion: AnalystOpinion,
*,
fetched_at: datetime,
source: str,
) -> int:
"""Record one fetch. Returns the new row id.
Always inserts — never upserts. A later fetch is evidence the
consensus moved, not a correction of the earlier one.
"""
row = AnalystConsensusHistory(
symbol=symbol.strip().upper(),
fetched_at=fetched_at,
strong_buy=opinion.strong_buy,
buy=opinion.buy,
hold=opinion.hold,
sell=opinion.sell,
strong_sell=opinion.strong_sell,
recommendation_mean=opinion.recommendation_mean,
recommendation_mean_is_derived=opinion.recommendation_mean_is_derived,
target_mean_price=opinion.target_mean_price,
target_high_price=opinion.target_high_price,
source=source,
)
with self._session_factory() as session:
session.add(row)
session.commit()
return row.id
[docs]
def latest_for_symbols(
self, symbols: Collection[str], *, max_age: timedelta, now: datetime
) -> dict[str, AnalystOpinion]:
"""The newest-per-symbol opinion, for symbols fetched within `max_age`.
A symbol with no row, or only a stale one, is simply absent from the
result — the same "this source contributes nothing" degrade
`discover_symbols` already gives an empty `themes` list, never an
exception.
"""
normalized = {s.strip().upper() for s in symbols}
if not normalized:
return {}
cutoff = now - max_age
# The newest `fetched_at` per symbol, then the row that matches it —
# two queries rather than a correlated subquery, for the same
# readability reason `NewsArchiveRepository` prefers a Python-side
# `_merge` over a single dense SQL expression.
latest_at = (
select(
AnalystConsensusHistory.symbol,
func.max(AnalystConsensusHistory.fetched_at).label("fetched_at"),
)
.where(
AnalystConsensusHistory.symbol.in_(normalized),
AnalystConsensusHistory.fetched_at >= cutoff,
)
.group_by(AnalystConsensusHistory.symbol)
.subquery()
)
statement = select(AnalystConsensusHistory).join(
latest_at,
(AnalystConsensusHistory.symbol == latest_at.c.symbol)
& (AnalystConsensusHistory.fetched_at == latest_at.c.fetched_at),
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
result: dict[str, AnalystOpinion] = {}
for row in rows:
result[row.symbol] = AnalystOpinion(
strong_buy=row.strong_buy,
buy=row.buy,
hold=row.hold,
sell=row.sell,
strong_sell=row.strong_sell,
recommendation_mean=row.recommendation_mean,
recommendation_mean_is_derived=row.recommendation_mean_is_derived,
target_mean_price=row.target_mean_price,
target_high_price=row.target_high_price,
)
return result
[docs]
def history(self) -> list[ConsensusHistoryRow]:
"""Every row, oldest first — the full history a rank-persistence
computation (issue #39) ranks over.
Unlike `latest_for_symbols`, this is not scoped to a symbol set or a
freshness window: the rank-persistence screen's whole point is
comparing a symbol's *older* readings against its newest one, so
filtering to "fresh" here would defeat it. Bounded by how often
`trader scan-universe` runs, not by index size, so a full-table read
is fine for a report command; there is no unbounded growth path this
needs to guard against yet.
"""
statement = select(AnalystConsensusHistory).order_by(
AnalystConsensusHistory.fetched_at.asc()
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
return [
ConsensusHistoryRow(
symbol=row.symbol,
fetched_at=row.fetched_at,
strong_buy=row.strong_buy,
recommendation_mean=row.recommendation_mean,
)
for row in rows
]