"""Financial Modeling Prep API — stable ratios/metrics + company news.""" import os import requests import streamlit as st from dotenv import load_dotenv from services.data_service import get_company_info, get_historical_ratios_yfinance, compute_ttm_ratios load_dotenv() BASE_URL = "https://financialmodelingprep.com" STABLE_BASE = f"{BASE_URL}/stable" LEGACY_BASE = f"{BASE_URL}/api/v3" def _api_key() -> str: return os.getenv("FMP_API_KEY", "") def _get(base_url: str, endpoint: str, params: dict | None = None) -> dict | list | None: params = params or {} params["apikey"] = _api_key() try: resp = requests.get(f"{base_url}{endpoint}", params=params, timeout=10) resp.raise_for_status() return resp.json() except Exception: return None @st.cache_data(ttl=3600) def get_key_ratios(ticker: str) -> dict: """Return TTM ratios for a ticker, computed from raw financial statements. All ratios are self-computed via compute_ttm_ratios() — no FMP calls. Forward P/E and dividend fallbacks come from yfinance's info dict. For edge cases, trailing P/E prefers the vendor-supplied value from the info dict when the self-computed statement-based figure is missing or materially inconsistent. This avoids obviously bad P/E outputs on tickers with restatements, near-zero earnings, or statement mapping quirks. """ ticker = ticker.upper() merged = {"symbol": ticker} computed = compute_ttm_ratios(ticker) if computed: merged.update(computed) # Forward P/E requires analyst estimates — can't compute from statements info = get_company_info(ticker) if info: trailing_pe_info = info.get("trailingPE") trailing_pe_computed = merged.get("peRatioTTM") if trailing_pe_info is not None: try: vendor_pe = float(trailing_pe_info) except (TypeError, ValueError): vendor_pe = None try: computed_pe = float(trailing_pe_computed) if trailing_pe_computed is not None else None except (TypeError, ValueError): computed_pe = None if vendor_pe is not None and vendor_pe > 0: if computed_pe is None or computed_pe <= 0: merged["peRatioTTM"] = vendor_pe else: # If the two values are wildly different, trust the vendor # trailing P/E. This prevents edge-case display bugs where a # malformed TTM net income makes P/E look duplicated/wrong. ratio_gap = max(vendor_pe, computed_pe) / max(min(vendor_pe, computed_pe), 1e-9) if ratio_gap > 2.0: merged["peRatioTTM"] = vendor_pe if info.get("forwardPE") is not None: merged["forwardPE"] = info["forwardPE"] # Fallback: dividends from info dict when cash-flow data is missing if merged.get("dividendYieldTTM") is None and info.get("dividendYield") is not None: merged["dividendYieldTTM"] = info["dividendYield"] payout_ratio_info = info.get("payoutRatio") if merged.get("dividendPayoutRatioTTM") is None and payout_ratio_info is not None: try: payout_ratio_value = float(payout_ratio_info) except (TypeError, ValueError): payout_ratio_value = None if payout_ratio_value is not None and payout_ratio_value > 0: merged["dividendPayoutRatioTTM"] = payout_ratio_value return merged if len(merged) > 1 else {} @st.cache_data(ttl=21600) def get_peers(ticker: str) -> list[str]: """Return comparable ticker symbols from FMP stable stock-peers endpoint.""" ticker = ticker.upper() data = _get(STABLE_BASE, "/stock-peers", params={"symbol": ticker}) if not isinstance(data, list): return [] peers = [] for row in data: symbol = str(row.get("symbol") or "").upper().strip() if symbol and symbol != ticker: peers.append(symbol) seen = set() deduped = [] for symbol in peers: if symbol not in seen: deduped.append(symbol) seen.add(symbol) return deduped[:8] @st.cache_data(ttl=3600) def get_ratios_for_tickers(tickers: list[str]) -> list[dict]: """Return merged TTM ratios/metrics rows for a list of tickers.""" results = [] for t in tickers: row = get_key_ratios(t) if row: row["symbol"] = t.upper() results.append(row) return results @st.cache_data(ttl=600) def get_company_news(ticker: str, limit: int = 20) -> list[dict]: """Return recent news articles for a ticker via legacy endpoint fallback.""" data = _get(LEGACY_BASE, "/stock_news", params={"tickers": ticker.upper(), "limit": limit}) if data and isinstance(data, list): return data return [] @st.cache_data(ttl=86400) def get_historical_ratios(ticker: str, limit: int = 10) -> list[dict]: """Annual historical valuation ratios (P/E, P/B, P/S, EV/EBITDA, etc.). Prism computes these from raw yfinance statements + price history so the methodology stays consistent with the rest of the app. """ rows = get_historical_ratios_yfinance(ticker.upper()) return rows[:limit] if rows else [] @st.cache_data(ttl=86400) def get_historical_key_metrics(ticker: str, limit: int = 10) -> list[dict]: """Annual historical key metrics (ROE, ROA, margins, debt/equity, etc.). Returned via the same self-computed historical dataset as get_historical_ratios(). """ rows = get_historical_ratios_yfinance(ticker.upper()) return rows[:limit] if rows else [] @st.cache_data(ttl=3600) def get_analyst_estimates(ticker: str) -> dict: """Return forward analyst estimates (annual only — quarterly requires premium).""" annual = _get(STABLE_BASE, "/analyst-estimates", params={"symbol": ticker.upper(), "limit": 5, "period": "annual"}) return { "annual": annual if isinstance(annual, list) else [], "quarterly": [], } @st.cache_data(ttl=3600) def get_insider_transactions(ticker: str, limit: int = 50) -> list[dict]: """Return recent insider buy/sell transactions.""" data = _get(LEGACY_BASE, "/insider-trading", params={"symbol": ticker.upper(), "limit": limit}) return data if isinstance(data, list) else [] @st.cache_data(ttl=3600) def get_sec_filings(ticker: str, limit: int = 30) -> list[dict]: """Return recent SEC filings (10-K, 10-Q, 8-K, etc.).""" data = _get(LEGACY_BASE, f"/sec_filings/{ticker.upper()}", params={"limit": limit}) return data if isinstance(data, list) else []