openbb-invest-api/technical_service.py

"""Technical analysis indicators via openbb-technical (local computation)."""

import asyncio
import logging
from typing import Any

from openbb import obb

logger = logging.getLogger(__name__)

PROVIDER = "yfinance"


async def get_technical_indicators(
    symbol: str, days: int = 200
) -> dict[str, Any]:
    """Compute key technical indicators for a symbol."""
    from datetime import datetime, timedelta

    start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")

    # Fetch historical data first
    hist = await asyncio.to_thread(
        obb.equity.price.historical,
        symbol,
        start_date=start,
        provider=PROVIDER,
    )

    if hist is None or hist.results is None:
        return {"symbol": symbol, "error": "No historical data available"}

    result: dict[str, Any] = {"symbol": symbol}

    # RSI (14-period)
    try:
        rsi = await asyncio.to_thread(obb.technical.rsi, data=hist.results, length=14)
        rsi_items = _extract_latest(rsi)
        result["rsi_14"] = rsi_items.get("RSI_14")
    except Exception:
        logger.warning("RSI calculation failed for %s", symbol, exc_info=True)
        result["rsi_14"] = None

    # MACD (12, 26, 9)
    try:
        macd = await asyncio.to_thread(
            obb.technical.macd, data=hist.results, fast=12, slow=26, signal=9
        )
        macd_items = _extract_latest(macd)
        result["macd"] = {
            "macd": macd_items.get("MACD_12_26_9"),
            "signal": macd_items.get("MACDs_12_26_9"),
            "histogram": macd_items.get("MACDh_12_26_9"),
        }
    except Exception:
        logger.warning("MACD calculation failed for %s", symbol, exc_info=True)
        result["macd"] = None

    # SMA (20, 50, 200)
    for period in [20, 50, 200]:
        try:
            sma = await asyncio.to_thread(
                obb.technical.sma, data=hist.results, length=period
            )
            sma_items = _extract_latest(sma)
            result[f"sma_{period}"] = sma_items.get(f"SMA_{period}")
        except Exception:
            logger.warning("SMA_%d failed for %s", period, symbol, exc_info=True)
            result[f"sma_{period}"] = None

    # EMA (12, 26)
    for period in [12, 26]:
        try:
            ema = await asyncio.to_thread(
                obb.technical.ema, data=hist.results, length=period
            )
            ema_items = _extract_latest(ema)
            result[f"ema_{period}"] = ema_items.get(f"EMA_{period}")
        except Exception:
            logger.warning("EMA_%d failed for %s", period, symbol, exc_info=True)
            result[f"ema_{period}"] = None

    # Bollinger Bands (20, 2)
    try:
        bbands = await asyncio.to_thread(
            obb.technical.bbands, data=hist.results, length=20, std=2
        )
        bb_items = _extract_latest(bbands)
        result["bollinger_bands"] = {
            "upper": bb_items.get("BBU_20_2.0"),
            "middle": bb_items.get("BBM_20_2.0"),
            "lower": bb_items.get("BBL_20_2.0"),
        }
    except Exception:
        logger.warning("Bollinger Bands failed for %s", symbol, exc_info=True)
        result["bollinger_bands"] = None

    # Add interpretation
    result["signals"] = _interpret_signals(result)

    return result


def _extract_latest(result: Any) -> dict[str, Any]:
    """Get the last row from a technical indicator result as a dict."""
    if result is None or result.results is None:
        return {}
    items = result.results
    if isinstance(items, list) and items:
        last = items[-1]
        return last.model_dump() if hasattr(last, "model_dump") else vars(last)
    return {}


def _interpret_signals(data: dict[str, Any]) -> list[str]:
    """Generate simple text signals from technical indicators."""
    signals: list[str] = []

    rsi = data.get("rsi_14")
    if rsi is not None:
        if rsi > 70:
            signals.append(f"RSI {rsi:.1f}: Overbought (bearish signal)")
        elif rsi < 30:
            signals.append(f"RSI {rsi:.1f}: Oversold (bullish signal)")
        else:
            signals.append(f"RSI {rsi:.1f}: Neutral")

    macd = data.get("macd")
    if macd and macd.get("histogram") is not None:
        hist = macd["histogram"]
        if hist > 0:
            signals.append("MACD histogram positive (bullish momentum)")
        else:
            signals.append("MACD histogram negative (bearish momentum)")

    sma_50 = data.get("sma_50")
    sma_200 = data.get("sma_200")
    if sma_50 is not None and sma_200 is not None:
        if sma_50 > sma_200:
            signals.append("Golden cross: SMA50 above SMA200 (bullish trend)")
        else:
            signals.append("Death cross: SMA50 below SMA200 (bearish trend)")

    return signals