openbb-invest-api/openbb_service.py

import asyncio
import logging
from datetime import datetime, timedelta
from typing import Any

from openbb import obb

logger = logging.getLogger(__name__)

PROVIDER = "yfinance"


def _to_dicts(result: Any) -> list[dict[str, Any]]:
    """Convert OBBject results to list of dicts."""
    if result is None or result.results is None:
        return []
    if isinstance(result.results, list):
        return [
            item.model_dump() if hasattr(item, "model_dump") else vars(item)
            for item in result.results
        ]
    if hasattr(result.results, "model_dump"):
        return [result.results.model_dump()]
    return [vars(result.results)]


def _first_or_empty(result: Any) -> dict[str, Any]:
    """Get first result as dict, or empty dict."""
    items = _to_dicts(result)
    return items[0] if items else {}


async def get_quote(symbol: str) -> dict:
    result = await asyncio.to_thread(
        obb.equity.price.quote, symbol, provider=PROVIDER
    )
    return _first_or_empty(result)


async def get_historical(symbol: str, days: int = 365) -> list[dict]:
    start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
    result = await asyncio.to_thread(
        obb.equity.price.historical,
        symbol,
        start_date=start,
        provider=PROVIDER,
    )
    items = _to_dicts(result)
    return [
        {**item, "date": str(item["date"])}
        if "date" in item and not isinstance(item["date"], str)
        else item
        for item in items
    ]


async def get_profile(symbol: str) -> dict:
    result = await asyncio.to_thread(
        obb.equity.profile, symbol, provider=PROVIDER
    )
    return _first_or_empty(result)


async def get_metrics(symbol: str) -> dict:
    result = await asyncio.to_thread(
        obb.equity.fundamental.metrics, symbol, provider=PROVIDER
    )
    return _first_or_empty(result)


async def get_income(symbol: str) -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.fundamental.income, symbol, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_balance(symbol: str) -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.fundamental.balance, symbol, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_cash_flow(symbol: str) -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.fundamental.cash, symbol, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_financials(symbol: str) -> dict:
    income, balance, cash_flow = await asyncio.gather(
        get_income(symbol),
        get_balance(symbol),
        get_cash_flow(symbol),
    )
    return {
        "symbol": symbol,
        "income": income,
        "balance": balance,
        "cash_flow": cash_flow,
    }


async def get_price_target(symbol: str) -> float | None:
    try:
        result = await asyncio.to_thread(
            obb.equity.estimates.price_target, symbol, provider="fmp"
        )
        items = _to_dicts(result)
        if items:
            return items[0].get("adj_price_target") or items[0].get("price_target")
    except Exception:
        logger.warning("Failed to get price target for %s", symbol, exc_info=True)
    return None


async def get_news(symbol: str) -> list[dict]:
    result = await asyncio.to_thread(
        obb.news.company, symbol, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_summary(symbol: str) -> dict:
    quote, profile, metrics, financials = await asyncio.gather(
        get_quote(symbol),
        get_profile(symbol),
        get_metrics(symbol),
        get_financials(symbol),
    )
    return {
        "quote": quote,
        "profile": profile,
        "metrics": metrics,
        "financials": financials,
    }


async def get_gainers() -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.discovery.gainers, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_losers() -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.discovery.losers, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_active() -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.discovery.active, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_undervalued() -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.discovery.undervalued_large_caps, provider=PROVIDER
    )
    return _to_dicts(result)


async def get_growth() -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.discovery.growth_tech, provider=PROVIDER
    )
    return _to_dicts(result)