feat: enhance data service with DCF and valuation support

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),