diff --git a/README.md b/README.md index 5d154f2..bf8fd88 100644 --- a/README.md +++ b/README.md @@ -134,10 +134,12 @@ Open the sidebar to set **Google API Key** and **SEC EDGAR Email**, then use the Updates are listed in **reverse chronological order (newest first)**. -| Date | Updates | -|------|---------| -| **2025-02-12** | **Hybrid architecture:** Send only Item 7 (MD&A) to Gemini; fetch financial metrics from **yfinance**. Add **HTML cleansing** (strip tags, collapse whitespace, remove page numbers). Change prompt to focus on management strategy, risks, and **sentiment** analysis. Add **find_toc.py** (SEC EDGAR HTML URL → Table of Contents extraction). Overhaul README for hybrid flow, Tech Stack, and Technical Challenge. | -| **2025-02-12** | Add **Update history (Changelog)** to README and include `find_toc.py` in Project Structure. | +| Date (UTC) | Updates | +|------------|---------| +| **2025-02-12 15:30** | **Dynamic Sector-Specific Analysis:** (1) **DuPont display:** Replace None/NaN with "N/A" in the table so missing rows (e.g. SBUX) do not break the layout. (2) **Sector & industry badge:** Auto-detect from `ticker.info` (sector/industry) and show Sector · Industry caption at top of Tab 1. (3) **Sector-specific metrics:** Technology (Rule of 40, FCF margin, R&D % of revenue), Retail/Consumer (inventory turnover, operating margin), Financials (ROE, ROA) — Tab 1 "Sector-specific metrics" block. (4) **Tab 2:** For Financial sector, add caption that FCF/EBITDA are less relevant and to refer to ROE/ROA in Tab 1. (5) **Gemini MD&A:** Add sector/industry args to `get_mda_comparative_insights`; prompt instructs extraction of industry-specific Non-GAAP KPIs (Same-Store Sales, ARR/NDR, DAU/MAU, etc.) in a markdown table. | +| **2025-02-12 11:00** | **Data robustness & TTM fallback:** (1) yfinance **KeyError/NaN handling:** `_get_row_series` with try/except for financial/balance-sheet row access; `_na(x)` for consistent "N/A" display. (2) **TTM fallback:** When annual `financials`/`balance_sheet` are missing or empty, use sum of first 4 quarters from `quarterly_financials` and latest quarter from `quarterly_balance_sheet` for DuPont/Altman. (3) **`get_sector_industry(ticker)`** added (cached sector/industry). (4) `get_dupont_altman_redflags_yoy` returns empty dict on exception; stronger column/date checks when building TTM. | +| **2025-02-12 09:00** | **Hybrid architecture:** Send only Item 7 (MD&A) to Gemini; fetch financial metrics from **yfinance**. Add **HTML cleansing** (strip tags, collapse whitespace, remove page numbers). Change prompt to focus on management strategy, risks, and **sentiment** analysis. Add **find_toc.py** (SEC EDGAR HTML URL → Table of Contents extraction). Overhaul README for hybrid flow, Tech Stack, and Technical Challenge. | +| **2025-02-12 08:45** | Add **Update history (Changelog)** to README and include `find_toc.py` in Project Structure. | | **2025-02-01** (approx.) | Selective extraction: send only Item 7 & 8 to the API. 429 handling: retry with 60s wait, "Analysis only" option, timeout/404/error messages. Switch to Google **Gemini** and add `GOOGLE_API_KEY` (.env and sidebar). CFA Investment Report, metrics table, two-step spinners. | | **2025-01-XX** (approx.) | Initial release: SEC EDGAR 10-K download, Item 7 & 8 extraction, LLM analysis, Streamlit UI. S&P 500 sample list (company name and ticker). | diff --git a/app.py b/app.py index 6840af8..026de72 100644 --- a/app.py +++ b/app.py @@ -17,6 +17,11 @@ import streamlit as st import pandas as pd from bs4 import BeautifulSoup +try: + import plotly.express as px +except ImportError: + px = None + try: from dotenv import load_dotenv load_dotenv() @@ -28,6 +33,23 @@ try: except ImportError: yf = None +# Company name → ticker for search/autocomplete (expand as needed) +COMPANY_LIST = [ + ("NVIDIA Corporation", "NVDA"), ("Apple Inc.", "AAPL"), ("Microsoft Corporation", "MSFT"), + ("Amazon.com Inc.", "AMZN"), ("Alphabet Inc.", "GOOGL"), ("Meta Platforms Inc.", "META"), + ("AMD", "AMD"), ("Intel Corporation", "INTC"), ("Qualcomm Inc.", "QCOM"), ("Tesla Inc.", "TSLA"), + ("Berkshire Hathaway", "BRK.B"), ("JPMorgan Chase", "JPM"), ("Visa Inc.", "V"), ("UnitedHealth", "UNH"), + ("Procter & Gamble", "PG"), ("Exxon Mobil", "XOM"), ("Johnson & Johnson", "JNJ"), ("Mastercard", "MA"), + ("Chevron", "CVX"), ("Home Depot", "HD"), ("Merck", "MRK"), ("AbbVie", "ABBV"), ("Costco", "COST"), + ("PepsiCo", "PEP"), ("Coca-Cola", "KO"), ("Pfizer", "PFE"), ("Walmart", "WMT"), ("Netflix", "NFLX"), + ("Adobe", "ADBE"), ("Salesforce", "CRM"), ("Comcast", "CMCSA"), ("Cisco", "CSCO"), ("Oracle", "ORCL"), + ("American Express", "AXP"), ("Bank of America", "BAC"), ("Wells Fargo", "WFC"), ("Verizon", "VZ"), + ("AT&T", "T"), ("Walt Disney", "DIS"), ("Nike", "NKE"), ("McDonald's", "MCD"), ("Starbucks", "SBUX"), + ("Goldman Sachs", "GS"), ("Morgan Stanley", "MS"), ("Target", "TGT"), ("Boeing", "BA"), ("IBM", "IBM"), +] +COMPANY_OPTIONS = [f"{t} - {n}" for n, t in COMPANY_LIST] +COMPANY_TICKER_MAP = {t: n for n, t in COMPANY_LIST} + def get_edgar_downloader(): from sec_edgar_downloader import Downloader @@ -156,6 +178,17 @@ def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path] return None +def find_all_10k_filing_dirs(download_root: Path, ticker: str) -> list: + """Return list of 10-K filing dirs sorted newest first (for multi-year comparison).""" + ticker_upper = ticker.upper() + for base in (download_root / "sec-edgar-filings", download_root): + path_10k = base / ticker_upper / "10-K" + if path_10k.exists(): + subdirs = sorted([d for d in path_10k.iterdir() if d.is_dir()], key=lambda x: x.name, reverse=True) + return subdirs + return [] + + def get_main_10k_text(filing_dir: Path) -> str: all_text = [] for ext in ("*.htm", "*.html", "*.txt"): @@ -200,6 +233,44 @@ def download_and_extract_item7_and_1a(ticker: str, email: str) -> tuple[str, str return full_text, item1a, item7 +def download_item7_latest_and_3y_ago(ticker: str, email: str) -> tuple[Optional[str], Optional[str], Optional[str], bool]: + """Download up to 5 10-Ks; extract Item 1A (latest only) and Item 7 from latest and from 3 years ago. + Returns (item1a_latest, item7_latest, item7_3y_ago, has_comparison). If < 4 filings, item7_3y_ago is None.""" + Downloader = get_edgar_downloader() + with tempfile.TemporaryDirectory() as tmpdir: + download_root = Path(tmpdir) + dl = Downloader("FQDC-10K-Analyzer", email, str(download_root)) + dl.get("10-K", ticker.upper(), limit=5, download_details=True) + filing_dirs = find_all_10k_filing_dirs(download_root, ticker) + if not filing_dirs: + raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.") + full_latest = get_main_10k_text(filing_dirs[0]) + if not full_latest: + raise ValueError("Could not extract text from the latest 10-K.") + item1a = find_item_section_generic( + full_latest, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80000 + ) + text_after_7 = full_latest[_find_section_start(full_latest, ITEM7_PATTERNS, 7):] if _find_section_start(full_latest, ITEM7_PATTERNS, 7) >= 0 else full_latest + item7_latest = find_item_section_generic( + text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000 + ) + if not item7_latest and text_after_7: + item7_latest = smart_chunk(text_after_7[:120000], max_chars=20000) + item7_3y_ago = None + has_comparison = False + if len(filing_dirs) >= 4: + full_3y = get_main_10k_text(filing_dirs[3]) + if full_3y: + text_3y = full_3y[_find_section_start(full_3y, ITEM7_PATTERNS, 7):] if _find_section_start(full_3y, ITEM7_PATTERNS, 7) >= 0 else full_3y + item7_3y_ago = find_item_section_generic( + text_3y, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000 + ) + if not item7_3y_ago and text_3y: + item7_3y_ago = smart_chunk(text_3y[:120000], max_chars=20000) + has_comparison = bool(item7_3y_ago) + return item1a or "", item7_latest or "", item7_3y_ago, has_comparison + + # ---------- Gemini (qualitative only) ---------- GEMINI_MODEL = "gemini-2.0-flash" RATE_LIMIT_WAIT_SEC = 60 @@ -270,10 +341,159 @@ Use clear headings. Do not invent figures. Keep the response focused and under 8 return response.text.strip() -# ---------- yfinance: DCF inputs ---------- +def get_mda_comparative_insights( + api_key: str, + item1a_text: str, + item7_latest: str, + item7_3y_ago: Optional[str], + ticker: str, + sector: Optional[str] = None, + industry: Optional[str] = None, +) -> str: + """Comparative analysis: if item7_3y_ago provided, compare MD&As over 3 years; else single-year. Sector-aware: extract industry-specific Non-GAAP KPIs.""" + model = get_gemini_model(api_key) + sector_label = (sector or "N/A").strip() + industry_label = (industry or "N/A").strip() + kpi_instruction = ( + f" Given that this company is in the **{sector_label}** sector" + + (f" (industry: {industry_label})" if industry_label != "N/A" else "") + + ", meticulously scan the MD&A to find and extract **industry-specific Non-GAAP KPIs** " + "(e.g. Same-Store Sales Growth for Retail, ARR/NDR for Software, DAU/MAU for Tech). Present these hidden KPIs in a **clean markdown table** with columns such as KPI name, value, and period if stated." + ) + if not item7_3y_ago or not item7_3y_ago.strip(): + combined = [] + if item1a_text: + combined.append(clean_text_for_llm(item1a_text)) + if item7_latest: + combined.append(clean_text_for_llm(item7_latest)) + combined_text = "\n\n---\n\n".join(combined) + combined_text = smart_chunk(combined_text, max_chars=22000) + user_prompt = f"""You are a senior equity analyst. Use British English. +The text below is from the latest 10-K for {ticker}: **Item 1A (Risk Factors)** and **Item 7 (MD&A)**. +Provide a concise report: 1) Management's Tone (Sentiment), 2) Key Strategic Shifts, 3) Major Hidden Risks.{kpi_instruction} +Use clear headings. Under 800 words.""" + full_content = f"""--- 10-K Excerpt ---\n\n{combined_text}\n\n---\n\n{user_prompt}""" + else: + latest_clean = smart_chunk(clean_text_for_llm(item7_latest), max_chars=12000) + past_clean = smart_chunk(clean_text_for_llm(item7_3y_ago), max_chars=12000) + user_prompt = f"""You are a senior equity analyst. Use British English. +Below are **Item 7 (Management's Discussion and Analysis)** from the 10-K for {ticker}: **LATEST YEAR** and **THREE YEARS AGO**. Perform a **Comparative Analysis**. + +1. **Core strategy**: What has changed in the company's stated strategy, priorities, or capital allocation between then and now? +2. **Emerging risks**: What new risks appear in the latest MD&A that were absent or less prominent 3 years ago? +3. **Management's tone**: How has the overall tone (confidence, caution, optimism) shifted? Quote 1–2 phrases from each period if relevant. +4. **Industry-specific KPIs**:{kpi_instruction} + +Use clear headings. Do not invent figures. Keep the response focused and under 900 words.""" + full_content = f"""--- MD&A LATEST YEAR ---\n\n{latest_clean}\n\n--- MD&A THREE YEARS AGO ---\n\n{past_clean}\n\n---\n\n{user_prompt}""" + try: + response = _generate_with_retry( + model, full_content, {"temperature": 0.3, "max_output_tokens": 4096} + ) + except Exception as api_err: + if _is_rate_limit_error(api_err): + raise RuntimeError("Rate limit exceeded. Please try again in a few minutes.") from api_err + raise + if not response or not response.text: + return "No analysis generated." + return response.text.strip() + + +# ---------- yfinance: raw statements & FCF = OCF - CapEx ---------- +def _safe_float(x) -> Optional[float]: + if x is None or (isinstance(x, float) and (x != x or pd.isna(x))): + return None + try: + return float(x) + except (TypeError, ValueError): + return None + + +def _get_row_series(df: pd.DataFrame, *names: str) -> Optional[pd.Series]: + if df is None or df.empty: + return None + for name in names: + try: + if name in df.index: + return df.loc[name].copy() + except (KeyError, TypeError): + continue + return None + + +@st.cache_data(ttl=300) +def get_sector_industry(ticker: str) -> dict: + """Return sector and industry from yfinance. Fallback to N/A.""" + if not yf: + return {"sector": "N/A", "industry": "N/A"} + try: + t = yf.Ticker(ticker.upper()) + info = t.info or {} + sector = (info.get("sector") or info.get("sectorDisp") or "N/A").strip() or "N/A" + industry = (info.get("industry") or info.get("industryDisp") or "N/A").strip() or "N/A" + return {"sector": sector, "industry": industry} + except Exception: + return {"sector": "N/A", "industry": "N/A"} + + +@st.cache_data(ttl=300) +def get_5yr_financial_trend(ticker: str) -> pd.DataFrame: + """Extract up to 5 years: Revenue, Net Income, Operating Margin, FCF (OCF - CapEx). Handles missing years.""" + if not yf: + return pd.DataFrame() + try: + t = yf.Ticker(ticker.upper()) + financials = t.financials # annual + cashflow = t.cashflow + if financials is None or financials.empty or cashflow is None or cashflow.empty: + return pd.DataFrame() + dates = sorted(financials.columns.tolist(), reverse=True)[:5] + ocf = _get_row_series(cashflow, "Operating Cash Flow", "Cash From Operating Activities", "Cash From Operations") + capx = _get_row_series(cashflow, "Capital Expenditure", "Capital Expenditures", "Purchase Of Property Plant And Equipment") + revenue = _get_row_series(financials, "Total Revenue", "Revenue", "Net Revenue") + ni = _get_row_series(financials, "Net Income", "Net Income Common Stockholders") + op_income = _get_row_series(financials, "Operating Income", "EBIT") + rows = [] + cashflow_cols = list(cashflow.columns) if cashflow is not None else [] + for d in dates: + yr = d.year if hasattr(d, "year") else int(str(d)[:4]) + rev = _safe_float(revenue.get(d)) if revenue is not None and d in revenue.index else None + net_i = _safe_float(ni.get(d)) if ni is not None and d in ni.index else None + op_i = _safe_float(op_income.get(d)) if op_income is not None and d in op_income.index else None + oper_margin = (op_i / rev * 100) if (op_i is not None and rev and rev != 0) else ((net_i / rev * 100) if (net_i is not None and rev and rev != 0) else None) + ocf_val = _safe_float(ocf.get(d)) if ocf is not None and d in ocf.index else None + if ocf_val is None and ocf is not None and cashflow_cols: + for c in cashflow_cols: + if (getattr(c, "year", None) or int(str(c)[:4])) == yr: + ocf_val = _safe_float(ocf.get(c)) + break + capx_val = _safe_float(capx.get(d)) if capx is not None and d in capx.index else None + if capx_val is None and capx is not None and cashflow_cols: + for c in cashflow_cols: + if (getattr(c, "year", None) or int(str(c)[:4])) == yr: + capx_val = _safe_float(capx.get(c)) + break + if ocf_val is not None and capx_val is not None: + fcf = ocf_val - capx_val + elif ocf_val is not None: + fcf = ocf_val + else: + fcf = None + rows.append({ + "Year": yr, + "Revenue": rev, + "Net Income": net_i, + "Operating Margin %": round(oper_margin, 2) if oper_margin is not None else None, + "FCF": fcf, + }) + return pd.DataFrame(rows) + except Exception: + return pd.DataFrame() + + @st.cache_data(ttl=300) def get_dcf_inputs(ticker: str) -> dict: - """Fetch FCF, Total Debt, Cash, Shares Outstanding for DCF. Returns dict or empty on failure.""" + """FCF = OCF - CapEx from cashflow; baseline = latest year. Debt, Cash, Shares from balance sheet/info.""" if not yf: return {} try: @@ -283,19 +503,16 @@ def get_dcf_inputs(ticker: str) -> dict: balance = t.balance_sheet if cashflow is None or cashflow.empty: return {} - fcf_row = None - for name in ("Free Cash Flow", "Cash From Operations"): - if name in cashflow.index: - fcf_row = cashflow.loc[name] - break - if fcf_row is None and len(cashflow.index) > 0: - fcf_row = cashflow.iloc[0] - latest_fcf = None - if fcf_row is not None and len(fcf_row) > 0: - try: - latest_fcf = float(fcf_row.iloc[0]) - except (TypeError, ValueError): - pass + ocf = _get_row_series(cashflow, "Operating Cash Flow", "Cash From Operating Activities", "Cash From Operations") + capx = _get_row_series(cashflow, "Capital Expenditure", "Capital Expenditures", "Purchase Of Property Plant And Equipment") + if ocf is None or len(ocf) == 0: + return {} + latest_date = ocf.index[0] + ocf_val = _safe_float(ocf.iloc[0]) + capx_val = _safe_float(capx.get(latest_date)) if capx is not None and latest_date in capx.index else (_safe_float(capx.iloc[0]) if capx is not None and len(capx) > 0 else None) + if capx_val is None: + capx_val = 0.0 + latest_fcf = (ocf_val - capx_val) if ocf_val is not None else None if latest_fcf is not None and (latest_fcf != latest_fcf or latest_fcf <= 0): latest_fcf = None total_debt = info.get("Total Debt") @@ -303,15 +520,9 @@ def get_dcf_inputs(ticker: str) -> dict: shares = info.get("Shares Outstanding") or info.get("Float Shares") if balance is not None and not balance.empty: if total_debt is None and "Total Debt" in balance.index: - try: - total_debt = float(balance.loc["Total Debt"].iloc[0]) - except (TypeError, ValueError, KeyError): - pass + total_debt = _safe_float(balance.loc["Total Debt"].iloc[0]) if cash is None and "Cash And Cash Equivalents" in balance.index: - try: - cash = float(balance.loc["Cash And Cash Equivalents"].iloc[0]) - except (TypeError, ValueError, KeyError): - pass + cash = _safe_float(balance.loc["Cash And Cash Equivalents"].iloc[0]) return { "fcf": latest_fcf, "total_debt": total_debt if total_debt is not None else 0, @@ -322,15 +533,15 @@ def get_dcf_inputs(ticker: str) -> dict: return {} -def dcf_intrinsic_value(fcf: float, wacc: float, terminal_growth: float, revenue_growth: float, years: int = 10) -> float: - """DCF: project FCF with revenue_growth, terminal value with terminal_growth, discount at WACC. Returns enterprise value.""" +def dcf_intrinsic_value(fcf: float, wacc: float, terminal_growth: float, fcf_growth: float, years: int = 5) -> float: + """5-year DCF: project FCF with fcf_growth, then terminal value; discount at WACC. Returns enterprise value.""" if fcf <= 0 or wacc <= terminal_growth: return 0.0 pv = 0.0 fcft = fcf for t in range(1, years + 1): pv += fcft / ((1 + wacc) ** t) - fcft *= (1 + revenue_growth) + fcft *= (1 + fcf_growth) terminal_fcf = fcft tv = terminal_fcf * (1 + terminal_growth) / (wacc - terminal_growth) pv += tv / ((1 + wacc) ** years) @@ -369,6 +580,208 @@ def get_comps_data(tickers: tuple) -> pd.DataFrame: return pd.DataFrame(rows) +# ---------- DuPont, Altman Z, Red Flags, YoY (2–3 years) ---------- +def _na(x): + """Return N/A for None/NaN, else value (for display).""" + if x is None or (isinstance(x, float) and (pd.isna(x) or x != x)): + return "N/A" + return x + + +@st.cache_data(ttl=300) +def get_dupont_altman_redflags_yoy(ticker: str) -> dict: + """Returns DuPont (3-step ROE), Altman Z-Score, red flags, YoY. Uses TTM if annual missing. Handles KeyError/NaN.""" + if not yf: + return {} + try: + t = yf.Ticker(ticker.upper()) + info = t.info or {} + fin = t.financials + bal = t.balance_sheet + if fin is None or fin.empty: + qfin = getattr(t, "quarterly_financials", None) + if qfin is not None and not qfin.empty and qfin.shape[1] >= 1: + fin = qfin.iloc[:, :4].sum(axis=1).to_frame("TTM") + else: + return {} + if bal is None or bal.empty: + qbal = getattr(t, "quarterly_balance_sheet", None) + if qbal is not None and not qbal.empty: + bal = qbal.iloc[:, :1] + else: + return {} + dates = sorted(fin.columns.tolist(), reverse=True)[:3] + if not dates: + return {} + rev = _get_row_series(fin, "Total Revenue", "Revenue", "Net Revenue") + ni = _get_row_series(fin, "Net Income", "Net Income Common Stockholders") + ebit = _get_row_series(fin, "Operating Income", "EBIT") + gross = _get_row_series(fin, "Gross Profit") + interest = _get_row_series(fin, "Interest Expense", "Interest Expense Net") + total_assets = _get_row_series(bal, "Total Assets") + total_equity = _get_row_series(bal, "Total Stockholder Equity", "Stockholders Equity", "Total Equity Gross Minority Interest") + current_assets = _get_row_series(bal, "Current Assets") + current_liab = _get_row_series(bal, "Current Liabilities") + retained = _get_row_series(bal, "Retained Earnings") + total_liab = _get_row_series(bal, "Total Liabilities") + market_cap = info.get("marketCap") or info.get("Market Cap") + def _v(s, d): + if s is None or d not in s.index: + return None + return _safe_float(s.get(d)) + rows = [] + for d in dates: + yr = d.year if hasattr(d, "year") else int(str(d)[:4]) + r = _v(rev, d) + net_i = _v(ni, d) + ta = _v(total_assets, d) + te = _v(total_equity, d) + if ta and ta > 0 and te and te > 0 and r and r != 0: + npm = (net_i / r * 100) if net_i is not None else None + at = r / ta if r and ta else None + em = ta / te if ta and te else None + roe = (net_i / te * 100) if (net_i and te) else (npm * at * em / 100 if (npm and at and em) else None) + else: + npm = at = em = roe = None + gross_p = _v(gross, d) + gross_margin = (gross_p / r * 100) if (gross_p and r and r != 0) else None + op_inc = _v(ebit, d) + op_margin = (op_inc / r * 100) if (op_inc and r and r != 0) else None + ca = _v(current_assets, d) + cl = _v(current_liab, d) + current_ratio = (ca / cl) if (ca and cl and cl != 0) else None + int_exp = _v(interest, d) + interest_cov = (op_inc / int_exp) if (op_inc and int_exp and int_exp != 0) else None + rows.append({ + "Year": yr, + "Revenue": r, "Net Income": net_i, + "NPM %": round(npm, 2) if npm is not None else None, + "Asset Turnover": round(at, 4) if at is not None else None, + "Equity Mult.": round(em, 2) if em is not None else None, + "ROE %": round(roe, 2) if roe is not None else None, + "Gross Margin %": round(gross_margin, 2) if gross_margin is not None else None, + "Operating Margin %": round(op_margin, 2) if op_margin is not None else None, + "Current Ratio": round(current_ratio, 2) if current_ratio is not None else None, + "Interest Coverage": round(interest_cov, 2) if interest_cov is not None else None, + }) + dupont_df = pd.DataFrame(rows) + yoy = [] + if len(dupont_df) >= 2: + for col in ["NPM %", "ROE %", "Gross Margin %", "Operating Margin %", "Current Ratio", "Interest Coverage"]: + if col not in dupont_df.columns: + continue + cur = dupont_df[col].iloc[0] + prev = dupont_df[col].iloc[1] + if cur is not None and prev is not None and prev != 0: + if "Margin" in col or "NPM" in col or "ROE" in col: + chg_bps = (cur - prev) * 100 # bps for % + yoy.append({"Ratio": col, "Latest": cur, "Prior": prev, "YoY (bps)": round(chg_bps, 0), "Comment": f"{'Improved' if chg_bps > 0 else 'Declined'} by {abs(round(chg_bps))} bps YoY"}) + else: + pct = (cur - prev) / abs(prev) * 100 + yoy.append({"Ratio": col, "Latest": cur, "Prior": prev, "YoY %": round(pct, 1), "Comment": f"{'Up' if pct > 0 else 'Down'} {abs(round(pct, 1))}% YoY"}) + latest_bal_d = bal.columns[0] + wc = (_v(current_assets, latest_bal_d) or 0) - (_v(current_liab, latest_bal_d) or 0) + ta_l = _v(total_assets, latest_bal_d) + re_l = _v(retained, latest_bal_d) + tl_l = _v(total_liab, latest_bal_d) + ebit_l = _v(ebit, fin.columns[0]) + sales_l = _v(rev, fin.columns[0]) + altman_z = None + if ta_l and ta_l > 0 and market_cap is not None and tl_l and tl_l != 0 and sales_l: + a = wc / ta_l + b = (re_l or 0) / ta_l + c = (ebit_l or 0) / ta_l + d = market_cap / tl_l + e = sales_l / ta_l + altman_z = 1.2 * a + 1.4 * b + 3.3 * c + 0.6 * d + 1.0 * e + red_flags = [] + if len(dupont_df) > 0: + row0 = dupont_df.iloc[0] + cr = row0.get("Current Ratio") + if cr is not None and cr < 1.0: + red_flags.append({"metric": "Current Ratio", "value": cr, "threshold": 1.0, "flag": "WARNING", "comment": "Current assets do not cover current liabilities; liquidity risk."}) + ic = row0.get("Interest Coverage") + if ic is not None and ic < 1.5: + red_flags.append({"metric": "Interest Coverage", "value": ic, "threshold": 1.5, "flag": "WARNING", "comment": "EBIT barely covers interest; default risk."}) + return { + "dupont": dupont_df, + "yoy": yoy, + "altman_z": round(altman_z, 2) if altman_z is not None else None, + "red_flags": red_flags, + } + except (KeyError, TypeError, ZeroDivisionError, IndexError) as e: + return {} + except Exception: + return {} + + +@st.cache_data(ttl=300) +def get_sector_specific_metrics(ticker: str, sector: str) -> dict: + """Technology: Rule of 40, R&D % revenue. Retail/Consumer: Inventory Turnover, Operating Margin. Financials: ROE, ROA.""" + if not yf: + return {} + try: + t = yf.Ticker(ticker.upper()) + info = t.info or {} + fin = t.financials + bal = t.balance_sheet + if fin is None or fin.empty: + fin = getattr(t, "quarterly_financials", None) + if fin is not None and not fin.empty: + fin = fin.iloc[:, :4].sum(axis=1).to_frame() + if bal is None or bal.empty: + bal = getattr(t, "quarterly_balance_sheet", None) + out = {} + sector_lower = (sector or "").lower() + if "technology" in sector_lower or "software" in sector_lower or "tech" in sector_lower: + rev = _get_row_series(fin, "Total Revenue", "Revenue", "Net Revenue") + ocf = _get_row_series(t.cashflow or getattr(t, "quarterly_cashflow", None), "Operating Cash Flow", "Cash From Operating Activities") + capx = _get_row_series(t.cashflow or getattr(t, "quarterly_cashflow", None), "Capital Expenditure", "Capital Expenditures") + rd = _get_row_series(fin, "Research And Development", "Research And Development Expense") + if rev is not None and len(rev) > 0: + r0 = _safe_float(rev.iloc[0]) + if ocf is not None and len(ocf) > 0 and capx is not None and len(capx) > 0: + fcf = _safe_float(ocf.iloc[0]) - _safe_float(capx.iloc[0]) + out["FCF Margin %"] = round(fcf / r0 * 100, 2) if r0 and fcf is not None else None + if rd is not None and len(rd) > 0: + out["R&D % of Revenue"] = round(_safe_float(rd.iloc[0]) / r0 * 100, 2) if r0 else None + rev_growth = None + if rev is not None and len(rev) >= 2: + cur, prev = _safe_float(rev.iloc[0]), _safe_float(rev.iloc[1]) + if prev and prev != 0: + rev_growth = (cur - prev) / prev * 100 + if rev_growth is not None and "FCF Margin %" in out and out["FCF Margin %"] is not None: + out["Rule of 40 (Rev Growth + FCF Margin)"] = round(rev_growth + out["FCF Margin %"], 1) + if "consumer" in sector_lower or "retail" in sector_lower or "cyclical" in sector_lower: + inv = _get_row_series(bal, "Inventory", "Total Inventory") + cogs = _get_row_series(fin, "Cost Of Revenue", "Cost Of Goods Sold", "Cost of Goods Sold") + rev = _get_row_series(fin, "Total Revenue", "Revenue", "Net Revenue") + op_inc = _get_row_series(fin, "Operating Income", "EBIT") + if inv is not None and len(inv) > 0 and cogs is not None and len(cogs) > 0: + inv0 = _safe_float(inv.iloc[0]) + cogs0 = _safe_float(cogs.iloc[0]) + out["Inventory Turnover"] = round(cogs0 / inv0, 2) if inv0 else None + if rev is not None and len(rev) > 0 and op_inc is not None and len(op_inc) > 0: + r0 = _safe_float(rev.iloc[0]) + op0 = _safe_float(op_inc.iloc[0]) + out["Operating Margin %"] = round(op0 / r0 * 100, 2) if r0 else None + if "financial" in sector_lower or "bank" in sector_lower or "insurance" in sector_lower: + ni = _get_row_series(fin, "Net Income", "Net Income Common Stockholders") + te = _get_row_series(bal, "Total Stockholder Equity", "Stockholders Equity", "Total Equity Gross Minority Interest") + ta = _get_row_series(bal, "Total Assets") + if ni is not None and te is not None and len(ni) > 0 and len(te) > 0: + te0 = _safe_float(te.iloc[0]) + ni0 = _safe_float(ni.iloc[0]) + out["ROE %"] = round(ni0 / te0 * 100, 2) if te0 else None + if ni is not None and ta is not None and len(ni) > 0 and len(ta) > 0: + ta0 = _safe_float(ta.iloc[0]) + ni0 = _safe_float(ni.iloc[0]) + out["ROA %"] = round(ni0 / ta0 * 100, 2) if ta0 else None + return out + except Exception: + return {} + + # ---------- Streamlit UI ---------- st.set_page_config(page_title="Financial Analysis Dashboard", layout="wide", initial_sidebar_state="expanded") @@ -398,21 +811,79 @@ with st.sidebar: value=os.environ.get("SEC_EDGAR_EMAIL", ""), help="Required for 10-K download.", ) - ticker = st.text_input("Primary Ticker", value="NVDA", max_chars=10).strip().upper() + st.markdown("**Ticker / Company search**") + search_term = st.text_input("Type ticker or company name", value="", key="ticker_search", placeholder="e.g. NVDA or NVIDIA") + search_upper = (search_term or "").strip().upper() + search_lower = (search_term or "").strip().lower() + filtered = [o for o in COMPANY_OPTIONS if search_upper in o.split(" - ")[0] or search_lower in o.lower()] if (search_upper or search_lower) else COMPANY_OPTIONS + default_idx = 0 + current_ticker = st.session_state.get("ticker", "NVDA") + for i, o in enumerate(filtered): + if o.startswith(current_ticker + " - "): + default_idx = i + break + selected = st.selectbox("Select company (ticker - name)", filtered, index=min(default_idx, len(filtered) - 1), key="company_select") + ticker_from_select = selected.split(" - ")[0].strip() if selected else "" + manual = st.text_input("Or enter ticker manually", value="", max_chars=10, key="manual_ticker").strip().upper() + ticker = manual if manual else ticker_from_select + if not ticker: + ticker = "NVDA" st.session_state["google_api_key"] = google_api_key st.session_state["sec_email"] = sec_email st.session_state["ticker"] = ticker - -if not ticker: - ticker = st.session_state.get("ticker", "NVDA") or "NVDA" + st.caption("Example: type _NVDA_ or _NVIDIA_ then select from list.") tab1, tab2, tab3 = st.tabs(["10-K & MD&A Insights", "3-Scenario DCF Valuation", "Industry Analysis & Comps"]) -# ----- Tab 1: 10-K & MD&A Insights ----- +# ----- Tab 1: Qualitative (MD&A) + Quantitative (DuPont, Altman Z, Red Flags, YoY) ----- with tab1: - st.subheader("10-K & MD&A Insights (Qualitative)") - st.markdown("Extract **Item 1A (Risk Factors)** and **Item 7 (MD&A)** from the latest 10-K. Gemini analyses: **Management's Tone**, **Strategic Shifts**, **Hidden Risks**.") - if st.button("Run 10-K Analysis", key="run_10k"): + st.subheader("10-K & MD&A Insights — Qualitative and Quantitative") + if ticker: + si = get_sector_industry(ticker) + sector, industry = si.get("sector", "N/A"), si.get("industry", "N/A") + st.caption(f"Sector: **{sector}** · Industry: **{industry}**") + st.markdown("**Quantitative** metrics below (DuPont ROE, Altman Z-Score, Red Flags, YoY trends). **Qualitative** analysis: run comparative MD&A with the button.") + if ticker: + q = get_dupont_altman_redflags_yoy(ticker) + if q: + st.markdown("#### Quantitative health (2–3 years)") + dupont_df = q.get("dupont") + if dupont_df is not None and not dupont_df.empty: + st.markdown("**3-Step DuPont (ROE = Net Profit Margin × Asset Turnover × Equity Multiplier)**") + display_cols = [c for c in ["Year", "NPM %", "Asset Turnover", "Equity Mult.", "ROE %"] if c in dupont_df.columns] + display_dupont = dupont_df[display_cols].copy().rename(columns={"Equity Mult.": "Equity Mult"}) + for col in display_dupont.columns: + display_dupont[col] = display_dupont[col].apply(lambda x: "N/A" if (x is None or (isinstance(x, float) and pd.isna(x))) else x) + st.dataframe(display_dupont, use_container_width=True, hide_index=True) + az = q.get("altman_z") + if az is not None: + st.metric("Altman Z-Score (distress / bankruptcy risk)", f"{az}", "Safe zone > 2.99; Grey 1.81–2.99; Distress < 1.81" if az < 1.81 else ("Grey zone" if az < 2.99 else "Safe zone")) + red_flags = q.get("red_flags") or [] + if red_flags: + st.markdown("**Red flags**") + for rf in red_flags: + st.warning(f"**{rf.get('flag', 'WARNING')}** — {rf.get('metric')}: {rf.get('value')} (threshold: {rf.get('threshold')}). {rf.get('comment', '')}") + elif dupont_df is not None and not dupont_df.empty: + st.success("No red flags triggered (Current Ratio ≥ 1.0, Interest Coverage ≥ 1.5).") + sector_metrics = get_sector_specific_metrics(ticker, sector) if ticker else {} + if sector_metrics: + st.markdown("**Sector-specific metrics**") + cols = st.columns(min(len(sector_metrics), 4)) + for i, (k, v) in enumerate(sector_metrics.items()): + with cols[i % len(cols)]: + disp = f"{v}" if v is not None else "N/A" + st.metric(k, disp, None) + yoy_list = q.get("yoy") or [] + if yoy_list: + st.markdown("**YoY ratio changes**") + for item in yoy_list: + st.caption(f"**{item.get('Ratio')}**: {item.get('Comment', '')}") + else: + st.info("Quantitative data not available for this ticker.") + st.markdown("---") + st.markdown("#### Qualitative: MD&A comparative analysis") + st.markdown("Download **latest 10-K** and **10-K from 3 years ago**. Extract **Item 7 (MD&A)** from both. Gemini: strategy shifts, emerging risks, management tone.") + if st.button("Run 10-K Comparative Analysis", key="run_10k"): if not ticker: st.error("Enter a ticker in the sidebar.") elif not st.session_state.get("google_api_key"): @@ -421,16 +892,27 @@ with tab1: st.error("Enter your SEC EDGAR email in the sidebar.") else: try: - with st.spinner("Downloading 10-K and extracting Item 1A & Item 7..."): - full_text, item1a, item7 = download_and_extract_item7_and_1a(ticker, st.session_state["sec_email"]) - with st.spinner("Running Gemini analysis (tone, strategy, risks)..."): - analysis = get_mda_insights( - st.session_state["google_api_key"], item1a, item7, ticker + with st.spinner("Downloading 10-Ks (latest + 3 years ago) and extracting Item 7..."): + item1a, item7_latest, item7_3y_ago, has_comparison = download_item7_latest_and_3y_ago( + ticker, st.session_state["sec_email"] + ) + if not has_comparison: + st.info("Only one or fewer 10-K filings available; showing single-year analysis.") + with st.spinner("Running Gemini (comparative or single-year analysis)..."): + si = get_sector_industry(ticker) + analysis = get_mda_comparative_insights( + st.session_state["google_api_key"], + item1a, + item7_latest, + item7_3y_ago, + ticker, + sector=si.get("sector"), + industry=si.get("industry"), ) st.success("Analysis complete.") st.markdown(analysis) - with st.expander("View raw excerpt (Item 1A + Item 7)"): - excerpt = (item1a or "") + "\n\n---\n\n" + (item7 or "") + with st.expander("View raw excerpt (Item 1A + Item 7 latest)"): + excerpt = (item1a or "") + "\n\n---\n\n" + (item7_latest or "") st.text(excerpt[:12000] + ("..." if len(excerpt) > 12000 else "")) except FileNotFoundError as e: st.error(str(e)) @@ -443,10 +925,53 @@ with tab1: with st.expander("Error details"): st.code(repr(e), language="text") -# ----- Tab 2: 3-Scenario DCF ----- +# ----- Tab 2: 5-Year Trend + 3-Scenario DCF ----- with tab2: - st.subheader("3-Scenario DCF Valuation (Quantitative)") - st.markdown("Uses **yfinance** for FCF, Debt, Cash, Shares. No Gemini. Adjust assumptions with sliders.") + st.subheader("5-Year Financial Trend & DCF Valuation") + if ticker: + si_t2 = get_sector_industry(ticker) + sector_t2 = (si_t2.get("sector") or "").lower() + is_financial = "financial" in sector_t2 or "bank" in sector_t2 or "insurance" in sector_t2 + else: + is_financial = False + df_trend = get_5yr_financial_trend(ticker) if ticker else pd.DataFrame() + if not df_trend.empty and len(df_trend) >= 1: + st.markdown("#### Key metrics (YoY % change)") + latest = df_trend.iloc[0] + prev = df_trend.iloc[1] if len(df_trend) >= 2 else None + def _yoy_pct(cur, prev_val): + if prev_val is None or cur is None or prev_val == 0: + return None + return (cur - prev_val) / abs(prev_val) * 100 + rev_yoy = _yoy_pct(latest.get("Revenue"), prev.get("Revenue") if prev is not None else None) + ni_yoy = _yoy_pct(latest.get("Net Income"), prev.get("Net Income") if prev is not None else None) + om_prev = prev.get("Operating Margin %") if prev is not None else None + om_cur = latest.get("Operating Margin %") + om_yoy = (om_cur - om_prev) if (om_cur is not None and om_prev is not None) else None + fcf_yoy = _yoy_pct(latest.get("FCF"), prev.get("FCF") if prev is not None else None) + m1, m2, m3, m4 = st.columns(4) + rev_val = latest.get("Revenue") + m1.metric("Revenue (latest yr)", f"${rev_val/1e9:.2f}B" if rev_val and rev_val >= 1e9 else (f"${rev_val/1e6:.0f}M" if rev_val else "—"), f"{rev_yoy:+.1f}% YoY" if rev_yoy is not None else None) + ni_val = latest.get("Net Income") + m2.metric("Net Income", f"${ni_val/1e9:.2f}B" if ni_val and abs(ni_val) >= 1e9 else (f"${ni_val/1e6:.0f}M" if ni_val is not None else "—"), f"{ni_yoy:+.1f}% YoY" if ni_yoy is not None else None) + om_val = latest.get("Operating Margin %") + m3.metric("Operating Margin %", f"{om_val:.1f}%" if om_val is not None else "—", f"{om_yoy:+.1f}pp YoY" if om_yoy is not None else None) + fcf_val = latest.get("FCF") + m4.metric("FCF", f"${fcf_val/1e9:.2f}B" if fcf_val and abs(fcf_val) >= 1e9 else (f"${fcf_val/1e6:.0f}M" if fcf_val is not None else "—"), f"{fcf_yoy:+.1f}% YoY" if fcf_yoy is not None else None) + st.caption("FCF = Operating Cash Flow − Capital Expenditure." + (" For Financials, FCF/EBITDA are less relevant; see ROE/ROA in Tab 1 sector-specific metrics." if is_financial else "")) + if len(df_trend) >= 2 and px is not None: + st.markdown("#### 5-year trend: Revenue & FCF") + df_plot = df_trend.copy() + df_plot["Revenue_M"] = (df_plot["Revenue"] / 1e6).round(1) + df_plot["FCF_M"] = (df_plot["FCF"] / 1e6).round(1) + fig = px.line(df_plot, x="Year", y=["Revenue_M", "FCF_M"], title="Revenue & Free Cash Flow ($M)") + fig.update_layout(yaxis_title="$M", legend_title="", hovermode="x unified") + fig.update_traces(line=dict(width=2)) + st.plotly_chart(fig, use_container_width=True) + elif ticker: + st.caption("5-year trend not available for this ticker. DCF section below uses latest FCF from yfinance.") + st.markdown("---") + st.markdown("#### DCF valuation: inputs & 3-scenario output") dcf_inputs = get_dcf_inputs(ticker) if ticker else {} if not dcf_inputs: st.warning("Could not fetch DCF inputs from yfinance. Check ticker or try again.") @@ -465,20 +990,20 @@ with tab2: base_growth = st.slider("Base Case FCF Growth (%)", -10.0, 30.0, 8.0, 0.5) / 100.0 bull_growth = base_growth + 0.02 bear_growth = base_growth - 0.02 - ev_base = dcf_intrinsic_value(fcf, wacc, term_growth, base_growth) - ev_bull = dcf_intrinsic_value(fcf, wacc, term_growth, bull_growth) - ev_bear = dcf_intrinsic_value(fcf, wacc, term_growth, bear_growth) + ev_base = dcf_intrinsic_value(fcf, wacc, term_growth, base_growth, years=5) + ev_bull = dcf_intrinsic_value(fcf, wacc, term_growth, bull_growth, years=5) + ev_bear = dcf_intrinsic_value(fcf, wacc, term_growth, bear_growth, years=5) equity_base = ev_base - total_debt + cash equity_bull = ev_bull - total_debt + cash equity_bear = ev_bear - total_debt + cash price_base = equity_base / shares if shares else 0 price_bull = equity_bull / shares if shares else 0 price_bear = equity_bear / shares if shares else 0 - st.markdown("#### Intrinsic Value per Share (3 Scenarios)") + st.markdown("**Intrinsic value per share**") c1, c2, c3 = st.columns(3) - c1.metric("Bull (+2% growth)", f"${price_bull:.2f}", "Base vs Bull") + c1.metric("Bull (+2% growth)", f"${price_bull:.2f}", f"+{(price_bull - price_base):.2f} vs Base") c2.metric("Base", f"${price_base:.2f}", "—") - c3.metric("Bear (-2% growth)", f"${price_bear:.2f}", "Base vs Bear") + c3.metric("Bear (−2% growth)", f"${price_bear:.2f}", f"{(price_bear - price_base):.2f} vs Base") df_dcf = pd.DataFrame({ "Scenario": ["Bull", "Base", "Bear"], "FCF Growth": [f"{bull_growth*100:.1f}%", f"{base_growth*100:.1f}%", f"{bear_growth*100:.1f}%"], @@ -486,12 +1011,12 @@ with tab2: }) st.dataframe(df_dcf, use_container_width=True, hide_index=True) else: - st.info("FCF or Shares Outstanding not available for this ticker. Try another.") + st.caption("FCF or Shares not available. FCF = OCF − CapEx from yfinance.") -# ----- Tab 3: Industry Comps ----- +# ----- Tab 3: Industry Comps (conditional formatting) ----- with tab3: st.subheader("Industry Analysis & Comps") - st.markdown("Enter **comma-separated competitor tickers** (e.g. `AMD, INTC, QCOM`). Multiples from **yfinance**.") + st.markdown("Enter **comma-separated competitor tickers**. Multiples from **yfinance**. Green = below peer average (undervalued), Red = above (overvalued).") comp_tickers = st.text_input("Competitor tickers", value="AMD, INTC, QCOM", key="comps").strip() if st.button("Load Comps", key="load_comps"): tickers_list = [t.strip().upper() for t in comp_tickers.split(",") if t.strip()] @@ -504,7 +1029,31 @@ with tab3: if df_comps.empty: st.warning("Could not fetch comps from yfinance.") else: - st.dataframe(df_comps, use_container_width=True, hide_index=True) + try: + styled = df_comps.style + for col in ["Forward P/E", "EV/EBITDA", "P/B"]: + if col not in df_comps.columns: + continue + s = pd.to_numeric(df_comps[col], errors="coerce") + avg = s.mean() + if pd.isna(avg): + continue + def color_fn(v, avg_val=avg): + if pd.isna(v): + return "" + try: + x = float(v) + except (TypeError, ValueError): + return "" + if x < avg_val: + return "background-color: rgba(0, 200, 83, 0.3); color: #0d5c2e" + if x > avg_val: + return "background-color: rgba(255, 82, 82, 0.3); color: #b71c1c" + return "" + styled = styled.map(lambda v: color_fn(v), subset=[col]) + st.dataframe(styled, use_container_width=True, hide_index=True) + except Exception: + st.dataframe(df_comps, use_container_width=True, hide_index=True) st.divider() with st.expander("S&P 500 sample — Company & Ticker"): diff --git a/requirements.txt b/requirements.txt index 8f0ab43..6d6836e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,4 @@ pandas>=2.0.0 lxml>=4.9.0 python-dotenv>=1.0.0 yfinance>=0.2.40 +plotly>=5.18.0