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
|
"""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.
"""
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:
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"]
if merged.get("dividendPayoutRatioTTM") is None and info.get("payoutRatio") is not None:
merged["dividendPayoutRatioTTM"] = info["payoutRatio"]
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.).
Falls back to yfinance-computed ratios if FMP returns empty (e.g. rate limit)."""
data = _get(STABLE_BASE, "/ratios", params={"symbol": ticker.upper(), "limit": limit})
if isinstance(data, list) and data:
return data
return get_historical_ratios_yfinance(ticker.upper())
@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.).
Falls back to yfinance-computed metrics if FMP returns empty (e.g. rate limit)."""
data = _get(STABLE_BASE, "/key-metrics", params={"symbol": ticker.upper(), "limit": limit})
if isinstance(data, list) and data:
return data
# yfinance fallback already covers all key metrics — return empty to avoid duplication
# (get_historical_ratios will have already provided the full merged dataset)
return []
@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 []
|
