"""News archive: fetch a span of history once, replay from it thereafter.
`BarCache`'s shape, for the same reason — one fetch, then a local read — but with
the opposite failure policy on refresh. There is no tail to re-fetch and no
`refresh` flag at all: an archived headline is evidence of what was published,
and re-reading it later would either change nothing or quietly substitute a
correction for the wording a stored decision was actually made on.
"""
import logging
from datetime import datetime
from typing import Protocol, runtime_checkable
from trader.news.base import ArchivedNewsItem, NewsItem
from trader.persistence.news_archive import NewsArchiveRepository
__all__ = ["NewsArchive", "NewsArchiveProvider"]
_logger = logging.getLogger("trader.news")
[docs]
@runtime_checkable
class NewsArchiveProvider(Protocol):
"""A news source that can answer for a *past* window.
Separate from `NewsProvider` because almost nothing can do this: yfinance
offers roughly the last ten items per ticker and no history at all, which is
the fact that made an archive table worth building in the first place.
"""
[docs]
def get_news_between(
self, symbol: str, start: datetime, end: datetime
) -> list[ArchivedNewsItem]:
"""Every archived item published in `[start, end)`."""
...
[docs]
class NewsArchive:
"""Ensures a time span is present locally, fetching only what is missing."""
def __init__(
self, repository: NewsArchiveRepository, provider: NewsArchiveProvider
) -> None:
self._repository = repository
self._provider = provider
[docs]
def ensure(self, symbol: str, start: datetime, end: datetime) -> list[NewsItem]:
"""Return archived news for `[start, end)`, fetching any uncovered span.
Raises:
MarketDataError: a fetch failed. Deliberately propagated: coverage is
recorded only after a fetch returns, so a failure leaves the span
missing and the next run retries it. Degrading quietly here would
write a permanent hole that reads as "this symbol had no news",
which is the one thing coverage exists to make impossible.
"""
for span_start, span_end in self._repository.missing_ranges(symbol, start, end):
self._fill(symbol, span_start, span_end)
return self._repository.load_items(symbol, start, end)
[docs]
def load(self, symbol: str, start: datetime, end: datetime) -> list[NewsItem]:
"""Read the archive without fetching anything.
What a replay uses. A replay that could reach the network would score the
model against news fetched today rather than against the pinned archive,
and would silently succeed while doing it.
"""
return self._repository.load_items(symbol, start, end)
def _fill(self, symbol: 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 genuinely quiet stretch being re-requested forever — for ITRG,
newsless in 138 of 189 measured weeks, that is most of the archive.
"""
fetched = self._provider.get_news_between(symbol, start, end)
inserted = self._repository.save_items(fetched)
self._repository.record_coverage(symbol, start, end)
_logger.info(
"%s: archived %d of %d news items for %s..%s",
symbol,
inserted,
len(fetched),
start.date(),
end.date(),
)