summaryrefslogtreecommitdiff
path: root/backend
diff options
context:
space:
mode:
Diffstat (limited to 'backend')
-rw-r--r--backend/app/services/data_service.py167
-rw-r--r--backend/tests/test_api.py140
2 files changed, 307 insertions, 0 deletions
diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py
index 16e062e..bf75bc7 100644
--- a/backend/app/services/data_service.py
+++ b/backend/app/services/data_service.py
@@ -113,6 +113,173 @@ def _safe_ratio(num: float | None, den: float | None) -> float | None:
return num / den
+def _build_income(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict:
+ if frame is None or frame.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(frame.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]]
+ if not quarterly:
+ col_labels.append("TTM")
+
+ def v(label: str) -> list[float | None]:
+ base = _row_vals(frame, label, n)
+ return base + ([_statement_ttm(frame_q, label)] if not quarterly else [])
+
+ def vm(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(frame, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(frame_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ rev = v("Total Revenue")
+ gross = v("Gross Profit")
+ net = v("Net Income")
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_row("Total Revenue", 0, True, rev),
+ _fin_row("Cost of Revenue", 1, False, v("Cost Of Revenue")),
+ _fin_row("Gross Profit", 0, True, gross),
+ _fin_margin("gross margin", [_safe_ratio(g, r) for g, r in zip(gross, rev)]),
+ _fin_row("Operating Expenses", 1, False, v("Operating Expense")),
+ _fin_row("Operating Income", 0, True, v("Operating Income")),
+ _fin_row("EBITDA", 1, False, vm("EBITDA", "Normalized EBITDA")),
+ _fin_row("Interest Expense", 1, False, v("Interest Expense")),
+ _fin_row("Pretax Income", 0, False, v("Pretax Income")),
+ _fin_row("Tax Provision", 1, False, v("Tax Provision")),
+ _fin_row("Net Income", 0, True, net),
+ _fin_margin("net margin", [_safe_ratio(ni, r) for ni, r in zip(net, rev)]),
+ _fin_row("EPS Basic", 1, False, v("Basic EPS")),
+ ],
+ }
+
+
+def _build_balance(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict:
+ if frame is None or frame.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(frame.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]]
+ if not quarterly:
+ col_labels.append("MRQ")
+
+ def v(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(frame, n, *labels)
+ if not quarterly:
+ val = None
+ for lbl in labels:
+ val = _balance_value(frame_q, lbl)
+ if val is not None:
+ break
+ base = base + [val]
+ return base
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_section("ASSETS"),
+ _fin_row("Current Assets", 0, True, v("Current Assets")),
+ _fin_row("Cash & Equivalents", 1, False, v("Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments")),
+ _fin_row("Short Term Investments", 1, False, v("Other Short Term Investments", "Short Term Investments")),
+ _fin_row("Receivables", 1, False, v("Receivables", "Net Receivables")),
+ _fin_row("Inventory", 1, False, v("Inventory")),
+ _fin_row("Total Assets", 0, True, v("Total Assets")),
+ _fin_section("LIABILITIES"),
+ _fin_row("Current Liabilities", 0, True, v("Current Liabilities")),
+ _fin_row("Accounts Payable", 1, False, v("Payables And Accrued Expenses", "Accounts Payable")),
+ _fin_row("Short Term Debt", 1, False, v("Current Debt", "Short Term Debt And Capital Lease Obligation")),
+ _fin_row("Long Term Debt", 1, False, v("Long Term Debt", "Long Term Debt And Capital Lease Obligation")),
+ _fin_row("Total Liabilities", 0, True, v("Total Liabilities Net Minority Interest", "Total Liabilities")),
+ _fin_section("EQUITY"),
+ _fin_row("Stockholders Equity", 0, True, v("Stockholders Equity", "Common Stock Equity")),
+ ],
+ }
+
+
+def _build_cash_flow(cf: pd.DataFrame, cf_q: pd.DataFrame, inc: pd.DataFrame, inc_q: pd.DataFrame, quarterly: bool) -> dict:
+ if cf is None or cf.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(cf.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in cf.columns[:n]]
+ if not quarterly:
+ col_labels.append("TTM")
+
+ def cv(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(cf, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(cf_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ def iv(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(inc, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(inc_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ op_cf = cv("Operating Cash Flow", "Cash Flow From Continuing Operating Activities")
+ capex = cv("Capital Expenditure")
+ # CapEx is negative in yfinance; FCF = Operating CF + CapEx
+ fcf = [a + b if a is not None and b is not None else None for a, b in zip(op_cf, capex)]
+ rev = iv("Total Revenue")
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_section("OPERATING"),
+ _fin_row("Net Income", 1, False, iv("Net Income")),
+ _fin_row("D&A", 1, False, cv("Depreciation And Amortization", "Reconciled Depreciation")),
+ _fin_row("Changes in Working Capital", 1, False, cv("Change In Working Capital")),
+ _fin_row("Operating Cash Flow", 0, True, op_cf),
+ _fin_section("INVESTING"),
+ _fin_row("CapEx", 1, False, capex),
+ _fin_row("Free Cash Flow", 0, True, fcf),
+ _fin_margin("FCF margin", [_safe_ratio(f, r) for f, r in zip(fcf, rev)]),
+ _fin_row("Investing Cash Flow", 0, True, cv("Investing Cash Flow", "Cash Flow From Continuing Investing Activities")),
+ _fin_section("FINANCING"),
+ _fin_row("Dividends Paid", 1, False, cv("Cash Dividends Paid", "Common Stock Dividend Paid")),
+ _fin_row("Buybacks", 1, False, cv("Repurchase Of Capital Stock", "Common Stock Repurchase")),
+ _fin_row("Financing Cash Flow", 0, True, cv("Financing Cash Flow", "Cash Flow From Continuing Financing Activities")),
+ _fin_row("Net Change in Cash", 0, True, cv("Changes In Cash", "End Cash Position")),
+ ],
+ }
+
+
+@cached(FINANCIALS_CACHE)
+def get_financials(symbol: str, period: str = "annual") -> dict:
+ sym = normalize_symbol(symbol)
+ quarterly = period == "quarterly"
+
+ inc = get_income_statement(sym, quarterly=quarterly)
+ bal = get_balance_sheet(sym, quarterly=quarterly)
+ cf = get_cash_flow(sym, quarterly=quarterly)
+
+ inc_q = get_income_statement(sym, quarterly=True) if not quarterly else inc
+ bal_q = get_balance_sheet(sym, quarterly=True) if not quarterly else bal
+ cf_q = get_cash_flow(sym, quarterly=True) if not quarterly else cf
+
+ return {
+ "period": period,
+ "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),
+ }
+
+
def _balance_value(frame: pd.DataFrame, *labels: str) -> float | None:
if frame is None or frame.empty:
return None
diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py
index 5d76dfd..f98b582 100644
--- a/backend/tests/test_api.py
+++ b/backend/tests/test_api.py
@@ -14,6 +14,7 @@ def clear_service_caches() -> None:
data_service.STATEMENT_CACHE.clear()
data_service.SHARES_CACHE.clear()
data_service.RATIO_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
def quarterly_frame(rows: dict[str, list[float]]) -> pd.DataFrame:
@@ -21,6 +22,11 @@ def quarterly_frame(rows: dict[str, list[float]]) -> pd.DataFrame:
return pd.DataFrame(rows, index=columns).T
+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 test_health() -> None:
assert main.health() == {"status": "ok"}
@@ -403,3 +409,137 @@ def test_overview_uses_computed_sources_and_ratios(monkeypatch) -> None:
assert overview["meta"]["sources"]["ratios.price_to_book"] == "computed"
assert overview["meta"]["field_availability"]["ratios.ev_to_ebitda"] is True
assert any(signal["key"] == "Valuation" and "24.5x" in signal["value"] for signal in overview["signals"])
+
+
+def test_build_income_annual_columns_and_ttm(monkeypatch) -> None:
+ data_service.STATEMENT_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
+
+ inc_annual = annual_frame({
+ "Total Revenue": [391_000.0, 383_300.0, 394_300.0, 365_800.0],
+ "Gross Profit": [180_700.0, 169_100.0, 170_800.0, 152_800.0],
+ "Net Income": [ 93_700.0, 97_000.0, 99_800.0, 94_700.0],
+ })
+ inc_q = quarterly_frame({
+ "Total Revenue": [100_000.0, 95_000.0, 98_000.0, 97_000.0],
+ "Gross Profit": [ 46_000.0, 44_000.0, 45_000.0, 43_000.0],
+ "Net Income": [ 24_000.0, 23_000.0, 24_000.0, 22_000.0],
+ })
+
+ monkeypatch.setattr(data_service, "get_income_statement",
+ lambda sym, quarterly=False: inc_q if quarterly else inc_annual)
+ monkeypatch.setattr(data_service, "get_balance_sheet",
+ lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_cash_flow",
+ lambda sym, quarterly=False: pd.DataFrame())
+
+ result = data_service.get_financials("AAPL", "annual")
+ income = result["income"]
+
+ assert income["columns"] == ["FY 2024", "FY 2023", "FY 2022", "FY 2021", "TTM"]
+ rev_row = next(r for r in income["rows"] if r["label"] == "Total Revenue")
+ assert rev_row["is_total"] is True
+ assert rev_row["values"][4] == 390_000.0 # sum of 4 quarters
+ margin_row = next(r for r in income["rows"] if r["label"] == "gross margin")
+ assert margin_row["is_margin"] is True
+ assert margin_row["values"][0] is not None # FY 2024 gross margin computed
+
+
+def test_build_income_quarterly_eight_columns(monkeypatch) -> None:
+ data_service.STATEMENT_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
+
+ cols = pd.to_datetime([
+ "2025-12-31","2025-09-30","2025-06-30","2025-03-31",
+ "2024-12-31","2024-09-30","2024-06-30","2024-03-31",
+ ])
+ inc_q8 = pd.DataFrame(
+ {"Total Revenue": [100_000.0]*8, "Net Income": [25_000.0]*8},
+ index=cols,
+ ).T
+
+ monkeypatch.setattr(data_service, "get_income_statement",
+ lambda sym, quarterly=False: inc_q8)
+ monkeypatch.setattr(data_service, "get_balance_sheet",
+ lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_cash_flow",
+ lambda sym, quarterly=False: pd.DataFrame())
+
+ result = data_service.get_financials("AAPL", "quarterly")
+ income = result["income"]
+ assert len(income["columns"]) == 8
+ assert income["columns"][0] == "Q4 2025"
+ assert "TTM" not in income["columns"]
+
+
+def test_build_balance_mrq_column(monkeypatch) -> None:
+ data_service.STATEMENT_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
+
+ bal_annual = annual_frame({"Total Assets": [364_900.0, 335_000.0, 352_800.0, 351_000.0]})
+ bal_q = quarterly_frame({"Total Assets": [371_900.0, 368_000.0, 360_000.0, 355_000.0]})
+
+ monkeypatch.setattr(data_service, "get_income_statement",
+ lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_balance_sheet",
+ lambda sym, quarterly=False: bal_q if quarterly else bal_annual)
+ monkeypatch.setattr(data_service, "get_cash_flow",
+ lambda sym, quarterly=False: pd.DataFrame())
+
+ result = data_service.get_financials("AAPL", "annual")
+ balance = result["balance"]
+ assert balance["columns"][-1] == "MRQ"
+ assets_row = next(r for r in balance["rows"] if r["label"] == "Total Assets")
+ assert assets_row["values"][-1] == 371_900.0 # MRQ value
+
+
+def test_build_cash_flow_fcf(monkeypatch) -> None:
+ data_service.STATEMENT_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
+
+ cf_annual = annual_frame({
+ "Operating Cash Flow": [118_300.0, 110_500.0, 122_200.0, 104_000.0],
+ "Capital Expenditure": [ -9_500.0, -10_900.0, -10_700.0, -8_600.0],
+ })
+ cf_q = quarterly_frame({
+ "Operating Cash Flow": [30_000.0, 29_000.0, 31_000.0, 28_000.0],
+ "Capital Expenditure": [-2_500.0, -2_400.0, -2_500.0, -2_400.0],
+ })
+ inc_annual = annual_frame({"Total Revenue": [391_000.0, 383_300.0, 394_300.0, 365_800.0]})
+ inc_q = quarterly_frame({"Total Revenue": [100_000.0, 95_000.0, 98_000.0, 97_000.0]})
+
+ def mock_cf(sym, quarterly=False):
+ return cf_q if quarterly else cf_annual
+
+ def mock_inc(sym, quarterly=False):
+ return inc_q if quarterly else inc_annual
+
+ monkeypatch.setattr(data_service, "get_income_statement", mock_inc)
+ monkeypatch.setattr(data_service, "get_balance_sheet", lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_cash_flow", mock_cf)
+
+ result = data_service.get_financials("AAPL", "annual")
+ cf = result["cash_flow"]
+
+ fcf_row = next(r for r in cf["rows"] if r["label"] == "Free Cash Flow")
+ assert fcf_row["is_total"] is True
+ # FY 2024: 118300 + (-9500) = 108800
+ assert fcf_row["values"][0] == 108_800.0
+ fcf_margin = next(r for r in cf["rows"] if r["label"] == "FCF margin")
+ assert fcf_margin["is_margin"] is True
+ assert fcf_margin["values"][0] is not None
+
+
+def test_get_financials_empty_statements(monkeypatch) -> None:
+ data_service.STATEMENT_CACHE.clear()
+ data_service.FINANCIALS_CACHE.clear()
+
+ monkeypatch.setattr(data_service, "get_income_statement", lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_balance_sheet", lambda sym, quarterly=False: pd.DataFrame())
+ monkeypatch.setattr(data_service, "get_cash_flow", lambda sym, quarterly=False: pd.DataFrame())
+
+ result = data_service.get_financials("AAPL", "annual")
+ assert result["income"]["columns"] == []
+ assert result["income"]["rows"] == []
+ assert result["balance"]["columns"] == []
+ assert result["cash_flow"]["columns"] == []
generated by cgit v1.3-2-g0d8e (git 2.53.0) at 2026-06-05 03:55:01 +0000