diff options
| -rw-r--r-- | components/valuation.py | 95 | ||||
| -rw-r--r-- | services/fmp_service.py | 81 |
2 files changed, 142 insertions, 34 deletions
diff --git a/components/valuation.py b/components/valuation.py index d7a84d4..170ae1f 100644 --- a/components/valuation.py +++ b/components/valuation.py @@ -27,6 +27,29 @@ FINANCIAL_INDUSTRY_KEYWORDS = ( "reit", ) +INDUSTRY_PEER_MAP = { + "consumer electronics": ["AAPL", "SONY", "DELL", "HPQ", "LOGI"], + "software - infrastructure": ["MSFT", "ORCL", "CRM", "NOW", "SNOW"], + "semiconductors": ["NVDA", "AMD", "AVGO", "QCOM", "INTC"], + "internet content & information": ["GOOGL", "META", "PINS", "SNAP", "RDDT"], + "banks - diversified": ["JPM", "BAC", "WFC", "C", "GS"], + "credit services": ["V", "MA", "AXP", "DFS", "COF"], + "insurance - diversified": ["BRK-B", "AIG", "ALL", "TRV", "CB"], + "reit - industrial": ["PLD", "PSA", "EXR", "COLD", "REXR"], +} +SECTOR_PEER_MAP = { + "Technology": ["AAPL", "MSFT", "NVDA", "ORCL", "ADBE"], + "Communication Services": ["GOOGL", "META", "NFLX", "TMUS", "DIS"], + "Consumer Cyclical": ["AMZN", "TSLA", "HD", "MCD", "NKE"], + "Consumer Defensive": ["WMT", "COST", "PG", "KO", "PEP"], + "Financial Services": ["JPM", "BAC", "WFC", "GS", "MS"], + "Healthcare": ["LLY", "UNH", "JNJ", "MRK", "PFE"], + "Industrials": ["GE", "CAT", "RTX", "UPS", "UNP"], + "Energy": ["XOM", "CVX", "COP", "SLB", "EOG"], + "Utilities": ["NEE", "DUK", "SO", "AEP", "XEL"], + "Real Estate": ["PLD", "AMT", "EQIX", "O", "SPG"], +} + def _is_financial_company(info: dict) -> bool: sector = str(info.get("sector") or "").strip() @@ -36,6 +59,26 @@ def _is_financial_company(info: dict) -> bool: return any(keyword in industry for keyword in FINANCIAL_INDUSTRY_KEYWORDS) +def _suggest_peer_tickers(ticker: str, info: dict) -> list[str]: + industry = str(info.get("industry") or "").strip().lower() + sector = str(info.get("sector") or "").strip() + + candidates = [] + if industry in INDUSTRY_PEER_MAP: + candidates.extend(INDUSTRY_PEER_MAP[industry]) + if not candidates and sector in SECTOR_PEER_MAP: + candidates.extend(SECTOR_PEER_MAP[sector]) + + candidates = [c.upper() for c in candidates if c.upper() != ticker.upper()] + seen = set() + deduped = [] + for c in candidates: + if c not in seen: + deduped.append(c) + seen.add(c) + return deduped[:8] + + def render_valuation(ticker: str): tab_ratios, tab_dcf, tab_comps, tab_analyst, tab_earnings = st.tabs([ "Key Ratios", "DCF Model", "Comps", "Analyst Targets", "Earnings History" @@ -85,7 +128,7 @@ def _render_ratios(ticker: str): ("Forward P/E", fmt_ratio(info.get("forwardPE")) if info.get("forwardPE") is not None else "—"), ("P/S (TTM)", r("priceToSalesRatioTTM", "priceToSalesTrailing12Months")), ("P/B", r("priceToBookRatioTTM", "priceToBook")), - ("EV/EBITDA", r("enterpriseValueMultipleTTM", "enterpriseToEbitda")), + ("EV/EBITDA", r("enterpriseValueMultipleTTM", "enterpriseToEbitda") if ratios.get("enterpriseValueMultipleTTM") is not None else r("evToEBITDATTM", "enterpriseToEbitda")), ("EV/Revenue", r("evToSalesTTM", "enterpriseToRevenue")), ]), ("Profitability", [ @@ -97,12 +140,12 @@ def _render_ratios(ticker: str): ("ROIC", r("returnOnInvestedCapitalTTM", fmt=fmt_pct)), ]), ("Leverage & Liquidity", [ - ("Debt/Equity", r("debtEquityRatioTTM", "debtToEquity")), + ("Debt/Equity", r("debtToEquityRatioTTM", "debtToEquity")), ("Current Ratio", r("currentRatioTTM", "currentRatio")), ("Quick Ratio", r("quickRatioTTM", "quickRatio")), - ("Interest Coverage", r("interestCoverageTTM")), + ("Interest Coverage", r("interestCoverageRatioTTM")), ("Dividend Yield", r("dividendYieldTTM", "dividendYield", fmt_pct)), - ("Payout Ratio", r("payoutRatioTTM", "payoutRatio", fmt_pct)), + ("Payout Ratio", r("dividendPayoutRatioTTM", "payoutRatio", fmt_pct)), ]), ] @@ -305,18 +348,40 @@ def _render_dcf(ticker: str): # ── Comps Table ────────────────────────────────────────────────────────────── def _render_comps(ticker: str): - peers = get_peers(ticker) - if not peers: - st.info("No comparable companies found. Check your FMP API key.") - return + info = get_company_info(ticker) + auto_peers = get_peers(ticker) + suggested_peers = _suggest_peer_tickers(ticker, info) + + default_peer_string = ", ".join(auto_peers or suggested_peers) + manual_peer_string = st.text_input( + "Peer tickers", + value=default_peer_string, + help="Edit the comparable-company set manually. Comma-separated tickers.", + key=f"peer_input_{ticker.upper()}", + ) - all_tickers = [ticker.upper()] + [p for p in peers[:9] if p != ticker.upper()] + if auto_peers: + st.caption("Using FMP-discovered peers.") + elif suggested_peers: + st.caption("Using Prism fallback peers based on sector/industry. Edit them if you want a tighter comp set.") + else: + st.caption("No automatic peer set found. Enter peer tickers manually to build a comps table.") + + manual_peers = [p.strip().upper() for p in manual_peer_string.split(",") if p.strip()] + peer_list = [] + seen = {ticker.upper()} + for peer in manual_peers: + if peer not in seen: + peer_list.append(peer) + seen.add(peer) + + all_tickers = [ticker.upper()] + peer_list[:9] with st.spinner("Loading comps..."): ratios_list = get_ratios_for_tickers(all_tickers) if not ratios_list: - st.info("Could not load ratios for peer companies.") + st.info("Could not load ratios for the selected peer companies.") return display_cols = { @@ -325,13 +390,19 @@ def _render_comps(ticker: str): "priceToSalesRatioTTM": "P/S", "priceToBookRatioTTM": "P/B", "enterpriseValueMultipleTTM": "EV/EBITDA", + "evToEBITDATTM": "EV/EBITDA", "netProfitMarginTTM": "Net Margin", "returnOnEquityTTM": "ROE", - "debtEquityRatioTTM": "D/E", + "debtToEquityRatioTTM": "D/E", } df = pd.DataFrame(ratios_list) - available = [c for c in display_cols if c in df.columns] + if "enterpriseValueMultipleTTM" not in df.columns and "evToEBITDATTM" in df.columns: + df["enterpriseValueMultipleTTM"] = df["evToEBITDATTM"] + if "debtToEquityRatioTTM" not in df.columns and "debtEquityRatioTTM" in df.columns: + df["debtToEquityRatioTTM"] = df["debtEquityRatioTTM"] + + available = [c for c in ["symbol", "peRatioTTM", "priceToSalesRatioTTM", "priceToBookRatioTTM", "enterpriseValueMultipleTTM", "netProfitMarginTTM", "returnOnEquityTTM", "debtToEquityRatioTTM"] if c in df.columns] df = df[available].rename(columns=display_cols) pct_cols = {"Net Margin", "ROE"} diff --git a/services/fmp_service.py b/services/fmp_service.py index bf31788..59d8215 100644 --- a/services/fmp_service.py +++ b/services/fmp_service.py @@ -1,65 +1,102 @@ -"""Financial Modeling Prep API — ratios, peers, company news.""" +"""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/api/v3" +BASE_URL = "https://financialmodelingprep.com" +STABLE_BASE = f"{BASE_URL}/stable" +LEGACY_BASE = f"{BASE_URL}/api/v3" def _api_key() -> str: - key = os.getenv("FMP_API_KEY", "") - return key + return os.getenv("FMP_API_KEY", "") -def _get(endpoint: str, params: dict = None) -> dict | list | None: +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 = 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 latest TTM key ratios.""" - data = _get(f"/ratios-ttm/{ticker.upper()}") - if data and isinstance(data, list) and len(data) > 0: - return data[0] - return {} + """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 list of comparable ticker symbols.""" - data = _get(f"/stock_peers", params={"symbol": ticker.upper()}) - if data and isinstance(data, list) and len(data) > 0: - return data[0].get("peersList", []) + """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 TTM ratios for a list of tickers (for comps table).""" + """Return merged TTM ratios/metrics rows for a list of tickers.""" results = [] for t in tickers: - data = _get(f"/ratios-ttm/{t}") - if data and isinstance(data, list) and len(data) > 0: - row = data[0] - row["symbol"] = t + 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.""" - data = _get("/stock_news", params={"tickers": ticker.upper(), "limit": limit}) + """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 [] |