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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
"""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]:
"""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.)."""
data = _get(LEGACY_BASE, f"/ratios/{ticker.upper()}", params={"limit": limit})
return data if isinstance(data, list) 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.)."""
data = _get(LEGACY_BASE, f"/key-metrics/{ticker.upper()}", params={"limit": limit})
return data if isinstance(data, list) else []
@st.cache_data(ttl=3600)
def get_analyst_estimates(ticker: str) -> dict:
"""Return annual and quarterly forward analyst estimates."""
annual = _get(LEGACY_BASE, f"/analyst-estimates/{ticker.upper()}", params={"limit": 5})
quarterly = _get(
LEGACY_BASE,
f"/analyst-estimates/{ticker.upper()}",
params={"limit": 10, "period": "quarter"},
)
return {
"annual": annual if isinstance(annual, list) else [],
"quarterly": quarterly if isinstance(quarterly, list) else [],
}
@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 []
|
