1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
import pandas as pd
import pytest
from fastapi.testclient import TestClient
from app import main
from app.services import data_service
def annual_frame(rows: dict[str, list[float]]) -> pd.DataFrame:
columns = pd.to_datetime(["2024-09-30", "2023-09-30", "2022-09-30", "2021-09-30"])
return pd.DataFrame(rows, index=columns).T
def quarterly_frame(rows: dict[str, list[float]]) -> pd.DataFrame:
columns = pd.to_datetime(["2025-12-31", "2025-09-30", "2025-06-30", "2025-03-31"])
return pd.DataFrame(rows, index=columns).T
def clear_valuation_cache() -> None:
data_service.VALUATION_CACHE.clear()
def advanced_payload() -> dict[str, float | int | list[float]]:
return {
"base_revenue": 1_000.0,
"revenue_growth": [0.10, 0.05],
"ebitda_margin": [0.30, 0.32],
"dna_pct_revenue": [0.05, 0.05],
"capex_pct_revenue": [0.07, 0.07],
"nwc_chg_pct_delta_rev": [0.02, 0.02],
"tax_rate": [0.25, 0.25],
"wacc": 0.10,
"terminal_growth": 0.03,
"projection_years": 2,
}
def test_simple_valuation_includes_five_by_five_sensitivity_grid(monkeypatch: pytest.MonkeyPatch) -> None:
# Given: historical free cash flow data for the simple DCF endpoint.
clear_valuation_cache()
cash_flow = annual_frame({
"Operating Cash Flow": [100.0, 90.0, 80.0, 70.0],
"Capital Expenditure": [-10.0, -9.0, -8.0, -7.0],
})
monkeypatch.setattr(data_service, "get_cash_flow", lambda symbol, quarterly=False: pd.DataFrame() if quarterly else cash_flow)
monkeypatch.setattr(data_service, "get_income_statement", lambda symbol, quarterly=False: pd.DataFrame())
monkeypatch.setattr(data_service, "get_balance_sheet", lambda symbol, quarterly=False: pd.DataFrame())
monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {"currentPrice": 150.0})
monkeypatch.setattr(data_service, "get_shares_outstanding", lambda symbol: 1_000_000_000.0)
# When: the simple valuation service is called.
result = data_service.get_valuation("ref")
# Then: the DCF result includes the mandated WACC x terminal-growth matrix.
dcf = result["dcf"]
sensitivity = dcf["sensitivity"]
assert result["currency"] == "USD"
assert result["currency_warning"] is None
assert sensitivity["wacc"] == [0.08, 0.09, 0.10, 0.11, 0.12]
assert sensitivity["terminal_growth"] == [0.015, 0.02, 0.03, 0.04, 0.045]
assert len(sensitivity["implied_prices"]) == 5
assert all(len(row) == 5 for row in sensitivity["implied_prices"])
assert sensitivity["implied_prices"][2][2] == pytest.approx(dcf["intrinsic_value_per_share"])
def test_advanced_valuation_endpoint_returns_inputs_and_sensitivity(monkeypatch: pytest.MonkeyPatch) -> None:
# Given: symbol-level balance sheet inputs and a base valuation shell.
clear_valuation_cache()
monkeypatch.setattr(
data_service,
"get_valuation",
lambda symbol: {
"symbol": "REF",
"current_price": 42.0,
"shares_outstanding": 100.0,
"dcf": {"available": False, "wacc": 0.10, "terminal_growth": 0.03, "projection_years": 5},
"ev_ebitda": {"available": False},
"ev_revenue": {"available": False},
"price_to_book": {"available": False},
},
)
monkeypatch.setattr(data_service, "get_shares_outstanding", lambda symbol: 100.0)
monkeypatch.setattr(
data_service,
"get_balance_sheet",
lambda symbol, quarterly=False: quarterly_frame({
"Total Debt": [50.0, 0.0, 0.0, 0.0],
"Cash And Cash Equivalents": [10.0, 0.0, 0.0, 0.0],
"Preferred Stock": [5.0, 0.0, 0.0, 0.0],
"Minority Interest": [3.0, 0.0, 0.0, 0.0],
}),
)
client = TestClient(main.app)
# When: the advanced valuation endpoint receives an explicit build payload.
response = client.post("/api/tickers/REF/valuation/advanced", json=advanced_payload())
# Then: it returns a full valuation response with base-case-centered sensitivity.
assert response.status_code == 200
body = response.json()
dcf = body["dcf"]
sensitivity = dcf["sensitivity"]
assert body["symbol"] == "REF"
assert body["currency"] == "USD"
assert body["currency_warning"] is None
assert dcf["advanced_inputs"] == advanced_payload()
assert dcf["intrinsic_value_per_share"] == pytest.approx(29.838067203890013)
assert sensitivity["wacc"] == [0.08, 0.09, 0.10, 0.11, 0.12]
assert sensitivity["terminal_growth"] == [0.015, 0.02, 0.03, 0.04, 0.045]
assert len(sensitivity["implied_prices"]) == 5
assert all(len(row) == 5 for row in sensitivity["implied_prices"])
assert sensitivity["implied_prices"][2][2] == pytest.approx(dcf["intrinsic_value_per_share"])
def test_advanced_valuation_endpoint_rejects_short_projection_arrays() -> None:
# Given: a payload whose per-year arrays do not cover projection_years.
payload = advanced_payload()
payload["revenue_growth"] = [0.10]
client = TestClient(main.app)
# When: the invalid payload is submitted.
response = client.post("/api/tickers/REF/valuation/advanced", json=payload)
# Then: FastAPI returns a validation response instead of an unhandled exception.
assert response.status_code == 422
assert "projection_years" in response.text
|
