Alpaca News API ingestion -> raw_news_items (feeds catalysts + Ollama)

The only news source was RSS (slow, regex ticker-scraping). Alpaca exposes
Benzinga news with a stable id + authoritative symbols list. Add it as a second
source landing into the SAME raw_news_items table, so recent_news_map (catalyst
prioritisation) and the Ollama enrichment (PR #2) pick it up with no other changes.

- AlpacaPaperClient.get_news(symbols, limit): GET /v1beta1/news (works on the
  existing data keys; verified live).
- research/ingestion/alpaca_news.py: ingest_alpaca_news lands new items, deduped
  by the Alpaca news id; never raises (news is an enhancement, not a dependency).
- run_live_paper step_news on the scheduler (ALPACA_NEWS_* env, default on, 120s),
  scoped to the tracked universe.

Verified live: passes landed ~34 then deduped; recent_news_map went from RSS-only
to 370 ticker->headline catalysts. tests/unit: 165 passed (+5).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -39,6 +39,13 @@ Alpha_Vantage_Api_Key=
 # added to the watchlist. Unset = disabled (no effect on the loop).
 # NEWS_FEEDS=prnewswire=https://www.prnewswire.com/rss/news-releases-list.rss,businesswire=https://feed.businesswire.com/rss/home/?rss=G1QFDERJXkJeGVtRWQ==
 
+# Alpaca News API ingestion (Benzinga) -> raw_news_items. Each item carries an
+# authoritative symbols list (no regex), feeding recent_news_map catalyst
+# prioritisation + the Ollama enrichment. Uses the existing Alpaca data keys.
+ALPACA_NEWS_ENABLED=1
+ALPACA_NEWS_INTERVAL_SECONDS=120
+ALPACA_NEWS_LIMIT=50
+
 # ============================================================
 # Live paper-trading orchestrator (run_live_paper.py)
 # ============================================================

alpaca_paper/client.py

@@ -337,6 +337,28 @@ class AlpacaPaperClient:
         )
         return payload.get("most_actives") or []
 
+    def get_news(
+        self,
+        symbols: list[str] | None = None,
+        limit: int = 50,
+        start_iso: str | None = None,
+        sort: str = "desc",
+    ) -> list[dict]:
+        """Latest market news (Benzinga). Each item carries a stable ``id`` and an
+        authoritative ``symbols`` list, so no regex ticker-scraping is needed.
+        Optional symbol filter (capped at 50 symbols/req); Alpaca caps ``limit`` at 50."""
+        params: dict = {
+            "limit": max(1, min(int(limit), 50)),
+            "sort": sort,
+            "exclude_contentless": "true",
+        }
+        if symbols:
+            params["symbols"] = ",".join(symbols[:50])
+        if start_iso:
+            params["start"] = start_iso
+        payload = self._data("GET", "/v1beta1/news", params=params)
+        return payload.get("news") or []
+
     def get_daily_bars(
         self, symbols: list[str], start_iso: str, end_iso: str | None = None
     ) -> dict[str, list[dict]]:

research/ingestion/alpaca_news.py

