openbb-invest-api/openbb_service.py

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

from openbb import obb

from obb_utils import to_list, first_or_empty

logger = logging.getLogger(__name__)

PROVIDER = "yfinance"


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_list(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_list(result)


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


async def get_cash_flow(symbol: str) -> list[dict]:
    result = await asyncio.to_thread(
        obb.equity.fundamental.cash, symbol, provider=PROVIDER
    )
    return to_list(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:
    """Get consensus analyst price target via yfinance."""
    import yfinance as yf

    def _fetch() -> float | None:
        t = yf.Ticker(symbol)
        return t.info.get("targetMeanPrice")

    try:
        return await asyncio.to_thread(_fetch)
    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_list(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_list(result)


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


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


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


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


async def get_upgrades_downgrades(symbol: str, limit: int = 20) -> list[dict]:
    """Get analyst upgrades/downgrades via yfinance."""
    import yfinance as yf

    def _fetch() -> list[dict[str, Any]]:
        t = yf.Ticker(symbol)
        df = t.upgrades_downgrades
        if df is None or df.empty:
            return []
        df = df.head(limit).reset_index()
        return [
            {
                "date": str(row.get("GradeDate", "")),
                "company": row.get("Firm"),
                "action": row.get("Action"),
                "from_grade": row.get("FromGrade"),
                "to_grade": row.get("ToGrade"),
                "price_target_action": row.get("priceTargetAction"),
                "current_price_target": row.get("currentPriceTarget"),
                "prior_price_target": row.get("priorPriceTarget"),
            }
            for _, row in df.iterrows()
        ]

    return await asyncio.to_thread(_fetch)


# --- Equity Fundamentals Extended (Group B) ---


async def get_management(symbol: str) -> list[dict]:
    """Get executive team info (name, title, compensation)."""
    result = await asyncio.to_thread(
        obb.equity.fundamental.management, symbol, provider=PROVIDER
    )
    return to_list(result)


async def get_dividends(symbol: str) -> list[dict]:
    """Get historical dividend records."""
    result = await asyncio.to_thread(
        obb.equity.fundamental.dividends, symbol, provider=PROVIDER
    )
    return to_list(result)


async def get_filings(symbol: str, form_type: str | None = None) -> list[dict]:
    """Get SEC filings (10-K, 10-Q, 8-K, etc.)."""
    kwargs: dict[str, Any] = {"symbol": symbol, "provider": "sec"}
    if form_type:
        kwargs["type"] = form_type
    result = await asyncio.to_thread(obb.equity.fundamental.filings, **kwargs)
    return to_list(result)


async def search_company(query: str) -> list[dict]:
    """Search for companies by name."""
    result = await asyncio.to_thread(obb.equity.search, query, provider="sec")
    return to_list(result)