"""One `NewsProvider` over two feeds.
**The two feeds are disjoint corpora, not substitutes.** Measured 2026-08-14:
Alpaca's news is 100% Benzinga and shared **0 of 10 headlines** with yfinance on
every symbol probed over the same 30-day window, and inside the 72-hour window
that actually reaches the prompt Alpaca was *thinner* than yfinance for every
small cap this app trades (ANNX 3 vs 4, ITRG 1 vs 3, QUBT 0 vs 2, IONQ 0 vs 10).
A straight swap to Alpaca would therefore have cut fresh news for exactly the
names where news drives the decision, pushing them into the strategy's
no-fresh-news skip — a change to the trading path disguised as a data-source
change. Merging is what keeps both corpora.
Three properties are load-bearing:
- **A per-feed quota**, or the chattier feed takes every slot. yfinance returns
far more items per hour for large caps, so Benzinga's would never appear.
- **A total order with a tiebreak.** `news_fingerprint` digests the *ordered*
set, so an unstable sort would change the fingerprint every cycle and defeat
the reuse gate whose entire job is not re-asking the model a question it has
already answered.
- **Failure isolation.** One feed down must cost half the corpus, not the
cycle. Both down must NOT read as "no news": the strategy's no-fresh-news
skip would then quietly stop entering on an outage.
"""
import logging
import math
from collections.abc import Sequence
from datetime import UTC, datetime, timedelta
from trader.errors import MarketDataError
from trader.news.base import DEFAULT_NEWS_WINDOW_HOURS, NewsItem, NewsProvider
__all__ = ["MergedNewsProvider"]
_logger = logging.getLogger("trader.news")
[docs]
class MergedNewsProvider:
"""Interleaves two feeds under the one-method `NewsProvider` Protocol.
`primary` is the feed whose items are preferred when the same story reaches
both — in production yfinance, which carried a summary on 10 of 10 items for
all eight symbols probed where Alpaca carried none on 24.
"""
def __init__(
self,
primary: NewsProvider,
secondary: NewsProvider,
*,
freshness_hours: float = DEFAULT_NEWS_WINDOW_HOURS,
) -> None:
self._primary = primary
self._secondary = secondary
self._freshness_hours = freshness_hours
[docs]
def get_recent_news(self, symbol: str, limit: int = 10) -> list[NewsItem]:
"""Most recent items first, capped at `limit`, drawn from both feeds.
Raises:
MarketDataError: **both** feeds failed. One failing is logged and
its half dropped.
"""
feeds = (self._primary, self._secondary)
halves: list[list[NewsItem]] = []
failures: list[str] = []
for provider in feeds:
try:
halves.append(self._newest_first(provider.get_recent_news(symbol, limit)))
except MarketDataError as exc:
# Named by class: the operator needs to know WHICH feed went
# away, since the surviving half looks like a healthy result.
_logger.warning(
"news feed %s failed for %s, dropping its half: %s",
type(provider).__name__,
symbol,
exc,
)
failures.append(f"{type(provider).__name__}: {exc}")
halves.append([])
if len(failures) == len(feeds):
raise MarketDataError(
f"Every news feed failed for {symbol} ({'; '.join(failures)})"
)
return self._merge(halves[0], halves[1], limit)
def _merge(
self, primary: list[NewsItem], secondary: list[NewsItem], limit: int
) -> list[NewsItem]:
"""Fresh items compete for the quota; stale ones only fill the tail.
**The quota is age-aware because an age-blind one measurably cost fresh
news.** Measured 2026-08-15 against the live feeds: a fixed 5/5 split
took IONQ from 10 fresh items to 8 and UUUU from 10 to 9, because
Benzinga's stale items held quota slots against fresh yfinance ones.
That is the same harm as swapping the feed outright — fewer fresh items
for exactly the names this app trades — in miniature, and it would have
shipped invisibly: the merged feed still returned ten items.
Stale items are ordered last, never dropped. The window belongs to the
strategy, which asks the model about a **held** position even with no
fresh news at all; filtering here would quietly take that away.
"""
cutoff = datetime.now(UTC) - timedelta(hours=self._freshness_hours)
fresh_primary, stale_primary = self._split(primary, cutoff)
fresh_secondary, stale_secondary = self._split(secondary, cutoff)
chosen = self._quota(fresh_primary, fresh_secondary, limit)
shortfall = limit - len(chosen)
if shortfall > 0:
tail = self._newest_first(stale_primary + stale_secondary)
chosen.extend(tail[:shortfall])
return self._ordered(self._dedupe(chosen))[:limit]
@staticmethod
def _split(
items: list[NewsItem], cutoff: datetime
) -> tuple[list[NewsItem], list[NewsItem]]:
fresh = [item for item in items if item.published_at >= cutoff]
stale = [item for item in items if item.published_at < cutoff]
return fresh, stale
@staticmethod
def _quota(
first: list[NewsItem], second: list[NewsItem], limit: int
) -> list[NewsItem]:
"""Half each, then top up from whichever side still has items.
The per-feed half is what stops the chattier feed taking every slot; the
top-up is what stops a sparse feed's unused half being wasted, which is
the norm for small caps.
"""
quota = math.ceil(limit / 2) if limit > 0 else 0
chosen = first[:quota] + second[:quota]
for spare in (first[quota:], second[quota:]):
shortfall = limit - len(chosen)
if shortfall <= 0:
break
chosen.extend(spare[:shortfall])
return chosen
@staticmethod
def _newest_first(items: Sequence[NewsItem]) -> list[NewsItem]:
"""Both providers promise this order; neither is trusted for it.
The quota is taken off the front of each list, so a feed that returned
oldest-first would spend its half on stale items.
"""
return sorted(items, key=lambda item: item.published_at, reverse=True)
@staticmethod
def _dedupe(items: Sequence[NewsItem]) -> list[NewsItem]:
"""One story, one item — matched on url **or** normalised title.
Both keys are needed: the same wire story reaches the two feeds under
different urls, and the same url can arrive with a reworded headline.
**This deliberately does not reorder anything.** An earlier version
visited candidates in a sorted order, which normalised the output into
the final tie order as a side effect — and that made the tiebreak in
`_ordered` redundant, so removing the tiebreak entirely left every test
passing. Ordering lives in exactly one place, and this is not it.
Duplicates are resolved by replacing in place instead, so the surviving
copy is the one with body text without the sequence being touched.
"""
kept: list[NewsItem] = []
slot_by_key: dict[str, int] = {}
for item in items:
keys = [" ".join(item.title.split()).casefold()]
if item.url:
keys.append(item.url)
slot = next((slot_by_key[key] for key in keys if key in slot_by_key), None)
if slot is None:
for key in keys:
slot_by_key[key] = len(kept)
kept.append(item)
continue
if kept[slot].summary is None and item.summary is not None:
# Measured: yfinance carried a summary on 10 of 10 items where
# Alpaca carried none on 24. The summarised copy is the one the
# prompt can actually read.
kept[slot] = item
for key in keys:
slot_by_key.setdefault(key, slot)
return kept
@staticmethod
def _ordered(items: list[NewsItem]) -> list[NewsItem]:
"""`published_at` descending, then url, then title.
Two stable passes rather than one composite key, because the primary
direction is descending and the tiebreaks ascend. The tiebreak is what
makes the merged order a function of the item set alone — see the
fingerprint note in this module's docstring.
"""
items.sort(key=lambda item: (item.url or "", item.title))
items.sort(key=lambda item: item.published_at, reverse=True)
return items