"""A local model as the ranker.
The first of two layers of failure containment. This one catches `TraderError`
and returns the deterministic fallback order with the error recorded; the
pipeline wraps the call in a second, outer `try`/`except` anyway, so a bug in
*this module* — a `TypeError`, not a `TraderError` — also degrades rather than
killing the cycle. Belt and braces, the same discipline
`trader/discovery/scan.py` and `trader/cli/main.py` apply to discovery, and for
the same reason: nothing about ranking may stop trading.
"""
import logging
from collections.abc import Sequence
from trader.errors import TraderError
from trader.llm.base import LlmProvider
from trader.llm.prompt import prompt_hash
from trader.llm.rank_prompt import (
RANK_PROMPT_VERSION,
RANK_SCHEMA,
build_rank_messages,
parse_ranking,
)
from trader.ranking.base import (
SOURCE_LLM,
SOURCE_SINGLE,
Candidate,
Ranking,
fallback_ranking,
)
__all__ = ["LlmRanker"]
_logger = logging.getLogger("trader.ranking")
[docs]
class LlmRanker:
"""Asks a language model to order approved candidates, best first."""
def __init__(self, llm: LlmProvider) -> None:
self._llm = llm
[docs]
def rank(self, candidates: Sequence[Candidate]) -> Ranking:
"""Order the candidates. Never raises.
Short-circuits at zero or one candidate: there is nothing to compare,
and the call costs roughly 15 seconds of a cycle for no information.
"""
if len(candidates) <= 1:
base = fallback_ranking(candidates)
return Ranking(
order=base.order,
source=SOURCE_SINGLE if candidates else base.source,
inputs={"rank_model": self._llm.model},
)
system, user = build_rank_messages(candidates)
inputs: dict[str, object] = {
"rank_model": self._llm.model,
"rank_prompt_version": RANK_PROMPT_VERSION,
"rank_prompt_sha256": prompt_hash(system, user),
}
try:
payload = self._llm.generate_json(system, user, RANK_SCHEMA)
except TraderError as exc:
# A model failure is a reason to spend capital in a less informed
# order, not a reason to stop spending it: every candidate here has
# already been approved and must still clear every guardrail, so a
# degraded order cannot unlock risk.
_logger.warning("ranking call failed, falling back: %s", exc)
base = fallback_ranking(candidates)
return Ranking(
order=base.order,
source=base.source,
inputs={**inputs, "rank_error": str(exc)},
)
order, notes, problems = parse_ranking(payload, candidates)
# Deliberately NOT logged here. `_allocate` logs the order and its
# source once per cycle; a second line saying nearly the same thing
# would make "one INFO line per cycle" false and would leave an
# operator unable to tell one ranking from two.
return Ranking(
order=order,
source=SOURCE_LLM,
notes=notes,
inputs={
**inputs,
"rank_raw_response": payload,
"rank_problems": problems,
},
)