import pandas as pd import pytest from fastapi import HTTPException from app import main from app.services import data_service def clear_service_caches() -> None: data_service.INFO_CACHE.clear() data_service.FAST_INFO_CACHE.clear() data_service.PRICE_CACHE.clear() data_service.HISTORY_CACHE.clear() 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: columns = pd.to_datetime(["2025-12-31", "2025-09-30", "2025-06-30", "2025-03-31"]) 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"} def test_search_smoke(monkeypatch) -> None: monkeypatch.setattr(main.data_service, "search_tickers", lambda q: [{"symbol": "AAPL", "name": "Apple Inc.", "exchange": "NASDAQ"}]) assert main.search("apple")[0]["symbol"] == "AAPL" def test_watchlist_smoke(tmp_path, monkeypatch) -> None: monkeypatch.setattr(main, "DB_PATH", tmp_path / "prism.db") monkeypatch.setattr(main.data_service, "get_company_info", lambda symbol: {"currentPrice": 100.0, "previousClose": 95.0}) res = main.add_watchlist_symbol("aapl") assert res["items"][0]["symbol"] == "AAPL" def test_mocked_ticker_overview(monkeypatch) -> None: monkeypatch.setattr( main.data_service, "get_ticker_overview", lambda symbol: { "profile": {"symbol": "AAPL", "name": "Apple Inc.", "sector": None, "industry": None, "exchange": "NASDAQ", "website": None, "summary": None}, "quote": {"price": 100.0, "prev_close": 98.0, "change": 2.0, "change_pct": 0.0204}, "signals": [], "stats": {"market_cap": None, "trailing_pe": None, "trailing_eps": None, "volume": None, "average_volume": None, "beta": None}, "ratios": { "price_to_book": None, "price_to_sales": None, "ev_to_sales": None, "ev_to_ebitda": None, "gross_margin_ttm": None, "operating_margin_ttm": None, "net_margin_ttm": None, "roe_ttm": None, "roa_ttm": None, "roic_ttm": None, "debt_to_equity": None, "current_ratio": None, "dividend_yield_ttm": None, "dividend_payout_ratio_ttm": None, }, "range_52w": {"low": None, "high": None, "price": 100.0}, "short_interest": {"short_percent_of_float": None, "short_ratio": None, "shares_short": None, "shares_short_prior_month": None, "shares_short_delta_pct": None}, "meta": {"status": "partial", "is_partial": True, "field_availability": {}, "sources": {}}, }, ) assert main.ticker_overview("AAPL")["profile"]["symbol"] == "AAPL" def test_service_overview_prefers_info_fields(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr( data_service, "get_company_info", lambda symbol: { "longName": "Apple Inc.", "exchange": "NMS", "currentPrice": 190.0, "previousClose": 188.0, "marketCap": 2_900_000_000_000, "trailingPE": 31.2, "trailingEps": 6.08, "volume": 50_000_000, "averageVolume": 60_000_000, "beta": 1.18, "fiftyTwoWeekHigh": 199.0, "fiftyTwoWeekLow": 164.0, }, ) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {"lastPrice": 1.0, "exchange": "NYQ"}) monkeypatch.setattr(data_service, "get_price_history", lambda symbol, period="1m": []) monkeypatch.setattr(data_service, "_pick_search_match", lambda symbol: {"symbol": "AAPL", "name": "Wrong", "exchange": "NYSE"}) monkeypatch.setattr(data_service, "get_profile_enrichment", lambda symbol: {}) overview = data_service.get_ticker_overview("AAPL") assert overview is not None assert overview["profile"]["name"] == "Apple Inc." assert overview["profile"]["exchange"] == "NASDAQ" assert overview["quote"]["price"] == 190.0 assert overview["stats"]["market_cap"] == 2_900_000_000_000 assert overview["meta"]["sources"]["profile.name"] == "info" assert overview["meta"]["sources"]["quote.price"] == "info" def test_service_overview_falls_back_to_fast_info(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {}) monkeypatch.setattr( data_service, "get_fast_info", lambda symbol: { "lastPrice": 100.0, "previousClose": 98.0, "marketCap": 2_000_000_000, "lastVolume": 1_500_000, "threeMonthAverageVolume": 1_250_000, "yearHigh": 130.0, "yearLow": 90.0, "exchange": "NMS", }, ) monkeypatch.setattr(data_service, "get_price_history", lambda symbol, period="1m": []) monkeypatch.setattr(data_service, "_pick_search_match", lambda symbol: {"symbol": "AAPL", "name": "Apple Inc.", "exchange": "NASDAQ"}) monkeypatch.setattr(data_service, "get_profile_enrichment", lambda symbol: {}) overview = data_service.get_ticker_overview("AAPL") assert overview is not None assert overview["quote"]["price"] == 100.0 assert overview["quote"]["prev_close"] == 98.0 assert overview["stats"]["average_volume"] == 1_250_000 assert overview["meta"]["sources"]["quote.price"] == "fast_info" assert overview["meta"]["sources"]["range_52w.high"] == "fast_info" def test_service_overview_falls_back_to_search_and_history(monkeypatch) -> None: clear_service_caches() month_history = [ {"date": "2026-01-01", "close": 98.0, "volume": 1000.0}, {"date": "2026-01-02", "close": 100.0, "volume": 1200.0}, ] year_history = month_history + [{"date": "2026-04-02", "close": 120.0, "volume": 900.0}] monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {}) monkeypatch.setattr(data_service, "_pick_search_match", lambda symbol: {"symbol": "AAPL", "name": "Apple Inc.", "exchange": "NMS"}) monkeypatch.setattr(data_service, "get_profile_enrichment", lambda symbol: {}) monkeypatch.setattr(data_service, "get_price_history", lambda symbol, period="1m": month_history if period == "1m" else year_history) overview = data_service.get_ticker_overview("AAPL") assert overview is not None assert overview["profile"]["name"] == "Apple Inc." assert overview["quote"]["price"] == 100.0 assert overview["range_52w"]["high"] == 120.0 assert overview["meta"]["sources"]["profile.name"] == "search" assert overview["meta"]["sources"]["quote.price"] == "history_recent" def test_service_overview_invalid_symbol(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {}) monkeypatch.setattr(data_service, "_pick_search_match", lambda symbol: {}) monkeypatch.setattr(data_service, "get_profile_enrichment", lambda symbol: {}) monkeypatch.setattr(data_service, "get_price_history", lambda symbol, period="1m": []) assert data_service.get_ticker_overview("BAD") is None def test_ticker_overview_404(monkeypatch) -> None: monkeypatch.setattr(main.data_service, "get_ticker_overview", lambda symbol: None) with pytest.raises(HTTPException) as exc: main.ticker_overview("INVALID") assert exc.value.status_code == 404 assert exc.value.detail == "ticker data unavailable" def test_ticker_overview_partial_response(monkeypatch) -> None: monkeypatch.setattr( main.data_service, "get_ticker_overview", lambda symbol: { "profile": {"symbol": "AAPL", "name": "Apple Inc.", "exchange": "NASDAQ", "sector": None, "industry": None, "website": None, "summary": None}, "quote": {"price": 100.0, "prev_close": 98.0, "change": 2.0, "change_pct": 0.0204}, "signals": [], "stats": {"market_cap": None, "trailing_pe": None, "trailing_eps": None, "volume": 1000.0, "average_volume": None, "beta": None}, "ratios": { "price_to_book": None, "price_to_sales": None, "ev_to_sales": None, "ev_to_ebitda": None, "gross_margin_ttm": None, "operating_margin_ttm": None, "net_margin_ttm": None, "roe_ttm": None, "roa_ttm": None, "roic_ttm": None, "debt_to_equity": None, "current_ratio": None, "dividend_yield_ttm": None, "dividend_payout_ratio_ttm": None, }, "range_52w": {"low": None, "high": None, "price": 100.0}, "short_interest": {"short_percent_of_float": None, "short_ratio": None, "shares_short": None, "shares_short_prior_month": None, "shares_short_delta_pct": None}, "meta": {"status": "partial", "is_partial": True, "field_availability": {"stats.market_cap": False}, "sources": {"profile.name": "search"}}, }, ) body = main.ticker_overview("AAPL") assert body["meta"]["is_partial"] is True assert body["profile"]["name"] == "Apple Inc." def test_ticker_history_period_mapping(monkeypatch) -> None: data_service.HISTORY_CACHE.clear() captured: list[str] = [] class DummyTicker: def __init__(self, symbol: str) -> None: self.symbol = symbol def history(self, period: str): captured.append(period) return pd.DataFrame( [{"Open": 1.0, "High": 1.0, "Low": 1.0, "Close": 1.0, "Volume": 1.0}], index=[pd.Timestamp("2026-01-01")], ) monkeypatch.setattr(data_service.yf, "Ticker", DummyTicker) assert len(data_service.get_price_history("AAPL", period="1m")) == 1 assert len(data_service.get_price_history("AAPL", period="3m")) == 1 assert len(data_service.get_price_history("AAPL", period="6m")) == 1 assert captured == ["1mo", "3mo", "6mo"] def test_compute_ttm_ratios_populates_overlapping_stats(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_latest_price", lambda symbol: 50.0) monkeypatch.setattr(data_service, "get_shares_outstanding", lambda symbol: 100.0) monkeypatch.setattr( data_service, "get_income_statement", lambda symbol, quarterly=False: quarterly_frame( { "Total Revenue": [1_000.0, 1_000.0, 1_000.0, 1_000.0], "Gross Profit": [500.0, 500.0, 500.0, 500.0], "Operating Income": [250.0, 250.0, 250.0, 250.0], "Net Income": [100.0, 100.0, 100.0, 100.0], "EBIT": [150.0, 150.0, 150.0, 150.0], "EBITDA": [200.0, 200.0, 200.0, 200.0], "Tax Provision": [20.0, 20.0, 20.0, 20.0], "Pretax Income": [120.0, 120.0, 120.0, 120.0], } ), ) monkeypatch.setattr( data_service, "get_balance_sheet", lambda symbol, quarterly=False: quarterly_frame( { "Stockholders Equity": [500.0, 0.0, 0.0, 0.0], "Total Assets": [1_200.0, 0.0, 0.0, 0.0], "Total Debt": [150.0, 0.0, 0.0, 0.0], "Current Assets": [300.0, 0.0, 0.0, 0.0], "Current Liabilities": [100.0, 0.0, 0.0, 0.0], "Cash And Cash Equivalents": [50.0, 0.0, 0.0, 0.0], } ), ) monkeypatch.setattr( data_service, "get_cash_flow", lambda symbol, quarterly=False: quarterly_frame({"Cash Dividends Paid": [-10.0, -10.0, -10.0, -10.0]}), ) ratios = data_service.compute_ttm_ratios("AAPL") assert ratios["market_cap"] == 5_000.0 assert ratios["trailing_eps"] == 4.0 assert ratios["trailing_pe"] == 12.5 assert ratios["price_to_book"] == 10.0 assert ratios["price_to_sales"] == 1.25 assert ratios["ev_to_sales"] == 1.275 assert ratios["gross_margin_ttm"] == 0.5 assert ratios["operating_margin_ttm"] == 0.25 assert ratios["net_margin_ttm"] == 0.1 assert ratios["roe_ttm"] == 0.8 assert round(ratios["roic_ttm"], 6) == round((600.0 * (1 - (80.0 / 480.0))) / 600.0, 6) assert ratios["debt_to_equity"] == 0.3 assert ratios["current_ratio"] == 3.0 assert ratios["dividend_yield_ttm"] == 0.008 assert ratios["dividend_payout_ratio_ttm"] == 0.1 def test_compute_ttm_ratios_guardrails_suppress_outliers(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr(data_service, "get_company_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_latest_price", lambda symbol: 1_000.0) monkeypatch.setattr(data_service, "get_shares_outstanding", lambda symbol: 100.0) monkeypatch.setattr( data_service, "get_income_statement", lambda symbol, quarterly=False: quarterly_frame( { "Total Revenue": [1.0, 1.0, 1.0, 1.0], "Net Income": [1.0, 1.0, 1.0, 1.0], "EBIT": [1.0, 1.0, 1.0, 1.0], "EBITDA": [100_000.0, 100_000.0, 100_000.0, 100_000.0], "Tax Provision": [0.0, 0.0, 0.0, 0.0], "Pretax Income": [1.0, 1.0, 1.0, 1.0], } ), ) monkeypatch.setattr( data_service, "get_balance_sheet", lambda symbol, quarterly=False: quarterly_frame( { "Stockholders Equity": [1.0, 0.0, 0.0, 0.0], "Total Assets": [10.0, 0.0, 0.0, 0.0], "Total Debt": [1_000.0, 0.0, 0.0, 0.0], "Cash And Cash Equivalents": [0.0, 0.0, 0.0, 0.0], } ), ) monkeypatch.setattr(data_service, "get_cash_flow", lambda symbol, quarterly=False: pd.DataFrame()) ratios = data_service.compute_ttm_ratios("AAPL") assert ratios["trailing_pe"] == 25_000.0 assert "price_to_book" not in ratios assert "price_to_sales" not in ratios assert "ev_to_sales" not in ratios assert "ev_to_ebitda" not in ratios def test_financials_schema_structure() -> None: from app.schemas import FinancialRow, FinancialStatement, FinancialsResponse row = FinancialRow(label="Revenue", indent=0, is_total=True, values=[1.0, 2.0, None]) assert row.label == "Revenue" assert row.is_total is True assert row.values[2] is None stmt = FinancialStatement(columns=["FY 2024", "TTM"], rows=[row]) assert len(stmt.columns) == 2 resp = FinancialsResponse( period="annual", income=stmt, balance=FinancialStatement(columns=[], rows=[]), cash_flow=FinancialStatement(columns=[], rows=[]), ) assert resp.period == "annual" def test_overview_uses_computed_sources_and_ratios(monkeypatch) -> None: clear_service_caches() monkeypatch.setattr( data_service, "get_company_info", lambda symbol: { "longName": "Apple Inc.", "exchange": "NMS", "currentPrice": 120.0, "previousClose": 118.0, }, ) monkeypatch.setattr(data_service, "get_fast_info", lambda symbol: {}) monkeypatch.setattr(data_service, "get_price_history", lambda symbol, period="1m": []) monkeypatch.setattr(data_service, "_pick_search_match", lambda symbol: {"symbol": "AAPL", "name": "Apple Inc.", "exchange": "NASDAQ"}) monkeypatch.setattr(data_service, "get_profile_enrichment", lambda symbol: {}) monkeypatch.setattr( data_service, "compute_ttm_ratios", lambda symbol: { "market_cap": 1_500_000_000.0, "trailing_pe": 24.5, "trailing_eps": 4.9, "price_to_book": 8.0, "price_to_sales": 6.2, "ev_to_sales": 6.8, "ev_to_ebitda": 22.1, "gross_margin_ttm": 0.44, "operating_margin_ttm": 0.27, "net_margin_ttm": 0.18, "roe_ttm": 0.31, "roa_ttm": 0.12, "roic_ttm": 0.19, "debt_to_equity": 0.42, "current_ratio": 1.8, "dividend_yield_ttm": 0.005, "dividend_payout_ratio_ttm": 0.12, }, ) overview = data_service.get_ticker_overview("AAPL") assert overview is not None assert overview["stats"]["trailing_pe"] == 24.5 assert overview["stats"]["market_cap"] == 1_500_000_000.0 assert overview["ratios"]["price_to_book"] == 8.0 assert overview["meta"]["sources"]["stats.trailing_pe"] == "computed" assert overview["meta"]["sources"]["stats.market_cap"] == "computed" 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"] == []