Source code for trader.ranking.base

"""What a ranker takes, what it returns, and the ordering used without one.

A ranker never sees a strategy, a broker, an executor or a repository — the
same discipline `daemon/` follows. It takes value objects and returns an order.
That is what lets it be tested without a model and swapped without touching the
pipeline.
"""

from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Protocol, runtime_checkable

from trader.ranking.comparables import Comparables

__all__ = [
    "SOURCE_CONFIDENCE_ORDERING_ENABLED",
    "SOURCE_FALLBACK_CONFIDENCE",
    "SOURCE_FALLBACK_SYMBOL",
    "SOURCE_LLM",
    "SOURCE_SINGLE",
    "Candidate",
    "Ranker",
    "Ranking",
    "fallback_ranking",
]

#: How an order was arrived at. Recorded on every decision row, so a log line
#: or a stored row never claims a ranking basis it did not have.
SOURCE_LLM = "llm"
SOURCE_FALLBACK_CONFIDENCE = "fallback:confidence"
SOURCE_FALLBACK_SYMBOL = "fallback:symbol"
SOURCE_SINGLE = "skipped:single-candidate"
#: `pipeline_config.confidence_ordering_enabled` forced this order, bypassing
#: whatever ranker was configured (issue #86). Distinct from
#: `SOURCE_FALLBACK_CONFIDENCE` on purpose: that source means "no ranker was
#: usable"; this one means "a ranker may well have been usable, an operator
#: chose not to use it" — collapsing the two would misreport why a cycle's
#: order was what it was.
SOURCE_CONFIDENCE_ORDERING_ENABLED = "confidence_ordering_enabled"

#: Sorts below any real confidence, including 0.0. A missing confidence must
#: not read as a high one — that would put the least-informed candidate first.
_ABSENT_CONFIDENCE = -1.0




@dataclass(frozen=True, slots=True)
class Candidate:
    """One approved BUY waiting on capital — the ranker's whole input.

    `confidence` is `float | None` rather than `float`: `LlmStrategy` puts one
    in `Signal.indicators`, and the rule strategies put nothing there. `None`
    means "this strategy does not report one", which is a different claim from
    0.0 and must stay distinguishable.
    """

    symbol: str
    confidence: float | None
    reason: str
    last_close: Decimal
    comparables: Comparables





@dataclass(frozen=True, slots=True)
class Ranking:
    """An order over candidate symbols, and how it was arrived at."""

    #: Best first. Always a permutation of exactly the candidate symbols.
    order: tuple[str, ...]
    source: str
    #: Symbol -> the ranker's own words about it. Empty on every fallback path.
    notes: Mapping[str, str] = field(default_factory=dict)
    #: Provenance for `decisions.inputs_json`.
    inputs: Mapping[str, object] = field(default_factory=dict)





@runtime_checkable
class Ranker(Protocol):
    """Orders approved candidates. Never raises, never vetoes, never sizes."""



    def rank(self, candidates: Sequence[Candidate]) -> Ranking:
        """Return an order over `candidates`, best first.

        The result must be a permutation of exactly the symbols passed in:
        never shorter, never longer, never containing a symbol nobody proposed.
        A ranker that could drop a candidate would be vetoing, which is
        deliberately not its job — every candidate has already been approved by
        its own per-symbol decision one call earlier.
        """
        ...






def fallback_ranking(candidates: Sequence[Candidate]) -> Ranking:
    """Confidence descending, ties broken on symbol ascending.

    Used whenever a model ranking is unavailable or unusable. Safe as a default
    precisely because ranking only *orders*: every candidate here has already
    passed its own decision and must still clear every guardrail, so a degraded
    order cannot unlock risk — it only spends the headroom in a less informed
    sequence. Refusing to trade at all on a ranker outage would make a dead
    endpoint a single point of failure for trading, which is the shape the
    discovery rules forbid.

    Ties break on symbol so a cycle is reproducible.
    """
    order = tuple(
        c.symbol
        for c in sorted(
            candidates,
            key=lambda c: (
                -(c.confidence if c.confidence is not None else _ABSENT_CONFIDENCE),
                c.symbol,
            ),
        )
    )
    # With no confidence anywhere, every comparison tied and this is really
    # alphabetical order. Saying "fallback:confidence" would claim a basis it
    # did not have.
    source = (
        SOURCE_FALLBACK_CONFIDENCE
        if any(c.confidence is not None for c in candidates)
        else SOURCE_FALLBACK_SYMBOL
    )
    return Ranking(order=order, source=source)