"""Repository for the historical news archive and its coverage spans.
Separate from `trader/persistence/news.py`-style live storage on purpose: there
is none. The live path records what it saw into `decisions.inputs_json` and
never reads a news table. This one accumulates backwards from Alpaca's archive
so a past decision can be replayed on the news it would have had.
"""
from collections.abc import Sequence
from datetime import UTC, datetime
from sqlalchemy import delete, select
from sqlalchemy.orm import Session, sessionmaker
from trader.news.base import ArchivedNewsItem, NewsItem
from trader.persistence.models import ArchivedNews, NewsArchiveCoverage
__all__ = ["NewsArchiveRepository"]
[docs]
class NewsArchiveRepository:
"""Reads and writes archived news, and tracks which spans were fetched."""
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
[docs]
def save_items(self, items: Sequence[ArchivedNewsItem]) -> int:
"""Insert archived items, skipping any already held. Returns inserted count.
The symbol comes from each item rather than from an argument, so a caller
cannot file one symbol's story under another's ticker — the mistake
`save_bars(symbol, ...)` leaves available by ignoring `Bar.symbol`.
An article already present is left exactly as it was. Benzinga corrects
stories in place under the same id, and the decision being replayed was
made on the original wording.
"""
if not items:
return 0
now = datetime.now(UTC)
with self._session_factory() as session:
symbols = {a.item.symbol for a in items}
held = {
(symbol, published_at, article_id)
for symbol, published_at, article_id in session.execute(
select(
ArchivedNews.symbol,
ArchivedNews.published_at,
ArchivedNews.article_id,
).where(ArchivedNews.symbol.in_(symbols))
).all()
}
rows: list[ArchivedNews] = []
for a in items:
key = (a.item.symbol, a.item.published_at, a.article_id)
# Dedupe against the input too: two copies of one article in a
# single call would both pass a database-only check and then
# collide on the unique constraint at commit.
if key in held:
continue
held.add(key)
rows.append(
ArchivedNews(
symbol=a.item.symbol,
article_id=a.article_id,
published_at=a.item.published_at,
title=a.item.title,
publisher=a.item.publisher,
summary=a.item.summary,
url=a.item.url,
content_type=a.item.content_type,
fetched_at=now,
)
)
session.add_all(rows)
session.commit()
return len(rows)
[docs]
def load_items(self, symbol: str, start: datetime, end: datetime) -> list[NewsItem]:
"""Archived items published in `[start, end)`, most recent first.
**Half-open, and that is the point.** The replay's claim is that the
model saw nothing published at or after the decision instant, so an
inclusive end would hand it the headline it is being scored for
anticipating. `load_bars` is inclusive at both ends because a bar
timestamp names a whole period; a publication instant names a moment.
Most recent first because that is the order the provider returns, the
prompt renders, and `news_fingerprint` digests. The `article_id`
tiebreak makes a same-second batch deterministic: without it two stories
published in the same second come back in whatever order SQLite scans,
and one replay could build a different prompt from the next.
"""
statement = (
select(ArchivedNews)
.where(
ArchivedNews.symbol == symbol,
ArchivedNews.published_at >= start,
ArchivedNews.published_at < end,
)
.order_by(ArchivedNews.published_at.desc(), ArchivedNews.article_id.desc())
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
return [
NewsItem(
symbol=r.symbol,
title=r.title,
publisher=r.publisher,
published_at=r.published_at,
summary=r.summary,
url=r.url,
content_type=r.content_type,
)
for r in rows
]
[docs]
def covered_ranges(self, symbol: str) -> list[tuple[datetime, datetime]]:
"""Merged, sorted spans known to have been fetched."""
statement = (
select(NewsArchiveCoverage)
.where(NewsArchiveCoverage.symbol == symbol)
.order_by(NewsArchiveCoverage.start_at.asc())
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
return _merge([(r.start_at, r.end_at) for r in rows])
[docs]
def record_coverage(self, symbol: str, start: datetime, end: datetime) -> None:
"""Record a fetched span, merging it into any it overlaps or touches.
Read, delete and insert in one transaction, for `BarRepository`'s
reason: split apart, two callers each rewrite the row set from a stale
read and the later commit silently discards the other's span.
"""
now = datetime.now(UTC)
with self._session_factory() as session:
rows = session.scalars(
select(NewsArchiveCoverage)
.where(NewsArchiveCoverage.symbol == symbol)
.order_by(NewsArchiveCoverage.start_at.asc())
).all()
merged = _merge([(r.start_at, r.end_at) for r in rows] + [(start, end)])
session.execute(
delete(NewsArchiveCoverage).where(NewsArchiveCoverage.symbol == symbol)
)
session.add_all(
NewsArchiveCoverage(symbol=symbol, start_at=s, end_at=e, fetched_at=now)
for s, e in merged
)
session.commit()
[docs]
def missing_ranges(
self, symbol: str, start: datetime, end: datetime
) -> list[tuple[datetime, datetime]]:
"""Sub-spans of `[start, end)` that have never been fetched.
Half-open like `load_items`, and unsnapped: news has no bar grid, so
there is no width to floor onto and no adjacency but exact contact.
"""
gaps: list[tuple[datetime, datetime]] = []
cursor = start
for covered_start, covered_end in self.covered_ranges(symbol):
if covered_end <= cursor:
continue
if covered_start >= end:
break
if covered_start > cursor:
gaps.append((cursor, covered_start))
cursor = max(cursor, covered_end)
if cursor >= end:
return gaps
if cursor < end:
gaps.append((cursor, end))
return gaps
def _merge(
ranges: list[tuple[datetime, datetime]],
) -> list[tuple[datetime, datetime]]:
"""Merge overlapping or exactly touching spans into a minimal sorted list.
No adjacency tolerance, unlike `BarRepository._merge`'s one bar-width. Two
news spans a minute apart are genuinely not contiguous — an article could
have been published in that minute — and merging them would report the hole
as covered, after which nothing would ever fetch it.
"""
if not ranges:
return []
ordered = sorted(ranges)
merged = [ordered[0]]
for start, end in ordered[1:]:
last_start, last_end = merged[-1]
if start <= last_end:
merged[-1] = (last_start, max(last_end, end))
else:
merged.append((start, end))
return merged