"""tests/test_yf_fallback.py — unit tests for ingestion/yf_fallback.py. Mocks yfinance and Alpha Vantage so tests are hermetic and fast. """ from __future__ import annotations from unittest.mock import MagicMock, patch import pandas as pd import pytest from ingestion.edgar import EdgarData from ingestion.yf_fallback import fill_missing_metrics # ── helpers ────────────────────────────────────────────────────────────────── def _make_edgar(**kwargs) -> EdgarData: defaults = dict( ticker="TSLA", company_name="Tesla, Inc.", filing_date="2025-07-24", period="Q22025", form_type="10-Q", revenue=25_500_000_000.0, revenue_yoy_pct=2.5, eps=0.91, gross_margin=0.18, operating_margin=0.06, free_cash_flow=1_000_000_000.0, mda_text="", risk_factors_text="", shares_diluted=None, effective_tax_rate=None, interest_expense=None, total_debt=None, dividends_paid=None, buybacks=None, capex=None, stockholders_equity=None, ) defaults.update(kwargs) return EdgarData(**defaults) def _income_df(*rows: tuple) -> pd.DataFrame: """Build a mock yfinance income-statement DataFrame. rows: [(name, value), ...] — one column (2025-06-30 period end). """ col = pd.Timestamp("2025-06-30") return pd.DataFrame( {col: {name: val for name, val in rows}} ) def _cashflow_df(cfo: float, capex: float) -> pd.DataFrame: col = pd.Timestamp("2025-06-30") return pd.DataFrame( {col: {"Operating Cash Flow": cfo, "Capital Expenditure": capex}} ) @pytest.fixture(autouse=True) def _clear_yf_cache(): """Reset module-level yfinance data cache between tests.""" import ingestion.yf_fallback as mod mod._YF_CACHE.clear() yield mod._YF_CACHE.clear() # ── no-op when all critical fields filled ──────────────────────────────────── def test_fill_missing_metrics_noop_when_all_critical_fields_filled(): edgar = _make_edgar() # revenue, eps, gross_margin, operating_margin, free_cash_flow all set with patch("ingestion.yf_fallback.fetch_earnings") as mock_av, \ patch("ingestion.yf_fallback.yf") as mock_yf: result = fill_missing_metrics(edgar) mock_av.assert_not_called() mock_yf.Ticker.assert_not_called() assert result.revenue == edgar.revenue assert result.eps == edgar.eps # ── EPS from Alpha Vantage ──────────────────────────────────────────────────── def test_fill_missing_metrics_fills_eps_from_alpha_vantage(): edgar = _make_edgar(eps=None) av_data = { "quarterlyEarnings": [ { "fiscalDateEnding": "2025-06-30", "reportedDate": "2025-07-23", "reportedEPS": "0.91", "estimatedEPS": "0.88", } ] } with patch("ingestion.yf_fallback.fetch_earnings", return_value=(av_data, None)), \ patch("ingestion.yf_fallback.yf"): result = fill_missing_metrics(edgar) assert result.eps == pytest.approx(0.91) assert result.metric_contexts["eps"]["source"] == "alpha_vantage" assert "eps:fallback_non_sec" in result.quality_warnings assert result.data_quality_status == "CHECK_REQUIRED" def test_fill_missing_metrics_eps_av_date_out_of_window_stays_none(): """AV entry >90 days before filing → not used, eps stays None.""" edgar = _make_edgar(eps=None, filing_date="2025-07-24") av_data = { "quarterlyEarnings": [ # ~200 days before filing date {"fiscalDateEnding": "2024-12-31", "reportedDate": "2025-02-01", "reportedEPS": "1.50"}, ] } with patch("ingestion.yf_fallback.fetch_earnings", return_value=(av_data, None)), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value.quarterly_financials = pd.DataFrame() mock_yf.Ticker.return_value.quarterly_cashflow = pd.DataFrame() result = fill_missing_metrics(edgar) assert result.eps is None def test_fill_missing_metrics_eps_av_error_falls_through_to_yfinance(): """When AV returns an error, yfinance income stmt is tried for EPS.""" col = pd.Timestamp("2025-06-30") income = pd.DataFrame({col: {"Total Revenue": 2e10, "Gross Profit": 4e9, "Operating Income": 1e9, "Diluted EPS": 0.88}}) cashflow = _cashflow_df(2e9, -5e8) edgar = _make_edgar(eps=None) fake_ticker = MagicMock() fake_ticker.quarterly_financials = income fake_ticker.quarterly_cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "rate limited")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker result = fill_missing_metrics(edgar) assert result.eps == pytest.approx(0.88) # ── Revenue + margins + FCF from yfinance ──────────────────────────────────── def test_fill_missing_metrics_fills_revenue_from_yfinance(): edgar = _make_edgar(revenue=None, gross_margin=None, operating_margin=None, free_cash_flow=None, eps=None) income = _income_df( ("Total Revenue", 25_500_000_000.0), ("Gross Profit", 4_590_000_000.0), ("Operating Income", 1_530_000_000.0), ("Diluted EPS", 0.91), ) cashflow = _cashflow_df(cfo=2_000_000_000.0, capex=-500_000_000.0) fake_ticker = MagicMock() fake_ticker.quarterly_financials = income fake_ticker.quarterly_cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "no key")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker result = fill_missing_metrics(edgar) assert result.revenue == 25_500_000_000.0 assert result.gross_margin == pytest.approx(4_590_000_000 / 25_500_000_000, rel=1e-3) assert result.operating_margin == pytest.approx(1_530_000_000 / 25_500_000_000, rel=1e-3) assert result.free_cash_flow == pytest.approx(1_500_000_000.0, rel=1e-3) assert result.eps == pytest.approx(0.91) assert result.metric_contexts["revenue"]["source"] == "yfinance" assert result.metric_contexts["free_cash_flow"]["statement"] == "cashflow" assert "revenue:fallback_non_sec" in result.quality_warnings def test_cashflow_uses_its_own_nearest_column(): edgar = _make_edgar(free_cash_flow=None) income_col = pd.Timestamp("2025-06-30") cashflow_col = pd.Timestamp("2025-06-29") income = pd.DataFrame({income_col: {"Total Revenue": edgar.revenue}}) cashflow = pd.DataFrame({cashflow_col: { "Operating Cash Flow": 2_000_000_000.0, "Capital Expenditure": -500_000_000.0, }}) fake_ticker = MagicMock() fake_ticker.quarterly_financials = income fake_ticker.quarterly_cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "no key")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker result = fill_missing_metrics(edgar) assert result.free_cash_flow == pytest.approx(1_500_000_000.0) assert result.metric_contexts["free_cash_flow"]["period_end"] == "2025-06-29" def test_fill_missing_metrics_yfinance_date_out_of_window_stays_none(): """yfinance column > 90 days before filing → not used, revenue stays None.""" edgar = _make_edgar(revenue=None, gross_margin=None, operating_margin=None, free_cash_flow=None, eps=None, filing_date="2025-07-24") # Column date is 2024-09-30 — ~300 days before filing far_col = pd.Timestamp("2024-09-30") income = pd.DataFrame({far_col: {"Total Revenue": 9e10, "Gross Profit": 2e10, "Operating Income": 8e9}}) cashflow = pd.DataFrame({far_col: {"Operating Cash Flow": 5e9, "Capital Expenditure": -1e9}}) fake_ticker = MagicMock() fake_ticker.quarterly_financials = income fake_ticker.quarterly_cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "no key")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker result = fill_missing_metrics(edgar) assert result.revenue is None def test_fill_missing_metrics_uses_annual_financials_for_10k(): """For 10-K filings, annual financials are used instead of quarterly.""" edgar = _make_edgar( revenue=None, gross_margin=None, operating_margin=None, free_cash_flow=None, eps=None, form_type="10-K", filing_date="2025-02-05", period="FY2024", ) col = pd.Timestamp("2024-12-31") income = pd.DataFrame({col: {"Total Revenue": 3e11, "Gross Profit": 1e11, "Operating Income": 9e10, "Diluted EPS": 5.0}}) cashflow = pd.DataFrame({col: {"Operating Cash Flow": 1.2e11, "Capital Expenditure": -1e10}}) fake_ticker = MagicMock() fake_ticker.financials = income fake_ticker.cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "no key")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker result = fill_missing_metrics(edgar) assert result.revenue == 3e11 assert result.eps == pytest.approx(5.0) # ── yfinance cache (one Ticker() call per ticker per run) ──────────────────── def test_fill_missing_metrics_yf_ticker_called_once_per_ticker(): """yf.Ticker is called at most once per ticker across multiple fill_missing_metrics calls.""" edgar1 = _make_edgar(revenue=None, gross_margin=None, operating_margin=None, free_cash_flow=None, eps=None, period="Q12025") edgar2 = _make_edgar(revenue=None, gross_margin=None, operating_margin=None, free_cash_flow=None, eps=None, period="Q22025") income = _income_df(("Total Revenue", 2e10)) cashflow = _cashflow_df(1e9, -2e8) fake_ticker = MagicMock() fake_ticker.quarterly_financials = income fake_ticker.quarterly_cashflow = cashflow with patch("ingestion.yf_fallback.fetch_earnings", return_value=(None, "no key")), \ patch("ingestion.yf_fallback.yf") as mock_yf: mock_yf.Ticker.return_value = fake_ticker fill_missing_metrics(edgar1) fill_missing_metrics(edgar2) assert mock_yf.Ticker.call_count == 1