"""Turning configured themes — and, since issue #27, analyst-scan candidates
— into tradable symbols.
Never raises. A search outage, a rate limit, a malformed response, an analyst
lookup timing out — each is caught, logged, and yields fewer candidates rather
than an exception. The fixed tickers must keep trading through a discovery
failure, which is the same discipline that makes every LLM failure resolve to
HOLD.
Failures are isolated per theme and per candidate: one broken theme does not
discard another theme's results, and one symbol whose analyst call fails does
not discard its four siblings. The analyst-scan source (issue #27) follows
the identical discipline and is isolated from theme search in both
directions: a broken scan never loses a theme hit, and vice versa.
"""
import logging
from collections.abc import Callable, Collection, Sequence
from datetime import datetime, timedelta
from trader.config.schema import DiscoverySettings
from trader.discovery.candidates import Candidate
from trader.discovery.filters import (
FilterOutcome,
apply_filters,
asset_gate,
is_strong_buy_consensus,
listing_gate,
opinion_gate,
)
from trader.domain import Bar
__all__ = ["ANALYST_SCAN_INDEX", "discover_symbols", "screen_for_entry"]
_logger = logging.getLogger("trader.discovery")
#: `trader scan-universe` supports only this index today (issue #27's Open
#: Question #4 defers which others); `discover_symbols` reads the same one.
ANALYST_SCAN_INDEX = "sp500"
[docs]
def discover_symbols(
*,
settings: DiscoverySettings,
search,
analysts,
broker,
bars_for: Callable[[str], Sequence[Bar]],
fixed_symbols: Collection[str],
now: datetime,
event_repository=None,
index_repository=None,
consensus_repository=None,
) -> list[Candidate]:
"""Search every theme, plus (issue #27) whatever the analyst scan found.
`index_repository`/`consensus_repository` are optional and, like
`event_repository`, duck-typed rather than imported as concrete
persistence classes — `discovery/` stays free of a `persistence/`
dependency. Left `None` (every existing caller), the analyst-scan source
simply contributes nothing, the same degrade `themes: []` already gives
theme search — neither source blocks the other.
"""
if not settings.enabled:
return []
fixed = {s.strip().upper() for s in fixed_symbols}
seen: set[str] = set()
accepted: list[Candidate] = []
for theme in settings.themes:
try:
hits = search.search(theme, limit=settings.max_candidates_per_theme)
except Exception as exc: # noqa: BLE001 - a broken theme must not lose others
_logger.warning("discovery: search failed for %r: %s", theme, exc)
continue
for hit in hits:
symbol = hit.symbol.strip().upper()
if symbol in fixed or symbol in seen:
continue
seen.add(symbol)
candidate = Candidate(
symbol=symbol,
theme=hit.theme,
origin=hit.origin,
headline=hit.headline,
url=hit.url,
)
_evaluate_and_accept(
candidate,
analysts,
broker,
bars_for,
settings,
accepted,
event_repository,
now,
)
if index_repository is not None and consensus_repository is not None:
_scan_analyst_candidates(
settings=settings,
analysts=analysts,
broker=broker,
bars_for=bars_for,
fixed=fixed,
seen=seen,
accepted=accepted,
event_repository=event_repository,
now=now,
index_repository=index_repository,
consensus_repository=consensus_repository,
)
return accepted
def _evaluate_and_accept(
candidate: Candidate,
analysts,
broker,
bars_for,
settings,
accepted,
event_repository,
now,
) -> None:
"""Run `_evaluate` and, on acceptance, log/append/record it. Shared by
both candidate sources so accepting a symbol means one thing everywhere."""
outcome = _evaluate(candidate, analysts, broker, bars_for, settings)
if outcome is None:
return
if not outcome.accepted:
_logger.info("discovery: rejected %s", outcome.reason)
return
_logger.info(
"discovery: %s from theme %r (%s)",
candidate.symbol,
candidate.theme,
candidate.origin,
)
accepted.append(candidate)
_record(event_repository, candidate, now)
def _scan_analyst_candidates(
*,
settings: DiscoverySettings,
analysts,
broker,
bars_for,
fixed: set[str],
seen: set[str],
accepted: list[Candidate],
event_repository,
now: datetime,
index_repository,
consensus_repository,
) -> None:
"""Symbols the S&P 500 scan found with a strong buy consensus (issue #27).
Upstream of `_evaluate`'s chain, not a replacement for it: this only
decides which index members are worth proposing as candidates at all —
every proposal still runs the identical `listing_gate`/`asset_gate`/
`opinion_gate`/price-volume chain every theme-search candidate gets,
including a fresh, live `analysts.get_opinion` call (the persisted
consensus here is only used for the upstream strong-buy check). Volume
stays bounded because only strong-buy survivors — a small subset of the
index — ever reach that live chain; the full index is never evaluated at
cycle scale, only queried from two cheap DB tables.
Failures degrade to "no candidates from this source", matching the outer
try/except `cli._discover_symbols_guarded` already wraps this whole
function in.
"""
try:
members = index_repository.members(ANALYST_SCAN_INDEX)
except Exception as exc: # noqa: BLE001 - a broken source must not lose theme results
_logger.warning("discovery: index membership read failed: %s", exc)
return
candidates = [m for m in members if m not in fixed and m not in seen]
if not candidates:
return
try:
opinions = consensus_repository.latest_for_symbols(
candidates,
max_age=timedelta(hours=settings.analyst_scan_max_age_hours),
now=now,
)
except Exception as exc: # noqa: BLE001 - a broken source must not lose theme results
_logger.warning("discovery: analyst consensus read failed: %s", exc)
return
for symbol, opinion in opinions.items():
if symbol in fixed or symbol in seen or not is_strong_buy_consensus(opinion):
continue
seen.add(symbol)
candidate = Candidate(
symbol=symbol,
theme="analyst_scan",
origin="analyst_scan",
recommendation_mean=opinion.recommendation_mean,
)
_evaluate_and_accept(
candidate,
analysts,
broker,
bars_for,
settings,
accepted,
event_repository,
now,
)
def _evaluate(candidate, analysts, broker, bars_for, settings):
"""Gather the inputs and run the chain, one gate at a time.
Lazily, so a candidate a cheap gate was always going to reject never
costs an expensive one: `get_asset` before `get_opinion` (~0.05-0.23s)
before `bars_for` (a 30-day fetch that also persists coverage rows).
`asset_gate`/`opinion_gate` are the exact functions `apply_filters` itself
composes, so a candidate that fails here is rejected for the identical
reason it would have been rejected by the eager, all-at-once call — the
laziness changes when each input is fetched, never what "pass" means.
`None` when any one lookup failed — a per-candidate failure drops only
this candidate, same as before.
"""
# Free, so it precedes every lookup: a warrant a theme scan turned up
# should not cost a broker call, an analyst call and a 30-day fetch to
# reject. Same function `apply_filters` composes, so "pass" means the same
# thing either way.
outcome = listing_gate(candidate)
if outcome is not None:
return outcome
try:
asset = broker.get_asset(candidate.symbol)
except Exception as exc: # noqa: BLE001 - one bad symbol must not lose the rest
_logger.warning("discovery: could not evaluate %s: %s", candidate.symbol, exc)
return None
outcome = asset_gate(candidate, asset)
if outcome is not None:
return outcome
try:
opinion = analysts.get_opinion(candidate.symbol)
except Exception as exc: # noqa: BLE001 - one bad symbol must not lose the rest
_logger.warning("discovery: could not evaluate %s: %s", candidate.symbol, exc)
return None
outcome = opinion_gate(candidate, opinion)
if outcome is not None:
return outcome
try:
bars = bars_for(candidate.symbol)
except Exception as exc: # noqa: BLE001 - one bad symbol must not lose the rest
_logger.warning("discovery: could not evaluate %s: %s", candidate.symbol, exc)
return None
return apply_filters(
candidate, asset=asset, opinion=opinion, bars=bars, settings=settings
)
def _record(event_repository, candidate: Candidate, now: datetime) -> None:
"""Audit only. A failed write must not lose the candidate."""
if event_repository is None:
return
try:
event_repository.record_discovery(
symbol=candidate.symbol,
theme=candidate.theme,
origin=candidate.origin,
headline=candidate.headline,
url=candidate.url,
discovered_at=now,
)
except Exception as exc: # noqa: BLE001 - audit must not break trading
_logger.warning("discovery: could not record %s: %s", candidate.symbol, exc)
[docs]
def screen_for_entry(
symbols,
*,
settings,
analysts,
broker,
bars_for,
) -> dict[str, FilterOutcome]:
"""Which of these symbols must not be bought, and why.
The operator's own `fixed_tickers` run through the same gates a discovered
candidate does — "I don't know that these are good purchases, I only know
I am interested in watching them". A symbol that fails is still evaluated
and still records a decision; only the *entry* is blocked, which is what
keeps a watchlist a watchlist.
Returns `{SYMBOL: FilterOutcome}` for failures only, the whole outcome
rather than just its `reason` text: `run_once` writes `outcome.reason`
into `decisions.reasoning` (for a human) and `outcome.code` into
`decisions.rejection_reason` (for a `GROUP BY`), and a single object is
what keeps those two in sync — two parallel dicts keyed by the same
symbols would be the "two fields that can contradict each other" shape
this codebase's own architecture review flags.
Never raises: a screening outage must not block trading, so a symbol
whose lookups fail is simply not blocked — the same fail-open direction
`discover_symbols` takes, and the opposite of the exposure seed, which
fails closed because guessing low there would spend money.
"""
blocked: dict[str, FilterOutcome] = {}
for symbol in symbols:
candidate = Candidate(
symbol=symbol.strip().upper(), theme="fixed_tickers", origin="fixed"
)
outcome = _evaluate(candidate, analysts, broker, bars_for, settings)
if outcome is not None and not outcome.accepted:
blocked[candidate.symbol] = outcome
_logger.info(
"screening: %s may not be bought — %s", candidate.symbol, outcome.reason
)
return blocked