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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
"""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
and float(payout_ratio_info) > 0
):
merged["dividendPayoutRatioTTM"] = payout_ratio_info
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 []
|
