"""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)