"""The ordered gate a discovered symbol must clear.
Cheapest first, and the expensive work is not here at all: `run_once` does the
400-day fetch and the backtest, and only survivors of this chain ever reach
it. An analyst lookup is 0.05-0.23s; the alternative is a 400-day fetch.
There is no trend filter. An earlier version combined a standalone
net-negative-analyst rejection with a conjunction of downtrend AND
net-negative analysts; the standalone rule always returned first, so the
conjunction could never fire — dead code, found by mutation testing
(replacing it with `if False` left the whole suite passing) and confirmed
against a real trace. `trend_lookback_days` and `_is_falling` were removed
rather than kept as dead config and dead code. See the design doc's amended
item 7 for the full account. The standalone analyst filter already serves the
operator's cost concern ("no reason to backtest something down 30 days AND
analysts say sell"), and does so more aggressively, since it doesn't require
the downtrend half at all.
"""
from collections.abc import Sequence
from dataclasses import dataclass
from trader.config.schema import DiscoverySettings
from trader.discovery.candidates import Candidate
from trader.domain import Bar, TradableAsset
from trader.marketdata.analysts import AnalystOpinion
__all__ = [
"DEFAULT_STRONG_BUY_MEAN_THRESHOLD",
"REASON_ANALYST_NET_NEGATIVE",
"REASON_NOT_COMMON_STOCK",
"REASON_NOT_TRADABLE",
"REASON_NO_PRICE_HISTORY",
"REASON_PRICE_FLOOR",
"REASON_VOLUME_FLOOR",
"FilterOutcome",
"apply_filters",
"asset_gate",
"is_strong_buy_consensus",
"listing_gate",
"opinion_gate",
]
#: Yahoo's consensus scale is 1 (strong buy) .. 5 (strong sell); `<= 2.0`
#: mirrors the sibling `stock-screener` repo's validated `<= 1.5` for "strong
#: buy", loosened slightly since this is a discovery signal feeding a model
#: that reasons further, not a final accept/reject (issue #27, design doc
#: decision #6). A tuning parameter, not an architectural one.
DEFAULT_STRONG_BUY_MEAN_THRESHOLD = 2.0
_US_EQUITY = "us_equity"
# NASDAQ's fifth-letter suffixes for listings that are not common stock:
# W a warrant, U a unit, R a right. Positional and only meaningful as the
# fifth character, which is why the check is not "contains a W" — `VTAK`,
# `SNOW`, `CLW` and `NUCL` are ordinary tickers.
_NON_COMMON_SUFFIXES = frozenset("WUR")
_COMMON_STOCK_LENGTH = 5
# One structured code per rejection this chain can produce. `decisions.
# rejection_reason` (issue #21) stores exactly one of these — never the
# free-text `reason`, which is for a human, not a `GROUP BY`. Every gate
# below sets its own code; `apply_filters`/`screen_for_entry` never invent
# one from prose, for the same reason `veto_class` in `run_once.py` is never
# derived from `reason` text: wording is free to change, a stored code must
# not silently change meaning under it.
REASON_NOT_COMMON_STOCK = "not_common_stock"
REASON_NOT_TRADABLE = "not_tradable"
REASON_ANALYST_NET_NEGATIVE = "analyst_net_negative"
REASON_NO_PRICE_HISTORY = "no_price_history"
REASON_PRICE_FLOOR = "price_floor"
REASON_VOLUME_FLOOR = "volume_floor"
[docs]
@dataclass(frozen=True, slots=True)
class FilterOutcome:
"""Whether a candidate survived, and the reason either way.
`code` is `None` on an accepted outcome — there is nothing to name — and
always one of the `REASON_*` constants above on a rejection. It is what
`screen_for_entry` hands `run_once` so a blocked `fixed_tickers` entry's
`decisions` row can carry a structured `rejection_reason` instead of only
the free-text `reason`.
"""
accepted: bool
reason: str
code: str | None = None
[docs]
def asset_gate(candidate: Candidate, asset: TradableAsset | None) -> FilterOutcome | None:
"""The tradability check alone. `None` means it passed.
Exposed (not `_`-private) rather than inlined only in `apply_filters`, so
that `discovery.scan._evaluate` can run this exact check the moment it
has an asset — before it ever fetches an analyst opinion or a bar
history — without duplicating the rule or reaching into a private name.
This stays the one place tradability is decided; `apply_filters` below
composes it with the other gates rather than re-deriving it.
"""
if asset is None:
return FilterOutcome(
False,
f"{candidate.symbol} is not a tradable asset at the broker",
code=REASON_NOT_TRADABLE,
)
if not asset.tradable or asset.asset_class != _US_EQUITY:
return FilterOutcome(
False,
f"{candidate.symbol} is not a tradable US equity "
f"(tradable={asset.tradable}, class={asset.asset_class})",
code=REASON_NOT_TRADABLE,
)
return None
[docs]
def listing_gate(candidate: Candidate) -> FilterOutcome | None:
"""Reject a listing whose symbol says it is not common stock.
The cheapest gate there is — a string check, no network — so `scan`
runs it before `get_asset`, and it is the reason a theme scan that turns
up a warrant costs nothing to reject it.
It exists because the broker cannot answer this question. Measured
2026-08-04: Alpaca reports `HQWWW` and `NUCLW` as `tradable=True,
asset_class='us_equity'`, exactly like the common shares they derive from,
so `asset_gate` passes them. Both came out of a real theme scan. A warrant
is a different instrument — it expires, it is leveraged, and an 8%
trailing stop on one is not the risk that percentage implies on a share.
Neither is it the same rule as the volume floor. That floor happened to
reject those two, but a liquid warrant on a hot name would clear it and
nothing else in the chain would notice it was not a share.
Scope, deliberately: NASDAQ's positional fifth-letter convention only.
NYSE's `.WS`-style suffixes are a separate rule and are not covered — a
symbol this gate passes is not thereby certified to be common stock.
"""
symbol = candidate.symbol.strip().upper()
if len(symbol) == _COMMON_STOCK_LENGTH and symbol[-1] in _NON_COMMON_SUFFIXES:
return FilterOutcome(
False,
f"{symbol} is not common stock: a five-letter symbol ending in "
f"{symbol[-1]} is a warrant, unit or right",
code=REASON_NOT_COMMON_STOCK,
)
return None
[docs]
def opinion_gate(
candidate: Candidate, opinion: AnalystOpinion | None
) -> FilterOutcome | None:
"""The analyst-consensus check alone. `None` means it passed.
Absent coverage is not disapproval — ETFs have no analysts at all — so a
`None` opinion always passes here. Exposed for the same reason as
`asset_gate`: it lets `_evaluate` stop before ever fetching 30 days of
bars for a symbol analysts already dislike.
"""
# The `opinion is not None` guard is explicit rather than left to fall
# through a truthiness check on `opinion` itself, because `None` (no
# analysts cover the symbol — ETFs, mainly) and an `AnalystOpinion` of all
# zeros are different claims. `0 > 0` is `False`, so an all-zero fixture
# would make `is_net_negative` false and this check would pass anyway —
# for the wrong reason: it would look identical to genuinely absent
# coverage and hide a filter that really does reject on missing data,
# rather than proving this one correctly lets missing data through.
if opinion is not None and opinion.is_net_negative:
return FilterOutcome(
False,
f"{candidate.symbol}: analyst consensus is net negative "
f"({opinion.sell + opinion.strong_sell} bearish against "
f"{opinion.strong_buy + opinion.buy} bullish)",
code=REASON_ANALYST_NET_NEGATIVE,
)
return None
[docs]
def is_strong_buy_consensus(
opinion: AnalystOpinion, *, mean_threshold: float = DEFAULT_STRONG_BUY_MEAN_THRESHOLD
) -> bool:
"""Whether analysts favor this symbol strongly enough to *propose* it.
Additive, not a replacement for `opinion_gate`: this decides whether an
index-membership symbol becomes a `Candidate` at all (issue #27's
`origin="analyst_scan"`); `opinion_gate`'s net-negative reject still runs
on every candidate afterward, unchanged, as the safety floor.
A named, tested predicate rather than arithmetic inlined at the call
site, per the design doc's decision #6 — so the two conditions this
checks (a low mean, or strong-buy outnumbering everything else combined)
stay a single, greppable definition of "strong buy consensus" instead of
each caller re-deriving it slightly differently.
"""
if opinion.recommendation_mean is not None and opinion.recommendation_mean <= (
mean_threshold
):
return True
return opinion.strong_buy > (
opinion.buy + opinion.hold + opinion.sell + opinion.strong_sell
)
[docs]
def apply_filters(
candidate: Candidate,
*,
asset: TradableAsset | None,
opinion: AnalystOpinion | None,
bars: Sequence[Bar],
settings: DiscoverySettings,
) -> FilterOutcome:
"""Run the chain, stopping at the first rejection.
Pure and unchanged from the caller's point of view: it still takes every
input already gathered and decides in one call. The laziness that skips
gathering an input a candidate was always going to be rejected before
reaching lives in `discovery.scan._evaluate`, which calls `listing_gate`,
`asset_gate` and `opinion_gate` — the exact same functions this composes —
before it ever fetches the asset, the opinion or the bars this needs.
"""
outcome = listing_gate(candidate)
if outcome is not None:
return outcome
outcome = asset_gate(candidate, asset)
if outcome is not None:
return outcome
outcome = opinion_gate(candidate, opinion)
if outcome is not None:
return outcome
if not bars:
return FilterOutcome(
False, f"{candidate.symbol}: no price history", code=REASON_NO_PRICE_HISTORY
)
last_close = bars[-1].close
if last_close < settings.min_price:
return FilterOutcome(
False,
f"{candidate.symbol}: price {last_close} is below the "
f"{settings.min_price} floor",
code=REASON_PRICE_FLOOR,
)
average_volume = sum(bar.volume for bar in bars) // len(bars)
if average_volume < settings.min_avg_volume:
return FilterOutcome(
False,
f"{candidate.symbol}: average volume {average_volume} is below "
f"the {settings.min_avg_volume} floor",
code=REASON_VOLUME_FLOOR,
)
return FilterOutcome(True, f"{candidate.symbol} cleared every filter")