mirror of
https://github.com/shawnkim1997/All-in-one-Financial-Analysis.git
synced 2026-08-16 11:58:07 +00:00
refactor: modular architecture v3.0 + SEC filing viewer fix + README
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>
This commit is contained in:
shawnkim1997
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
7ce5661569
commit
d337c63976
@@ -0,0 +1,173 @@
|
||||
from typing import Optional
|
||||
import pandas as pd
|
||||
import streamlit as st
|
||||
from utils.formatting import _safe_float
|
||||
from config.constants import INCOME_ROW_MAP, BALANCE_ROW_MAP, CASHFLOW_ROW_MAP
|
||||
|
||||
try:
|
||||
import yfinance as yf
|
||||
except ImportError:
|
||||
yf = None
|
||||
|
||||
try:
|
||||
from yahooquery import Ticker as YQTicker
|
||||
except ImportError:
|
||||
YQTicker = None
|
||||
|
||||
|
||||
def _yq_df_to_our_shape(df: pd.DataFrame, row_map: list, date_col: str = "asOfDate") -> Optional[pd.DataFrame]:
|
||||
"""Convert yahooquery DataFrame (rows=periods, columns=line items) to our shape: index=line names, columns=dates."""
|
||||
if df is None or df.empty or date_col not in df.columns:
|
||||
return None
|
||||
df = df.dropna(subset=[date_col]).sort_values(date_col, ascending=False).head(5)
|
||||
if df.empty:
|
||||
return None
|
||||
dates = df[date_col].astype(str).str[:10].tolist()
|
||||
data = {}
|
||||
for our_name, yq_col in row_map:
|
||||
cols = (yq_col,) if isinstance(yq_col, str) else yq_col
|
||||
val_col = next((c for c in cols if c in df.columns), None)
|
||||
if val_col is None:
|
||||
data[our_name] = [None] * len(dates)
|
||||
continue
|
||||
data[our_name] = [_safe_float(v) for v in df[val_col].tolist()]
|
||||
out = pd.DataFrame(data, index=dates).T
|
||||
out.columns = dates
|
||||
return out
|
||||
|
||||
|
||||
def _share_issued_from_yq_balance(df_bal: pd.DataFrame) -> Optional[pd.Series]:
|
||||
"""Try OrdinarySharesNumber then ShareIssued for shares outstanding in yahooquery balance."""
|
||||
if df_bal is None or df_bal.empty:
|
||||
return None
|
||||
for col in ("OrdinarySharesNumber", "ShareIssued"):
|
||||
if col in df_bal.columns and "asOfDate" in df_bal.columns:
|
||||
s = df_bal.set_index("asOfDate")[col].sort_index(ascending=False)
|
||||
s.index = s.index.astype(str).str[:10]
|
||||
return s.reindex(s.index) # keep as series with date index
|
||||
return None
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def _get_annual_financials_balance_cashflow_yahooquery(ticker: str) -> tuple:
|
||||
"""Fetch income, balance, cash flow from yahooquery. Return (fin_df, bal_df, cf_df) with index=line items, columns=dates. TTM fallback if annual insufficient."""
|
||||
if not YQTicker or not ticker:
|
||||
return (None, None, None)
|
||||
try:
|
||||
yq = YQTicker(ticker.upper())
|
||||
inc_a = yq.income_statement(frequency="a", trailing=False)
|
||||
bal_a = yq.balance_sheet(frequency="a", trailing=False)
|
||||
cf_a = yq.cash_flow(frequency="a", trailing=False)
|
||||
if inc_a is None or inc_a.empty or bal_a is None or bal_a.empty:
|
||||
inc_q = yq.income_statement(frequency="q", trailing=False)
|
||||
bal_q = yq.balance_sheet(frequency="q", trailing=False)
|
||||
cf_q = yq.cash_flow(frequency="q", trailing=False)
|
||||
# Build TTM: need at least 2 periods for Piotroski/Radar; use last 4Q and previous 4Q when 8+ quarters
|
||||
if inc_q is not None and not inc_q.empty and len(inc_q) >= 4:
|
||||
ttm0 = inc_q.head(4).sum(numeric_only=True)
|
||||
row0 = ttm0.to_dict() if hasattr(ttm0, "to_dict") else dict(ttm0)
|
||||
row0["asOfDate"] = inc_q["asOfDate"].iloc[0] if "asOfDate" in inc_q.columns else "TTM0"
|
||||
rows_inc = [row0]
|
||||
if len(inc_q) >= 8:
|
||||
ttm1 = inc_q.iloc[4:8].sum(numeric_only=True)
|
||||
row1 = ttm1.to_dict() if hasattr(ttm1, "to_dict") else dict(ttm1)
|
||||
row1["asOfDate"] = inc_q["asOfDate"].iloc[4] if "asOfDate" in inc_q.columns else "TTM1"
|
||||
rows_inc.append(row1)
|
||||
inc_a = pd.DataFrame(rows_inc)
|
||||
if bal_q is not None and not bal_q.empty:
|
||||
bal_a = bal_q.head(2) if (bal_a is None or bal_a.empty) else bal_a
|
||||
if cf_q is not None and not cf_q.empty and len(cf_q) >= 4 and (cf_a is None or cf_a.empty):
|
||||
ttm0_cf = cf_q.head(4).sum(numeric_only=True)
|
||||
row0_cf = ttm0_cf.to_dict() if hasattr(ttm0_cf, "to_dict") else dict(ttm0_cf)
|
||||
row0_cf["asOfDate"] = cf_q["asOfDate"].iloc[0] if "asOfDate" in cf_q.columns else "TTM0"
|
||||
rows_cf = [row0_cf]
|
||||
if len(cf_q) >= 8:
|
||||
ttm1_cf = cf_q.iloc[4:8].sum(numeric_only=True)
|
||||
row1_cf = ttm1_cf.to_dict() if hasattr(ttm1_cf, "to_dict") else dict(ttm1_cf)
|
||||
row1_cf["asOfDate"] = cf_q["asOfDate"].iloc[4] if "asOfDate" in cf_q.columns else "TTM1"
|
||||
rows_cf.append(row1_cf)
|
||||
cf_a = pd.DataFrame(rows_cf)
|
||||
fin_df = _yq_df_to_our_shape(inc_a, INCOME_ROW_MAP)
|
||||
bal_df = _yq_df_to_our_shape(bal_a, BALANCE_ROW_MAP)
|
||||
if bal_df is not None and "Share Issued" not in bal_df.index and bal_a is not None and not bal_a.empty:
|
||||
for sh_col in ("OrdinarySharesNumber", "ShareIssued"):
|
||||
if sh_col in bal_a.columns:
|
||||
row = {"Share Issued": [_safe_float(bal_a[sh_col].iloc[0])]}
|
||||
if bal_df is not None and not bal_df.empty:
|
||||
d = str(bal_a["asOfDate"].iloc[0])[:10] if "asOfDate" in bal_a.columns else bal_df.columns[0]
|
||||
extra = pd.DataFrame(row, index=[d]).T
|
||||
extra.columns = [d]
|
||||
bal_df = pd.concat([bal_df, extra], axis=0)
|
||||
break
|
||||
cf_df = _yq_df_to_our_shape(cf_a, CASHFLOW_ROW_MAP)
|
||||
return (fin_df, bal_df, cf_df)
|
||||
except Exception:
|
||||
return (None, None, None)
|
||||
|
||||
|
||||
# ---------- Raw statements & FCF = OCF - CapEx ----------
|
||||
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
|
||||
|
||||
|
||||
def _fin_or_bal_empty(df) -> bool:
|
||||
"""True if DataFrame is missing, empty, or has no columns (e.g. yfinance returned empty)."""
|
||||
return df is None or df.empty or (hasattr(df, "columns") and len(df.columns) == 0)
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def _get_annual_financials_balance_cashflow(ticker: str) -> tuple:
|
||||
"""Return (fin_df, bal_df, cf_df). Uses yahooquery first; if missing/fail, falls back to yfinance with TTM when needed."""
|
||||
if not ticker:
|
||||
return (None, None, None)
|
||||
fin_df, bal_df, cf_df = _get_annual_financials_balance_cashflow_yahooquery(ticker)
|
||||
if fin_df is not None and not fin_df.empty and bal_df is not None and not bal_df.empty:
|
||||
return (fin_df, bal_df, cf_df)
|
||||
if not yf:
|
||||
return (None, None, None)
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
fin = getattr(t, "financials", None)
|
||||
bal = getattr(t, "balance_sheet", None)
|
||||
cf = getattr(t, "cashflow", None)
|
||||
if _fin_or_bal_empty(fin):
|
||||
qf = getattr(t, "quarterly_financials", None)
|
||||
if qf is not None and not qf.empty:
|
||||
n = len(qf.columns)
|
||||
if n >= 8:
|
||||
c0 = qf.iloc[:, :4].sum(axis=1)
|
||||
c1 = qf.iloc[:, 4:8].sum(axis=1)
|
||||
fin = pd.concat([c0, c1], axis=1)
|
||||
fin.columns = ["TTM0", "TTM1"]
|
||||
elif n >= 5:
|
||||
c0 = qf.iloc[:, :4].sum(axis=1)
|
||||
c1 = qf.iloc[:, 4:n].sum(axis=1)
|
||||
fin = pd.concat([c0, c1], axis=1)
|
||||
fin.columns = ["TTM0", "TTM1"]
|
||||
else:
|
||||
fin = qf.iloc[:, : min(4, n)].sum(axis=1).to_frame("TTM0")
|
||||
if _fin_or_bal_empty(bal):
|
||||
qb = getattr(t, "quarterly_balance_sheet", None)
|
||||
if qb is not None and not qb.empty:
|
||||
n = len(qb.columns)
|
||||
bal = qb.iloc[:, : min(2, n)].copy()
|
||||
if bal.shape[1] == 1:
|
||||
bal.columns = ["B0"]
|
||||
else:
|
||||
bal.columns = ["B0", "B1"]
|
||||
if _fin_or_bal_empty(cf):
|
||||
qc = getattr(t, "quarterly_cashflow", None)
|
||||
if qc is not None and not qc.empty:
|
||||
n = len(qc.columns)
|
||||
cf = qc.iloc[:, : min(4, n)].sum(axis=1).to_frame("TTM0")
|
||||
return (fin, bal, cf)
|
||||
except Exception:
|
||||
return (None, None, None)
|
||||
@@ -0,0 +1,206 @@
|
||||
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
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
import streamlit as st
|
||||
import pandas as pd
|
||||
from utils.formatting import _safe_float
|
||||
|
||||
try:
|
||||
import yfinance as yf
|
||||
except ImportError:
|
||||
yf = None
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_technical_indicators(ticker: str) -> dict:
|
||||
"""RSI(14), SMA(50), SMA(200), support/resistance, 52-week range."""
|
||||
out = {"rsi_14": None, "sma_50": None, "sma_200": None, "current_price": None, "support": None, "resistance": None, "52w_high": None, "52w_low": None}
|
||||
if not yf or not ticker:
|
||||
return out
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
hist = t.history(period="1y")
|
||||
if hist is None or hist.empty or len(hist) < 14:
|
||||
return out
|
||||
close = hist["Close"]
|
||||
out["current_price"] = float(close.iloc[-1])
|
||||
delta = close.diff()
|
||||
gain = delta.where(delta > 0, 0).rolling(window=14).mean()
|
||||
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
|
||||
rs = gain / loss
|
||||
rsi = 100 - (100 / (1 + rs))
|
||||
out["rsi_14"] = round(float(rsi.iloc[-1]), 1) if not pd.isna(rsi.iloc[-1]) else None
|
||||
if len(close) >= 50:
|
||||
out["sma_50"] = round(float(close.rolling(50).mean().iloc[-1]), 2)
|
||||
if len(close) >= 200:
|
||||
out["sma_200"] = round(float(close.rolling(200).mean().iloc[-1]), 2)
|
||||
out["52w_high"] = round(float(close.max()), 2)
|
||||
out["52w_low"] = round(float(close.min()), 2)
|
||||
recent = close.tail(20)
|
||||
out["support"] = round(float(recent.min()), 2)
|
||||
out["resistance"] = round(float(recent.max()), 2)
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
|
||||
|
||||
@st.cache_data(ttl=600)
|
||||
def get_risk_analysis(ticker: str) -> list:
|
||||
"""Risk factors with estimated EPS impact."""
|
||||
if not yf or not ticker:
|
||||
return []
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = t.info or {}
|
||||
eps = info.get("trailingEps") or info.get("forwardEps") or 1.0
|
||||
beta = info.get("beta") or 1.0
|
||||
debt_equity = info.get("debtToEquity") or 0
|
||||
margin = info.get("operatingMargins") or 0
|
||||
risks = []
|
||||
impact = round(eps * (beta - 1) * 0.1, 2) if beta > 1 else round(eps * 0.05, 2)
|
||||
risks.append({"risk": "Market / Macro Risk", "severity": "High" if beta > 1.3 else "Medium", "eps_impact": f"-${abs(impact):.2f}", "description": f"Beta {beta:.2f}"})
|
||||
comp_impact = round(eps * 0.08, 2)
|
||||
risks.append({"risk": "Competitive Pressure", "severity": "High" if margin < 0.15 else "Medium", "eps_impact": f"-${abs(comp_impact):.2f}", "description": f"Op margin {margin*100:.1f}%"})
|
||||
lev_impact = round(eps * 0.06, 2) if debt_equity and debt_equity > 100 else round(eps * 0.03, 2)
|
||||
risks.append({"risk": "Financial / Leverage", "severity": "High" if (debt_equity or 0) > 150 else ("Medium" if (debt_equity or 0) > 80 else "Low"), "eps_impact": f"-${abs(lev_impact):.2f}", "description": f"D/E {debt_equity:.0f}%" if debt_equity else "D/E N/A"})
|
||||
risks.append({"risk": "Regulatory / Legal", "severity": "Medium", "eps_impact": f"-${abs(round(eps * 0.05, 2)):.2f}", "description": "Regulatory changes"})
|
||||
risks.append({"risk": "Currency / FX", "severity": "Medium", "eps_impact": f"-${abs(round(eps * 0.04, 2)):.2f}", "description": "FX exposure"})
|
||||
risks.append({"risk": "Supply Chain", "severity": "Medium", "eps_impact": f"-${abs(round(eps * 0.05, 2)):.2f}", "description": "Component/logistics risk"})
|
||||
return risks
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
@st.cache_data(ttl=120)
|
||||
def _get_ticker_bar_data() -> list:
|
||||
"""Fetch major index/crypto prices for top ticker bar."""
|
||||
items = []
|
||||
tickers_bar = {"S&P 500": "^GSPC", "NASDAQ": "^IXIC", "KOSPI": "^KS11", "NIKKEI": "^N225", "BTC": "BTC-USD", "ETH": "ETH-USD"}
|
||||
for label, sym in tickers_bar.items():
|
||||
try:
|
||||
t = yf.Ticker(sym)
|
||||
info = t.info or {}
|
||||
price = info.get("regularMarketPrice") or info.get("previousClose") or 0
|
||||
prev = info.get("regularMarketPreviousClose") or info.get("previousClose") or price
|
||||
change_pct = ((price - prev) / prev * 100) if prev else 0
|
||||
items.append({"label": label, "price": price, "change": change_pct})
|
||||
except Exception:
|
||||
items.append({"label": label, "price": 0, "change": 0})
|
||||
return items
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def _fetch_news_rss(ticker_sym: str, company_name: str = "") -> list:
|
||||
"""Fetch news from Google News RSS. Returns list of {title, source, url, published}."""
|
||||
import feedparser
|
||||
items = []
|
||||
query = ticker_sym if not company_name else company_name
|
||||
try:
|
||||
feed = feedparser.parse(f"https://news.google.com/rss/search?q={query}+stock&hl=en-US&gl=US&ceid=US:en")
|
||||
for entry in (feed.entries or [])[:15]:
|
||||
items.append({
|
||||
"title": entry.get("title", ""),
|
||||
"source": entry.get("source", {}).get("title", "Google News") if hasattr(entry.get("source", ""), "get") else "Google News",
|
||||
"url": entry.get("link", ""),
|
||||
"published": entry.get("published", ""),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
return items
|
||||
+285
@@ -0,0 +1,285 @@
|
||||
from typing import Optional
|
||||
import pandas as pd
|
||||
import streamlit as st
|
||||
from utils.formatting import _safe_float, _na
|
||||
from data.financials import _get_row_series, _get_annual_financials_balance_cashflow
|
||||
|
||||
try:
|
||||
import yfinance as yf
|
||||
except ImportError:
|
||||
yf = None
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_comps_data(tickers: tuple) -> pd.DataFrame:
|
||||
"""Fetch Forward P/E, EV/EBITDA, P/B using forwardPE, enterpriseToEbitda, priceToBook. Missing → None (display as N/A). Robust per-ticker error handling."""
|
||||
if not yf:
|
||||
return pd.DataFrame()
|
||||
rows = []
|
||||
for sym in tickers:
|
||||
sym = str(sym).strip().upper()
|
||||
if not sym:
|
||||
continue
|
||||
try:
|
||||
t = yf.Ticker(sym)
|
||||
info = t.info or {}
|
||||
forward_pe = info.get("forwardPE") or info.get("Forward PE") or info.get("trailingPE") or info.get("Trailing PE")
|
||||
ev_ebitda = info.get("enterpriseToEbitda")
|
||||
if ev_ebitda is None:
|
||||
ev, ebitda = info.get("enterpriseValue"), info.get("ebitda")
|
||||
if ev is not None and ebitda is not None and ebitda != 0:
|
||||
ev_ebitda = ev / ebitda
|
||||
pb = info.get("priceToBook") or info.get("Price To Book")
|
||||
rows.append({
|
||||
"Ticker": sym,
|
||||
"Forward P/E": round(float(forward_pe), 2) if forward_pe is not None and _safe_float(forward_pe) is not None else None,
|
||||
"EV/EBITDA": round(float(ev_ebitda), 2) if ev_ebitda is not None and _safe_float(ev_ebitda) is not None else None,
|
||||
"P/B": round(float(pb), 2) if pb is not None and _safe_float(pb) is not None else None,
|
||||
})
|
||||
except Exception:
|
||||
rows.append({"Ticker": sym, "Forward P/E": None, "EV/EBITDA": None, "P/B": None})
|
||||
if not rows:
|
||||
return pd.DataFrame()
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
@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 yahooquery then yfinance with TTM fallback."""
|
||||
try:
|
||||
fin, bal, _ = _get_annual_financials_balance_cashflow(ticker)
|
||||
if fin is None or fin.empty or bal is None or bal.empty:
|
||||
return {}
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = t.info or {}
|
||||
# TTM columns: keep order TTM0 (current), TTM1 (prior). Else use date sort (newest first).
|
||||
col_list = fin.columns.tolist()
|
||||
if col_list and str(col_list[0]).startswith("TTM"):
|
||||
dates = col_list[:3]
|
||||
else:
|
||||
dates = sorted(col_list, 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 i, d in enumerate(dates):
|
||||
yr = int(str(d)[:4]) if (isinstance(d, str) and str(d)[:4].isdigit()) else (d.year if hasattr(d, "year") else (2024 - i))
|
||||
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)
|
||||
if op_inc is not None and int_exp is not None and int_exp != 0:
|
||||
_ic = op_inc / int_exp
|
||||
interest_cov = round(_ic, 2) if (_ic == _ic and not (isinstance(_ic, float) and (pd.isna(_ic) or _ic != _ic))) else None
|
||||
else:
|
||||
interest_cov = None # N/A when Interest Expense is 0 or missing (avoid nan%)
|
||||
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": interest_cov,
|
||||
})
|
||||
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 and not (pd.isna(cur) or pd.isna(prev)):
|
||||
if "Margin" in col or "NPM" in col or "ROE" in col:
|
||||
chg_pp = (cur - prev) # percentage point change (e.g. 7.0 = 7%)
|
||||
if pd.isna(chg_pp) or chg_pp != chg_pp:
|
||||
continue
|
||||
yoy.append({"Ratio": col, "Latest": cur, "Prior": prev, "YoY (pp)": round(chg_pp, 2), "Comment": f"{'Improved' if chg_pp > 0 else 'Declined'} by {abs(chg_pp):.1f}% YoY"})
|
||||
else:
|
||||
pct = (cur - prev) / abs(prev) * 100
|
||||
if pd.isna(pct) or pct != pct:
|
||||
continue
|
||||
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_quarterly_momentum(ticker: str) -> dict:
|
||||
"""Last 4 quarters Revenue and Net Income from quarterly_financials; QoQ growth for most recent quarter. Returns {df, qoq_revenue_pct, qoq_ni_pct} or empty."""
|
||||
out = {"df": None, "qoq_revenue_pct": None, "qoq_ni_pct": None}
|
||||
if not yf or not ticker:
|
||||
return out
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
qfin = getattr(t, "quarterly_financials", None)
|
||||
if qfin is None or qfin.empty or len(qfin.columns) < 2:
|
||||
return out
|
||||
rev = _get_row_series(qfin, "Total Revenue", "Revenue", "Net Revenue")
|
||||
ni = _get_row_series(qfin, "Net Income", "Net Income Common Stockholders")
|
||||
if rev is None and ni is None:
|
||||
return out
|
||||
cols = list(qfin.columns)[:4]
|
||||
rows = []
|
||||
for c in cols:
|
||||
try:
|
||||
if hasattr(c, "strftime"):
|
||||
q = (c.month - 1) // 3 + 1 if hasattr(c, "month") else 1
|
||||
label = c.strftime("%Y") + f"-Q{q}"
|
||||
else:
|
||||
label = str(c)[:12]
|
||||
except Exception:
|
||||
label = str(c)[:12]
|
||||
r_val = _safe_float(rev.loc[c]) if rev is not None and c in rev.index else None
|
||||
n_val = _safe_float(ni.loc[c]) if ni is not None and c in ni.index else None
|
||||
rows.append({"Quarter": label, "Revenue": r_val, "Net Income": n_val})
|
||||
out["df"] = pd.DataFrame(rows)
|
||||
if len(rows) >= 2:
|
||||
r0, r1 = rows[0].get("Revenue"), rows[1].get("Revenue")
|
||||
n0, n1 = rows[0].get("Net Income"), rows[1].get("Net Income")
|
||||
if r0 is not None and r1 is not None and r1 != 0:
|
||||
out["qoq_revenue_pct"] = round((r0 - r1) / abs(r1) * 100, 1)
|
||||
if n0 is not None and n1 is not None and n1 != 0:
|
||||
out["qoq_ni_pct"] = round((n0 - n1) / abs(n1) * 100, 1)
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_quarterly_ratio_changes(ticker: str) -> list:
|
||||
"""QoQ ratio changes: NPM %, ROE %, Gross Margin %, Operating Margin %, Current Ratio, Interest Coverage. Latest quarter vs previous. Returns list of {Metric, Current, Change, Trend}."""
|
||||
out = []
|
||||
if not yf or not ticker:
|
||||
return out
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
qf = getattr(t, "quarterly_financials", None)
|
||||
qb = getattr(t, "quarterly_balance_sheet", None)
|
||||
if qf is None or qf.empty or qb is None or qb.empty or len(qf.columns) < 2 or len(qb.columns) < 2:
|
||||
return out
|
||||
rev = _get_row_series(qf, "Total Revenue", "Revenue", "Net Revenue")
|
||||
ni = _get_row_series(qf, "Net Income", "Net Income Common Stockholders")
|
||||
gross = _get_row_series(qf, "Gross Profit")
|
||||
ebit = _get_row_series(qf, "Operating Income", "EBIT")
|
||||
interest = _get_row_series(qf, "Interest Expense", "Interest Expense Net")
|
||||
ta = _get_row_series(qb, "Total Assets")
|
||||
te = _get_row_series(qb, "Total Stockholder Equity", "Stockholders Equity", "Total Equity Gross Minority Interest")
|
||||
ca = _get_row_series(qb, "Current Assets")
|
||||
cl = _get_row_series(qb, "Current Liabilities")
|
||||
def v(s, col):
|
||||
if s is None or col not in s.index:
|
||||
return None
|
||||
return _safe_float(s.get(col))
|
||||
c0, c1 = qf.columns[0], qf.columns[1]
|
||||
b0, b1 = qb.columns[0], qb.columns[1]
|
||||
r0, r1 = v(rev, c0), v(rev, c1)
|
||||
n0, n1 = v(ni, c0), v(ni, c1)
|
||||
g0, g1 = v(gross, c0), v(gross, c1)
|
||||
e0, e1 = v(ebit, c0), v(ebit, c1)
|
||||
i0, i1 = v(interest, c0), v(interest, c1)
|
||||
ta0, ta1 = v(ta, b0), v(ta, b1)
|
||||
te0, te1 = v(te, b0), v(te, b1)
|
||||
ca0, ca1 = v(ca, b0), v(ca, b1)
|
||||
cl0, cl1 = v(cl, b0), v(cl, b1)
|
||||
npm0 = (n0 / r0 * 100) if (n0 is not None and r0 and r0 != 0) else None
|
||||
npm1 = (n1 / r1 * 100) if (n1 is not None and r1 and r1 != 0) else None
|
||||
roe0 = (n0 / te0 * 100) if (n0 is not None and te0 and te0 != 0) else None
|
||||
roe1 = (n1 / te1 * 100) if (n1 is not None and te1 and te1 != 0) else None
|
||||
gm0 = (g0 / r0 * 100) if (g0 is not None and r0 and r0 != 0) else None
|
||||
gm1 = (g1 / r1 * 100) if (g1 is not None and r1 and r1 != 0) else None
|
||||
om0 = (e0 / r0 * 100) if (e0 is not None and r0 and r0 != 0) else None
|
||||
om1 = (e1 / r1 * 100) if (e1 is not None and r1 and r1 != 0) else None
|
||||
cr0 = (ca0 / cl0) if (ca0 is not None and cl0 and cl0 != 0) else None
|
||||
cr1 = (ca1 / cl1) if (ca1 is not None and cl1 and cl1 != 0) else None
|
||||
ic0 = (e0 / i0) if (e0 is not None and i0 and i0 != 0) else None
|
||||
ic1 = (e1 / i1) if (e1 is not None and i1 and i1 != 0) else None
|
||||
def row(metric, cur, prev, is_pct_point=False):
|
||||
if cur is None:
|
||||
return None
|
||||
cur_str = f"{round(cur, 2):.2f}"
|
||||
if prev is None or (is_pct_point and prev != prev):
|
||||
return {"Metric": metric, "Current Value": cur_str, "Change": "—", "Trend": "—"}
|
||||
if is_pct_point:
|
||||
chg = cur - prev
|
||||
else:
|
||||
chg = ((cur - prev) / abs(prev) * 100) if prev != 0 else 0
|
||||
trend = "↑" if chg > 0 else ("↓" if chg < 0 else "—")
|
||||
chg_str = f"{chg:+.1f}%" if not is_pct_point else f"{chg:+.1f} pp"
|
||||
return {"Metric": metric, "Current Value": cur_str, "Change": chg_str, "Trend": trend}
|
||||
for name, cur, prev, is_pp in [
|
||||
("NPM %", npm0, npm1, True), ("ROE %", roe0, roe1, True), ("Gross Margin %", gm0, gm1, True),
|
||||
("Operating Margin %", om0, om1, True), ("Current Ratio", cr0, cr1, False), ("Interest Coverage", ic0, ic1, False),
|
||||
]:
|
||||
r = row(name, cur, prev, is_pp)
|
||||
if r:
|
||||
out.append(r)
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
+261
@@ -0,0 +1,261 @@
|
||||
from typing import Optional
|
||||
import pandas as pd
|
||||
import streamlit as st
|
||||
from utils.formatting import _safe_float
|
||||
from data.financials import _get_row_series, _get_annual_financials_balance_cashflow
|
||||
from data.ratios import get_dupont_altman_redflags_yoy
|
||||
|
||||
try:
|
||||
import yfinance as yf
|
||||
except ImportError:
|
||||
yf = None
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_income_statement_sankey_data(ticker: str) -> dict:
|
||||
"""Latest year (or TTM): Revenue, COGS, Gross Profit, OpEx, Operating Income, Tax/Interest/Other, Net Income. Uses yahooquery then yfinance."""
|
||||
out = {"revenue": 0, "cogs": 0, "gross_profit": 0, "opex": 0, "operating_income": 0, "tax_interest_other": 0, "net_income": 0}
|
||||
fin, _, _ = _get_annual_financials_balance_cashflow(ticker)
|
||||
if fin is None or fin.empty:
|
||||
return out
|
||||
try:
|
||||
rev = _get_row_series(fin, "Total Revenue", "Revenue", "Net Revenue")
|
||||
cogs = _get_row_series(fin, "Cost Of Revenue", "Cost Of Goods Sold")
|
||||
gross = _get_row_series(fin, "Gross Profit")
|
||||
op_inc = _get_row_series(fin, "Operating Income", "EBIT")
|
||||
ni = _get_row_series(fin, "Net Income", "Net Income Common Stockholders")
|
||||
if rev is None or len(rev) == 0:
|
||||
return out
|
||||
d = rev.index[0]
|
||||
revenue = abs(_safe_float(rev.get(d)) or 0)
|
||||
cogs_val = abs(_safe_float(cogs.get(d)) if cogs is not None and d in cogs.index else 0) or 0
|
||||
gross_val = _safe_float(gross.get(d)) if gross is not None and d in gross.index else None
|
||||
if gross_val is None and revenue and cogs_val is not None:
|
||||
gross_val = revenue - cogs_val
|
||||
elif gross_val is None:
|
||||
gross_val = revenue
|
||||
gross_val = abs(gross_val) if gross_val is not None else 0
|
||||
op_inc_val = _safe_float(op_inc.get(d)) if op_inc is not None and d in op_inc.index else None
|
||||
op_inc_val = op_inc_val if op_inc_val is not None else 0
|
||||
ni_val = _safe_float(ni.get(d)) if ni is not None and d in ni.index else None
|
||||
ni_val = ni_val if ni_val is not None else 0
|
||||
opex_val = max(0, gross_val - op_inc_val) if (gross_val >= op_inc_val) else 0
|
||||
tax_interest_other = max(0, op_inc_val - ni_val) if (op_inc_val - ni_val) > 0 else abs(min(0, op_inc_val - ni_val))
|
||||
out["revenue"] = max(revenue, 1)
|
||||
out["cogs"] = min(cogs_val, revenue - 1e-6)
|
||||
out["gross_profit"] = gross_val
|
||||
out["opex"] = opex_val
|
||||
out["operating_income"] = op_inc_val
|
||||
out["tax_interest_other"] = tax_interest_other
|
||||
out["net_income"] = ni_val
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
|
||||
|
||||
# sankey_data_from_ai, piotroski_from_ai, radar_metrics_from_ai → data/scores_ai.py
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_radar_metrics_normalized(ticker: str) -> dict:
|
||||
"""ROE, Current Ratio, Asset Turnover, Equity Mult, Revenue YoY. Normalized to 0-100 for radar. Returns {theta: [...], r: [...], labels: [...]} or empty."""
|
||||
if not ticker:
|
||||
return {}
|
||||
q = get_dupont_altman_redflags_yoy(ticker)
|
||||
if not q:
|
||||
return {}
|
||||
dupont_df = q.get("dupont")
|
||||
if dupont_df is None or dupont_df.empty or len(dupont_df) < 2:
|
||||
return {}
|
||||
row0 = dupont_df.iloc[0]
|
||||
row1 = dupont_df.iloc[1]
|
||||
roe = row0.get("ROE %") or 0
|
||||
cr = row0.get("Current Ratio") or 0
|
||||
at = row0.get("Asset Turnover") or 0
|
||||
em = row0.get("Equity Mult.") or 0
|
||||
rev0 = dupont_df["Revenue"].iloc[0] if "Revenue" in dupont_df.columns else None
|
||||
rev1 = dupont_df["Revenue"].iloc[1] if "Revenue" in dupont_df.columns else None
|
||||
rev_yoy = ((rev0 - rev1) / rev1 * 100) if (rev0 and rev1 and rev1 != 0) else 0
|
||||
def norm_roe(x):
|
||||
if x is None: return 50
|
||||
return min(100, max(0, (x + 10) / 40 * 100))
|
||||
def norm_cr(x):
|
||||
if x is None: return 50
|
||||
return min(100, max(0, x / 3 * 100))
|
||||
def norm_at(x):
|
||||
if x is None: return 50
|
||||
return min(100, max(0, x * 50))
|
||||
def norm_em(x):
|
||||
if x is None: return 50
|
||||
return min(100, max(0, (x - 0.5) / 2.5 * 100))
|
||||
def norm_yoy(x):
|
||||
if x is None: return 50
|
||||
return min(100, max(0, (x + 20) / 50 * 100))
|
||||
return {
|
||||
"theta": ["Profitability (ROE)", "Liquidity (Curr.Ratio)", "Efficiency (Asset Turn.)", "Solvency (Equity Mult.)", "Growth (Rev YoY)"],
|
||||
"r": [norm_roe(roe), norm_cr(cr), norm_at(at), norm_em(em), norm_yoy(rev_yoy)],
|
||||
"labels": ["Profitability (ROE)", "Liquidity (Curr.Ratio)", "Efficiency (Asset Turn.)", "Solvency (Equity Mult.)", "Growth (Rev YoY)"],
|
||||
}
|
||||
|
||||
|
||||
def _build_radar_figure(ticker: str) -> "go.Figure":
|
||||
"""Plotly radar chart from ticker data."""
|
||||
from utils.charts import _build_radar_common
|
||||
data = get_radar_metrics_normalized(ticker)
|
||||
if not data or not data.get("r"):
|
||||
return None
|
||||
return _build_radar_common(data["theta"], data["r"])
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_piotroski_fscore(ticker: str) -> dict:
|
||||
"""Piotroski F-Score (0-9) from last 2 periods. Uses yahooquery then yfinance with TTM fallback. Returns score + criteria + used_ttm."""
|
||||
out = {"score": 0, "criteria": [], "used_ttm": False}
|
||||
fin, bal, cf = _get_annual_financials_balance_cashflow(ticker)
|
||||
if fin is None or fin.empty or bal is None or bal.empty:
|
||||
return out
|
||||
if cf is None or cf.empty:
|
||||
cf = pd.DataFrame()
|
||||
try:
|
||||
ncol = min(2, len(fin.columns))
|
||||
rev = _get_row_series(fin, "Total Revenue", "Revenue")
|
||||
ni = _get_row_series(fin, "Net Income", "Net Income Common Stockholders")
|
||||
gross = _get_row_series(fin, "Gross Profit")
|
||||
ta = _get_row_series(bal, "Total Assets")
|
||||
lt_debt = _get_row_series(bal, "Long Term Debt")
|
||||
ca = _get_row_series(bal, "Current Assets")
|
||||
cl = _get_row_series(bal, "Current Liabilities")
|
||||
ocf = _get_row_series(cf, "Operating Cash Flow", "Cash From Operating Activities") if not cf.empty else None
|
||||
shares = _get_row_series(bal, "Share Issued") or _get_row_series(bal, "Ordinary Shares Number")
|
||||
if shares is None and yf:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = getattr(t, "info", None) or {}
|
||||
sh_info = info.get("sharesOutstanding") or info.get("Shares Outstanding")
|
||||
if sh_info is not None:
|
||||
try:
|
||||
sh_float = float(sh_info)
|
||||
shares = pd.Series([sh_float] * ncol, index=fin.columns[:ncol])
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
def v0(s):
|
||||
if s is None or len(s) == 0:
|
||||
return None
|
||||
x = _safe_float(s.iloc[0])
|
||||
return x if (x is not None and x == x and not (isinstance(x, float) and pd.isna(x))) else None
|
||||
def v1(s):
|
||||
if s is None or len(s) < 2:
|
||||
return None
|
||||
x = _safe_float(s.iloc[1])
|
||||
return x if (x is not None and x == x and not (isinstance(x, float) and pd.isna(x))) else None
|
||||
ni0, ni1 = v0(ni), v1(ni)
|
||||
ocf0 = v0(ocf) if ocf is not None else None
|
||||
ta0, ta1 = v0(ta), v1(ta)
|
||||
roa0 = (ni0 / ta0 * 100) if (ni0 is not None and ta0 is not None and ta0 != 0) else None
|
||||
roa1 = (ni1 / ta1 * 100) if (ni1 is not None and ta1 is not None and ta1 != 0) else None
|
||||
c1 = (ni0 is not None and ni0 > 0)
|
||||
c2 = (ocf0 is not None and ocf0 > 0)
|
||||
c3 = (roa0 is not None and roa1 is not None and roa0 > roa1)
|
||||
c4 = (ocf0 is not None and ni0 is not None and ocf0 > ni0)
|
||||
lt0 = v0(lt_debt) or 0
|
||||
lt1 = v1(lt_debt) or 0
|
||||
c5 = (ta0 is not None and ta0 != 0 and ta1 is not None and ta1 != 0 and (lt0 / ta0) < (lt1 / ta1))
|
||||
cl0, cl1 = v0(cl), v1(cl)
|
||||
ca0, ca1 = v0(ca), v1(ca)
|
||||
cr0 = (ca0 / cl0) if (ca0 is not None and cl0 is not None and cl0 != 0) else None
|
||||
cr1 = (ca1 / cl1) if (ca1 is not None and cl1 is not None and cl1 != 0) else None
|
||||
c6 = (cr0 is not None and cr1 is not None and cr0 > cr1)
|
||||
sh0, sh1 = v0(shares), v1(shares)
|
||||
c7 = (sh0 is not None and sh1 is not None and sh0 <= sh1) if (sh0 is not None and sh1 is not None) else True
|
||||
rev0, rev1 = v0(rev), v1(rev)
|
||||
gm0 = (v0(gross) / rev0 * 100) if (gross is not None and rev0 is not None and rev0 != 0) else None
|
||||
gm1 = (v1(gross) / rev1 * 100) if (gross is not None and rev1 is not None and rev1 != 0) else None
|
||||
c8 = (gm0 is not None and gm1 is not None and gm0 > gm1)
|
||||
at0 = (rev0 / ta0) if (rev0 is not None and ta0 is not None and ta0 != 0) else None
|
||||
at1 = (rev1 / ta1) if (rev1 is not None and ta1 is not None and ta1 != 0) else None
|
||||
c9 = (at0 is not None and at1 is not None and at0 > at1)
|
||||
criteria = [
|
||||
("Net Income > 0 (profitability)", c1),
|
||||
("Operating Cash Flow > 0 (cash generative)", c2),
|
||||
("ROA increased vs prior period (improving returns)", c3),
|
||||
("OCF > Net Income (earnings quality, less accruals)", c4),
|
||||
("Leverage decreased: LT Debt/Assets lower (less debt)", c5),
|
||||
("Current Ratio improved (better liquidity)", c6),
|
||||
("No dilution: shares unchanged or lower (no equity raise)", c7),
|
||||
("Gross Margin improved (pricing power)", c8),
|
||||
("Asset Turnover improved (efficiency)", c9),
|
||||
]
|
||||
score = sum(1 for _, p in criteria if p)
|
||||
out["score"] = score
|
||||
out["criteria"] = criteria
|
||||
out["used_ttm"] = bool(fin is not None and hasattr(fin, "columns") and len(fin.columns) > 0 and any(str(c).startswith("TTM") for c in fin.columns))
|
||||
return out
|
||||
except Exception:
|
||||
out["used_ttm"] = False
|
||||
return out
|
||||
|
||||
|
||||
@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 {}
|
||||
@@ -0,0 +1,98 @@
|
||||
"""
|
||||
AI-derived score/chart helper functions: build Sankey, Piotroski, Radar data from Gemini-extracted financials.
|
||||
"""
|
||||
from utils.charts import _radar_norm
|
||||
|
||||
|
||||
def sankey_data_from_ai(ai_dict: dict) -> dict:
|
||||
"""Build Sankey input dict from get_sec_financials_llm result (current_yr). Gross Profit = Revenue - CostOfRevenue; Operating Income = Gross Profit - OperatingExpenses."""
|
||||
out = {"revenue": 0, "cogs": 0, "gross_profit": 0, "opex": 0, "operating_income": 0, "tax_interest_other": 0, "net_income": 0}
|
||||
cur = (ai_dict or {}).get("current_yr") or {}
|
||||
revenue = max(0, (cur.get("Revenue") or 0))
|
||||
cogs = max(0, min(cur.get("CostOfRevenue") or 0, revenue - 1e-6))
|
||||
gross_profit = revenue - cogs
|
||||
opex = max(0, cur.get("OperatingExpenses") or 0)
|
||||
operating_income = gross_profit - opex
|
||||
net_income = cur.get("NetIncome") or 0
|
||||
tax_interest_other = max(0, operating_income - net_income) if operating_income > net_income else abs(min(0, operating_income - net_income))
|
||||
out["revenue"] = max(revenue, 1)
|
||||
out["cogs"] = cogs
|
||||
out["gross_profit"] = gross_profit
|
||||
out["opex"] = opex
|
||||
out["operating_income"] = operating_income
|
||||
out["tax_interest_other"] = tax_interest_other
|
||||
out["net_income"] = net_income
|
||||
return out
|
||||
|
||||
|
||||
def piotroski_from_ai(ai_dict: dict) -> dict:
|
||||
"""Piotroski F-Score (0-9) from AI-extracted current_yr vs previous_yr. Returns {score, criteria, used_ttm: True}."""
|
||||
out = {"score": 0, "criteria": [], "used_ttm": True}
|
||||
cur = (ai_dict or {}).get("current_yr") or {}
|
||||
prev = (ai_dict or {}).get("previous_yr") or {}
|
||||
if not cur:
|
||||
return out
|
||||
def v(d, k): return (d.get(k) or 0)
|
||||
ni0, ni1 = v(cur, "NetIncome"), v(prev, "NetIncome")
|
||||
ocf0 = v(cur, "OperatingCashFlow")
|
||||
ta0, ta1 = v(cur, "TotalAssets"), v(prev, "TotalAssets")
|
||||
roa0 = (ni0 / ta0 * 100) if ta0 and ta0 != 0 else None
|
||||
roa1 = (ni1 / ta1 * 100) if ta1 and ta1 != 0 else None
|
||||
c1 = ni0 > 0
|
||||
c2 = ocf0 > 0
|
||||
c3 = (roa0 is not None and roa1 is not None and roa0 > roa1)
|
||||
c4 = ocf0 > ni0
|
||||
lt0, lt1 = v(cur, "LongTermDebt"), v(prev, "LongTermDebt")
|
||||
c5 = (ta0 and ta1 and (lt0 / ta0) < (lt1 / ta1)) if ta0 and ta1 else False
|
||||
ca0, ca1 = v(cur, "CurrentAssets"), v(prev, "CurrentAssets")
|
||||
cl0, cl1 = v(cur, "CurrentLiabilities"), v(prev, "CurrentLiabilities")
|
||||
cr0 = (ca0 / cl0) if cl0 and cl0 != 0 else None
|
||||
cr1 = (ca1 / cl1) if cl1 and cl1 != 0 else None
|
||||
c6 = (cr0 is not None and cr1 is not None and cr0 > cr1)
|
||||
sh0, sh1 = v(cur, "SharesOutstanding"), v(prev, "SharesOutstanding")
|
||||
c7 = (sh0 <= sh1) if (sh0 and sh1) else True
|
||||
rev0, rev1 = v(cur, "Revenue"), v(prev, "Revenue")
|
||||
gm0 = ((rev0 - v(cur, "CostOfRevenue")) / rev0 * 100) if rev0 and rev0 != 0 else None
|
||||
gm1 = ((rev1 - v(prev, "CostOfRevenue")) / rev1 * 100) if rev1 and rev1 != 0 else None
|
||||
c8 = (gm0 is not None and gm1 is not None and gm0 > gm1)
|
||||
at0 = (rev0 / ta0) if rev0 and ta0 and ta0 != 0 else None
|
||||
at1 = (rev1 / ta1) if rev1 and ta1 and ta1 != 0 else None
|
||||
c9 = (at0 is not None and at1 is not None and at0 > at1)
|
||||
criteria = [
|
||||
("Net Income > 0 (profitability)", c1),
|
||||
("Operating Cash Flow > 0 (cash generative)", c2),
|
||||
("ROA increased vs prior period (improving returns)", c3),
|
||||
("OCF > Net Income (earnings quality, less accruals)", c4),
|
||||
("Leverage decreased: LT Debt/Assets lower (less debt)", c5),
|
||||
("Current Ratio improved (better liquidity)", c6),
|
||||
("No dilution: shares unchanged or lower (no equity raise)", c7),
|
||||
("Gross Margin improved (pricing power)", c8),
|
||||
("Asset Turnover improved (efficiency)", c9),
|
||||
]
|
||||
out["score"] = sum(1 for _, p in criteria if p)
|
||||
out["criteria"] = criteria
|
||||
return out
|
||||
|
||||
|
||||
def radar_metrics_from_ai(ai_dict: dict) -> dict:
|
||||
"""ROE, Current Ratio, Asset Turnover, Equity Mult, Revenue YoY from AI dict; normalized 0-100 for radar. Equity proxy: TotalAssets - CurrentLiabilities - LongTermDebt."""
|
||||
cur = (ai_dict or {}).get("current_yr") or {}
|
||||
prev = (ai_dict or {}).get("previous_yr") or {}
|
||||
if not cur:
|
||||
return {}
|
||||
eq0 = (cur.get("TotalAssets") or 0) - (cur.get("CurrentLiabilities") or 0) - (cur.get("LongTermDebt") or 0)
|
||||
if eq0 <= 0:
|
||||
eq0 = (cur.get("TotalAssets") or 0) * 0.5
|
||||
roe = (cur.get("NetIncome") or 0) / eq0 * 100 if eq0 else 0
|
||||
ca, cl = cur.get("CurrentAssets") or 0, cur.get("CurrentLiabilities") or 0
|
||||
current_ratio = (ca / cl) if cl and cl != 0 else 0
|
||||
ta = cur.get("TotalAssets") or 1
|
||||
asset_turnover = (cur.get("Revenue") or 0) / ta
|
||||
equity_mult = (cur.get("TotalAssets") or 0) / eq0 if eq0 else 0
|
||||
rev0, rev1 = cur.get("Revenue") or 0, prev.get("Revenue") or 0
|
||||
rev_yoy = ((rev0 - rev1) / rev1 * 100) if rev1 and rev1 != 0 else 0
|
||||
return {
|
||||
"theta": ["Profitability (ROE)", "Liquidity (Curr.Ratio)", "Efficiency (Asset Turn.)", "Solvency (Equity Mult.)", "Growth (Rev YoY)"],
|
||||
"r": _radar_norm(roe, current_ratio, asset_turnover, equity_mult, rev_yoy),
|
||||
"labels": ["Profitability (ROE)", "Liquidity (Curr.Ratio)", "Efficiency (Asset Turn.)", "Solvency (Equity Mult.)", "Growth (Rev YoY)"],
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
"""
|
||||
SEC 10-K download + section extraction: download via sec-edgar-downloader, extract items, cache.
|
||||
"""
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from config.constants import (
|
||||
ITEM1A_PATTERNS, ITEM3_PATTERNS, ITEM7_PATTERNS, ITEM8_PATTERNS, ITEM9A_PATTERNS,
|
||||
)
|
||||
from data.sec_parser import (
|
||||
find_item_section_generic, _find_section_start,
|
||||
_extract_item_from_full, clean_text_for_llm, smart_chunk,
|
||||
)
|
||||
from data.sec_fetcher import (
|
||||
get_edgar_downloader, find_downloaded_10k_path, find_all_10k_filing_dirs,
|
||||
get_main_10k_text, _load_10k_html_from_cache, _get_main_10k_html_file,
|
||||
_save_10k_html_to_cache, _load_10k_from_cache, _save_10k_to_cache,
|
||||
)
|
||||
|
||||
|
||||
def download_and_extract_all_items(ticker: str, email: str) -> dict:
|
||||
"""Download latest 10-K, extract Item 1A, 3, 7, 9A; clean and return (and optionally cache).
|
||||
Also saves the raw HTML file to data/TICKER_latest_raw.html for the native viewer."""
|
||||
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=1, download_details=True)
|
||||
filing_dir = find_downloaded_10k_path(download_root, ticker)
|
||||
if not filing_dir:
|
||||
raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.")
|
||||
full_text = get_main_10k_text(filing_dir)
|
||||
if not full_text:
|
||||
raise ValueError("Could not extract text from the 10-K.")
|
||||
# Save raw HTML to persistent cache while temp dir is still open
|
||||
if not _load_10k_html_from_cache(ticker):
|
||||
main_html_path = _get_main_10k_html_file(filing_dir)
|
||||
if main_html_path:
|
||||
try:
|
||||
with open(main_html_path, "r", encoding="utf-8", errors="replace") as _f:
|
||||
_save_10k_html_to_cache(ticker, _f.read())
|
||||
except Exception:
|
||||
pass
|
||||
item1a = find_item_section_generic(full_text, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80000)
|
||||
item3 = _extract_item_from_full(full_text, ITEM3_PATTERNS, 3, ["Legal", "Proceedings"], max_chars=40000)
|
||||
item9a = _extract_item_from_full(full_text, ITEM9A_PATTERNS, 9, ["Controls", "Procedures", "Internal"], max_chars=40000)
|
||||
start7 = _find_section_start(full_text, ITEM7_PATTERNS, 7)
|
||||
text_after_7 = full_text[start7:] if start7 >= 0 else full_text
|
||||
item7 = find_item_section_generic(text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000)
|
||||
if not item7 and text_after_7:
|
||||
item7 = text_after_7[:120000]
|
||||
item8 = _extract_item_from_full(full_text, ITEM8_PATTERNS, 8, ["Financial Statements", "Supplementary Data"], max_chars=200000)
|
||||
data = {
|
||||
"item1a": clean_text_for_llm(item1a or ""),
|
||||
"item3": clean_text_for_llm(item3 or ""),
|
||||
"item9a": clean_text_for_llm(item9a or ""),
|
||||
"item7": clean_text_for_llm(item7 or ""),
|
||||
"item8": clean_text_for_llm(item8 or ""),
|
||||
}
|
||||
_save_10k_to_cache(ticker, data)
|
||||
return data
|
||||
|
||||
|
||||
def get_10k_sections(ticker: str, email: str) -> tuple:
|
||||
"""Return (sections dict, status). status = 'cache' if loaded from file else 'downloaded'."""
|
||||
cached = _load_10k_from_cache(ticker)
|
||||
if cached is not None:
|
||||
return cached, "cache"
|
||||
return download_and_extract_all_items(ticker, email), "downloaded"
|
||||
|
||||
|
||||
def download_and_extract_item7_and_1a(ticker: str, email: str) -> tuple:
|
||||
"""Fetch 10-K and return full_text, Item 1A (Risk Factors), Item 7 (MD&A). Uses cache when available."""
|
||||
sections, _ = get_10k_sections(ticker, email)
|
||||
return "", sections.get("item1a", "") or "", sections.get("item7", "") or ""
|
||||
|
||||
|
||||
def download_item7_latest_and_3y_ago(ticker: str, email: str) -> tuple:
|
||||
"""Download up to 5 10-Ks; extract Item 1A (latest only) and Item 7 from latest and from 3 years ago."""
|
||||
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
|
||||
@@ -0,0 +1,252 @@
|
||||
"""
|
||||
SEC EDGAR download, fetch, cache: 10-K download via sec-edgar-downloader, EDGAR API HTML fetch, disk cache.
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
import tempfile
|
||||
import requests
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from utils.prefs import _DATA_DIR
|
||||
from config.constants import (
|
||||
ITEM1A_PATTERNS, ITEM3_PATTERNS, ITEM7_PATTERNS, ITEM8_PATTERNS, ITEM9A_PATTERNS,
|
||||
)
|
||||
from data.sec_parser import (
|
||||
extract_text_from_file, find_item_section_generic, _find_section_start,
|
||||
_extract_item_from_full, clean_text_for_llm,
|
||||
)
|
||||
|
||||
|
||||
def get_edgar_downloader():
|
||||
from sec_edgar_downloader import Downloader
|
||||
return Downloader
|
||||
|
||||
|
||||
def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path]:
|
||||
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)
|
||||
if subdirs:
|
||||
return subdirs[0]
|
||||
for base in (download_root / "sec-edgar-filings", download_root):
|
||||
if not base.exists():
|
||||
continue
|
||||
for company_dir in base.iterdir():
|
||||
if not company_dir.is_dir():
|
||||
continue
|
||||
path_10k = company_dir / "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)
|
||||
if subdirs:
|
||||
return subdirs[0]
|
||||
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"):
|
||||
for path in filing_dir.rglob(ext):
|
||||
try:
|
||||
t = extract_text_from_file(path)
|
||||
if len(t) > 1000:
|
||||
all_text.append((path, t))
|
||||
except Exception:
|
||||
continue
|
||||
if not all_text:
|
||||
return ""
|
||||
_, main_text = max(all_text, key=lambda x: len(x[1]))
|
||||
return main_text
|
||||
|
||||
|
||||
def _get_10k_cache_path(ticker: str) -> Path:
|
||||
"""Path for cached 10-K sections: data/TICKER_latest.json."""
|
||||
_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||||
return _DATA_DIR / f"{ticker.upper()}_latest.json"
|
||||
|
||||
|
||||
def _get_10k_html_cache_path(ticker: str) -> Path:
|
||||
"""Path for cached raw 10-K HTML: data/TICKER_latest_raw.html."""
|
||||
_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||||
return _DATA_DIR / f"{ticker.upper()}_latest_raw.html"
|
||||
|
||||
|
||||
def _load_10k_html_from_cache(ticker: str) -> Optional[str]:
|
||||
"""Load raw 10-K HTML from data/TICKER_latest_raw.html. Returns None if missing."""
|
||||
path = _get_10k_html_cache_path(ticker)
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||||
return f.read()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _save_10k_html_to_cache(ticker: str, html: str) -> None:
|
||||
"""Save raw 10-K HTML to data/TICKER_latest_raw.html."""
|
||||
path = _get_10k_html_cache_path(ticker)
|
||||
_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8", errors="replace") as f:
|
||||
f.write(html)
|
||||
|
||||
|
||||
def fetch_sec_filing_html(ticker: str, filing_type: str = "10-K") -> dict:
|
||||
"""Fetch raw SEC filing HTML from EDGAR public API.
|
||||
|
||||
Returns dict with keys:
|
||||
- html: str | None (the raw HTML content)
|
||||
- error: str | None (human-readable error for st.error())
|
||||
- doc_url: str | None (final document URL for reference)
|
||||
- source: 'cache' | 'edgar_api' | None
|
||||
"""
|
||||
# Check disk cache (only for 10-K for backward compat)
|
||||
if filing_type == "10-K":
|
||||
cached = _load_10k_html_from_cache(ticker)
|
||||
if cached:
|
||||
return {"html": cached, "error": None, "doc_url": None, "source": "cache"}
|
||||
|
||||
# SEC requires: "Company Name (contact@email.com)" format
|
||||
headers = {
|
||||
"User-Agent": "FQDC-Terminal (atlas-terminal@fqdc.io)",
|
||||
"Accept-Encoding": "gzip, deflate",
|
||||
}
|
||||
|
||||
# Step 1: ticker -> CIK
|
||||
url_tickers = "https://www.sec.gov/files/company_tickers.json"
|
||||
r = requests.get(url_tickers, headers={**headers, "Accept": "application/json"}, timeout=15)
|
||||
if not r.ok:
|
||||
return {"html": None, "error": f"SEC tickers lookup failed: HTTP {r.status_code}", "doc_url": url_tickers, "source": None}
|
||||
cik = None
|
||||
for entry in r.json().values():
|
||||
if entry.get("ticker", "").upper() == ticker.upper():
|
||||
cik = str(entry["cik_str"]).zfill(10)
|
||||
break
|
||||
if not cik:
|
||||
return {"html": None, "error": f"Ticker '{ticker}' not found in SEC company_tickers.json", "doc_url": None, "source": None}
|
||||
|
||||
# Step 2: find latest filing of requested type + primaryDocument
|
||||
url_submissions = f"https://data.sec.gov/submissions/CIK{cik}.json"
|
||||
r = requests.get(url_submissions, headers={**headers, "Accept": "application/json"}, timeout=15)
|
||||
if not r.ok:
|
||||
return {"html": None, "error": f"SEC submissions API failed: HTTP {r.status_code} for CIK {cik}", "doc_url": url_submissions, "source": None}
|
||||
filings = r.json().get("filings", {}).get("recent", {})
|
||||
forms = filings.get("form", [])
|
||||
accessions = filings.get("accessionNumber", [])
|
||||
primary_docs = filings.get("primaryDocument", [])
|
||||
filing_dates = filings.get("filingDate", [])
|
||||
|
||||
accession = None
|
||||
main_doc = None
|
||||
filing_date = None
|
||||
for form, acc, pdoc, fdate in zip(forms, accessions, primary_docs, filing_dates):
|
||||
if form == filing_type:
|
||||
accession = acc.replace("-", "")
|
||||
main_doc = pdoc
|
||||
filing_date = fdate
|
||||
break
|
||||
if not accession or not main_doc:
|
||||
return {"html": None, "error": f"No '{filing_type}' filing found for {ticker} (CIK {cik})", "doc_url": None, "source": None}
|
||||
|
||||
# Step 3: download the primary .htm document
|
||||
doc_url = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{accession}/{main_doc}"
|
||||
r = requests.get(doc_url, headers={**headers, "Accept": "text/html,application/xhtml+xml"}, timeout=90)
|
||||
if not r.ok:
|
||||
return {"html": None, "error": f"SEC document download failed: HTTP {r.status_code} for {doc_url}", "doc_url": doc_url, "source": None}
|
||||
|
||||
html_content = r.text
|
||||
if not html_content or len(html_content) < 500:
|
||||
return {"html": None, "error": f"SEC returned empty/tiny document ({len(html_content)} bytes) from {doc_url}", "doc_url": doc_url, "source": None}
|
||||
|
||||
# Cache to disk (10-K only)
|
||||
if filing_type == "10-K":
|
||||
_save_10k_html_to_cache(ticker, html_content)
|
||||
|
||||
return {"html": html_content, "error": None, "doc_url": doc_url, "source": "edgar_api", "filing_date": filing_date}
|
||||
|
||||
|
||||
def _wrap_edgar_html_for_iframe(raw_html: str, ticker: str) -> str:
|
||||
"""Inject a minimal CSS reset so EDGAR HTML renders cleanly inside components.html()."""
|
||||
inject_css = """
|
||||
<style>
|
||||
/* Soft readability reset for EDGAR documents */
|
||||
body {
|
||||
font-family: 'Times New Roman', Times, serif;
|
||||
font-size: 14px;
|
||||
line-height: 1.6;
|
||||
color: #1a1a1a;
|
||||
background: #ffffff;
|
||||
margin: 16px 24px;
|
||||
max-width: 1100px;
|
||||
}
|
||||
table { border-collapse: collapse; width: 100%; margin: 8px 0; font-size: 13px; }
|
||||
td, th { border: 1px solid #ccc; padding: 4px 8px; vertical-align: top; }
|
||||
th { background: #f0f0f0; font-weight: bold; }
|
||||
p { margin: 6px 0; }
|
||||
h1, h2, h3, h4 { color: #111; margin: 12px 0 6px; }
|
||||
a { color: #1155cc; }
|
||||
hr { border: none; border-top: 1px solid #ddd; margin: 12px 0; }
|
||||
</style>
|
||||
"""
|
||||
# If the HTML has a <head>, inject after it. Otherwise prepend.
|
||||
if "<head>" in raw_html.lower():
|
||||
raw_html = raw_html.replace("<head>", f"<head>{inject_css}", 1)
|
||||
elif "<html" in raw_html.lower():
|
||||
raw_html = raw_html.replace("<html", f"<html", 1)
|
||||
raw_html = inject_css + raw_html
|
||||
else:
|
||||
raw_html = f"<html><head>{inject_css}</head><body>{raw_html}</body></html>"
|
||||
return raw_html
|
||||
|
||||
|
||||
def _load_10k_from_cache(ticker: str) -> Optional[dict]:
|
||||
"""Load Item 1A, 3, 7, 8, 9A (plain text) from data/ticker_latest.json. Returns None if missing."""
|
||||
path = _get_10k_cache_path(ticker)
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _save_10k_to_cache(ticker: str, data: dict) -> None:
|
||||
"""Save cleaned 10-K sections to data/ticker_latest.json."""
|
||||
path = _get_10k_cache_path(ticker)
|
||||
_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=0)
|
||||
|
||||
|
||||
def _get_main_10k_html_file(filing_dir: Path) -> Optional[Path]:
|
||||
"""Return the Path of the largest .htm/.html file in the filing dir (the main document)."""
|
||||
candidates = []
|
||||
for ext in ("*.htm", "*.html"):
|
||||
for p in filing_dir.rglob(ext):
|
||||
try:
|
||||
candidates.append((p.stat().st_size, p))
|
||||
except Exception:
|
||||
pass
|
||||
if not candidates:
|
||||
return None
|
||||
candidates.sort(reverse=True)
|
||||
return candidates[0][1]
|
||||
|
||||
|
||||
# Download & extraction functions moved to data/sec_downloader.py:
|
||||
# download_and_extract_all_items, get_10k_sections,
|
||||
# download_and_extract_item7_and_1a, download_item7_latest_and_3y_ago
|
||||
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
SEC 10-K text extraction: HTML parsing, item section extraction, text cleaning.
|
||||
"""
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
from config.constants import (
|
||||
ITEM1A_PATTERNS, ITEM3_PATTERNS, ITEM7_PATTERNS, ITEM8_PATTERNS, ITEM9A_PATTERNS,
|
||||
)
|
||||
|
||||
|
||||
def _slice_html_items_1a_to_9a(raw_html: str) -> str:
|
||||
"""Fast string slice: keep only Item 1A through end of Item 9A to avoid parsing 50MB+ full file. Uses .find()/regex on raw string only."""
|
||||
if not raw_html or len(raw_html) < 5000:
|
||||
return raw_html
|
||||
start = -1
|
||||
for needle in ("Item 1A", "ITEM 1A", "Item 1a"):
|
||||
i = raw_html.find(needle)
|
||||
if i != -1 and (start == -1 or i < start):
|
||||
start = i
|
||||
if start == -1:
|
||||
m = re.search(r"Item\s+1A\s", raw_html, re.IGNORECASE)
|
||||
start = m.start() if m else 0
|
||||
else:
|
||||
start = max(0, start - 200)
|
||||
search_region = raw_html[start:]
|
||||
end_match = re.search(r"Item\s+10\s|Item\s+12\s|Part\s+III\b|PART\s+III\b", search_region, re.IGNORECASE)
|
||||
end = start + end_match.start() if end_match else len(raw_html)
|
||||
end = min(end, start + 8_000_000)
|
||||
return raw_html[start:end]
|
||||
|
||||
|
||||
def _extract_text_from_html_string(html_str: str) -> str:
|
||||
"""Parse HTML string with lxml; drop table/img/svg/style/script immediately to reduce memory and speed."""
|
||||
if not html_str or not html_str.strip():
|
||||
return ""
|
||||
try:
|
||||
soup = BeautifulSoup(html_str, "lxml")
|
||||
except Exception:
|
||||
soup = BeautifulSoup(html_str, "html.parser")
|
||||
for tag in soup.find_all(["table", "img", "svg", "style", "script"]):
|
||||
tag.decompose()
|
||||
return soup.get_text(separator="\n", strip=True)
|
||||
|
||||
|
||||
def extract_text_from_html(html_path: Path) -> str:
|
||||
try:
|
||||
with open(html_path, "r", encoding="utf-8", errors="replace") as f:
|
||||
raw = f.read()
|
||||
except Exception:
|
||||
with open(html_path, "r", encoding="latin-1", errors="replace") as f:
|
||||
raw = f.read()
|
||||
chunk = _slice_html_items_1a_to_9a(raw)
|
||||
return _extract_text_from_html_string(chunk)
|
||||
|
||||
|
||||
def extract_text_from_file(file_path: Path) -> str:
|
||||
suf = file_path.suffix.lower()
|
||||
if suf in (".htm", ".html"):
|
||||
return extract_text_from_html(file_path)
|
||||
if suf == ".txt":
|
||||
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
|
||||
text = f.read()
|
||||
text = re.sub(r"<[^>]+>", " ", text)
|
||||
text = re.sub(r"\s+", " ", text)
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def _find_section_start(text: str, patterns: list, item_num: int) -> int:
|
||||
for pat in patterns:
|
||||
m = re.search(pat, text, re.IGNORECASE)
|
||||
if m:
|
||||
return m.start()
|
||||
m = re.search(r"\bItem\s+" + str(item_num) + r"\b", text, re.IGNORECASE)
|
||||
return m.start() if m else -1
|
||||
|
||||
|
||||
def find_item_section_generic(text: str, patterns: list, item_num: int, title_keywords: list, max_chars: int = 120000) -> str:
|
||||
start = _find_section_start(text, patterns, item_num)
|
||||
if start == -1:
|
||||
pattern = re.compile(
|
||||
r"\bItem\s+" + str(item_num) + r"\b[.\s]*[^\n]*(" + "|".join(re.escape(k) for k in title_keywords) + r")?",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
match = pattern.search(text)
|
||||
if not match:
|
||||
return ""
|
||||
start = match.start()
|
||||
next_item = re.search(r"\n\s*Item\s+\d+[A-Z]?\s+", text[start + 100:], re.IGNORECASE)
|
||||
if next_item:
|
||||
end = start + 100 + next_item.start()
|
||||
else:
|
||||
end = min(start + max_chars, len(text))
|
||||
return text[start:end].strip()
|
||||
|
||||
|
||||
def clean_text_for_llm(html_content: str) -> str:
|
||||
"""Aggressive cleaning for LLM: strip tables/code, collapse whitespace, drop non-ASCII. Uses lxml for speed; drops table/img/svg/style/script."""
|
||||
if not html_content or not html_content.strip():
|
||||
return ""
|
||||
try:
|
||||
soup = BeautifulSoup(html_content, "lxml")
|
||||
for tag in soup.find_all(["table", "img", "style", "script", "svg", "math"]):
|
||||
tag.decompose()
|
||||
text = soup.get_text(separator=" ")
|
||||
except Exception:
|
||||
text = re.sub(r"<[^>]+>", " ", html_content)
|
||||
text = re.sub(r"\s+", " ", text)
|
||||
text = " ".join(text.split())
|
||||
text = re.sub(r"[^\x20-\x7E\n]", " ", text)
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
lines = []
|
||||
for line in text.split("\n"):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if re.fullmatch(r"\d+", line) or re.fullmatch(r"[\.\-\s\-]+", line):
|
||||
continue
|
||||
if re.match(r"^(page\s+\d+|\d+)\s*$", line, re.IGNORECASE) and len(line) < 20:
|
||||
continue
|
||||
lines.append(line)
|
||||
result = " ".join(lines)
|
||||
result = re.sub(r"\s+", " ", result).strip()
|
||||
return result
|
||||
|
||||
|
||||
def smart_chunk(section: str, max_chars: int = 10000, head_ratio: float = 0.5) -> str:
|
||||
"""Limit payload for Gemini; 10k chars ~ 2.5k tokens for fast response."""
|
||||
if not section or len(section) <= max_chars:
|
||||
return section
|
||||
head_size = int(max_chars * head_ratio)
|
||||
tail_size = max_chars - head_size - 100
|
||||
return section[:head_size] + " [ ... middle omitted ... ] " + section[-tail_size:]
|
||||
|
||||
|
||||
def _extract_item_from_full(text: str, patterns: list, item_num: int, keywords: list, max_chars: int = 60000) -> str:
|
||||
"""Extract one item section from full 10-K text."""
|
||||
start = _find_section_start(text, patterns, item_num)
|
||||
if start < 0:
|
||||
pattern = re.compile(r"\bItem\s+" + str(item_num) + r"[A-Z]?\b[.\s]*[^\n]*", re.IGNORECASE)
|
||||
match = pattern.search(text)
|
||||
start = match.start() if match else -1
|
||||
if start < 0:
|
||||
return ""
|
||||
next_item = re.search(r"\n\s*Item\s+\d+[A-Z]?\s+", text[start + 100:], re.IGNORECASE)
|
||||
end = start + 100 + next_item.start() if next_item else min(start + max_chars, len(text))
|
||||
return text[start:end].strip()
|
||||
@@ -0,0 +1,172 @@
|
||||
from typing import Optional
|
||||
import pandas as pd
|
||||
import streamlit as st
|
||||
from utils.formatting import _safe_float
|
||||
from data.financials import _get_row_series
|
||||
|
||||
try:
|
||||
import yfinance as yf
|
||||
except ImportError:
|
||||
yf = None
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_analyst_consensus(ticker: str) -> dict:
|
||||
"""Fetch analyst consensus from yfinance."""
|
||||
out = {"targetMeanPrice": None, "targetHighPrice": None, "targetLowPrice": None, "recommendationKey": "N/A", "revenueGrowth": "N/A", "earningsGrowth": "N/A", "numberOfAnalystOpinions": "N/A", "currentPrice": None}
|
||||
if not yf or not ticker:
|
||||
return out
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = t.info or {}
|
||||
for key in ("targetMeanPrice", "targetHighPrice", "targetLowPrice"):
|
||||
v = info.get(key)
|
||||
if v is not None:
|
||||
try:
|
||||
out[key] = float(v)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
out["currentPrice"] = info.get("currentPrice") or info.get("regularMarketPrice") or info.get("previousClose")
|
||||
out["numberOfAnalystOpinions"] = info.get("numberOfAnalystOpinions") or "N/A"
|
||||
rec = info.get("recommendationKey") or info.get("recommendation")
|
||||
if rec is not None:
|
||||
out["recommendationKey"] = str(rec)
|
||||
rg = info.get("revenueGrowth")
|
||||
if rg is not None:
|
||||
try:
|
||||
out["revenueGrowth"] = f"{float(rg) * 100:.1f}%"
|
||||
except (TypeError, ValueError):
|
||||
out["revenueGrowth"] = str(rg)
|
||||
eg = info.get("earningsGrowth")
|
||||
if eg is not None:
|
||||
try:
|
||||
out["earningsGrowth"] = f"{float(eg) * 100:.1f}%"
|
||||
except (TypeError, ValueError):
|
||||
out["earningsGrowth"] = str(eg)
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_dcf_smart_defaults(ticker: str) -> dict:
|
||||
"""Smart default assumptions: WACC from CAPM (Beta), Terminal Growth = 2.5%, FCF Growth from revenueGrowth/earningsGrowth or 8%."""
|
||||
out = {"wacc_pct": 10.0, "term_growth_pct": 2.5, "fcf_growth_pct": 8.0}
|
||||
if not yf or not ticker:
|
||||
return out
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = t.info or {}
|
||||
beta = info.get("beta")
|
||||
if beta is None:
|
||||
beta = 1.0
|
||||
else:
|
||||
try:
|
||||
beta = float(beta)
|
||||
except (TypeError, ValueError):
|
||||
beta = 1.0
|
||||
risk_free = 4.0
|
||||
market_risk_premium = 5.0
|
||||
calculated_wacc = risk_free + (beta * market_risk_premium)
|
||||
out["wacc_pct"] = round(min(20.0, max(4.0, calculated_wacc)), 1)
|
||||
out["term_growth_pct"] = 2.5
|
||||
rev_growth = info.get("revenueGrowth") or info.get("earningsGrowth")
|
||||
if rev_growth is not None:
|
||||
try:
|
||||
g = float(rev_growth)
|
||||
out["fcf_growth_pct"] = round(min(30.0, max(-10.0, g * 100)), 1)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return out
|
||||
except Exception:
|
||||
return out
|
||||
|
||||
|
||||
@st.cache_data(ttl=300)
|
||||
def get_fcff_fcfe_valuation(ticker: str) -> dict:
|
||||
"""FCFF/FCFE two-stage valuation model. Returns dict with fcff, fcfe, and per-share values."""
|
||||
if not yf:
|
||||
return {}
|
||||
try:
|
||||
t = yf.Ticker(ticker.upper())
|
||||
info = t.info or {}
|
||||
fin = t.financials
|
||||
bal = t.balance_sheet
|
||||
cf = t.cashflow
|
||||
if fin is None or fin.empty:
|
||||
return {}
|
||||
# Get latest year data
|
||||
rev = _get_row_series(fin, "Total Revenue", "Revenue")
|
||||
ebit = _get_row_series(fin, "EBIT", "Operating Income")
|
||||
ni = _get_row_series(fin, "Net Income", "Net Income Common Stockholders")
|
||||
ocf = _get_row_series(cf, "Operating Cash Flow", "Cash From Operating Activities") if cf is not None else None
|
||||
capx = _get_row_series(cf, "Capital Expenditure", "Capital Expenditures") if cf is not None else None
|
||||
dep = _get_row_series(cf, "Depreciation And Amortization", "Depreciation & Amortization") if cf is not None else None
|
||||
|
||||
r0 = _safe_float(rev.iloc[0]) if rev is not None and len(rev) > 0 else None
|
||||
ebit0 = _safe_float(ebit.iloc[0]) if ebit is not None and len(ebit) > 0 else None
|
||||
ni0 = _safe_float(ni.iloc[0]) if ni is not None and len(ni) > 0 else None
|
||||
ocf0 = _safe_float(ocf.iloc[0]) if ocf is not None and len(ocf) > 0 else None
|
||||
capx0 = abs(_safe_float(capx.iloc[0]) or 0) if capx is not None and len(capx) > 0 else 0
|
||||
dep0 = _safe_float(dep.iloc[0]) if dep is not None and len(dep) > 0 else 0
|
||||
|
||||
# Tax rate estimation
|
||||
tax_expense = _get_row_series(fin, "Tax Provision", "Income Tax Expense")
|
||||
pretax = _get_row_series(fin, "Pretax Income", "Income Before Tax")
|
||||
tax_rate = 0.21 # default US corporate
|
||||
if tax_expense is not None and pretax is not None and len(tax_expense) > 0 and len(pretax) > 0:
|
||||
te = _safe_float(tax_expense.iloc[0])
|
||||
pt = _safe_float(pretax.iloc[0])
|
||||
if pt and pt > 0 and te is not None:
|
||||
tax_rate = min(max(te / pt, 0.05), 0.40)
|
||||
|
||||
# Balance sheet items
|
||||
total_debt_series = _get_row_series(bal, "Total Debt") if bal is not None else None
|
||||
cash_series = _get_row_series(bal, "Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments") if bal is not None else None
|
||||
equity_series = _get_row_series(bal, "Total Stockholder Equity", "Stockholders Equity", "Total Equity Gross Minority Interest") if bal is not None else None
|
||||
|
||||
total_debt = _safe_float(total_debt_series.iloc[0]) if total_debt_series is not None and len(total_debt_series) > 0 else 0
|
||||
cash_val = _safe_float(cash_series.iloc[0]) if cash_series is not None and len(cash_series) > 0 else 0
|
||||
equity_val = _safe_float(equity_series.iloc[0]) if equity_series is not None and len(equity_series) > 0 else 0
|
||||
|
||||
shares = info.get("sharesOutstanding") or info.get("impliedSharesOutstanding") or 1
|
||||
beta = info.get("beta") or 1.0
|
||||
|
||||
# FCFF = EBIT(1-t) + D&A - CapEx - delta WC (approximate)
|
||||
fcff = None
|
||||
if ebit0 is not None:
|
||||
fcff = ebit0 * (1 - tax_rate) + (dep0 or 0) - capx0
|
||||
|
||||
# FCFE = Net Income + D&A - CapEx - delta WC + Net Borrowing (approximate as NI + D&A - CapEx)
|
||||
fcfe = None
|
||||
if ni0 is not None:
|
||||
fcfe = ni0 + (dep0 or 0) - capx0
|
||||
|
||||
# WACC components
|
||||
rf = 0.045 # risk-free rate
|
||||
erp = 0.055 # equity risk premium
|
||||
cost_of_equity = rf + beta * erp
|
||||
cost_of_debt = 0.05 # approximate
|
||||
if total_debt and equity_val and (total_debt + equity_val) > 0:
|
||||
debt_weight = total_debt / (total_debt + equity_val)
|
||||
equity_weight = equity_val / (total_debt + equity_val)
|
||||
else:
|
||||
debt_weight, equity_weight = 0.2, 0.8
|
||||
|
||||
wacc = equity_weight * cost_of_equity + debt_weight * cost_of_debt * (1 - tax_rate)
|
||||
|
||||
# Margins
|
||||
fcff_margin = (fcff / r0 * 100) if fcff and r0 and r0 > 0 else None
|
||||
fcfe_margin = (fcfe / r0 * 100) if fcfe and r0 and r0 > 0 else None
|
||||
|
||||
return {
|
||||
"fcff": fcff, "fcfe": fcfe, "fcff_margin": fcff_margin, "fcfe_margin": fcfe_margin,
|
||||
"ebit": ebit0, "net_income": ni0, "revenue": r0,
|
||||
"tax_rate": tax_rate * 100, "depreciation": dep0, "capex": capx0,
|
||||
"total_debt": total_debt, "cash": cash_val, "equity": equity_val,
|
||||
"shares": shares, "beta": beta,
|
||||
"wacc": wacc * 100, "cost_of_equity": cost_of_equity * 100, "cost_of_debt": cost_of_debt * 100,
|
||||
"debt_weight": debt_weight * 100, "equity_weight": equity_weight * 100,
|
||||
}
|
||||
except Exception:
|
||||
return {}
|
||||
Reference in New Issue
Block a user