diff options
Diffstat (limited to 'backend/app/services/data_service.py')
| -rw-r--r-- | backend/app/services/data_service.py | 294 |
1 files changed, 287 insertions, 7 deletions
diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py index caa8294..cb53ea8 100644 --- a/backend/app/services/data_service.py +++ b/backend/app/services/data_service.py @@ -6,13 +6,17 @@ import math import os import statistics from collections import defaultdict -from typing import Any +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any, Sequence, TypedDict import httpx import pandas as pd import yfinance as yf from cachetools import TTLCache, cached +from app.schemas import AdvancedDcfInputs + SEARCH_CACHE = TTLCache(maxsize=128, ttl=60) INFO_CACHE = TTLCache(maxsize=256, ttl=300) FAST_INFO_CACHE = TTLCache(maxsize=256, ttl=300) @@ -31,6 +35,7 @@ BETA_CACHE = TTLCache(maxsize=256, ttl=3600) SHORT_CACHE = TTLCache(maxsize=256, ttl=3600) FINANCIALS_CACHE = TTLCache(maxsize=128, ttl=3600) VALUATION_CACHE = TTLCache(maxsize=128, ttl=3600) +WACC_CACHE = TTLCache(maxsize=128, ttl=3600) HIST_RATIOS_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) RATIOS_ENDPOINT_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) SECTOR_BENCHMARK_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) @@ -45,6 +50,25 @@ _SHARE_LABELS = ( "Share Issued", "Common Stock Shares Outstanding", ) +_SIMPLE_WACC_RANGE = [0.08, 0.09, 0.10, 0.11, 0.12] +_SIMPLE_TERMINAL_GROWTH_RANGE = [0.015, 0.02, 0.03, 0.04, 0.045] +_WACC_OFFSETS = [-0.02, -0.01, 0.0, 0.01, 0.02] +_TERMINAL_GROWTH_OFFSETS = [-0.015, -0.01, 0.0, 0.01, 0.015] + + +class _SensitivityGrid(TypedDict): + wacc: list[float] + terminal_growth: list[float] + implied_prices: list[list[float | None]] + + +@dataclass(frozen=True, slots=True) +class _DcfCapitalClaims: + shares_outstanding: float + total_debt: float + cash: float + preferred_equity: float + minority_interest: float def normalize_symbol(symbol: str) -> str: @@ -365,12 +389,12 @@ def _run_dcf( } projected = [base_fcf * ((1 + growth_rate) ** yr) for yr in range(1, projection_years + 1)] - discounted = [fcf / ((1 + wacc) ** i) for i, fcf in enumerate(projected, start=1)] + discounted = [fcf / ((1 + wacc) ** (year - 0.5)) for year, fcf in enumerate(projected, start=1)] fcf_pv_sum = float(sum(discounted)) terminal_fcf = float(projected[-1]) * (1 + terminal_growth) terminal_value = terminal_fcf / (wacc - terminal_growth) - terminal_value_pv = terminal_value / ((1 + wacc) ** projection_years) + terminal_value_pv = terminal_value / ((1 + wacc) ** (projection_years - 0.5)) enterprise_value = fcf_pv_sum + terminal_value_pv total_debt = float(total_debt or 0.0) @@ -396,6 +420,143 @@ def _run_dcf( } +def _run_dcf_explicit_build( + base_revenue: float, + revenue_growth: Sequence[float], + ebitda_margin: Sequence[float], + dna_pct_revenue: Sequence[float], + capex_pct_revenue: Sequence[float], + nwc_chg_pct_delta_rev: Sequence[float], + tax_rate: Sequence[float], + wacc: float, + terminal_growth: float, + shares_outstanding: float, + total_debt: float, + cash: float, + preferred_equity: float, + minority_interest: float, + projection_years: int, +) -> dict: + if base_revenue <= 0: + return {"available": True, "error": "Base revenue must be greater than 0."} + if shares_outstanding <= 0: + return {"available": True, "error": "Shares outstanding must be greater than 0."} + if wacc <= 0: + return {"available": True, "error": "WACC must be greater than 0%."} + if terminal_growth >= wacc: + return {"available": True, "error": "Terminal growth must be lower than WACC."} + if projection_years <= 0: + return {"available": True, "error": "Projection years must be greater than 0."} + + assumptions = ( + revenue_growth, + ebitda_margin, + dna_pct_revenue, + capex_pct_revenue, + nwc_chg_pct_delta_rev, + tax_rate, + ) + if any(len(values) < projection_years for values in assumptions): + return {"available": True, "error": "Projection assumptions must cover every projection year."} + + revenue = float(base_revenue) + projected_fcf: list[float] = [] + for year in range(projection_years): + prior_revenue = revenue + revenue = prior_revenue * (1 + revenue_growth[year]) + ebitda = revenue * ebitda_margin[year] + dna = revenue * dna_pct_revenue[year] + ebit = ebitda - dna + nopat = ebit * (1 - tax_rate[year]) + capex = revenue * capex_pct_revenue[year] + delta_rev = 0.0 if year == 0 else revenue - prior_revenue + nwc_change = delta_rev * nwc_chg_pct_delta_rev[year] + projected_fcf.append(nopat + dna - capex - nwc_change) + + discounted_fcf = [ + fcf / ((1 + wacc) ** (year + 0.5)) + for year, fcf in enumerate(projected_fcf) + ] + fcf_pv_sum = float(sum(discounted_fcf)) + + terminal_fcf = projected_fcf[-1] * (1 + terminal_growth) + terminal_value = terminal_fcf / (wacc - terminal_growth) + terminal_value_pv = terminal_value / ((1 + wacc) ** (projection_years - 0.5)) + + enterprise_value = fcf_pv_sum + terminal_value_pv + net_debt = float(total_debt or 0.0) - float(cash or 0.0) + equity_value = enterprise_value - net_debt - float(preferred_equity or 0.0) - float(minority_interest or 0.0) + + return { + "available": True, + "intrinsic_value_per_share": equity_value / shares_outstanding, + "enterprise_value": enterprise_value, + "equity_value": equity_value, + "net_debt": net_debt, + "cash_and_equivalents": float(cash or 0.0), + "total_debt": float(total_debt or 0.0), + "terminal_value_pv": terminal_value_pv, + "fcf_pv_sum": fcf_pv_sum, + "base_fcf": projected_fcf[0], + "wacc": wacc, + "terminal_growth": terminal_growth, + "projection_years": projection_years, + } + + +def _centered_rate_range(center: float, offsets: Sequence[float]) -> list[float]: + return [round(center + offset, 4) for offset in offsets] + + +def _dcf_sensitivity_grid( + run_case: Callable[[float, float], dict[str, bool | str | float | None]], + wacc_values: Sequence[float], + terminal_growth_values: Sequence[float], +) -> _SensitivityGrid: + implied_prices: list[list[float | None]] = [] + for wacc in wacc_values: + row: list[float | None] = [] + for terminal_growth in terminal_growth_values: + result = run_case(wacc, terminal_growth) + price = result.get("intrinsic_value_per_share") + row.append(float(price) if isinstance(price, int | float) and math.isfinite(float(price)) else None) + implied_prices.append(row) + return { + "wacc": list(wacc_values), + "terminal_growth": list(terminal_growth_values), + "implied_prices": implied_prices, + } + + +def _advanced_dcf_sensitivity_grid( + inputs: AdvancedDcfInputs, + claims: _DcfCapitalClaims, +) -> _SensitivityGrid: + wacc_values = _centered_rate_range(inputs.wacc, _WACC_OFFSETS) + terminal_growth_values = _centered_rate_range(inputs.terminal_growth, _TERMINAL_GROWTH_OFFSETS) + + def run_case(wacc: float, terminal_growth: float) -> dict[str, bool | str | float | None]: + return _run_dcf_explicit_build( + base_revenue=inputs.base_revenue, + revenue_growth=inputs.revenue_growth, + ebitda_margin=inputs.ebitda_margin, + dna_pct_revenue=inputs.dna_pct_revenue, + capex_pct_revenue=inputs.capex_pct_revenue, + nwc_chg_pct_delta_rev=inputs.nwc_chg_pct_delta_rev, + tax_rate=inputs.tax_rate, + wacc=wacc, + terminal_growth=terminal_growth, + shares_outstanding=claims.shares_outstanding, + total_debt=claims.total_debt, + cash=claims.cash, + preferred_equity=claims.preferred_equity, + minority_interest=claims.minority_interest, + projection_years=inputs.projection_years, + ) + + return _dcf_sensitivity_grid(run_case, wacc_values, terminal_growth_values) + + def _run_ev_ebitda( ebitda: float, total_debt: float, @@ -472,6 +633,8 @@ def get_valuation(symbol: str) -> dict: inc_q = get_income_statement(sym, quarterly=True) bal_q = get_balance_sheet(sym, quarterly=True) info = get_company_info(sym) + fast_info = get_fast_info(sym) + currency_metadata = _currency_metadata(info, fast_info, prefer_reporting=True) shares = get_shares_outstanding(sym) current_price = _safe_float(info.get("currentPrice")) @@ -513,6 +676,20 @@ def get_valuation(symbol: str) -> dict: elif "error" in dcf_raw: dcf_out = {"available": True, "error": dcf_raw["error"], "wacc": 0.10, "terminal_growth": 0.03, "projection_years": 5} else: + sensitivity = _dcf_sensitivity_grid( + lambda wacc, terminal_growth: _run_dcf( + fcf_series=fcf_series, + shares_outstanding=shares, + wacc=wacc, + terminal_growth=terminal_growth, + total_debt=total_debt, + cash_and_equivalents=cash, + preferred_equity=preferred, + minority_interest=minority, + ), + _SIMPLE_WACC_RANGE, + _SIMPLE_TERMINAL_GROWTH_RANGE, + ) dcf_out = { "available": True, "intrinsic_value_per_share": dcf_raw.get("intrinsic_value_per_share"), @@ -528,6 +705,7 @@ def get_valuation(symbol: str) -> dict: "wacc": 0.10, "terminal_growth": 0.03, "projection_years": 5, + "sensitivity": sensitivity, } common = dict( @@ -559,6 +737,8 @@ def get_valuation(symbol: str) -> dict: return { "symbol": sym, + "currency": currency_metadata["currency"], + "currency_warning": currency_metadata["currency_warning"], "current_price": current_price, "shares_outstanding": shares, "dcf": dcf_out, @@ -568,10 +748,55 @@ def get_valuation(symbol: str) -> dict: } +def get_advanced_valuation(symbol: str, inputs: AdvancedDcfInputs) -> dict: + sym = normalize_symbol(symbol) + valuation = get_valuation(sym) + shares = get_shares_outstanding(sym) + balance_sheet = get_balance_sheet(sym, quarterly=True) + + claims = _DcfCapitalClaims( + shares_outstanding=shares or 0.0, + total_debt=_balance_value(balance_sheet, "Total Debt") or 0.0, + cash=_balance_value( + balance_sheet, + "Cash And Cash Equivalents", + "Cash Cash Equivalents And Short Term Investments", + ) or 0.0, + preferred_equity=_balance_value(balance_sheet, "Preferred Stock") or 0.0, + minority_interest=_balance_value(balance_sheet, "Minority Interest") or 0.0, + ) + dcf_raw = _run_dcf_explicit_build( + base_revenue=inputs.base_revenue, + revenue_growth=inputs.revenue_growth, + ebitda_margin=inputs.ebitda_margin, + dna_pct_revenue=inputs.dna_pct_revenue, + capex_pct_revenue=inputs.capex_pct_revenue, + nwc_chg_pct_delta_rev=inputs.nwc_chg_pct_delta_rev, + tax_rate=inputs.tax_rate, + wacc=inputs.wacc, + terminal_growth=inputs.terminal_growth, + shares_outstanding=claims.shares_outstanding, + total_debt=claims.total_debt, + cash=claims.cash, + preferred_equity=claims.preferred_equity, + minority_interest=claims.minority_interest, + projection_years=inputs.projection_years, + ) + dcf_raw["sensitivity"] = _advanced_dcf_sensitivity_grid(inputs, claims) + dcf_raw["advanced_inputs"] = inputs.model_dump() + valuation["dcf"] = dcf_raw + valuation["symbol"] = sym + valuation["shares_outstanding"] = shares + return valuation + + @cached(FINANCIALS_CACHE) def get_financials(symbol: str, period: str = "annual") -> dict: sym = normalize_symbol(symbol) quarterly = period == "quarterly" + info = get_company_info(sym) + fast_info = get_fast_info(sym) + currency_metadata = _currency_metadata(info, fast_info, prefer_reporting=True) inc = get_income_statement(sym, quarterly=quarterly) bal = get_balance_sheet(sym, quarterly=quarterly) @@ -583,6 +808,8 @@ def get_financials(symbol: str, period: str = "annual") -> dict: return { "period": period, + "currency": currency_metadata["currency"], + "currency_warning": currency_metadata["currency_warning"], "income": _build_income(inc, inc_q, quarterly), "balance": _build_balance(bal, bal_q, quarterly), "cash_flow": _build_cash_flow(cf, cf_q, inc, inc_q, quarterly), @@ -618,6 +845,13 @@ def _statement_ttm(frame: pd.DataFrame, *labels: str) -> float | None: return None +@cached(WACC_CACHE) +def compute_wacc(symbol: str) -> dict[str, str | bool | float | None]: + from app.services.wacc_service import compute_wacc_value + + return compute_wacc_value(symbol) + + 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 @@ -641,9 +875,12 @@ def _find_price_at_date(price_history: list[dict], target: "pd.Timestamp") -> fl @cached(HIST_RATIOS_CACHE) -def compute_historical_ratios(symbol: str) -> dict[str, list[float | None]]: +def compute_historical_ratios(symbol: str) -> dict[str, Any]: """Per-fiscal-year ratios from annual statements, oldest-first (up to 4 points).""" sym = normalize_symbol(symbol) + info = get_company_info(sym) + fast_info = get_fast_info(sym) + currency_metadata = _currency_metadata(info, fast_info, prefer_reporting=True) inc_a = get_income_statement(sym, quarterly=False) bal_a = get_balance_sheet(sym, quarterly=False) cf_a = get_cash_flow(sym, quarterly=False) @@ -747,7 +984,9 @@ def compute_historical_ratios(symbol: str) -> dict[str, list[float | None]]: result["price_to_book"].append(_cap_ratio(market_cap / equity, 0, 100) if market_cap and equity and equity > 0 else None) result["price_to_sales"].append(_cap_ratio(market_cap / revenue, 0, 100) if market_cap and revenue and revenue > 0 else None) - return {k: list(reversed(v)) for k, v in result.items()} + historical = {k: list(reversed(v)) for k, v in result.items()} + historical.update(currency_metadata) + return historical @cached(RATIOS_ENDPOINT_CACHE) @@ -806,6 +1045,8 @@ def get_ratios(symbol: str) -> dict: return {"value": val, "spark": spark, "vs_sector": vs_sector} return { + "currency": ttm.get("currency"), + "currency_warning": ttm.get("currency_warning"), "pe_ttm": point("trailing_pe", "trailing_pe"), "ev_ebitda": point("ev_to_ebitda", "ev_to_ebitda"), "gross_margin": point("gross_margin_ttm", "gross_margin"), @@ -1005,6 +1246,38 @@ def get_fast_info(symbol: str) -> dict[str, Any]: return {} +def _currency_metadata( + info: dict[str, Any], + fast_info: dict[str, Any], + *, + prefer_reporting: bool, +) -> dict[str, str | None]: + def _normalize_currency(value: Any) -> str | None: + text = str(value or "").strip().upper() + return text or None + + trading_currency = _normalize_currency(fast_info.get("currency")) or _normalize_currency(info.get("currency")) + reporting_currency = ( + _normalize_currency(info.get("financialCurrency")) + or _normalize_currency(fast_info.get("currency")) + or _normalize_currency(info.get("currency")) + ) + + if prefer_reporting: + currency = reporting_currency or trading_currency or "USD" + else: + currency = trading_currency or reporting_currency or "USD" + + currency_warning: str | None = None + if trading_currency and reporting_currency and trading_currency != reporting_currency: + currency_warning = ( + f"Trading currency {trading_currency} may differ from reporting currency {reporting_currency}; " + "values are not FX-normalized." + ) + + return {"currency": currency, "currency_warning": currency_warning} + + @cached(PRICE_CACHE) def get_latest_price(symbol: str) -> float | None: """Return latest close price, falling back to quote fields in info.""" @@ -1366,6 +1639,8 @@ def _build_profile(sym: str, info: dict[str, Any], fast_info: dict[str, Any], se def compute_ttm_ratios(symbol: str) -> dict[str, Any]: sym = normalize_symbol(symbol) info = get_company_info(sym) + fast_info = get_fast_info(sym) + currency_metadata = _currency_metadata(info, fast_info, prefer_reporting=True) 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) @@ -1373,7 +1648,7 @@ def compute_ttm_ratios(symbol: str) -> dict[str, Any]: cash_flow = get_cash_flow(sym, quarterly=True) if income is None or income.empty: - return {} + return dict(currency_metadata) revenue = _statement_ttm(income, "Total Revenue") gross_profit = _statement_ttm(income, "Gross Profit") @@ -1397,6 +1672,8 @@ def compute_ttm_ratios(symbol: str) -> dict[str, Any]: trailing_eps = net_income / shares ratios: dict[str, Any] = {} + ratios["currency"] = currency_metadata["currency"] + ratios["currency_warning"] = currency_metadata["currency_warning"] ratios["market_cap"] = market_cap ratios["trailing_eps"] = trailing_eps @@ -1458,7 +1735,7 @@ def compute_ttm_ratios(symbol: str) -> dict[str, Any]: if 0 <= payout < 10: ratios["dividend_payout_ratio_ttm"] = payout - return {key: value for key, value in ratios.items() if value is not None} + return {key: value for key, value in ratios.items() if value is not None or key in ("currency", "currency_warning")} @cached(BETA_CACHE) @@ -1695,6 +1972,7 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: info = get_company_info(sym) search_match = _pick_search_match(sym) fast_info = get_fast_info(sym) + currency_metadata = _currency_metadata(info, fast_info, prefer_reporting=False) month_history = get_price_history(sym, period="1m") year_history = get_price_history(sym, period="1y") computed = compute_ttm_ratios(sym) @@ -1760,6 +2038,8 @@ def get_ticker_overview(symbol: str) -> dict[str, Any] | None: is_partial = not all(field_availability.values()) return { + "currency": currency_metadata["currency"], + "currency_warning": currency_metadata["currency_warning"], "profile": profile, "quote": quote, "signals": build_signals(info, computed), |