stock-oracle/app/services/news/alpaca_news_client.py

"""
Alpaca News (Benzinga backend) REST client.

Endpoint: GET /v1beta1/news on data.alpaca.markets
Auth: same APCA-API-KEY-ID/APCA-API-SECRET-KEY as market data.
Free-tier history: ~30 days. Cumulative archive must be self-built via daily
ingest.

Rate limit handled via existing per-process token-bucket pattern (200 req/min
shared with the market-data client where this matters; News calls are far
less frequent so a small private bucket is sufficient).
"""

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, AsyncIterator

import httpx

from app.core.config import settings

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class AlpacaNewsArticle:
    id: int
    headline: str
    summary: str | None
    url: str | None
    author: str | None
    created_at: datetime
    updated_at: datetime
    source: str | None  # vendor source tag (e.g. "benzinga")
    symbols: list[str]
    images: list[dict[str, Any]]
    content: str | None

    @classmethod
    def from_payload(cls, payload: dict[str, Any]) -> "AlpacaNewsArticle":
        return cls(
            id=int(payload["id"]),
            headline=payload.get("headline") or "",
            summary=payload.get("summary"),
            url=payload.get("url"),
            author=payload.get("author"),
            created_at=_parse_iso(payload["created_at"]),
            updated_at=_parse_iso(payload.get("updated_at") or payload["created_at"]),
            source=payload.get("source"),
            symbols=list(payload.get("symbols") or []),
            images=list(payload.get("images") or []),
            content=payload.get("content"),
        )


def _parse_iso(s: str) -> datetime:
    """Parse Alpaca timestamps (RFC 3339, may end in 'Z')."""
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    dt = datetime.fromisoformat(s)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


class AlpacaNewsClient:
    """Alpaca News API client. Reuses ALPACA_API_KEY/SECRET."""

    def __init__(
        self,
        api_key: str | None = None,
        secret_key: str | None = None,
        base_url: str | None = None,
        max_requests_per_min: int = 60,
    ):
        self.api_key = api_key or settings.ALPACA_API_KEY
        self.secret_key = secret_key or settings.ALPACA_SECRET_KEY
        self.base_url = (base_url or settings.ALPACA_NEWS_BASE_URL).rstrip("/")
        self._max_rpm = max_requests_per_min
        self._request_times: list[float] = []
        self._client: httpx.AsyncClient | None = None
        self._lock = asyncio.Lock()

    def is_configured(self) -> bool:
        return bool(self.api_key) and bool(self.secret_key)

    async def _get_client(self) -> httpx.AsyncClient:
        if self._client is None or self._client.is_closed:
            self._client = httpx.AsyncClient(
                base_url=self.base_url,
                headers={
                    "APCA-API-KEY-ID": self.api_key,
                    "APCA-API-SECRET-KEY": self.secret_key,
                },
                timeout=30.0,
            )
        return self._client

    async def close(self) -> None:
        if self._client and not self._client.is_closed:
            await self._client.aclose()

    async def _wait_for_rate_limit(self) -> None:
        async with self._lock:
            now = time.monotonic()
            self._request_times = [t for t in self._request_times if now - t < 60]
            if len(self._request_times) >= self._max_rpm:
                sleep_for = 60 - (now - self._request_times[0]) + 0.1
                if sleep_for > 0:
                    logger.debug(f"AlpacaNews rate limit reached, sleeping {sleep_for:.1f}s")
                    await asyncio.sleep(sleep_for)
            self._request_times.append(time.monotonic())

    async def _request(
        self,
        path: str,
        params: dict[str, Any] | None = None,
        retries: int = 3,
    ) -> dict[str, Any]:
        await self._wait_for_rate_limit()
        client = await self._get_client()

        last_exc: Exception | None = None
        for attempt in range(retries):
            try:
                resp = await client.get(path, params=params)
                if resp.status_code == 429:
                    wait = 2 ** attempt
                    logger.warning(f"AlpacaNews 429 — retrying in {wait}s (attempt {attempt + 1})")
                    await asyncio.sleep(wait)
                    continue
                if resp.status_code >= 500:
                    wait = 2 ** attempt
                    logger.warning(f"AlpacaNews {resp.status_code} — retrying in {wait}s (attempt {attempt + 1})")
                    await asyncio.sleep(wait)
                    continue
                resp.raise_for_status()
                return resp.json()
            except httpx.HTTPStatusError as exc:
                last_exc = exc
                if attempt < retries - 1 and exc.response.status_code in (429, 500, 502, 503, 504):
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
            except (httpx.ConnectError, httpx.ReadTimeout) as exc:
                last_exc = exc
                if attempt < retries - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
        raise last_exc  # type: ignore[misc]

    async def fetch_news(
        self,
        symbols: list[str] | None = None,
        start: datetime | None = None,
        end: datetime | None = None,
        include_content: bool = True,
        sort: str = "desc",
        page_limit: int = 50,
    ) -> AsyncIterator[AlpacaNewsArticle]:
        """
        Yield articles for the given symbols (or the whole stream if None) over
        [start, end]. Handles pagination via `page_token`.

        Alpaca page_size max = 50.
        """
        params: dict[str, Any] = {
            "limit": min(page_limit, 50),
            "sort": sort,
            "include_content": str(include_content).lower(),
        }
        if symbols:
            params["symbols"] = ",".join(s.strip().upper() for s in symbols if s)
        if start is not None:
            params["start"] = _to_rfc3339(start)
        if end is not None:
            params["end"] = _to_rfc3339(end)

        page_token: str | None = None
        while True:
            if page_token:
                params["page_token"] = page_token
            else:
                params.pop("page_token", None)

            payload = await self._request("/v1beta1/news", params=params)
            for raw in payload.get("news", []) or []:
                try:
                    yield AlpacaNewsArticle.from_payload(raw)
                except Exception as e:
                    logger.warning(f"AlpacaNews payload parse failed: {e} — {raw.get('id')}")

            page_token = payload.get("next_page_token")
            if not page_token:
                return


def _to_rfc3339(dt: datetime) -> str:
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    # Alpaca accepts RFC-3339 with "Z" or "+00:00"
    return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")