"""Forward returns and max adverse excursion, from one fetch per symbol.
**Structurally isolated from the prompt path.** This module must import
nothing that builds a model prompt and nothing that loads point-in-time news
or bars for one — enforced by a source-text check in this package's test
suite — because the scoring rule requires that nothing computed here can leak
into a prompt. Stored bars are dividend-adjusted and adjusted history is
retroactively mutable (CLAUDE.md), so both endpoints of every horizon come
from a *single* fetch: mixing a close written months ago with one written
today would put an invisible step at the join.
"""
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Protocol
from trader.domain import Bar
from trader.replay.sessions import session_grid, session_offset, trading_date
__all__ = [
"HORIZONS",
"ForwardPrices",
"fetch_forward_prices",
"forward_return",
"max_adverse_excursion",
]
#: The pre-registered horizons. N=5 is the headline; the rest are reported
#: alongside on every run so N cannot be swapped after seeing results.
HORIZONS: tuple[int, ...] = (1, 5, 10, 21)
class _Provider(Protocol):
def get_history(
self, symbol: str, start: date, end: date, interval: str = "1d"
) -> list[Bar]: ...
[docs]
class ForwardPrices:
"""One symbol's closes and lows, from one fetch, for scoring at any horizon."""
__slots__ = ("closes", "fetched_at", "grid", "lows", "symbol")
def __init__(
self,
symbol: str,
fetched_at: datetime,
closes: dict[date, Decimal],
lows: dict[date, Decimal],
grid: list[date],
) -> None:
self.symbol = symbol
self.fetched_at = fetched_at
self.closes = closes
self.lows = lows
self.grid = grid
[docs]
def fetch_forward_prices(
provider: _Provider, symbol: str, start: date, end: date
) -> ForwardPrices:
"""Fetch `symbol`'s bars once and index them for every horizon.
One call, one adjustment basis. `fetched_at` is recorded so a run's
provenance shows exactly when this basis was pinned.
"""
bars = provider.get_history(symbol, start, end, interval="1d")
closes: dict[date, Decimal] = {}
lows: dict[date, Decimal] = {}
for b in bars:
day = trading_date(b.timestamp)
closes[day] = b.close
lows[day] = b.low
return ForwardPrices(
symbol=symbol,
fetched_at=datetime.now(UTC),
closes=closes,
lows=lows,
grid=session_grid(bars),
)
[docs]
def forward_return(prices: ForwardPrices, entry_day: date, n: int) -> Decimal | None:
"""`close(exit) / close(entry) - 1`, off the fetched session grid.
Returns `None` when the horizon runs past the fetched data — never a
truncated return, which would silently understate a long horizon.
Raises:
ValueError: `entry_day` is not a session on the fetched grid.
"""
exit_day = session_offset(prices.grid, entry_day, n)
if exit_day is None:
return None
return prices.closes[exit_day] / prices.closes[entry_day] - 1
[docs]
def max_adverse_excursion(
prices: ForwardPrices, entry_day: date, n: int
) -> Decimal | None:
"""The worst intraday dip after entry, entry day excluded, exit day included.
`min(low) / close(entry) - 1` over the sessions strictly after `entry_day`
up to and including the exit day. The entry day's own low must not count:
it happened before the decision, not after it.
"""
exit_day = session_offset(prices.grid, entry_day, n)
if exit_day is None:
return None
entry_index = prices.grid.index(entry_day)
exit_index = prices.grid.index(exit_day)
window = prices.grid[entry_index + 1 : exit_index + 1]
if not window:
return None
worst_low = min(prices.lows[day] for day in window)
return worst_low / prices.closes[entry_day] - 1