openbb-invest-api/alphavantage_service.py

"""Alpha Vantage API client for news sentiment analysis."""

import logging
from typing import Any

import httpx

from config import settings

logger = logging.getLogger(__name__)

BASE_URL = "https://www.alphavantage.co/query"
TIMEOUT = 15.0


def _is_configured() -> bool:
    return bool(settings.alphavantage_api_key)


async def get_news_sentiment(
    symbol: str, limit: int = 50
) -> dict[str, Any]:
    """Get news articles with per-ticker sentiment scores.

    Returns articles with sentiment_score (-1 to 1),
    sentiment_label (Bearish/Bullish/Neutral), and relevance_score.
    Free tier: 25 requests/day.
    """
    if not _is_configured():
        return {
            "configured": False,
            "message": "Set INVEST_API_ALPHAVANTAGE_API_KEY to enable news sentiment",
        }

    params = {
        "function": "NEWS_SENTIMENT",
        "tickers": symbol,
        "limit": limit,
        "apikey": settings.alphavantage_api_key,
    }

    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        resp = await client.get(BASE_URL, params=params)
        resp.raise_for_status()
        data = resp.json()

    if "Error Message" in data or "Note" in data:
        msg = data.get("Error Message") or data.get("Note", "")
        logger.warning("Alpha Vantage error: %s", msg)
        return {"configured": True, "error": msg, "articles": []}

    feed = data.get("feed", [])
    articles = [_parse_article(article, symbol) for article in feed]
    overall = _compute_overall_sentiment(articles)

    return {
        "configured": True,
        "symbol": symbol,
        "article_count": len(articles),
        "overall_sentiment": overall,
        "articles": articles,
    }


def _parse_article(article: dict[str, Any], symbol: str) -> dict[str, Any]:
    """Extract relevant fields from a single news article."""
    ticker_sentiment = _find_ticker_sentiment(
        article.get("ticker_sentiment", []), symbol
    )

    return {
        "title": article.get("title"),
        "url": article.get("url"),
        "source": article.get("source"),
        "published": article.get("time_published"),
        "summary": article.get("summary"),
        "overall_sentiment_score": _safe_float(
            article.get("overall_sentiment_score")
        ),
        "overall_sentiment_label": article.get("overall_sentiment_label"),
        "ticker_sentiment_score": ticker_sentiment.get("score"),
        "ticker_sentiment_label": ticker_sentiment.get("label"),
        "ticker_relevance": ticker_sentiment.get("relevance"),
        "topics": [
            t.get("topic") for t in article.get("topics", [])
        ],
    }


def _find_ticker_sentiment(
    sentiments: list[dict[str, Any]], symbol: str
) -> dict[str, Any]:
    """Find the sentiment entry matching the requested ticker."""
    upper = symbol.upper()
    for s in sentiments:
        ticker = s.get("ticker", "")
        if ticker.upper() == upper:
            return {
                "score": _safe_float(s.get("ticker_sentiment_score")),
                "label": s.get("ticker_sentiment_label"),
                "relevance": _safe_float(s.get("relevance_score")),
            }
    return {"score": None, "label": None, "relevance": None}


def _safe_float(val: Any) -> float | None:
    """Convert value to float, returning None on failure."""
    if val is None:
        return None
    try:
        return float(val)
    except (ValueError, TypeError):
        return None


def _compute_overall_sentiment(
    articles: list[dict[str, Any]],
) -> dict[str, Any]:
    """Compute aggregate sentiment stats from parsed articles."""
    scores = [
        a["ticker_sentiment_score"]
        for a in articles
        if a["ticker_sentiment_score"] is not None
    ]
    if not scores:
        return {
            "avg_score": None,
            "label": "Unknown",
            "bullish_count": 0,
            "bearish_count": 0,
            "neutral_count": 0,
            "total_scored": 0,
        }

    avg = sum(scores) / len(scores)
    bullish = sum(1 for s in scores if s >= 0.15)
    bearish = sum(1 for s in scores if s <= -0.15)
    neutral = len(scores) - bullish - bearish

    if avg >= 0.35:
        label = "Bullish"
    elif avg >= 0.15:
        label = "Somewhat-Bullish"
    elif avg >= -0.15:
        label = "Neutral"
    elif avg >= -0.35:
        label = "Somewhat-Bearish"
    else:
        label = "Bearish"

    return {
        "avg_score": round(avg, 4),
        "label": label,
        "bullish_count": bullish,
        "bearish_count": bearish,
        "neutral_count": neutral,
        "total_scored": len(scores),
    }