Source code for trader.replay.bars_at

"""The bars a prompt may see at a past instant.

The subtlety that makes this its own module: a `1d` bar is stamped at
exchange midnight, so the decision day's own bar is timestamped 04:00Z —
*before* a 14:00Z decision instant — while its close had not yet happened. A
plain `timestamp < as_of` filter therefore leaks one day of the future,
silently, on every row. The bound is the last session strictly before the
decision day.
"""

from collections.abc import Sequence
from datetime import UTC, datetime, timedelta
from typing import Protocol

from trader.domain import Bar
from trader.errors import ReplayError
from trader.replay.sessions import trading_date

__all__ = ["bars_before"]

#: How far back to ask the repository for. Wide enough that `lookback` is
#: always the binding constraint — 30 sessions is six calendar weeks, and 400
#: calendar days covers that with room for holidays and a thin symbol's gaps.
_LOAD_DAYS = 400


class _BarRepo(Protocol):
    def load_bars(
        self, symbol: str, interval: str, start: datetime, end: datetime
    ) -> list[Bar]: ...




def bars_before(
    repository: _BarRepo,
    symbol: str,
    as_of: datetime,
    lookback: int,
    interval: str = "1d",
) -> list[Bar]:
    """The last `lookback` bars closing strictly before `as_of`'s trading day.

    Oldest first, because `build_bar_summary` slices the tail and renders in
    that order.

    Raises:
        ReplayError: the repository returned a bar on or after `as_of`'s
            trading day. Asserted rather than filtered: a leak here is the
            scoring rule's "silently invent skill" case, and the run must
            stop.
    """
    if as_of.tzinfo is None:
        raise ValueError(f"as_of must be tz-aware UTC, got {as_of!r}")
    decision_day = trading_date(as_of)
    day_start = datetime(
        decision_day.year, decision_day.month, decision_day.day, tzinfo=UTC
    )
    rows: Sequence[Bar] = repository.load_bars(
        symbol,
        interval,
        as_of - timedelta(days=_LOAD_DAYS),
        day_start - timedelta(seconds=1),
    )
    for candidate in rows:
        if trading_date(candidate.timestamp) >= decision_day:
            raise ReplayError(
                f"{symbol}: bar timestamped at or after the decision day "
                f"({candidate.timestamp.isoformat()}, trading day "
                f"{trading_date(candidate.timestamp)} >= {decision_day})"
            )
    ordered = sorted(rows, key=lambda b: b.timestamp)
    return ordered[-lookback:] if lookback > 0 else ordered