@@ -0,0 +1,122 @@
+"""Alpaca News API ingestion -> raw_news_items (the SAME landing table as RSS).
+
+Benzinga-sourced headlines arrive with a stable ``id`` and an authoritative
+``symbols`` list, so unlike rss.py we get real tickers (no regex scraping) and a
+reliable dedupe key. Lands into raw_fetch_attempts + raw_news_items exactly like
+rss.py, so the existing readers (recent_news_map catalyst prioritisation, the
+Ollama enrichment in PR #2) pick it up with no other changes.
+
+Stdlib-only transform; never raises — news is an enhancement, not a dependency.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+logger = logging.getLogger(__name__)
+
+PARSER_VERSION = "alpaca-news-v1"
+
+
+@dataclass
+class AlpacaNewsResult:
+    source: str = "alpaca"
+    fetch_attempt_id: str = ""
+    item_count: int = 0
+    error: str | None = None
+    new_items: list[str] = field(default_factory=list)
+
+
+def normalize_item(item: dict) -> dict:
+    """One Alpaca news item -> raw_news_items row fields."""
+    symbols = sorted({s for s in (item.get("symbols") or []) if s})
+    return {
+        "news_id": str(item.get("id") or ""),
+        "url": item.get("url") or "",
+        "title": (item.get("headline") or "").strip(),
+        # created_at is the publish time; fall back to updated_at
+        "published": item.get("created_at") or item.get("updated_at") or "",
+        "snippet": (item.get("summary") or "")[:500],
+        "tickers": ",".join(symbols),
+        "src": item.get("source") or "alpaca",
+    }
+
+
+def _dedupe_hash(n: dict) -> str:
+    """Stable per-item key: the Alpaca news id when present, else url+title."""
+    basis = n["news_id"] or f"{n['url']}|{n['title']}"
+    return hashlib.sha256(f"alpaca|{basis}".encode()).hexdigest()
+
+
+def ingest_alpaca_news(
+    con,
+    client,
+    symbols: list[str] | None = None,
+    limit: int = 50,
+    ingest_run_id: str | None = None,
+) -> AlpacaNewsResult:
+    """Fetch latest Alpaca news (optionally filtered to ``symbols``) and land new
+    items into raw_news_items. Deduped by the Alpaca news id. Never raises."""
+    result = AlpacaNewsResult()
+    attempt_id = str(uuid.uuid4())
+    result.fetch_attempt_id = attempt_id
+    run_id = ingest_run_id or str(uuid.uuid4())
+    fetched_at = datetime.now(timezone.utc).replace(tzinfo=None)
+
+    try:
+        items = client.get_news(symbols=symbols or None, limit=limit) or []
+    except Exception as exc:  # noqa: BLE001
+        result.error = str(exc)
+        items = []
+        logger.warning("alpaca news fetch failed: %s", exc)
+
+    try:
+        con.execute(
+            """
+            INSERT INTO raw_fetch_attempts (
+                id, source, capability, fetched_at, ingest_run_id,
+                http_status, item_count, error_msg, payload_hash, parser_version
+            ) VALUES (?, 'alpaca', 'news_api', ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [attempt_id, fetched_at, run_id,
+             200 if not result.error else None,
+             len(items), result.error, None, PARSER_VERSION],
+        )
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("alpaca news attempt-row insert failed: %s", exc)
+
+    for item in items:
+        n = normalize_item(item)
+        if not n["news_id"] and not n["title"]:
+            continue
+        item_hash = _dedupe_hash(n)
+        try:
+            exists = con.execute(
+                "SELECT 1 FROM raw_news_items WHERE payload_hash = ? LIMIT 1",
+                [item_hash],
+            ).fetchone()
+            if exists:
+                continue
+            item_id = str(uuid.uuid4())
+            con.execute(
+                """
+                INSERT INTO raw_news_items (
+                    id, fetch_attempt_id, source, raw_url, raw_title,
+                    raw_published_at, raw_body_snippet, raw_tickers,
+                    payload_hash, parser_version, fetched_at, ingest_run_id
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                [item_id, attempt_id, f"alpaca:{n['src']}", n["url"], n["title"],
+                 n["published"], n["snippet"], n["tickers"], item_hash,
+                 PARSER_VERSION, fetched_at, run_id],
+            )
+            result.new_items.append(item_id)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("alpaca news item insert failed: %s", exc)
+
+    result.item_count = len(items)
+    return result

run_live_paper.py

@@ -49,6 +49,7 @@ from research.ingestion.market_data import (
 from research.ingestion.scheduler import Scheduler
 from research.ingestion.watcher_task import ResearchWatchlistProvider
 from research.ingestion.discovery import run_discovery, screen_universe
+from research.ingestion.alpaca_news import ingest_alpaca_news
 from research.multi_schema import open_research_db
 from runtime.exit_manager import LiveExitManager
 from runtime.halt_guard import is_halt_gap
@@ -767,7 +768,25 @@ def main(argv: list[str] | None = None) -> int:
         return (f"EOD FLATTEN closed [{closed}]"
                 + (f" errors={res['errors']} (will retry)" if res["errors"] else ""))
 
+    def step_news():
+        """Land latest Alpaca/Benzinga news for the tracked universe into
+        raw_news_items (feeds recent_news_map catalyst prioritisation + the Ollama
+        enrichment). Off the hot path; never blocks trading."""
+        if not client:
+            return "skipped (no keys)"
+        res = ingest_alpaca_news(
+            rt["research_con"], client, symbols=symbols,
+            limit=int(os.environ.get("ALPACA_NEWS_LIMIT", "50")),
+        )
+        if res.error:
+            return f"ERROR {res.error}"
+        if res.new_items:
+            return f"+{len(res.new_items)} new / {res.item_count} fetched"
+        return None
+
     scheduler = Scheduler()
+    scheduler.add("news", step_news, float(os.environ.get("ALPACA_NEWS_INTERVAL_SECONDS", "120")),
+                  enabled=_flag("ALPACA_NEWS_ENABLED", "1"))
     scheduler.add("discover", step_discover,
                   float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")),
                   enabled=_flag("DISCOVER_ENABLED", "1"))

tests/unit/research/test_alpaca_news.py

@@ -0,0 +1,81 @@
+"""Alpaca News API ingestion -> raw_news_items (same landing table as RSS)."""
+
+import pytest
+
+from research.ingestion.alpaca_news import ingest_alpaca_news, normalize_item
+from storage.db import get_connection
+
+
+@pytest.fixture
+def con():
+    c = get_connection(":memory:")
+    yield c
+    c.close()
+
+
+class _FakeClient:
+    def __init__(self, news=None, raises=False):
+        self._news = news or []
+        self.raises = raises
+        self.calls = []
+
+    def get_news(self, symbols=None, limit=50):
+        self.calls.append({"symbols": symbols, "limit": limit})
+        if self.raises:
+            raise RuntimeError("news api down")
+        return self._news
+
+
+def _item(nid, headline, symbols, **kw):
+    d = {"id": nid, "headline": headline, "symbols": symbols,
+         "url": f"http://x/{nid}", "summary": "snippet",
+         "created_at": "2026-06-20T14:00:00Z", "source": "benzinga"}
+    d.update(kw)
+    return d
+
+
+def test_normalize_uses_alpaca_symbols_as_tickers():
+    n = normalize_item(_item("1", "  ICCM jumps on data  ", ["ICCM", "CDT", "ICCM"]))
+    assert n["news_id"] == "1"
+    assert n["title"] == "ICCM jumps on data"      # stripped
+    assert n["tickers"] == "CDT,ICCM"              # sorted + deduped, no regex
+    assert n["src"] == "benzinga"
+    assert n["published"] == "2026-06-20T14:00:00Z"
+
+
+def test_ingest_lands_rows_with_alpaca_source(con):
+    client = _FakeClient([_item("100", "WPRT moves", ["WPRT"]),
+                          _item("101", "CDT boost", ["CDT", "ICCM"])])
+    r = ingest_alpaca_news(con, client, symbols=["WPRT", "CDT"])
+    assert r.item_count == 2 and len(r.new_items) == 2 and r.error is None
+    rows = con.execute(
+        "SELECT source, raw_tickers, raw_title FROM raw_news_items "
+        "WHERE source LIKE 'alpaca%' ORDER BY raw_title").fetchall()
+    assert len(rows) == 2
+    assert rows[0][0] == "alpaca:benzinga"
+    assert any("WPRT" in (t or "") for _, t, _ in rows)
+
+
+def test_ingest_dedupes_on_second_pass(con):
+    client = _FakeClient([_item("200", "same story", ["AAA"])])
+    assert len(ingest_alpaca_news(con, client).new_items) == 1
+    # same Alpaca id -> nothing new the second time
+    r2 = ingest_alpaca_news(con, client)
+    assert len(r2.new_items) == 0 and r2.item_count == 1
+
+
+def test_ingest_never_raises_when_client_fails(con):
+    client = _FakeClient(raises=True)
+    r = ingest_alpaca_news(con, client)            # must not raise
+    assert r.error is not None and r.new_items == []
+    # the failed attempt is still recorded (with the error) for observability
+    att = con.execute(
+        "SELECT error_msg FROM raw_fetch_attempts WHERE source='alpaca'").fetchall()
+    assert att and att[0][0]
+
+
+def test_ingest_forwards_symbol_filter(con):
+    client = _FakeClient([])
+    ingest_alpaca_news(con, client, symbols=["AAA", "BBB"], limit=10)
+    assert client.calls[0]["symbols"] == ["AAA", "BBB"]
+    assert client.calls[0]["limit"] == 10