File size: 10,845 Bytes
35676b4 7880373 35676b4 7880373 35676b4 | 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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 | """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
|