diff --git a/.env.example b/.env.example index fd899bb..b0c0327 100644 --- a/.env.example +++ b/.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) # ============================================================ diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index 5e9d582..1702ba7 100644 --- a/alpaca_paper/client.py +++ b/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]]: diff --git a/research/ingestion/alpaca_news.py b/research/ingestion/alpaca_news.py new file mode 100644 index 0000000..b134a69 --- /dev/null +++ b/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 diff --git a/run_live_paper.py b/run_live_paper.py index 0121e6d..3990906 100644 --- a/run_live_paper.py +++ b/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")) diff --git a/tests/unit/research/test_alpaca_news.py b/tests/unit/research/test_alpaca_news.py new file mode 100644 index 0000000..4eae34c --- /dev/null +++ b/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