diff options
Diffstat (limited to 'backend/app')
| -rw-r--r-- | backend/app/schemas.py | 18 | ||||
| -rw-r--r-- | backend/app/services/data_service.py | 313 |
2 files changed, 310 insertions, 21 deletions
diff --git a/backend/app/schemas.py b/backend/app/schemas.py index db0076d..2ee6dac 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -39,6 +39,23 @@ class OverviewStats(BaseModel): beta: float | None = None +class OverviewRatios(BaseModel): + price_to_book: float | None = None + price_to_sales: float | None = None + ev_to_sales: float | None = None + ev_to_ebitda: float | None = None + gross_margin_ttm: float | None = None + operating_margin_ttm: float | None = None + net_margin_ttm: float | None = None + roe_ttm: float | None = None + roa_ttm: float | None = None + roic_ttm: float | None = None + debt_to_equity: float | None = None + current_ratio: float | None = None + dividend_yield_ttm: float | None = None + dividend_payout_ratio_ttm: float | None = None + + class Range52Week(BaseModel): low: float | None = None high: float | None = None @@ -75,6 +92,7 @@ class TickerOverview(BaseModel): quote: Quote signals: list[Signal] = Field(default_factory=list) stats: OverviewStats + ratios: OverviewRatios range_52w: Range52Week short_interest: ShortInterest meta: OverviewMeta diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py index ae078cd..6f8587f 100644 --- a/backend/app/services/data_service.py +++ b/backend/app/services/data_service.py @@ -18,10 +18,18 @@ PRICE_CACHE = TTLCache(maxsize=256, ttl=300) HISTORY_CACHE = TTLCache(maxsize=256, ttl=300) INTRADAY_CACHE = TTLCache(maxsize=128, ttl=60) MARKET_CACHE = TTLCache(maxsize=8, ttl=300) +STATEMENT_CACHE = TTLCache(maxsize=256, ttl=3600) +SHARES_CACHE = TTLCache(maxsize=256, ttl=3600) +RATIO_CACHE = TTLCache(maxsize=256, ttl=3600) PERIODS = {"1m", "3m", "6m", "1y", "5y"} YF_PERIOD_MAP = {"1m": "1mo", "3m": "3mo", "6m": "6mo", "1y": "1y", "5y": "5y"} _XMAP = {"NYQ": "NYSE", "NMS": "NASDAQ", "NGM": "NASDAQ", "NCM": "NASDAQ", "ASE": "AMEX"} +_SHARE_LABELS = ( + "Ordinary Shares Number", + "Share Issued", + "Common Stock Shares Outstanding", +) def normalize_symbol(symbol: str) -> str: @@ -55,6 +63,46 @@ def _json_value(value: Any) -> Any: return value +def _cap_ratio(value: float | None, lower: float, upper: float) -> float | None: + if value is None or value <= lower or value >= upper: + return None + return value + + +def _balance_value(frame: pd.DataFrame, *labels: str) -> float | None: + if frame is None or frame.empty: + return None + for label in labels: + if label not in frame.index: + continue + series = pd.to_numeric(frame.loc[label], errors="coerce").dropna() + if series.empty: + continue + value = _safe_float(series.iloc[0]) + if value is not None: + return value + return None + + +def _statement_ttm(frame: pd.DataFrame, *labels: str) -> float | None: + if frame is None or frame.empty: + return None + for label in labels: + if label not in frame.index: + continue + series = pd.to_numeric(frame.loc[label].iloc[:4], errors="coerce").dropna() + if len(series) == 4: + value = _safe_float(series.sum()) + if value is not None: + return value + return None + + +def _latest_share_count(balance_sheet: pd.DataFrame) -> float | None: + shares = _balance_value(balance_sheet, *_SHARE_LABELS) + return shares if shares is not None and shares > 0 else None + + def _pick_search_match(symbol: str) -> dict[str, Any]: sym = normalize_symbol(symbol) results = search_tickers(sym) @@ -97,8 +145,7 @@ def get_company_info(symbol: str) -> dict[str, Any]: info = yf.Ticker(sym).info or {} if not isinstance(info, dict): return {} - cleaned = {str(k): _json_value(v) for k, v in info.items()} - return cleaned + return {str(k): _json_value(v) for k, v in info.items()} except Exception: return {} @@ -187,6 +234,76 @@ def get_intraday_history(symbol: str, period: str, interval: str) -> list[dict[s return [] +@cached(STATEMENT_CACHE) +def get_income_statement(symbol: str, quarterly: bool = False) -> pd.DataFrame: + try: + ticker = yf.Ticker(normalize_symbol(symbol)) + frame = ticker.quarterly_income_stmt if quarterly else ticker.income_stmt + return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() + except Exception: + return pd.DataFrame() + + +@cached(STATEMENT_CACHE) +def get_balance_sheet(symbol: str, quarterly: bool = False) -> pd.DataFrame: + try: + ticker = yf.Ticker(normalize_symbol(symbol)) + frame = ticker.quarterly_balance_sheet if quarterly else ticker.balance_sheet + return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() + except Exception: + return pd.DataFrame() + + +@cached(STATEMENT_CACHE) +def get_cash_flow(symbol: str, quarterly: bool = False) -> pd.DataFrame: + try: + ticker = yf.Ticker(normalize_symbol(symbol)) + frame = ticker.quarterly_cashflow if quarterly else ticker.cashflow + return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() + except Exception: + return pd.DataFrame() + + +@cached(SHARES_CACHE) +def get_shares_outstanding(symbol: str) -> float | None: + sym = normalize_symbol(symbol) + info = get_company_info(sym) + for key in ("sharesOutstanding", "impliedSharesOutstanding"): + shares = _safe_float(info.get(key)) + if shares is not None and shares > 0: + return shares + + fast_info = get_fast_info(sym) + shares = _safe_float(fast_info.get("shares")) + if shares is not None and shares > 0: + return shares + + balance_sheet = get_balance_sheet(sym, quarterly=True) + shares = _latest_share_count(balance_sheet) + if shares is not None: + return shares + + try: + history = yf.Ticker(sym).get_shares_full(start="2000-01-01") + if isinstance(history, pd.Series): + values = pd.to_numeric(history, errors="coerce").dropna() + if not values.empty: + latest = _safe_float(values.iloc[-1]) + if latest is not None and latest > 0: + return latest + except Exception: + pass + return None + + +def get_market_cap_computed(symbol: str, price: float | None = None, shares: float | None = None) -> float | None: + latest_price = price if price is not None else get_latest_price(symbol) + share_count = shares if shares is not None else get_shares_outstanding(symbol) + if latest_price is not None and latest_price > 0 and share_count is not None and share_count > 0: + return latest_price * share_count + return None + + def _history_rows(df: pd.DataFrame, include_time: bool) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for idx, row in df.iterrows(): @@ -246,9 +363,11 @@ def build_quote(info: dict[str, Any], symbol: str) -> dict[str, Any]: return {"price": price, "prev_close": prev_close, "change": change, "change_pct": change_pct} -def build_signals(info: dict[str, Any]) -> list[dict[str, str]]: +def build_signals(info: dict[str, Any], ratios: dict[str, Any]) -> list[dict[str, str]]: signals: list[dict[str, str]] = [] pe = _safe_float(info.get("trailingPE")) + if pe is None: + pe = _safe_float(ratios.get("trailing_pe")) if pe is not None and pe > 0: if pe < 15: signals.append({"key": "Valuation", "state": "pos", "value": f"P/E {pe:.1f}x", "description": "Attractive multiple"}) @@ -260,18 +379,25 @@ def build_signals(info: dict[str, Any]) -> list[dict[str, str]]: signals.append({"key": "Valuation", "state": "neu", "value": "P/E unavailable", "description": "No trailing earnings"}) _ratio_signal(signals, "Growth", info.get("revenueGrowth"), 0.10, 0.0, "Strong top-line growth", "Low but positive growth", "Contracting revenue") - _ratio_signal(signals, "Profit", info.get("profitMargins"), 0.15, 0.05, "High net margin", "Moderate net margin", "Thin or negative margin") + + profit = _safe_float(info.get("profitMargins")) + if profit is None: + profit = _safe_float(ratios.get("net_margin_ttm")) + _ratio_signal(signals, "Profit", profit, 0.15, 0.05, "High net margin", "Moderate net margin", "Thin or negative margin") debt_to_equity = _safe_float(info.get("debtToEquity")) if debt_to_equity is not None: - de_x = debt_to_equity / 100.0 - if de_x < 0.5: + debt_to_equity = debt_to_equity / 100.0 + else: + debt_to_equity = _safe_float(ratios.get("debt_to_equity")) + if debt_to_equity is not None: + if debt_to_equity < 0.5: state, desc = "pos", "Low leverage" - elif de_x < 2.0: + elif debt_to_equity < 2.0: state, desc = "warn", "Moderate leverage" else: state, desc = "neg", "High leverage" - signals.append({"key": "Leverage", "state": state, "value": f"D/E {de_x:.2f}x", "description": desc}) + signals.append({"key": "Leverage", "state": state, "value": f"D/E {debt_to_equity:.2f}x", "description": desc}) return signals @@ -295,7 +421,8 @@ def _ratio_signal( state, desc = "warn", warn_desc else: state, desc = "neg", negative_desc - signals.append({"key": key, "state": state, "value": f"{ratio * 100:+.0f}%" if key == "Growth" else f"{ratio * 100:.0f}%", "description": desc}) + formatted = f"{ratio * 100:+.0f}%" if key == "Growth" else f"{ratio * 100:.0f}%" + signals.append({"key": key, "state": state, "value": formatted, "description": desc}) def _field(source_map: dict[str, dict[str, Any]], field_sources: dict[str, str], name: str, *candidates: tuple[str, str]) -> Any: @@ -412,11 +539,111 @@ def _build_profile(sym: str, info: dict[str, Any], fast_info: dict[str, Any], se } +@cached(RATIO_CACHE) +def compute_ttm_ratios(symbol: str) -> dict[str, Any]: + sym = normalize_symbol(symbol) + info = get_company_info(sym) + price = _safe_float(info.get("currentPrice") or info.get("regularMarketPrice")) or get_latest_price(sym) + shares = get_shares_outstanding(sym) + income = get_income_statement(sym, quarterly=True) + balance = get_balance_sheet(sym, quarterly=True) + cash_flow = get_cash_flow(sym, quarterly=True) + + if income is None or income.empty: + return {} + + revenue = _statement_ttm(income, "Total Revenue") + gross_profit = _statement_ttm(income, "Gross Profit") + operating_income = _statement_ttm(income, "Operating Income") + net_income = _statement_ttm(income, "Net Income") + ebit = _statement_ttm(income, "EBIT") + ebitda = _statement_ttm(income, "EBITDA", "Normalized EBITDA") + tax_provision = _statement_ttm(income, "Tax Provision") + pretax_income = _statement_ttm(income, "Pretax Income") + + equity = _balance_value(balance, "Stockholders Equity", "Common Stock Equity") + total_assets = _balance_value(balance, "Total Assets") + total_debt = _balance_value(balance, "Total Debt", "Long Term Debt And Capital Lease Obligation") + current_assets = _balance_value(balance, "Current Assets") + current_liabilities = _balance_value(balance, "Current Liabilities") + cash = _balance_value(balance, "Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments") or 0.0 + + market_cap = get_market_cap_computed(sym, price=price, shares=shares) + trailing_eps = None + if shares is not None and shares > 0 and net_income is not None: + trailing_eps = net_income / shares + + ratios: dict[str, Any] = {} + ratios["market_cap"] = market_cap + ratios["trailing_eps"] = trailing_eps + + if revenue and revenue > 0: + if gross_profit is not None: + ratios["gross_margin_ttm"] = gross_profit / revenue + if operating_income is not None: + ratios["operating_margin_ttm"] = operating_income / revenue + if net_income is not None: + ratios["net_margin_ttm"] = net_income / revenue + + if equity and equity > 0 and net_income is not None: + roe = net_income / equity + if abs(roe) < 10: + ratios["roe_ttm"] = roe + + if total_assets and total_assets > 0 and net_income is not None: + roa = net_income / total_assets + if abs(roa) < 10: + ratios["roa_ttm"] = roa + + if ebit is not None and pretax_income not in (None, 0): + effective_tax_rate = max(0.0, (tax_provision or 0.0) / pretax_income) + invested_capital = (equity or 0.0) + (total_debt or 0.0) - cash + if invested_capital > 0: + roic = (ebit * (1 - effective_tax_rate)) / invested_capital + if abs(roic) < 10: + ratios["roic_ttm"] = roic + + if market_cap and market_cap > 0: + if net_income and net_income > 0: + ratios["trailing_pe"] = market_cap / net_income + if revenue and revenue > 0: + ratios["price_to_sales"] = _cap_ratio(market_cap / revenue, 0, 100) + if equity and equity > 0: + ratios["price_to_book"] = _cap_ratio(market_cap / equity, 0, 100) + + enterprise_value = market_cap + (total_debt or 0.0) - cash + if revenue and revenue > 0: + ratios["ev_to_sales"] = _cap_ratio(enterprise_value / revenue, 0, 100) + if ebitda and ebitda > 1e6: + ratios["ev_to_ebitda"] = _cap_ratio(enterprise_value / ebitda, 0, 500) + + if equity and equity > 0 and total_debt is not None: + ratios["debt_to_equity"] = _cap_ratio(total_debt / equity, -1, 100) + + if current_liabilities and current_liabilities > 0 and current_assets is not None: + ratios["current_ratio"] = current_assets / current_liabilities + + dividends_paid = _statement_ttm(cash_flow, "Cash Dividends Paid", "Common Stock Dividend Paid") + if dividends_paid is not None: + dividends_paid = abs(dividends_paid) + if market_cap and market_cap > 0: + div_yield = dividends_paid / market_cap + if 0 <= div_yield < 1: + ratios["dividend_yield_ttm"] = div_yield + if net_income and net_income > 0: + payout = dividends_paid / net_income + if 0 <= payout < 10: + ratios["dividend_payout_ratio_ttm"] = payout + + return {key: value for key, value in ratios.items() if value is not None} + + def _build_quote_and_stats( info: dict[str, Any], fast_info: dict[str, Any], month_history: list[dict[str, Any]], year_history: list[dict[str, Any]], + computed: dict[str, Any], field_sources: dict[str, str], ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: month_snapshot = _history_snapshot(month_history) @@ -426,6 +653,7 @@ def _build_quote_and_stats( "fast_info": fast_info, "history_recent": month_snapshot, "history_year": year_snapshot, + "computed": computed, } price = _safe_float( @@ -478,9 +706,11 @@ def _build_quote_and_stats( ("history_recent", "averageVolume"), ) ) - market_cap = _safe_float(_field(source_map, field_sources, "stats.market_cap", ("info", "marketCap"), ("fast_info", "marketCap"))) - trailing_pe = _safe_float(_field(source_map, field_sources, "stats.trailing_pe", ("info", "trailingPE"))) - trailing_eps = _safe_float(_field(source_map, field_sources, "stats.trailing_eps", ("info", "trailingEps"))) + market_cap = _safe_float( + _field(source_map, field_sources, "stats.market_cap", ("info", "marketCap"), ("fast_info", "marketCap"), ("computed", "market_cap")) + ) + trailing_pe = _safe_float(_field(source_map, field_sources, "stats.trailing_pe", ("info", "trailingPE"), ("computed", "trailing_pe"))) + trailing_eps = _safe_float(_field(source_map, field_sources, "stats.trailing_eps", ("info", "trailingEps"), ("computed", "trailing_eps"))) beta = _safe_float(_field(source_map, field_sources, "stats.beta", ("info", "beta"))) range_low = _safe_float( _field( @@ -513,23 +743,46 @@ def _build_quote_and_stats( "average_volume": average_volume, "beta": beta, }, - { - "low": range_low, - "high": range_high, - "price": price, - }, + {"low": range_low, "high": range_high, "price": price}, ) +def _build_ratios(computed: dict[str, Any], field_sources: dict[str, str]) -> dict[str, Any]: + ratios: dict[str, Any] = {} + keys = ( + "price_to_book", + "price_to_sales", + "ev_to_sales", + "ev_to_ebitda", + "gross_margin_ttm", + "operating_margin_ttm", + "net_margin_ttm", + "roe_ttm", + "roa_ttm", + "roic_ttm", + "debt_to_equity", + "current_ratio", + "dividend_yield_ttm", + "dividend_payout_ratio_ttm", + ) + for key in keys: + value = _safe_float(computed.get(key)) + ratios[key] = value + if value is not None: + field_sources[f"ratios.{key}"] = "computed" + return ratios + + def _has_any_overview_data( profile: dict[str, Any], quote: dict[str, Any], stats: dict[str, Any], + ratios: dict[str, Any], range_52w: dict[str, Any], short_interest: dict[str, Any], field_sources: dict[str, str], ) -> bool: - for bucket in (profile, quote, stats, range_52w, short_interest): + for bucket in (profile, quote, stats, ratios, range_52w, short_interest): for key, value in bucket.items(): if key == "symbol": continue @@ -549,10 +802,13 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: fast_info = get_fast_info(sym) month_history = get_price_history(sym, period="1m") year_history = get_price_history(sym, period="1y") + computed = compute_ttm_ratios(sym) field_sources: dict[str, str] = {} profile = _build_profile(sym, info, fast_info, search_match, field_sources) - quote, stats, range_52w = _build_quote_and_stats(info, fast_info, month_history, year_history, field_sources) + quote, stats, range_52w = _build_quote_and_stats(info, fast_info, month_history, year_history, computed, field_sources) + ratios = _build_ratios(computed, field_sources) + short = _safe_int(info.get("sharesShort")) short_prior = _safe_int(info.get("sharesShortPriorMonth")) short_delta = None @@ -566,7 +822,7 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: "shares_short_delta_pct": short_delta, } - if not _has_any_overview_data(profile, quote, stats, range_52w, short_interest, field_sources): + if not _has_any_overview_data(profile, quote, stats, ratios, range_52w, short_interest, field_sources): return None field_availability = { @@ -584,6 +840,20 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: "stats.volume": stats.get("volume") is not None, "stats.average_volume": stats.get("average_volume") is not None, "stats.beta": stats.get("beta") is not None, + "ratios.price_to_book": ratios.get("price_to_book") is not None, + "ratios.price_to_sales": ratios.get("price_to_sales") is not None, + "ratios.ev_to_sales": ratios.get("ev_to_sales") is not None, + "ratios.ev_to_ebitda": ratios.get("ev_to_ebitda") is not None, + "ratios.gross_margin_ttm": ratios.get("gross_margin_ttm") is not None, + "ratios.operating_margin_ttm": ratios.get("operating_margin_ttm") is not None, + "ratios.net_margin_ttm": ratios.get("net_margin_ttm") is not None, + "ratios.roe_ttm": ratios.get("roe_ttm") is not None, + "ratios.roa_ttm": ratios.get("roa_ttm") is not None, + "ratios.roic_ttm": ratios.get("roic_ttm") is not None, + "ratios.debt_to_equity": ratios.get("debt_to_equity") is not None, + "ratios.current_ratio": ratios.get("current_ratio") is not None, + "ratios.dividend_yield_ttm": ratios.get("dividend_yield_ttm") is not None, + "ratios.dividend_payout_ratio_ttm": ratios.get("dividend_payout_ratio_ttm") is not None, "range_52w.low": range_52w.get("low") is not None, "range_52w.high": range_52w.get("high") is not None, } @@ -592,8 +862,9 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: return { "profile": profile, "quote": quote, - "signals": build_signals(info), + "signals": build_signals(info, computed), "stats": stats, + "ratios": ratios, "range_52w": range_52w, "short_interest": short_interest, "meta": { |