feat: add valuation math helpers and VALUATION_CACHE

1 files changed, 57 insertions, 0 deletions

diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py
index f7188f7..9a88005 100644
--- a/backend/app/services/data_service.py
+++ b/backend/app/services/data_service.py
@@ -27,6 +27,7 @@ RATIO_CACHE = TTLCache(maxsize=256, ttl=3600)
 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)
 
 PERIODS = {"1m", "3m", "6m", "1y", "2y", "5y"}
 YF_PERIOD_MAP = {"1m": "1mo", "3m": "3mo", "6m": "6mo", "1y": "1y", "2y": "2y", "5y": "5y"}
@@ -265,6 +266,62 @@ def _build_cash_flow(cf: pd.DataFrame, cf_q: pd.DataFrame, inc: pd.DataFrame, in
     }
 
 
+_GROWTH_FLOOR = -0.50
+_GROWTH_CAP = 0.50
+_GROWTH_MIN_BASE = 1e-9
+
+
+def _cap_growth(value: float) -> float:
+    return max(_GROWTH_FLOOR, min(_GROWTH_CAP, float(value)))
+
+
+def _dcf_capped_growth_rate(fcf_series: "pd.Series") -> float | None:
+    historical = fcf_series.sort_index().dropna().astype(float).values
+    if len(historical) < 2:
+        return None
+    rates = []
+    for i in range(1, len(historical)):
+        prev, curr = float(historical[i - 1]), float(historical[i])
+        if abs(prev) < _GROWTH_MIN_BASE:
+            continue
+        if prev <= 0 or curr <= 0:
+            continue
+        rates.append((curr - prev) / prev)
+    if not rates:
+        return None
+    raw = float(pd.Series(rates).median())
+    return _cap_growth(raw)
+
+
+def _build_fcf_series(cf_annual: "pd.DataFrame") -> "pd.Series | None":
+    if cf_annual is None or cf_annual.empty:
+        return None
+    op_labels = ("Operating Cash Flow", "Cash Flow From Continuing Operating Activities")
+    op_row = None
+    for label in op_labels:
+        if label in cf_annual.index:
+            op_row = pd.to_numeric(cf_annual.loc[label], errors="coerce")
+            break
+    if op_row is None or "Capital Expenditure" not in cf_annual.index:
+        return None
+    capex_row = pd.to_numeric(cf_annual.loc["Capital Expenditure"], errors="coerce")
+    fcf = (op_row + capex_row).dropna().sort_index()
+    return fcf if not fcf.empty else None
+
+
+def _build_multiple_result(raw: dict) -> dict:
+    if not raw:
+        return {"available": False}
+    return {
+        "available": True,
+        "implied_price_per_share": raw.get("implied_price_per_share"),
+        "implied_ev": raw.get("implied_ev"),
+        "equity_value": raw.get("equity_value"),
+        "net_debt": raw.get("net_debt"),
+        "multiple_used": raw.get("target_multiple_used"),
+    }
+
+
 @cached(FINANCIALS_CACHE)
 def get_financials(symbol: str, period: str = "annual") -> dict:
     sym = normalize_symbol(symbol)