"""tests/test_analytics_deltas.py — unit tests for analytics/deltas.py. Uses a temporary SQLite DB with synthetic data. No network calls; alphavantage is mocked where needed. """ import pytest from pathlib import Path from unittest.mock import patch from analytics.deltas import ( compute_metric_deltas, compute_eps_surprise, compute_guidance_change, compute_risk_diff, build_quarter_snapshot, MetricDelta, EpsSurprise, GuidanceChange, QuarterSnapshot, _derive_virtual_q4_row, ) from storage import metrics_db from storage.metrics_db import init_db, upsert_metrics # --------------------------------------------------------------------------- # Synthetic test data # --------------------------------------------------------------------------- QUARTERLY_ROWS = [ # (period, filing_date, revenue, eps, gross_margin, operating_margin, # free_cash_flow, capex, buybacks, dividends_paid, total_debt, # shares_diluted, guidance_disclosed) ("Q12025", "2025-02-01", 100e9, 2.50, 0.30, 0.20, 20e9, 5e9, 8e9, 2e9, 50e9, 15_000e6, 1), # latest ("Q12024", "2024-02-01", 90e9, 2.20, 0.28, 0.18, 18e9, 4.5e9, 6e9, 1.8e9, 55e9, 15_200e6, 1), # YoY peer ("Q42024", "2024-11-01", 95e9, 2.40, 0.29, 0.19, 19e9, 4.8e9, 7e9, 1.9e9, 52e9, 15_100e6, 1), # QoQ peer ("Q32024", "2024-08-01", 85e9, 2.10, 0.27, 0.17, 17e9, 4.2e9, 5e9, 1.7e9, 57e9, 15_300e6, 0), ("Q22024", "2024-05-01", 80e9, 2.00, 0.26, 0.16, 15e9, 4.0e9, 4e9, 1.5e9, 60e9, 15_400e6, 0), ] def _make_row(period, filing_date, revenue, eps, gross_margin, operating_margin, free_cash_flow, capex, buybacks, dividends_paid, total_debt, shares_diluted, guidance_disclosed): return { "ticker": "TEST", "company_name": "Test Corp", "period": period, "filing_date": filing_date, "form_type": "10-Q", "revenue": revenue, "revenue_yoy_pct": None, "eps": eps, "gross_margin": gross_margin, "operating_margin": operating_margin, "free_cash_flow": free_cash_flow, "capex": capex, "buybacks": buybacks, "dividends_paid": dividends_paid, "total_debt": total_debt, "shares_diluted": shares_diluted, "guidance_disclosed": guidance_disclosed, "guidance_text": None, "ingested_at": "2026-05-05T00:00:00", "effective_tax_rate": None, "interest_expense": None, "stockholders_equity": None, } # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def tmp_metrics_db(tmp_path, monkeypatch): """Create a temp SQLite DB, patch DB_PATH, and insert synthetic rows.""" db_path = tmp_path / "metrics.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() for row_args in QUARTERLY_ROWS: upsert_metrics(_make_row(*row_args)) yield db_path @pytest.fixture def single_row_db(tmp_path, monkeypatch): """Temp DB with exactly one row — no comparison possible.""" db_path = tmp_path / "metrics_single.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() upsert_metrics(_make_row( "Q12025", "2025-02-01", 100e9, 2.50, 0.30, 0.20, 20e9, 5e9, 8e9, 2e9, 50e9, 15_000e6, 1, )) # Override ticker so it doesn't collide with "TEST" row = _make_row("Q12025", "2025-02-01", 100e9, 2.50, 0.30, 0.20, 20e9, 5e9, 8e9, 2e9, 50e9, 15_000e6, 1) row["ticker"] = "SINGLE" upsert_metrics(row) yield db_path @pytest.fixture def withdrawn_guidance_db(tmp_path, monkeypatch): """Temp DB where the latest row has guidance_disclosed=0 but prior has 1.""" db_path = tmp_path / "metrics_withdrawn.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() # latest: guidance withdrawn row_latest = _make_row("Q12025", "2025-02-01", 100e9, 2.50, 0.30, 0.20, 20e9, 5e9, 8e9, 2e9, 50e9, 15_000e6, 0) row_latest["ticker"] = "WTEST" # prior: guidance was present row_prior = _make_row("Q42024", "2024-11-01", 95e9, 2.40, 0.29, 0.19, 19e9, 4.8e9, 7e9, 1.9e9, 52e9, 15_100e6, 1) row_prior["ticker"] = "WTEST" upsert_metrics(row_latest) upsert_metrics(row_prior) yield db_path # --------------------------------------------------------------------------- # Helper # --------------------------------------------------------------------------- def _find_delta(deltas: list, label: str) -> MetricDelta: """Return the first MetricDelta whose label matches, or raise AssertionError.""" for d in deltas: if d.label == label: return d raise AssertionError(f"No MetricDelta with label={label!r}; available: {[d.label for d in deltas]}") # --------------------------------------------------------------------------- # compute_metric_deltas # --------------------------------------------------------------------------- def test_compute_metric_deltas_yoy(tmp_metrics_db): deltas = compute_metric_deltas("TEST") assert deltas, "Expected non-empty list of MetricDelta" assert any(d.period_basis == "YoY" for d in deltas), "Expected at least one YoY delta" rev = _find_delta(deltas, "Revenue") assert rev.period_basis == "YoY" assert abs(rev.current - 100.0) < 0.01, f"current={rev.current}" assert abs(rev.prior - 90.0) < 0.01, f"prior={rev.prior}" assert abs(rev.delta_pct - 11.11) < 0.1, f"delta_pct={rev.delta_pct}" assert rev.direction == "up" assert rev.favorable is True def test_compute_metric_deltas_margin_pp(tmp_metrics_db): deltas = compute_metric_deltas("TEST") op = _find_delta(deltas, "Op. Margin") # 0.20 - 0.18 = 0.02 * 100 = 2.0 pp assert abs(op.delta_pct - 2.0) < 0.01, f"delta_pct={op.delta_pct}" assert op.unit == "pp" def test_compute_metric_deltas_debt_direction(tmp_metrics_db): deltas = compute_metric_deltas("TEST") debt = _find_delta(deltas, "Total Debt") # current=50B, prior=55B → down → favorable (lower debt is good) assert debt.direction == "down" assert debt.favorable is True assert abs(debt.delta_pct - (-9.09)) < 0.1, f"delta_pct={debt.delta_pct}" def test_compute_metric_deltas_single_quarter(single_row_db): """Only one row → no comparison possible → empty list.""" result = compute_metric_deltas("SINGLE") assert result == [] def test_compute_metric_deltas_empty(): """Ticker not in DB → returns empty list.""" result = compute_metric_deltas("NONEXISTENT") assert result == [] def test_qoq_requires_exact_adjacent_period(tmp_path, monkeypatch): db_path = tmp_path / "metrics_gap.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() for period, filing_date, revenue in ( ("Q32025", "2025-10-20", 120e9), ("Q12025", "2025-04-20", 100e9), ): row = _make_row( period, filing_date, revenue, None, None, None, None, None, None, None, None, None, 0, ) row["ticker"] = "GAP" upsert_metrics(row) assert compute_metric_deltas("GAP") == [] def test_qoq_uses_exact_q4_for_q1(tmp_path, monkeypatch): db_path = tmp_path / "metrics_adjacent.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() for period, filing_date, revenue in ( ("Q12025", "2025-04-20", 100e9), ("Q42024", "2025-02-01", 80e9), ): row = _make_row( period, filing_date, revenue, None, None, None, None, None, None, None, None, None, 0, ) row["ticker"] = "ADJ" upsert_metrics(row) revenue = _find_delta(compute_metric_deltas("ADJ"), "Revenue") assert revenue.period_basis == "QoQ" assert revenue.prior == 80.0 def test_virtual_q4_derives_additive_metrics_and_margins(): quarters = [ {"period": "Q12025", "revenue": 20.0, "gross_margin": 0.50, "operating_margin": 0.25, "free_cash_flow": 3.0, "capex": 1.0, "buybacks": 1.0, "dividends_paid": 0.2}, {"period": "Q22025", "revenue": 25.0, "gross_margin": 0.52, "operating_margin": 0.28, "free_cash_flow": 4.0, "capex": 1.2, "buybacks": 1.5, "dividends_paid": 0.2}, {"period": "Q32025", "revenue": 30.0, "gross_margin": 0.54, "operating_margin": 0.30, "free_cash_flow": 5.0, "capex": 1.3, "buybacks": 2.0, "dividends_paid": 0.2}, ] annual = { "period": "FY2025", "form_type": "10-K", "filing_date": "2026-02-01", "revenue": 110.0, "gross_margin": 0.55, "operating_margin": 0.31, "free_cash_flow": 20.0, "capex": 5.0, "buybacks": 7.0, "dividends_paid": 0.8, "eps": 10.0, "shares_diluted": 100.0, } q4 = _derive_virtual_q4_row(annual, quarters) assert q4["period"] == "Q42025" assert q4["revenue"] == 35.0 assert q4["free_cash_flow"] == 8.0 assert q4["capex"] == 1.5 assert q4["eps"] is None expected_gp = (110.0 * 0.55 - (20.0 * 0.50 + 25.0 * 0.52 + 30.0 * 0.54)) / 35.0 assert q4["gross_margin"] == pytest.approx(expected_gp) # --------------------------------------------------------------------------- # compute_eps_surprise # --------------------------------------------------------------------------- _SURPRISE_DATA = { "quarterlyEarnings": [ {"surprisePercentage": "5.2"}, {"surprisePercentage": "3.1"}, {"surprisePercentage": "2.8"}, {"surprisePercentage": "-1.5"}, {"surprisePercentage": "4.0"}, ] } @patch("ingestion.alphavantage.fetch_earnings", return_value=(_SURPRISE_DATA, None)) def test_compute_eps_surprise_beat_streak(mock_fetch): result = compute_eps_surprise("TEST") assert result is not None assert abs(result.latest_beat_pct - 5.2) < 0.01 assert result.beat_streak == 3, f"beat_streak={result.beat_streak}" # avg of first 4: (5.2 + 3.1 + 2.8 + -1.5) / 4 = 9.6 / 4 = 2.4 assert abs(result.avg_4q_surprise - 2.4) < 0.01, f"avg_4q={result.avg_4q_surprise}" @patch("ingestion.alphavantage.fetch_earnings", return_value=({"quarterlyEarnings": []}, None)) def test_compute_eps_surprise_returns_none_on_empty(mock_fetch): result = compute_eps_surprise("TEST") assert result is None # --------------------------------------------------------------------------- # compute_guidance_change # --------------------------------------------------------------------------- def test_compute_guidance_change_maintained(tmp_metrics_db): # Q12025 guidance_disclosed=1, Q42024 guidance_disclosed=1 → maintained result = compute_guidance_change("TEST", {}) assert result.disclosed_change == "maintained" def test_compute_guidance_change_withdrawn(withdrawn_guidance_db): # Q12025 guidance_disclosed=0, Q42024 guidance_disclosed=1 → withdrawn result = compute_guidance_change("WTEST", {}) assert result.disclosed_change == "withdrawn" # --------------------------------------------------------------------------- # compute_risk_diff # --------------------------------------------------------------------------- def test_compute_risk_diff_counts_new(): brief = { "risks_categorized": [ {"is_new_this_filing": True}, {"is_new_this_filing": False}, {"is_new_this_filing": True}, ] } assert compute_risk_diff(brief) == 2 def test_compute_risk_diff_empty_brief(): assert compute_risk_diff({}) == 0 def test_compute_risk_diff_none(): assert compute_risk_diff(None) == 0 # --------------------------------------------------------------------------- # build_quarter_snapshot # --------------------------------------------------------------------------- @patch("analytics.deltas.compute_metric_deltas", return_value=[]) def test_build_quarter_snapshot_none_on_no_data(mock_deltas, tmp_path, monkeypatch): """No DB rows → returns None.""" db_path = tmp_path / "empty.db" monkeypatch.setattr(metrics_db, "DB_PATH", db_path) init_db() result = build_quarter_snapshot("NONEXISTENT", {}) assert result is None @patch("analytics.deltas.compute_eps_surprise", return_value=None) def test_build_quarter_snapshot_success(mock_eps, tmp_metrics_db): snapshot = build_quarter_snapshot("TEST", {}) assert snapshot is not None assert isinstance(snapshot, QuarterSnapshot) assert snapshot.ticker == "TEST" assert snapshot.period == "Q12025" assert snapshot.filing_date == "2025-02-01"