Source code for trader.marketdata.cache

"""Bar cache: fetch a window once, reuse it thereafter.

Backtests re-read the same history repeatedly. Fetching every time is slow,
rate-limited, and — because Yahoo revises history — makes results drift for
reasons unrelated to the code under test.
"""

import logging
from datetime import datetime, timedelta

from trader.domain import Bar
from trader.errors import MarketDataError
from trader.marketdata.base import MarketDataProvider
from trader.persistence.bars import BarRepository

__all__ = ["BarCache"]

_logger = logging.getLogger("trader.marketdata")


[docs] class BarCache: """Ensures a date window is present locally, fetching only what is missing."""
[docs] def __init__( self, repository: BarRepository, provider: MarketDataProvider, refresh_tail: timedelta | None = None, ) -> None: """ Args: refresh_tail: how far back from a requested window's end to re-fetch on every call. `None` keeps the cache strictly fetch-once, which is what backtests want. The trading path sets it, because the current day's bar is written partial — its high, low and close are wherever price stood at the first fetch of the day — and `save_bars` skips timestamps it already holds, so nothing would ever correct it to the real close. The policy lives here rather than at each call site so a second caller cannot forget it. """ self._repository = repository self._provider = provider self._refresh_tail = refresh_tail # (symbol, interval) -> the `end` an automatic tail refresh last ran # for (issue #101). `run_once.py`'s symbol/strategy loop is `for # symbol: for strategy:`, so every strategy calls `ensure` for the # same symbol with the identical `decided_at` as `end` -- confirmed # live, all decisions in one cycle share one `decided_at` down to the # microsecond. A later cycle's different `end` naturally busts this; # no explicit "new cycle" signal is needed. Never applied to # `refresh=True`, the explicit operator-invoked path. self._tail_refreshed_at: dict[tuple[str, str], datetime] = {}
[docs] def ensure( self, symbol: str, start: datetime, end: datetime, interval: str = "1d", refresh: bool = False, ) -> list[Bar]: """Return bars for the window, fetching any span not already covered.""" if refresh: self._replace(symbol, interval, start, end) else: # `max` so a tail wider than the window cannot fetch bars the # caller never asked for. tail_start = ( max(start, end - self._refresh_tail) if self._refresh_tail is not None else None ) tail_covered_by_fill = False for span_start, span_end in self._repository.missing_ranges( symbol, interval, start, end ): self._fill(symbol, interval, span_start, span_end) # If the gap just fetched already spans the whole tail, that # data is fresh from this call and a second, immediate # refetch of the same span would be pure waste — the whole # window is missing on the first `ensure` of any symbol, so # this is the common case, not a rare one. if ( tail_start is not None and span_start.date() <= tail_start.date() and span_end.date() >= end.date() ): tail_covered_by_fill = True if tail_start is not None and not tail_covered_by_fill: key = (symbol, interval) if self._tail_refreshed_at.get(key) != end: self._refresh_tail_or_degrade(symbol, interval, tail_start, end) self._tail_refreshed_at[key] = end return self._repository.load_bars(symbol, interval, start, end)
def _refresh_tail_or_degrade( self, symbol: str, interval: str, start: datetime, end: datetime ) -> None: """Best-effort tail refresh: automatic, so a provider outage here must degrade to the cached bars rather than block evaluation. Nobody asked for fresh data on this specific call — `refresh_tail` did, on every call, as a matter of policy. `ensure(..., refresh=True)` is the operator-invoked path and does not come through here: there, someone did ask, so a failure must keep propagating. """ try: self._replace(symbol, interval, start, end) except MarketDataError as exc: _logger.warning( "%s: automatic tail refresh of %s..%s at %s failed (%s); " "falling back to the %d bars already cached", symbol, start.date(), end.date(), interval, exc, len(self._repository.load_bars(symbol, interval, start, end)), ) def _fill(self, symbol: str, interval: str, start: datetime, end: datetime) -> None: """Fetch a span never fetched before, and record it as covered. Coverage is recorded even when the fetch returns nothing, which is what stops a window containing no trading days being re-requested forever. """ fetched = self._provider.get_history(symbol, start.date(), end.date(), interval) self._repository.save_bars(symbol, interval, fetched) self._repository.record_coverage(symbol, interval, start, end) def _replace( self, symbol: str, interval: str, start: datetime, end: datetime ) -> None: """Re-fetch a span already held, replacing it only if the fetch succeeded. Fetch before delete, deliberately. The original order deleted first, so a provider outage between the two turned a transient network failure into permanent loss of the series that gates live entries — and an empty result is not evidence the history is gone. """ fetched = self._provider.get_history(symbol, start.date(), end.date(), interval) if not fetched: _logger.warning( "%s: refresh of %s..%s at %s returned no bars; keeping the %d " "cached bars rather than deleting them", symbol, start.date(), end.date(), interval, len(self._repository.load_bars(symbol, interval, start, end)), ) return self._repository.delete_bars(symbol, interval, start, end) self._repository.save_bars(symbol, interval, fetched) self._repository.record_coverage(symbol, interval, start, end)