"""Building the ranking prompt, and parsing what comes back.
Two rules define this module, and they are the same two that define
`prompt.py` — which is why it lives next to it.
**Nothing here raises.** Every malformed, missing, hallucinated, duplicated or
omitted response shape degrades to a valid permutation of exactly the
candidates passed in, with the fault recorded. The ranker cannot create a
trade — every candidate was already approved by its own decision and must still
clear every guardrail — so a degraded order is safe in a way a degraded
*decision* would not be. But it must be visible: a guard that silently swallows
a fault leaves nothing able to tell a good ranking from a rescued one.
**The prompt is reconstructible.** `RANK_PROMPT_VERSION` is stored with every
decision alongside the inputs. **Bump it whenever the wording changes**, or
rankings either side of the change become indistinguishable. It is deliberately
separate from `PROMPT_TEMPLATE_VERSION`: the two prompts change independently,
and this work does not touch the decision prompt at all.
"""
import logging
from collections.abc import Sequence
from trader.ranking.base import Candidate, fallback_ranking
__all__ = [
"RANK_PROMPT_VERSION",
"RANK_SCHEMA",
"build_rank_messages",
"parse_ranking",
]
#: Bump on any wording change. Stored with every ranked decision.
RANK_PROMPT_VERSION = "1"
_logger = logging.getLogger("trader.ranking")
#: Note the absence of a `rank` integer. Array order is authoritative, so
#: order and rank cannot contradict each other — the same reasoning that
#: removed `position_open: bool` from `Strategy.evaluate`.
RANK_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {
"ranking": {
"type": "array",
"items": {
"type": "object",
"properties": {
"symbol": {"type": "string"},
"reason": {"type": "string"},
},
"required": ["symbol", "reason"],
},
}
},
"required": ["ranking"],
}
_SYSTEM = """You are a disciplined long-only trading assistant allocating a \
limited amount of capital.
Every candidate below has already been approved for purchase. Your only job is
to put them in order, best first, for a short-term long entry. There may not be
enough capital for all of them, so the order decides which get funded.
Rules:
- You may only respond with the required JSON object.
- Include every candidate exactly once. Do not drop any, and do not add a
symbol that is not listed.
- You are ordering, not vetoing. "None of these" is not an available answer.
- Judge the evidence in front of you. Do not assume facts you were not given.
- A confidently written reason is not the same as a strong opportunity. Weigh
the numbers against the wording.
"""
def _candidate_block(candidate: Candidate) -> list[str]:
"""One candidate as prompt lines."""
comparables = candidate.comparables
confidence = (
"not reported" if candidate.confidence is None else f"{candidate.confidence:.2f}"
)
position = (
"flat window, undefined"
if comparables.range_position_pct is None
else f"{comparables.range_position_pct}% of its range"
)
return [
f"{candidate.symbol}",
f" Latest close: {candidate.last_close}",
f" Change over the last {comparables.bars_considered} bars: "
f"{comparables.window_change_pct}%",
f" Position in that window: {position}",
f" Strategy confidence: {confidence}",
f" Why it was approved: {candidate.reason}",
]
[docs]
def build_rank_messages(candidates: Sequence[Candidate]) -> tuple[str, str]:
"""Return `(system, user)` messages for one ranking.
Candidates are presented in **symbol order**, not confidence order.
Deterministic, so `prompt_hash` replay works — and position bias being
real, pre-sorting by confidence would make the laziest possible response
(repeat the given order) indistinguishable from a considered ranking.
Since distrust of raw confidence is the whole reason a comparative call
exists, hiding that failure would defeat the purpose.
"""
lines = [
f"Rank these {len(candidates)} candidates, best first.",
"",
]
for candidate in sorted(candidates, key=lambda c: c.symbol):
lines += _candidate_block(candidate)
lines.append("")
lines.append("Return every symbol above exactly once, best first.")
return _SYSTEM, "\n".join(lines)
[docs]
def parse_ranking(
payload: object, candidates: Sequence[Candidate]
) -> tuple[tuple[str, ...], dict[str, str], list[str]]:
"""Turn a model response into `(order, notes, problems)`. Never raises.
The returned order is **always a permutation of exactly the candidate
symbols** — never shorter, never longer, never containing a symbol nobody
proposed. Recognised symbols keep the order given; unknown ones are dropped;
duplicates keep their first place; anything omitted is appended in fallback
order. `problems` records every degradation, because a silent rescue is
indistinguishable from a good ranking.
"""
known = {c.symbol.upper(): c.symbol for c in candidates}
fallback = fallback_ranking(candidates).order
order: list[str] = []
notes: dict[str, str] = {}
problems: list[str] = []
raw = payload.get("ranking") if isinstance(payload, dict) else None
if not isinstance(raw, list):
problems.append(
f"response had no usable 'ranking' array (got {type(raw).__name__}); "
"falling back to the deterministic order"
)
raw = []
for element in raw:
if not isinstance(element, dict):
problems.append(f"ranking entry {element!r} is not an object; skipped")
continue
raw_symbol = element.get("symbol")
if not isinstance(raw_symbol, str) or not raw_symbol.strip():
problems.append(f"ranking entry {element!r} has no symbol; skipped")
continue
symbol = known.get(raw_symbol.strip().upper())
if symbol is None:
problems.append(f"{raw_symbol.strip().upper()} was not a candidate; dropped")
continue
if symbol in order:
problems.append(f"{symbol} appeared more than once; duplicate ignored")
continue
order.append(symbol)
reason = element.get("reason")
notes[symbol] = reason.strip() if isinstance(reason, str) else ""
missing = [symbol for symbol in fallback if symbol not in order]
if missing:
problems.append(
f"{', '.join(missing)} missing from the ranking; appended in confidence order"
)
order.extend(missing)
if problems:
_logger.warning("rank response needed repair: %s", "; ".join(problems))
return tuple(order), notes, problems