mirror of
https://github.com/shawnkim1997/All-in-one-Financial-Analysis.git
synced 2026-08-05 06:47:44 +00:00
shawnkim1997
d337c63976
Architecture (3,909-line monolith → 28 focused modules, all < 300 lines):
- config/: constants.py (company lists, row maps, Damodaran baselines), theme.py (CSS/HTML)
- utils/: prefs, formatting, ticker, dcf, charts, ui_helpers
- data/: sec_parser, sec_fetcher, sec_downloader, financials, fundamentals,
valuation, ratios, scores, scores_ai, market
- ai/: gemini_core, gemini_sec, gemini_insights
- views/: sidebar, tab1_quant, tab1_ai, tab1_filings, tab2_dcf,
tab3_comps, tab4_news, tab5_markets, tab6_crypto, tab7_technical
- app.py: thin orchestrator (~118 lines)
- Strict unidirectional dependency graph (no circular imports)
- All @st.cache_data TTLs and st.session_state keys preserved identically
SEC filing viewer fix:
- Rebuilt EDGAR fetch chain: company_tickers.json → CIK → submissions API
→ filings.recent.primaryDocument[] (replaces deprecated directory.item)
- Filing type selectbox (10-K, 10-Q, 8-K, 20-F, 6-K) connected to backend
- Native HTML rendered via streamlit.components.v1.html() with CSS reset
- Errors surfaced explicitly with st.error()
- DART direct links restored for Korean-listed companies
.gitignore: data/ → data/*.json + data/*.html (preserve Python modules)
README: full rewrite for master's portfolio — 7-tab layout, architecture
diagram, modular structure tree, technical challenges, design rationale
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
207 lines
10 KiB
Python
207 lines
10 KiB
Python
from typing import Optional
|
|
import pandas as pd
|
|
import streamlit as st
|
|
from utils.formatting import _safe_float
|
|
|
|
try:
|
|
import yfinance as yf
|
|
except ImportError:
|
|
yf = None
|
|
|
|
from data.financials import _get_row_series, _get_annual_financials_balance_cashflow
|
|
|
|
|
|
@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:
|
|
"""FCF, Cash, Total Debt, Shares: from yahooquery (via _get_annual_financials) or yfinance fallback."""
|
|
out = {"fcf": None, "total_debt": 0.0, "cash": 0.0, "shares": None}
|
|
if not ticker:
|
|
return out
|
|
try:
|
|
fin, bal, cf = _get_annual_financials_balance_cashflow(ticker)
|
|
if bal is not None and not bal.empty and cf is not None and not cf.empty:
|
|
sh = _get_row_series(bal, "Share Issued")
|
|
out["shares"] = _safe_float(sh.iloc[0]) if sh is not None and len(sh) > 0 else None
|
|
td = _get_row_series(bal, "Total Debt")
|
|
out["total_debt"] = float(td.iloc[0] or 0) if td is not None and len(td) > 0 else 0.0
|
|
cash_s = _get_row_series(bal, "Cash And Cash Equivalents")
|
|
out["cash"] = float(cash_s.iloc[0] or 0) if cash_s is not None and len(cash_s) > 0 else 0.0
|
|
ocf = _get_row_series(cf, "Operating Cash Flow")
|
|
capx = _get_row_series(cf, "Capital Expenditure")
|
|
if ocf is not None and len(ocf) > 0:
|
|
ocf_val = _safe_float(ocf.iloc[0])
|
|
capx_val = _safe_float(capx.iloc[0]) if capx is not None and len(capx) > 0 else 0.0
|
|
if ocf_val is not None:
|
|
out["fcf"] = ocf_val - (capx_val or 0)
|
|
if out.get("fcf") is not None or out.get("shares") is not None:
|
|
return out
|
|
except Exception:
|
|
pass
|
|
if not yf:
|
|
return out
|
|
try:
|
|
t = yf.Ticker(ticker.upper())
|
|
info = t.info or {}
|
|
fast_info = getattr(t, "fast_info", None)
|
|
cashflow = getattr(t, "cashflow", None)
|
|
if cashflow is None or cashflow.empty:
|
|
cashflow = getattr(t, "quarterly_cashflow", None)
|
|
balance = getattr(t, "balance_sheet", None)
|
|
if balance is None or balance.empty:
|
|
balance = getattr(t, "quarterly_balance_sheet", None)
|
|
|
|
# ----- Shares Outstanding: multi-step fallback (no manual by default) -----
|
|
shares = None
|
|
if fast_info is not None:
|
|
try:
|
|
s = getattr(fast_info, "shares", None)
|
|
if s is None and hasattr(fast_info, "get"):
|
|
s = fast_info.get("shares")
|
|
if s is not None and float(s) > 0:
|
|
shares = float(s)
|
|
except (TypeError, ValueError, AttributeError):
|
|
pass
|
|
if shares is None:
|
|
for key in ("sharesOutstanding", "Shares Outstanding", "impliedSharesOutstanding", "Float Shares"):
|
|
s = info.get(key)
|
|
if s is not None and float(s) > 0:
|
|
shares = float(s)
|
|
break
|
|
if shares is None and balance is not None and not balance.empty:
|
|
try:
|
|
if "Share Issued" in balance.index:
|
|
shares = _safe_float(balance.loc["Share Issued"].iloc[0])
|
|
if (shares is None or shares <= 0) and "Ordinary Shares Number" in balance.index:
|
|
shares = _safe_float(balance.loc["Ordinary Shares Number"].iloc[0])
|
|
except (KeyError, TypeError, IndexError):
|
|
pass
|
|
out["shares"] = shares if (shares is not None and shares > 0) else None
|
|
|
|
# ----- Total Debt: fast_info → info → balance -----
|
|
total_debt = None
|
|
if fast_info is not None:
|
|
try:
|
|
d = getattr(fast_info, "total_debt", None) or (fast_info.get("total_debt") if hasattr(fast_info, "get") else None)
|
|
if d is not None and float(d) >= 0:
|
|
total_debt = float(d)
|
|
except (TypeError, ValueError, AttributeError):
|
|
pass
|
|
if total_debt is None:
|
|
total_debt = info.get("Total Debt")
|
|
if total_debt is None and balance is not None and not balance.empty:
|
|
try:
|
|
if "Total Debt" in balance.index:
|
|
total_debt = _safe_float(balance.loc["Total Debt"].iloc[0])
|
|
except (KeyError, TypeError, IndexError):
|
|
pass
|
|
out["total_debt"] = float(total_debt) if total_debt is not None else 0.0
|
|
|
|
# ----- Cash: fast_info → info → balance -----
|
|
cash = None
|
|
if fast_info is not None:
|
|
try:
|
|
c = getattr(fast_info, "cash", None) or (fast_info.get("cash") if hasattr(fast_info, "get") else None)
|
|
if c is not None and float(c) >= 0:
|
|
cash = float(c)
|
|
except (TypeError, ValueError, AttributeError):
|
|
pass
|
|
if cash is None:
|
|
cash = info.get("Cash And Cash Equivalents") or info.get("Cash")
|
|
if cash is None and balance is not None and not balance.empty:
|
|
try:
|
|
for row in ("Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments", "Cash"):
|
|
if row in balance.index:
|
|
cash = _safe_float(balance.loc[row].iloc[0])
|
|
if cash is not None:
|
|
break
|
|
except (KeyError, TypeError, IndexError):
|
|
pass
|
|
out["cash"] = float(cash) if cash is not None else 0.0
|
|
|
|
# ----- Base FCF = OCF - CapEx -----
|
|
ocf = _get_row_series(cashflow, "Operating Cash Flow", "Cash From Operating Activities", "Cash From Operations") if cashflow is not None else None
|
|
capx = _get_row_series(cashflow, "Capital Expenditure", "Capital Expenditures", "Purchase Of Property Plant And Equipment") if cashflow is not None else None
|
|
if ocf is not None and len(ocf) > 0:
|
|
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 hasattr(capx, "index") and latest_date in getattr(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
|
|
if ocf_val is not None:
|
|
latest_fcf = ocf_val - capx_val
|
|
if latest_fcf == latest_fcf and not (isinstance(latest_fcf, float) and pd.isna(latest_fcf)):
|
|
out["fcf"] = latest_fcf
|
|
return out
|
|
except Exception:
|
|
return out
|