"""Building the decision prompt, and parsing what comes back.
Two rules define this module.
**Everything that can go wrong becomes HOLD.** An unknown action, a missing
field, a non-numeric confidence — none of them raise, and none of them trade. A
misbehaving model must never be able to open a position, and the safest thing a
long-only system can do is nothing.
**The prompt is reconstructible.** `PROMPT_TEMPLATE_VERSION` is stored with
every decision alongside the inputs, so the exact prompt can be rebuilt later
and re-run against a different model. `prompt_hash` then proves the rebuild
matches what was actually sent. **Bump the version whenever the wording
changes**, or old decisions become silently unreproducible.
"""
import hashlib
import math
from collections.abc import Sequence
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from trader.domain import Bar, Position
from trader.indicators import atr, bollinger, donchian_high, donchian_low, rsi, sma
from trader.marketdata.analysts import AnalystOpinion
from trader.news.base import NewsItem
from trader.reporting.returns import PERCENT_QUANTUM, position_return_pct, return_pct
from trader.strategies.base import Action
__all__ = [
"DECISION_SCHEMA",
"PROMPT_TEMPLATE_VERSION",
"build_bar_summary",
"build_messages",
"build_technical_summary",
"parse_decision",
"prompt_hash",
]
#: Bump on any wording change. Stored with every decision.
#: "3" (2026-08-13): the prompt now states the current date and labels each
#: headline's relative age, and stale items are withheld. See issue #5.
#: "4" (2026-08-15): the wording is unchanged, but the news reaching it comes
#: from two merged feeds (yfinance plus Alpaca/Benzinga, 0 of 10 shared
#: headlines) instead of one. The corpus is what the model reasons over, so a
#: decision made on the merged feed is not comparable to one made on yfinance
#: alone — this version is the only thing that keeps the two sets of rows
#: distinguishable. See issue #7.
#: "5" (2026-08-19): analyst consensus (strong buy/buy/hold/sell/strong sell
#: counts) is now shown for every symbol, where discovery already fetched it
#: to reject net-negative candidates and then discarded it. See issue #32.
#: "6" (2026-08-19): the analyst-ratings line gains a recommendation mean
#: (Yahoo's 1..5 consensus scale, labelled when this app derived it rather
#: than yfinance reporting it) and target-price upside off the latest close
#: already in context — no extra fetch. See issue #27.
#: "7" (2026-08-23): a "Technical signals" block is added — RSI(14),
#: Bollinger-band and Donchian-channel position, distance from the 50- and
#: 200-day moving averages, and ATR(14) — computed once per symbol from bars
#: already in the local cache, the same indicators three shadow strategies
#: already compute and previously threw away. See issue #63. "Performance
#: versus SPY and versus sector" (also named in issue #63's source inventory)
#: is deliberately NOT included: SPY's own bars are a different symbol's data
#: never passed to `evaluate()`, and "sector" needs a symbol->sector taxonomy
#: this codebase has never fetched — both are a new data source/fetch, which
#: contradicts the "no new data source, no new fetch" framing issue #63 itself
#: uses to scope this change in over items 1/6/7 of the same inventory. Left
#: for a follow-up issue with its own collaborator-injection design, the same
#: way `analyst_provider` was designed rather than folded into this change.
#: "8" (2026-08-25): the technical-signals block gains a relative-strength
#: line — the symbol's 20-day return minus a benchmark's (default SPY) same-
#: window return. This is the follow-up issue "7"'s own comment named: SPY's
#: bars now do reach `evaluate()`, via `LlmStrategy.set_benchmark_bars` (a
#: duck-typed collaborator `run_once.py` calls before `evaluate()`, not a
#: `Strategy` Protocol change), fed from the SAME `BarCache` the cycle
#: already populated for SPY as a `fixed_tickers` entry — still no new fetch,
#: no new data source. See issue #87.
#: "9" (2026-08-31): a "Next earnings date" line is added, from a new
#: `EarningsProvider` (yfinance's `.calendar`, see `marketdata/earnings.py`)
#: -- this one IS a new fetch, unlike "7"/"8". Context only: never gates or
#: blocks an entry in v1, the same posture "5"/"6" shipped analyst consensus
#: with. See issue #89.
PROMPT_TEMPLATE_VERSION = "9"
#: RSI period. Matches `rsi_revert_14`'s configured default, so the number the
#: model is shown is the same number that shadow strategy already computes.
_TECH_RSI_PERIOD = 14
#: Bollinger period and band width. Matches `bollinger_revert_20`'s default.
_TECH_BOLLINGER_PERIOD = 20
_TECH_BOLLINGER_NUM_STD = 2
#: Donchian entry/exit periods. Matches `turtle_20_10`'s default.
_TECH_DONCHIAN_ENTRY_PERIOD = 20
_TECH_DONCHIAN_EXIT_PERIOD = 10
#: Trend moving averages: docs/ideas.md's "position against the 50- and
#: 200-day averages" (issue #63), a conventional pair with no shadow-strategy
#: precedent to match.
_TECH_SHORT_MA_PERIOD = 50
_TECH_LONG_MA_PERIOD = 200
#: ATR period. 14 is the conventional default (Wilder's own), and no shadow
#: strategy computes ATR today, so there is no existing configured value to
#: match.
_TECH_ATR_PERIOD = 14
#: Relative-strength lookback, in trading days. 20 matches the existing
#: Bollinger/Donchian-entry precedent above rather than introducing a fourth
#: distinct period this module has to justify separately (issue #87).
_TECH_RELATIVE_STRENGTH_PERIOD = 20
#: Default benchmark when a caller passes benchmark bars without naming the
#: symbol they came from. `build_technical_summary`'s own callers always
#: supply both together in practice; this exists so the two params can't
#: silently drift (a value with no name to render).
_DEFAULT_BENCHMARK_SYMBOL = "SPY"
#: Below this, `_age_phrase` reads "just now" rather than "0h ago".
_JUST_NOW_HOURS = 1
#: At or above this, `_age_phrase` switches from hours to whole days, so a
#: three-week-old item reads "21 days ago" rather than "504h ago".
_DAYS_PHRASE_HOURS = 48
DECISION_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["buy", "sell", "hold"]},
"confidence": {"type": "number"},
"reason": {"type": "string"},
},
"required": ["action", "confidence", "reason"],
}
_SYSTEM = """You are a disciplined long-only trading assistant.
For the symbol you are given, decide one of:
buy - open a long position
sell - close an existing long position
hold - do nothing
Rules:
- You may only respond with the required JSON object.
- hold is always an acceptable answer, and is the right answer when the
evidence is weak or mixed.
- Only recommend sell when a position is already open.
- Judge the evidence in front of you. Do not assume facts you were not given.
"""
[docs]
def build_bar_summary(bars: Sequence[Bar], lookback: int) -> dict[str, object]:
"""A compact, JSON-safe description of recent price action.
Prices are strings, not floats: they are `Decimal` everywhere else in this
codebase and this dict is written to `decisions.inputs_json`, where a float
would both lose precision and misrepresent what the model was shown.
"""
window = list(bars[-lookback:]) if lookback > 0 else list(bars)
closes = [str(b.close) for b in window]
highs = [b.high for b in window]
lows = [b.low for b in window]
return {
"bars_considered": len(window),
"recent_closes": closes,
"last_close": closes[-1] if closes else None,
"window_high": str(max(highs)) if highs else None,
"window_low": str(min(lows)) if lows else None,
"first_timestamp": window[0].timestamp.isoformat() if window else None,
"last_timestamp": window[-1].timestamp.isoformat() if window else None,
}
def _n_day_return(window: Sequence[Bar], period: int) -> Decimal | None:
"""The close-to-close return over the trailing `period` trading days.
`None` when `window` is too short to look `period` bars back — the same
"not yet available" contract every other reading in this module uses,
never a fabricated number from a shorter window than asked for.
"""
if len(window) <= period:
return None
latest = window[-1].close
prior = window[-1 - period].close
return return_pct(latest - prior, prior)
[docs]
def build_technical_summary(
bars: Sequence[Bar],
benchmark_bars: Sequence[Bar] | None = None,
benchmark_symbol: str = _DEFAULT_BENCHMARK_SYMBOL,
) -> dict[str, object]:
"""Technical, trend and volatility readings, computed once per symbol.
Unlike `build_bar_summary`, which is deliberately sliced to the prompt's
recent-closes window, this reads the *full* `bars` sequence `evaluate()`
already has — a 200-day moving average needs 200 bars, not the 30 the
prompt narrates. No new fetch: `bars` is exactly what the pipeline already
passed in (issue #63), and `benchmark_bars`, when given, is the same
benchmark symbol's bars the cycle's `BarCache` already holds for its own
evaluation as a `fixed_tickers` entry — also no new fetch (issue #87).
Same JSON-safety discipline as `build_bar_summary`: every price-valued
output is a `str`, not a `float`, because this dict is written to
`decisions.inputs_json`. RSI and the two channel-position percentages are
dimensionless/percentage readings and stay `float`/`str`-of-`Decimal`
respectively, matching how `rsi_revert`/`bollinger_revert`/`turtle`
already type the same numbers.
A period whose warmup is not yet satisfied, or a division that would be
by a zero-width band or channel, reports `None` for that one field rather
than a fabricated number — "not yet available" and "0" are different
claims, and `build_messages` renders the two differently. One field being
unavailable never blocks the others: a young symbol with no 200-day
history still gets RSI, Bollinger and Donchian readings.
`relative_strength_pct` is `None` whenever either leg is unavailable —
`bars` too short for its own `_TECH_RELATIVE_STRENGTH_PERIOD`-day return,
or `benchmark_bars` missing/empty/too short for the same. It never
reports one symbol's return standing in for a "relative" reading that
only had one side of the comparison ("missing is not zero", same
discipline as absent analyst coverage). `relative_strength_benchmark_
symbol` is set only alongside a computed value, so the render layer never
has a benchmark name with no number behind it.
"""
if not bars:
return {
"rsi_14": None,
"bollinger_pct_b": None,
"donchian_pct": None,
"sma_50": None,
"sma_50_distance_pct": None,
"sma_200": None,
"sma_200_distance_pct": None,
"atr_14": None,
"atr_pct_of_price": None,
"relative_strength_pct": None,
"relative_strength_benchmark_symbol": None,
}
window = list(bars)
close = window[-1].close
rsi_value = rsi(window, _TECH_RSI_PERIOD)[-1]
band = bollinger(window, _TECH_BOLLINGER_PERIOD, _TECH_BOLLINGER_NUM_STD)[-1]
# Percent of the band's own width, 0% = lower band, 100% = upper band —
# `return_pct(pl, basis)` is exactly "close - lower, as a percentage of
# (upper - lower)", the same "one function, not two formulas" reasoning
# `position_return_pct` documents at the top of this module.
bollinger_pct_b = (
return_pct(close - band.lower, band.upper - band.lower)
if band is not None
else None
)
entry = donchian_high(window, _TECH_DONCHIAN_ENTRY_PERIOD)[-1]
exit_channel = donchian_low(window, _TECH_DONCHIAN_EXIT_PERIOD)[-1]
donchian_pct = (
return_pct(close - exit_channel, entry - exit_channel)
if entry is not None and exit_channel is not None
else None
)
sma_short = sma(window, _TECH_SHORT_MA_PERIOD)[-1]
sma_short_distance = (
return_pct(close - sma_short, sma_short) if sma_short is not None else None
)
sma_long = sma(window, _TECH_LONG_MA_PERIOD)[-1]
sma_long_distance = (
return_pct(close - sma_long, sma_long) if sma_long is not None else None
)
atr_value = atr(window, _TECH_ATR_PERIOD)[-1]
# Normalised against price, not shown as a raw dollar figure alone: a raw
# ATR is not comparable across symbols at very different price levels, and
# "how volatile is this relative to its own price" is what the trailing
# stop's uniform-percentage problem (docs/ideas.md item 5) actually needs.
atr_pct = return_pct(atr_value, close) if atr_value is not None else None
relative_strength_pct: Decimal | None = None
relative_strength_benchmark_symbol: str | None = None
symbol_return = _n_day_return(window, _TECH_RELATIVE_STRENGTH_PERIOD)
if symbol_return is not None and benchmark_bars:
benchmark_return = _n_day_return(
list(benchmark_bars), _TECH_RELATIVE_STRENGTH_PERIOD
)
if benchmark_return is not None:
relative_strength_pct = symbol_return - benchmark_return
relative_strength_benchmark_symbol = benchmark_symbol
return {
"rsi_14": rsi_value,
"bollinger_pct_b": str(bollinger_pct_b) if bollinger_pct_b is not None else None,
"donchian_pct": str(donchian_pct) if donchian_pct is not None else None,
"sma_50": str(sma_short) if sma_short is not None else None,
"sma_50_distance_pct": (
str(sma_short_distance) if sma_short_distance is not None else None
),
"sma_200": str(sma_long) if sma_long is not None else None,
"sma_200_distance_pct": (
str(sma_long_distance) if sma_long_distance is not None else None
),
"atr_14": str(atr_value) if atr_value is not None else None,
"atr_pct_of_price": str(atr_pct) if atr_pct is not None else None,
"relative_strength_pct": (
str(relative_strength_pct) if relative_strength_pct is not None else None
),
"relative_strength_benchmark_symbol": relative_strength_benchmark_symbol,
}
def _technical_lines(technicals: dict[str, object] | None) -> list[str]:
"""Render `build_technical_summary`'s dict as prompt text.
Every line degrades to its own "not available" wording rather than being
silently omitted: an absent line reads as "nothing to consider" (the same
reasoning `build_messages` already documents for the no-news case), and a
model shown four working indicators and silence on the fifth has no way to
tell "not computed" from "computed and uninteresting".
"""
if not technicals:
return ["Technical signals: not available."]
def _num(key: str, label: str, suffix: str = "") -> str:
value = technicals.get(key)
if value is None:
return f"{label}: not available."
return f"{label}: {value}{suffix}"
rsi_value = technicals.get("rsi_14")
rsi_line = (
f"RSI (14): {rsi_value:.1f}"
if isinstance(rsi_value, (int, float))
else "RSI (14): not available."
)
lines = [
"Technical signals:",
rsi_line,
_num(
"bollinger_pct_b",
"Bollinger-band position",
"% of band width (0% = lower band, 100% = upper band)",
),
_num(
"donchian_pct",
"Donchian-channel position",
"% of channel width (0% = 10-day low, 100% = 20-day high)",
),
]
sma_50 = technicals.get("sma_50")
sma_50_distance = technicals.get("sma_50_distance_pct")
if sma_50 is None or sma_50_distance is None:
lines.append("50-day moving average: not available.")
else:
lines.append(
f"50-day moving average: {sma_50} ({sma_50_distance}% vs. latest close)"
)
sma_200 = technicals.get("sma_200")
sma_200_distance = technicals.get("sma_200_distance_pct")
if sma_200 is None or sma_200_distance is None:
lines.append("200-day moving average: not available.")
else:
lines.append(
f"200-day moving average: {sma_200} ({sma_200_distance}% vs. latest close)"
)
atr_14 = technicals.get("atr_14")
atr_pct = technicals.get("atr_pct_of_price")
if atr_14 is None or atr_pct is None:
lines.append("Average True Range (14): not available.")
else:
lines.append(f"Average True Range (14): {atr_14} ({atr_pct}% of latest close)")
relative_strength_pct = technicals.get("relative_strength_pct")
relative_strength_benchmark = technicals.get("relative_strength_benchmark_symbol")
if relative_strength_pct is None or relative_strength_benchmark is None:
lines.append("Relative strength: not available.")
else:
lines.append(
f"Relative strength vs {relative_strength_benchmark} "
f"({_TECH_RELATIVE_STRENGTH_PERIOD}d): {relative_strength_pct}%."
)
return lines
def _position_lines(position: Position) -> list[str]:
"""Describe the holding, including whether it is up or down.
`position_return_pct` (`trader/reporting.py`) is the same function
`trader positions`/`trader account` call for the identical figure, so the
operator and the model never see two different numbers for "up 7%". It
returns `None` only for a zero cost basis, which `build_messages` must
never raise on — a prompt is still owed even for a position gifted at
zero cost — so that case falls back to `0.00%` here, deliberately, rather
than propagating the `None`.
"""
pct = position_return_pct(position)
if pct is None:
pct = Decimal(0).quantize(PERCENT_QUANTUM)
return [
f"Position held: {position.quantity} shares, "
f"entry {position.avg_entry_price}, now {position.current_price}, "
f"unrealized P/L {position.unrealized_pl:+} ({pct:+}%)",
]
def _age_phrase(now: datetime, published_at: datetime) -> str:
"""How old an item is, in words a model reads correctly.
An absolute timestamp requires knowing today's date to interpret, and
measured, the model narrated three-week-old headlines as live catalysts
when only the timestamp was given. A future-dated item — publisher clock
skew, real in the stored data — reads "just now" rather than a negative
age, which would read as nonsense or as the future.
"""
hours = (now - published_at).total_seconds() / 3600
if hours < _JUST_NOW_HOURS:
return "just now"
if hours < _DAYS_PHRASE_HOURS:
return f"{int(hours)}h ago"
return f"{int(hours // 24)} days ago"
def _recommendation_mean_line(analyst: AnalystOpinion) -> str | None:
"""Yahoo's 1 (strong buy) .. 5 (strong sell) consensus scale, or `None`.
Labelled "derived" when this app computed it from the five counts rather
than yfinance reporting it directly (issue #27) — the model is told which
kind of number it is seeing, the same honesty `recommendation_mean_is_
derived` exists for in `decisions.inputs_json`.
"""
if analyst.recommendation_mean is None:
return None
derived = (
" (derived from the counts above)"
if analyst.recommendation_mean_is_derived
else ""
)
return f"Analyst recommendation mean: {analyst.recommendation_mean:.2f}{derived}."
def _target_price_line(analyst: AnalystOpinion, last_close: object) -> str | None:
"""Target-price upside off the close already in `bar_summary` — no extra
fetch. `None` when either the target or the close is unavailable; a
fabricated upside from a missing close would be worse than no line."""
if analyst.target_mean_price is None or last_close is None:
return None
try:
close = Decimal(str(last_close))
except InvalidOperation:
return None
if close <= 0:
return None
upside = (analyst.target_mean_price - close) / close * 100
high_part = (
f", high target {analyst.target_high_price}"
if analyst.target_high_price is not None
else ""
)
return (
f"Analyst target price: mean {analyst.target_mean_price} "
f"({upside:+.1f}% vs. latest close){high_part}."
)
def _earnings_line(earnings: date | None, now: datetime) -> str:
""" "Next earnings date: ... (in N days)." or "not available." (issue #89).
`now` is the same clock `build_messages`'s caller already passes for the
"Current date and time" line and the news-age labels — no new time
source. `N` can be negative (an earnings date already passed but not yet
refreshed in the cache) or zero; neither is filtered here, the same
"-0.0h" discipline `docs/invariants.md#the-news-recency-window`
documents for the news window: the model is shown what is known, not a
version of it clamped to look tidy.
"""
if earnings is None:
return "Next earnings date: not available."
days = (earnings - now.date()).days
return f"Next earnings date: {earnings.isoformat()} (in {days} days)."
[docs]
def build_messages(
symbol: str,
bar_summary: dict[str, object],
news: Sequence[NewsItem],
position: Position | None,
*,
now: datetime,
max_age_hours: float,
withheld: int = 0,
analyst: AnalystOpinion | None = None,
technicals: dict[str, object] | None = None,
earnings: date | None = None,
) -> tuple[str, str]:
"""Return `(system, user)` messages for one decision.
`news` is expected to be **already filtered** to the recency window by
`partition_by_age`; `withheld` is how many items that filter removed.
Filtering is the caller's job because the caller also has to record the
stale items in `inputs_json` — yfinance keeps no archive, so the
decision-time snapshot is the only copy that will ever exist.
`analyst` being `None` means "no coverage available" — it deliberately
does not distinguish a symbol with no analysts (most ETFs) from a fetch
that failed, the same non-distinction `AnalystProvider.get_opinion`
already makes at the source. The caller records which one happened in
`inputs_json`; the model is only ever told whether it has the data.
`technicals` is `build_technical_summary`'s dict, or `None`/`{}` when the
caller could not compute it (issue #63) — both render "not available",
the same non-distinction `analyst` already makes, for the same reason:
the model is only ever told whether it has the data, never why it does
not.
`earnings` (issue #89) is the next known earnings date, or `None` — the
same non-distinction again: "no upcoming earnings scheduled" (an ETF)
and "the fetch failed" both render "not available.", and the caller
records which one happened via `earnings_error` in `inputs_json`.
**Context only — never gates or blocks an entry in v1**, matching how
analyst consensus shipped in "5"/"6" before any gate was built on it.
"""
lines = [
f"Current date and time: {now.strftime('%Y-%m-%d %H:%M UTC')}",
f"Symbol: {symbol}",
]
if position is None:
lines.append("Position currently open: no")
else:
lines += _position_lines(position)
lines += [
"",
"Recent daily closes (oldest to newest): "
+ ", ".join(str(c) for c in bar_summary.get("recent_closes") or []),
f"Latest close: {bar_summary.get('last_close')}",
f"Window high: {bar_summary.get('window_high')}",
f"Window low: {bar_summary.get('window_low')}",
"",
]
if analyst is None:
lines.append("Analyst coverage: none available.")
else:
lines.append(
"Analyst ratings: "
f"{analyst.strong_buy} strong buy, {analyst.buy} buy, "
f"{analyst.hold} hold, {analyst.sell} sell, "
f"{analyst.strong_sell} strong sell."
)
mean_line = _recommendation_mean_line(analyst)
if mean_line is not None:
lines.append(mean_line)
target_line = _target_price_line(analyst, bar_summary.get("last_close"))
if target_line is not None:
lines.append(target_line)
lines.append(_earnings_line(earnings, now))
lines.append("")
lines += _technical_lines(technicals)
lines.append("")
if news:
# The caveat is load-bearing. Measured on 2026-07-31, only 6 of 10 AAPL
# items and 1 of 10 SPY items actually named the symbol; the rest were
# general market commentary. Presenting them as company news is how a
# fluent model is led into narrating noise.
lines.append(
"Recent headlines. These are the latest items the news feed "
"returned for this symbol, and they may or may not actually "
"concern it — weigh them accordingly:"
)
for item in news:
when = item.published_at.strftime("%Y-%m-%d %H:%M UTC")
age = _age_phrase(now, item.published_at)
lines.append(f"- [{age} | {when}] {item.title} ({item.publisher})")
if item.summary:
lines.append(f" {item.summary}")
if withheld:
lines.append(f"({withheld} older items were withheld as stale.)")
else:
# Not "no recent news was available": that was false for ANNX, where
# ten items existed and every one was three weeks old.
empty = f"No news published in the last {max_age_hours:g} hours."
if withheld:
empty += f" ({withheld} older items were withheld as stale.)"
lines.append(empty)
lines += ["", "Decide: buy, sell, or hold."]
return _SYSTEM, "\n".join(lines)
[docs]
def prompt_hash(system: str, user: str) -> str:
"""A stable digest of the exact prompt sent.
Stored so a later rebuild from `inputs_json` can be proven identical to
what was actually sent — which is what catches a template edited since.
"""
digest = hashlib.sha256()
digest.update(system.encode())
digest.update(b"\x00")
digest.update(user.encode())
return digest.hexdigest()
[docs]
def parse_decision(payload: dict[str, object]) -> tuple[Action, float, str]:
"""Turn a model response into `(action, confidence, reason)`.
Never raises. Every failure mode resolves to HOLD with a reason saying
what was wrong, because this is the boundary between an unreliable model
and code that can spend money.
"""
raw_action = payload.get("action")
reason = str(payload.get("reason") or "").strip()
confidence = 0.0
raw_confidence = payload.get("confidence")
if isinstance(raw_confidence, (int, float)) and not isinstance(raw_confidence, bool):
# `json.loads` accepts the bare `NaN`/`Infinity` literals by default,
# and `OllamaProvider.generate_json` passes them through unmodified.
# Without this guard, `max(0.0, min(1.0, float('nan')))` evaluates to
# 1.0 — garbage input silently becoming *maximum* confidence, which is
# the one direction that matters here. Same hazard as the finiteness
# guard in `brokers/alpaca.py`'s `_to_decimal`.
candidate = float(raw_confidence)
if math.isfinite(candidate):
confidence = max(0.0, min(1.0, candidate))
normalized = str(raw_action).strip().lower() if raw_action is not None else ""
try:
action = Action(normalized)
except ValueError:
return (
Action.HOLD,
confidence,
f"Model returned an unusable action {raw_action!r}; holding. "
f"Model said: {reason or '(no reason given)'}",
)
return action, confidence, reason or f"Model chose {action.value} with no reason."