66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
"""Financial Modeling Prep (FMP) fundamental data provider."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_BASE_URL = "https://financialmodelingprep.com/api/v3/key-metrics-ttm"
|
|
|
|
|
|
def _safe_float(val: object) -> float | None:
|
|
"""Convert a value to float, returning None for missing/invalid values."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
return float(val)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
class FMPProvider(FundamentalsProvider):
|
|
"""Fetches fundamental data from Financial Modeling Prep key-metrics-ttm endpoint."""
|
|
|
|
def __init__(self, api_key: str, timeout: float = 10.0) -> None:
|
|
self._api_key = api_key
|
|
self._client = httpx.AsyncClient(timeout=timeout)
|
|
|
|
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
|
|
"""Fetch TTM key metrics from FMP."""
|
|
try:
|
|
resp = await self._client.get(
|
|
f"{_BASE_URL}/{ticker}",
|
|
params={"apikey": self._api_key},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
if not isinstance(data, list) or len(data) == 0:
|
|
logger.warning("FMP returned empty response for %s", ticker)
|
|
return None
|
|
|
|
item = data[0]
|
|
|
|
return FundamentalsSnapshot(
|
|
ticker=ticker,
|
|
eps_ttm=_safe_float(item.get("netIncomePerShareTTM")),
|
|
pe_ratio=_safe_float(item.get("peRatioTTM")),
|
|
peg_ratio=_safe_float(item.get("pegRatioTTM")),
|
|
revenue_growth_yoy=_safe_float(item.get("revenueGrowth")),
|
|
profit_margin=_safe_float(item.get("netProfitMarginTTM")),
|
|
debt_to_equity=_safe_float(item.get("debtToEquityTTM")),
|
|
market_cap=_safe_float(item.get("marketCapTTM")),
|
|
fetched_at=datetime.now(timezone.utc),
|
|
)
|
|
except Exception:
|
|
logger.exception("FMP fetch failed for %s", ticker)
|
|
return None
|
|
|
|
async def close(self) -> None:
|
|
await self._client.aclose()
|