"""A trailing stop with a hard floor, plus LLM-bounded discretion over the trail.
requirements.md 6.1's "Trailing stop with floor" and 6.3's companion-markdown
config format. Deliberately not the same mechanism as the protective-stop
ratchet in `trader/execution/` (§8, the OTO/atomic-protection path): that one
protects a fill this app already made and runs unconditionally on every held
position, regardless of which strategy owns it. This is a *signal generator*
— one more `Strategy` competing for the SELL decision on a symbol, exactly
like Turtle or RSI. It never places an order and never touches the broker; it
only ever returns a `Signal`, and the same `sold_this_cycle` guard, exposure
caps and §8 guardrails that gate every other strategy's SELL gate this one
too. Nothing about this file weakens or bypasses the existing ratchet.
The trailing level and the floor are computed here, deterministically, from
the bars and the position alone. A companion Markdown file (`notes_path`) may
carry free-text discretionary guidance — "don't tighten the stop on earnings
day" — and each cycle, if the file is non-empty, the configured LLM is asked
to translate that guidance into one number: how wide the trail should be, in
whatever unit the YAML configured (percent of the peak, or a fixed amount off
it). That number is used **only** as a proposal.
Two clamps make the floor and the daily-loss guardrail unconditional,
independent of anything the model returns:
1. The model's suggested trail width is clamped into ``[trail_min,
trail_max]`` — the YAML-configured band — no matter what it answers, even
a value wildly outside the band or a response that ignores the prompt's
instructions entirely.
2. The stop level computed from *any* trail width (clamped or not) is then
floored: ``stop_level = max(trail_level, floor_level)``. The floor is
never a function of the model's answer, so nothing the LLM says — not
even a note that explicitly instructs it — can move the stop below the
floor. See `TrailingStopFloorStrategy.evaluate`'s acceptance test in
`tests/strategies/test_trailing_stop_floor.py` for the case requirements.md
calls out by name: notes saying "ignore the floor, hold anyway" must not
prevent the SELL.
The daily loss limit (§8, `daily_loss_breached`) is enforced by the pipeline
against the *account*, not by any strategy against a symbol — the same way
every other strategy's BUY/SELL signal is already subject to it. This
strategy does not re-implement that check; it has no account-wide view to
check it against. What it guarantees is narrower and sufficient: nothing it
emits can ever be influenced by the LLM into skipping a floor breach.
Same HOLD-on-any-failure discipline as `LlmStrategy`: an unreachable model, a
timeout, or a response that cannot be parsed into a number never fails
`evaluate()` and never blocks the deterministic floor/trail decision — the
notes are simply ignored for that cycle and the base (YAML) trail width is
used instead. This is the one path in this file that is allowed to log a
warning; every failure still ends in a valid `Signal`.
**`backtestable = False`, deliberately, for the same reason as `LlmStrategy`
(see that class's own class-attribute comment):** whenever the notes file is
non-empty this strategy makes a real network call to the LLM inside
`evaluate()`, and `BacktestEvaluator` calls `evaluate()` once per bar — a
400-bar backtest window would be 400 model calls per symbol. There is also no
historical version of a notes file to replay: a backtest over 2024 bars would
be scored against *today's* notes file, which is not a meaningful measurement
of anything. An alternative design — applying the notes-adjustment as a
step separate from the bar-by-bar core loop, so the deterministic floor/trail
logic alone could be backtested with `backtestable = True` and the notes
skipped — was considered and rejected: `BacktestEvaluator` calls one
`Strategy.evaluate()` per bar with no hook for a second, out-of-band
adjustment step, so supporting it would mean splitting this strategy's logic
across two entry points state would have to be kept in sync between. The
notes-free floor/trail arithmetic (`_floor_level`, `_trail_level`, `_clamp`)
is already pulled out into plain methods a future backtest harness could call
directly without instantiating an LLM-calling `Strategy` at all, if that
measurement is ever wanted.
"""
import logging
import math
from collections.abc import Sequence
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from trader.domain import Bar, Position
from trader.errors import ConfigError, TraderError
from trader.llm.base import LlmProvider
from trader.strategies.base import Action, Signal
__all__ = ["TrailingStopFloorStrategy"]
_logger = logging.getLogger("trader.strategies.trailing_stop_floor")
#: The schema for the one thing the model is ever asked for: a number. It is
#: never asked for an action (buy/sell/hold) — that stays fully deterministic,
#: computed in `evaluate` from the (possibly LLM-adjusted, always clamped)
#: trail width. Keeping the model out of the action entirely is what makes the
#: floor unconditional: there is no "the model said hold" branch to bypass it.
TRAIL_ADJUST_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {
"trail_value": {"type": "number"},
"reason": {"type": "string"},
},
"required": ["trail_value", "reason"],
}
_ZERO = Decimal(0)
_HUNDRED = Decimal(100)
def _to_decimal_param(
strategy_id: str, name: str, value: object, *, minimum: Decimal | None = None
) -> Decimal:
"""A configured YAML number as `Decimal`, never via `float()`.
YAML parses a bare decimal literal (`8.5`) as a Python `float` before this
code ever sees it — that step is outside this app's control, same as
every other percentage in `PipelineConfig`. What this function guarantees
is that *this* code never calls `float()` on it: `Decimal(str(value))` is
the same conversion `LlmStrategy.__init__` already uses for
`reuse_price_band_pct`, and it is exact where `Decimal(value)` on a float
would carry that float's binary-fraction error into a number that gates
money.
`bool` is rejected explicitly because it is an `int` subclass — a YAML
`true` would otherwise silently become `Decimal(1)`.
"""
if isinstance(value, bool):
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r} (bool)."
)
if isinstance(value, Decimal):
candidate = value
elif isinstance(value, (int, float, str)):
try:
candidate = Decimal(str(value))
except InvalidOperation:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
) from None
else:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
)
# `Decimal("NaN")` and `Decimal("Infinity")` both parse without error and
# then make every comparison downstream either always-False or
# always-True — the same trap `_finite_number` in `llm_strategy.py` guards
# against for `max_news_age_hours`, and YAML's `.nan`/`.inf` reach here as
# exactly those strings via `Decimal(str(float("nan")))`.
if not candidate.is_finite():
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a finite number, got {value!r}."
)
if minimum is not None and candidate < minimum:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be at least {minimum}, got {value!r}."
)
return candidate
def _clamp(value: Decimal, lo: Decimal, hi: Decimal) -> Decimal:
"""`value` restricted to `[lo, hi]`. `lo <= hi` is an invariant `__init__`
already enforces, so this never has to guess which bound is which."""
return max(lo, min(hi, value))
def _parse_trail_value(payload: dict[str, object]) -> Decimal | None:
"""The model's suggested trail width, or `None` if it cannot be trusted.
Mirrors `parse_decision`'s discipline in `llm/prompt.py`: a bare `<=`
check would let `float("nan")` through (`nan <= anything` is always
`False`), and `json.loads` accepts the bare `NaN`/`Infinity` literals that
a misbehaving model can emit. Never raises — every failure here becomes
`None`, and the caller's response is to fall back to the base trail
width, not to fail the cycle.
"""
raw = payload.get("trail_value")
if isinstance(raw, bool) or not isinstance(raw, (int, float)):
return None
if not math.isfinite(raw):
return None
try:
return Decimal(str(raw))
except InvalidOperation:
return None
@dataclass(frozen=True, slots=True)
class _TrailAdjustment:
"""What the notes-adjustment step decided this cycle, for `last_inputs`."""
value: Decimal
llm_reason: str | None
clamped: bool
[docs]
class TrailingStopFloorStrategy:
"""A trailing-stop exit, floor-protected, with bounded LLM discretion.
Exit-only: this strategy never originates a BUY. Its "position reference"
(requirements 6.1) is `tickers` — the symbol(s) it manages — plus the fact
that it only ever acts on a symbol it is already holding. A symbol it
manages but does not currently hold, and a symbol it does not manage at
all, both resolve to HOLD, for different stated reasons (see `evaluate`).
"""
#: See the module docstring for the full reasoning.
backtestable = False
def __init__(
self,
id: str, # noqa: A002 - matches the Strategy Protocol
llm: LlmProvider,
# Defaults to `()`, not a required positional, so a config that omits
# it reaches the `not tickers` check below and raises `ConfigError`
# naming the field — rather than the registry's
# `TrailingStopFloorStrategy(id=..., llm=llm, **kwargs)` raising a bare
# `TypeError: missing 1 required positional argument` with no strategy
# id and no field name in it.
tickers: Sequence[str] = (),
trail_pct: object | None = None,
trail_amount: object | None = None,
trail_min: object | None = None,
trail_max: object | None = None,
floor_price: object | None = None,
floor_pct: object | None = None,
notes_path: str | None = None,
lookback_bars: int = 60,
) -> None:
if not tickers:
raise ConfigError(f"Strategy {id!r}: tickers must be a non-empty list.")
normalized_tickers = tuple(str(t).strip().upper() for t in tickers)
if any(not t for t in normalized_tickers):
raise ConfigError(f"Strategy {id!r}: tickers must not contain blank entries.")
# --- Trail: exactly one of a percentage-of-peak or a fixed amount. ---
if (trail_pct is None) == (trail_amount is None):
raise ConfigError(
f"Strategy {id!r}: specify exactly one of trail_pct or trail_amount."
)
if trail_pct is not None:
self._trail_mode = "pct"
self._trail_unit = "%"
base_trail = _to_decimal_param(id, "trail_pct", trail_pct, minimum=_ZERO)
if base_trail <= 0:
raise ConfigError(
f"Strategy {id!r}: trail_pct must be > 0, got {trail_pct!r}."
)
else:
self._trail_mode = "amount"
self._trail_unit = ""
base_trail = _to_decimal_param(
id, "trail_amount", trail_amount, minimum=_ZERO
)
if base_trail <= 0:
raise ConfigError(
f"Strategy {id!r}: trail_amount must be > 0, got {trail_amount!r}."
)
self._trail_value = base_trail
# The band the LLM's answer is clamped into. Defaulting each bound to
# the base trail value (rather than, say, 0 and infinity) means an
# operator who configures no band at all gets zero discretion by
# default — granting the model any room to move is an opt-in, not the
# unedited-config behaviour. Same "safe unless configured otherwise"
# shape as `mode: shadow`.
min_value = (
_to_decimal_param(id, "trail_min", trail_min, minimum=_ZERO)
if trail_min is not None
else base_trail
)
max_value = (
_to_decimal_param(id, "trail_max", trail_max, minimum=_ZERO)
if trail_max is not None
else base_trail
)
if min_value > max_value:
raise ConfigError(
f"Strategy {id!r}: trail_min ({min_value}) must be <= "
f"trail_max ({max_value})."
)
if not (min_value <= base_trail <= max_value):
raise ConfigError(
f"Strategy {id!r}: the base trail ({base_trail}) must lie "
f"within [trail_min, trail_max] = [{min_value}, {max_value}]."
)
self._trail_min = min_value
self._trail_max = max_value
# --- Floor: exactly one of an absolute price or a percent of entry. ---
if (floor_price is None) == (floor_pct is None):
raise ConfigError(
f"Strategy {id!r}: specify exactly one of floor_price or floor_pct."
)
if floor_price is not None:
price = _to_decimal_param(id, "floor_price", floor_price, minimum=_ZERO)
if price <= 0:
raise ConfigError(
f"Strategy {id!r}: floor_price must be > 0, got {floor_price!r}."
)
self._floor_price: Decimal | None = price
self._floor_pct: Decimal | None = None
else:
pct = _to_decimal_param(id, "floor_pct", floor_pct, minimum=_ZERO)
if not (_ZERO <= pct < _HUNDRED):
raise ConfigError(
f"Strategy {id!r}: floor_pct must be within [0, 100), "
f"got {floor_pct!r}."
)
self._floor_price = None
self._floor_pct = pct
if lookback_bars < 1:
raise ConfigError(
f"Strategy {id!r}: lookback_bars must be >= 1, got {lookback_bars!r}."
)
self.id = id
self._llm = llm
self._tickers = normalized_tickers
self._notes_path = notes_path
self._lookback_bars = lookback_bars
# Mutable per-instance state the pipeline reads after `evaluate()`,
# same contract as `LlmStrategy.last_inputs`: what the decision was
# based on, for `decisions.inputs_json`.
self.last_inputs: dict[str, object] = {}
[docs]
def warmup_bars(self) -> int:
return self._lookback_bars
# ------------------------------------------------------------------
# Deterministic core. No LLM, no I/O — a future backtest harness (see the
# module docstring's `backtestable` reasoning) can call these directly.
# ------------------------------------------------------------------
def _floor_level(self, position: Position) -> Decimal:
"""The hard floor for this position. Never a function of the LLM."""
if self._floor_price is not None:
return self._floor_price
floor_pct = self._floor_pct
if floor_pct is None:
# Unreachable: `__init__` requires exactly one of `floor_price`/
# `floor_pct`. Raised rather than asserted so this stays true even
# if the process runs with `python -O`, which strips `assert`.
raise ConfigError(f"Strategy {self.id!r}: no floor is configured.")
return position.avg_entry_price * (1 - floor_pct / _HUNDRED)
def _trail_level(self, peak: Decimal, effective_trail: Decimal) -> Decimal:
"""Where the trail alone would place the stop, given a trail width
already clamped into `[trail_min, trail_max]`."""
if self._trail_mode == "amount":
return peak - effective_trail
return peak * (1 - effective_trail / _HUNDRED)
# ------------------------------------------------------------------
[docs]
def evaluate(self, bars: Sequence[Bar], position: Position | None) -> Signal:
if not bars:
self.last_inputs = {"error": "no bars available"}
return Signal(Action.HOLD, "No bars available to evaluate.")
symbol = bars[-1].symbol
self.last_inputs = {
"symbol": symbol,
"managed_tickers": list(self._tickers),
"trail_mode": self._trail_mode,
"base_trail_value": str(self._trail_value),
"trail_min": str(self._trail_min),
"trail_max": str(self._trail_max),
}
if symbol not in self._tickers:
reason = (
f"{symbol} is not among the tickers this instance manages "
f"({', '.join(self._tickers)})."
)
self.last_inputs["skipped"] = "symbol not managed"
return Signal(Action.HOLD, reason)
if position is None:
self.last_inputs["position_open"] = False
return Signal(
Action.HOLD,
f"No open position in {symbol}; this strategy only manages an "
"existing holding's exit and never originates an entry.",
)
self.last_inputs["position_open"] = True
self.last_inputs["position"] = {
"quantity": str(position.quantity),
"avg_entry_price": str(position.avg_entry_price),
"current_price": str(position.current_price),
}
window = (
list(bars[-self._lookback_bars :]) if self._lookback_bars > 0 else list(bars)
)
peak = max(b.high for b in window)
close = bars[-1].close
self.last_inputs["peak"] = str(peak)
self.last_inputs["close"] = str(close)
floor_level = self._floor_level(position)
self.last_inputs["floor_level"] = str(floor_level)
adjustment = self._adjust_trail(symbol, position, floor_level, peak, close)
effective = adjustment.value
self.last_inputs["effective_trail_value"] = str(effective)
if adjustment.clamped:
self.last_inputs["trail_clamped"] = True
if adjustment.llm_reason is not None:
self.last_inputs["llm_reason"] = adjustment.llm_reason
trail_level = self._trail_level(peak, effective)
# The floor is a hard lower bound on the stop, applied AFTER any LLM
# influence and independent of it: nothing above this line can ever
# reach this comparison already knowing what the LLM said and
# deciding to skip it. `max` — never `min` — because the app is
# long-only and the stop must never sit below the floor no matter how
# wide the trail computes.
stop_level = max(trail_level, floor_level)
self.last_inputs["trail_level"] = str(trail_level)
self.last_inputs["stop_level"] = str(stop_level)
indicators: dict[str, Decimal | float] = {
"close": close,
"peak": peak,
"floor_level": floor_level,
"trail_level": trail_level,
"stop_level": stop_level,
"effective_trail_value": effective,
}
unit = self._trail_unit
detail = (
f"stop {stop_level} (floor {floor_level}, trail {effective}{unit} "
f"off peak {peak})"
)
if close <= stop_level:
return Signal(
Action.SELL, f"{symbol} close {close} at/below {detail}.", indicators
)
return Signal(Action.HOLD, f"{symbol} close {close} above {detail}.", indicators)
def _adjust_trail(
self,
symbol: str,
position: Position,
floor_level: Decimal,
peak: Decimal,
close: Decimal,
) -> _TrailAdjustment:
"""The trail width to use this cycle: the base, or the model's
notes-informed suggestion — always clamped into the configured band.
**Never raises, and every failure falls back to the base (YAML) trail
width**, exactly the discipline `LlmStrategy.evaluate` documents for
its own model call: an empty/unreadable notes file, an unreachable
model, a timeout, or a response that cannot be parsed into a number
all resolve to "ignore the notes this cycle," never to an exception
that would abort `evaluate()` before the floor comparison runs.
"""
notes = self._read_notes()
self.last_inputs["notes_present"] = bool(notes)
if not notes:
self.last_inputs["skipped_model_call"] = "no notes"
return _TrailAdjustment(self._trail_value, None, clamped=False)
try:
system, user = self._build_messages(
symbol, position, floor_level, peak, close, notes
)
self.last_inputs["prompt"] = user
payload = self._llm.generate_json(system, user, TRAIL_ADJUST_SCHEMA)
self.last_inputs["raw_response"] = payload
suggested = _parse_trail_value(payload)
except TraderError as exc:
_logger.warning(
"%s: model call failed for %s, using base trail: %s", self.id, symbol, exc
)
self.last_inputs["error"] = str(exc)
return _TrailAdjustment(self._trail_value, None, clamped=False)
except Exception as exc: # noqa: BLE001 - belt-and-braces, see docstring
# Same shape as `LlmStrategy`'s own broad catches: an
# unanticipated wrong-typed response (`AttributeError`/
# `KeyError`/`TypeError`) must not escape `evaluate()` and abort
# the symbol loop before the ratchet or reconciliation run for
# every OTHER symbol this cycle.
_logger.warning(
"%s: notes adjustment failed for %s, using base trail: %s",
self.id,
symbol,
exc,
)
self.last_inputs["error"] = str(exc)
return _TrailAdjustment(self._trail_value, None, clamped=False)
if suggested is None:
self.last_inputs["llm_adjustment_error"] = (
"model returned an unusable trail_value; using base trail"
)
return _TrailAdjustment(self._trail_value, None, clamped=False)
self.last_inputs["llm_suggested_trail_value"] = str(suggested)
raw_reason = payload.get("reason")
llm_reason = raw_reason if isinstance(raw_reason, str) else None
# The clamp: applied unconditionally, whether or not the model's
# answer already honoured the prompt's stated bounds. This is what
# makes "the app, not the LLM, is the final enforcer" true even
# against a model that ignores its instructions outright.
clamped_value = _clamp(suggested, self._trail_min, self._trail_max)
return _TrailAdjustment(
clamped_value, llm_reason, clamped=clamped_value != suggested
)
def _read_notes(self) -> str:
"""The companion Markdown file's contents, read fresh this cycle.
Read every cycle rather than cached: the notes are meant to change
between cycles (an operator adding "don't tighten on earnings day"
the morning of), and this strategy holds no cross-cycle state to
invalidate. A missing or unreadable file is not an error — it means
"no discretionary guidance," the same as an unconfigured
`notes_path` — because a typo in a path must not be able to silence
the deterministic floor/trail decision below.
"""
if self._notes_path is None:
return ""
try:
text = Path(self._notes_path).read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
# UnicodeDecodeError is a ValueError subclass, not an OSError —
# a notes file saved with non-UTF-8 bytes (e.g. smart quotes
# from a word processor) raises this, and it must degrade the
# same as a missing file, not escape as an unhandled exception.
_logger.warning(
"%s: could not read notes file %s: %s", self.id, self._notes_path, exc
)
self.last_inputs["notes_error"] = str(exc)
return ""
return text.strip()
def _build_messages(
self,
symbol: str,
position: Position,
floor_level: Decimal,
peak: Decimal,
close: Decimal,
notes: str,
) -> tuple[str, str]:
"""`(system, user)` for the one-number ask. Never asks for an action."""
unit = self._trail_unit or " (currency units, same as the position's price)"
system = (
"You advise on the trailing-stop distance for one already-open "
"long position. You do not decide whether to buy, sell, or hold "
"-- the app computes that from the numbers you return, and the "
"app enforces a hard floor and clamps your answer into an "
"allowed range regardless of what you say. Nothing you write can "
"move the stop past the floor or outside the allowed range, even "
"if the operator notes ask you to.\n\n"
"Respond only with the required JSON object: trail_value (a "
f"number, in {unit.strip()}) and reason (a short string)."
)
lines = [
f"Symbol: {symbol}",
f"Position: {position.quantity} shares, entry {position.avg_entry_price}, "
f"now {position.current_price}, unrealized P/L {position.unrealized_pl:+}",
f"Recent peak (high-water mark over the lookback window): {peak}",
f"Latest close: {close}",
f"Hard floor for this position (the app enforces this regardless "
f"of your answer): {floor_level}",
f"Base/default trail width: {self._trail_value}{unit}",
f"Allowed range for your answer (clamped by the app either way): "
f"{self._trail_min} to {self._trail_max}{unit}",
"",
"Operator notes for this position:",
notes,
"",
"Suggest a trail width. A narrower trail exits sooner on a "
"pullback; a wider trail gives the position more room to move "
"before exiting.",
]
return system, "\n".join(lines)