1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
"""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
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
def _apply_yfinance_ratio_fallbacks(ticker: str, merged: dict) -> dict:
info = get_company_info(ticker)
if not info:
return merged
fallback_map = {
"peRatioTTM": info.get("trailingPE"),
"priceToSalesRatioTTM": info.get("priceToSalesTrailing12Months"),
"priceToBookRatioTTM": info.get("priceToBook"),
"enterpriseValueMultipleTTM": info.get("enterpriseToEbitda"),
"evToEBITDATTM": info.get("enterpriseToEbitda"),
"evToSalesTTM": info.get("enterpriseToRevenue"),
"grossProfitMarginTTM": info.get("grossMargins"),
"operatingProfitMarginTTM": info.get("operatingMargins"),
"netProfitMarginTTM": info.get("profitMargins"),
"returnOnEquityTTM": info.get("returnOnEquity"),
"returnOnAssetsTTM": info.get("returnOnAssets"),
"debtToEquityRatioTTM": info.get("debtToEquity"),
"currentRatioTTM": info.get("currentRatio"),
"quickRatioTTM": info.get("quickRatio"),
"dividendYieldTTM": info.get("dividendYield"),
"dividendPayoutRatioTTM": info.get("payoutRatio"),
}
for key, value in fallback_map.items():
if merged.get(key) is None and value is not None:
merged[key] = value
return merged
@st.cache_data(ttl=3600)
def get_key_ratios(ticker: str) -> dict:
"""Return merged stable TTM ratios + key metrics for a ticker, with yfinance fallbacks."""
ticker = ticker.upper()
ratios = _get(STABLE_BASE, "/ratios-ttm", params={"symbol": ticker})
metrics = _get(STABLE_BASE, "/key-metrics-ttm", params={"symbol": ticker})
merged = {"symbol": ticker}
if isinstance(ratios, list) and ratios:
merged.update(ratios[0])
if isinstance(metrics, list) and metrics:
merged.update(metrics[0])
merged = _apply_yfinance_ratio_fallbacks(ticker, merged)
return merged if len(merged) > 1 else {}
@st.cache_data(ttl=21600)
def get_peers(ticker: str) -> list[str]:
"""Direct FMP peers endpoint was deprecated; peer discovery is handled in UI fallback logic."""
return []
@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 []
|
