diff --git a/src/findata/api/_b3_common.py b/src/findata/api/_b3_common.py new file mode 100644 index 0000000..564e50d --- /dev/null +++ b/src/findata/api/_b3_common.py @@ -0,0 +1,26 @@ +"""Shared B3 live-quote helpers for the REST and MCP layers. + +``yfinance`` is imported lazily so the rest of the API keeps working without +the optional ``[b3]`` extra installed. Both ``routers/b3.py`` and ``mcp_app.py`` +resolve the quotes module through here, so the install hint lives in one place. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import HTTPException + +MAX_TICKERS = 20 + + +def resolve_quotes() -> Any: + """Return the yfinance-backed quotes module, or raise 503 if the extra is missing.""" + try: + from findata.sources.b3 import quotes + except ImportError as exc: # pragma: no cover - only without the [b3] extra + raise HTTPException( + status_code=503, + detail="B3 live quotes need the optional extra: pip install 'openfindata[b3]'", + ) from exc + return quotes diff --git a/src/findata/api/mcp_app.py b/src/findata/api/mcp_app.py index d43a48d..a026e5f 100644 --- a/src/findata/api/mcp_app.py +++ b/src/findata/api/mcp_app.py @@ -37,6 +37,7 @@ from fastapi.responses import JSONResponse from pydantic import BaseModel, Field +from findata.api._b3_common import MAX_TICKERS, resolve_quotes from findata.registry import lookup from findata.sources.anbima import indices as anbima_src from findata.sources.aneel import leiloes @@ -66,7 +67,6 @@ router = APIRouter() -_MAX_TICKERS = 20 _MIN_YEAR_B3_COTAHIST = 1986 # B3 publishes COTAHIST since 1986 _RGF_MAX_PERIOD = 3 # RGF quadrimestre runs 1..3 @@ -384,16 +384,6 @@ async def cvm_structured_fund( # ── B3: Bolsa ───────────────────────────────────────────────────── -def _b3_quotes() -> Any: - try: - from findata.sources.b3 import quotes - except ImportError as exc: # pragma: no cover, only without the [b3] extra - raise HTTPException( - 503, "Live quotes need the optional extra: pip install 'openfindata[b3]'" - ) from exc - return quotes - - @router.get( "/b3/quote", operation_id="b3_quote", @@ -408,12 +398,12 @@ async def b3_quote( """Current quote(s) from the optional yfinance-backed source. For canonical, official end-of-day history use ``b3_cotahist`` instead. """ - quotes = _b3_quotes() + quotes = resolve_quotes() ticker_list = [t.strip() for t in tickers.split(",") if t.strip()] if not ticker_list: raise HTTPException(400, "at least one ticker is required") - if len(ticker_list) > _MAX_TICKERS: - raise HTTPException(400, f"max {_MAX_TICKERS} tickers per request") + if len(ticker_list) > MAX_TICKERS: + raise HTTPException(400, f"max {MAX_TICKERS} tickers per request") if len(ticker_list) == 1: return await quotes.get_quote(ticker_list[0]) return await quotes.get_multiple_quotes(ticker_list) @@ -655,11 +645,7 @@ async def anbima_tool( if dataset == "ettj": return await anbima_src.get_ettj(data) if dataset == "debentures": - rows = await anbima_src.get_debentures(data) - if emissor: - needle = emissor.upper() - rows = [r for r in rows if needle in r.emissor.upper()] - return rows[:limit] + return (await anbima_src.get_debentures(data, emissor=emissor))[:limit] fam = anbima_src.IMAFamily(family) if family else None return (await anbima_src.get_ima(fam))[:limit] diff --git a/src/findata/api/routers/anbima.py b/src/findata/api/routers/anbima.py index a92346c..6af0e87 100644 --- a/src/findata/api/routers/anbima.py +++ b/src/findata/api/routers/anbima.py @@ -41,10 +41,7 @@ async def debentures( limit: int = Query(default=500, ge=1, le=5000), ) -> list[anbima.DebentureQuote]: """Daily secondary-market quotes for outstanding debentures.""" - rows = await anbima.get_debentures(data) - if emissor: - needle = emissor.upper() - rows = [r for r in rows if needle in r.emissor.upper()] + rows = await anbima.get_debentures(data, emissor=emissor) return rows[:limit] diff --git a/src/findata/api/routers/b3.py b/src/findata/api/routers/b3.py index 41d0ffc..48a1f96 100644 --- a/src/findata/api/routers/b3.py +++ b/src/findata/api/routers/b3.py @@ -12,29 +12,18 @@ from fastapi import APIRouter, HTTPException, Path, Query +from findata.api._b3_common import MAX_TICKERS, resolve_quotes from findata.sources.b3 import cotahist, indices router = APIRouter(prefix="/b3", tags=["B3 - Bolsa"]) -_MAX_TICKERS_PER_REQUEST = 20 _CURRENT_YEAR = date.today().year -def _quotes() -> Any: - try: - from findata.sources.b3 import quotes - except ImportError as exc: # pragma: no cover — triggered only without yfinance - raise HTTPException( - status_code=503, - detail=("B3 support is disabled. Install with: pip install 'openfindata[b3]'"), - ) from exc - return quotes - - @router.get("/quote/{ticker}") async def get_quote(ticker: str) -> Any: """Get current stock quote for a B3 ticker (e.g., PETR4, VALE3, WEGE3).""" - quotes = _quotes() + quotes = resolve_quotes() try: return await quotes.get_quote(ticker) except HTTPException: @@ -56,7 +45,7 @@ async def get_history( ), ) -> Any: """Get historical price data for a B3 stock.""" - quotes = _quotes() + quotes = resolve_quotes() try: return await quotes.get_history(ticker, period, interval) except HTTPException: @@ -70,14 +59,14 @@ async def get_multiple_quotes( tickers: str = Query(..., description="Comma-separated tickers (e.g., PETR4,VALE3,ITUB4)"), ) -> Any: """Get current quotes for multiple B3 tickers.""" - quotes = _quotes() + quotes = resolve_quotes() ticker_list = [t.strip() for t in tickers.split(",") if t.strip()] if not ticker_list: raise HTTPException(status_code=400, detail="At least one ticker is required") - if len(ticker_list) > _MAX_TICKERS_PER_REQUEST: + if len(ticker_list) > MAX_TICKERS: raise HTTPException( status_code=400, - detail=f"Max {_MAX_TICKERS_PER_REQUEST} tickers per request", + detail=f"Max {MAX_TICKERS} tickers per request", ) try: return await quotes.get_multiple_quotes(ticker_list) diff --git a/src/findata/sources/anbima/indices.py b/src/findata/sources/anbima/indices.py index 0682303..1c8c527 100644 --- a/src/findata/sources/anbima/indices.py +++ b/src/findata/sources/anbima/indices.py @@ -555,7 +555,9 @@ async def _fetch_anbima_txt(url: str, label: str, d: date) -> str | None: # ── Debêntures (daily TXT) ─────────────────────────────────────── -async def get_debentures(data_referencia: date | None = None) -> list[DebentureQuote]: +async def get_debentures( + data_referencia: date | None = None, emissor: str | None = None +) -> list[DebentureQuote]: """Daily secondary-market quotes for outstanding debentures.""" d = data_referencia or date.today() ymd = d.strftime("%y%m%d") @@ -597,6 +599,9 @@ async def get_debentures(data_referencia: date | None = None) -> list[DebentureQ ) if not header_seen: logger.warning("debentures file for %s had no recognisable header row", d) + if emissor: + needle = emissor.upper() + rows = [r for r in rows if needle in r.emissor.upper()] return rows