Spaces:
Sleeping
Sleeping
| import pytest | |
| def env(tmp_path, monkeypatch): | |
| """Wire both the snapshot archive and the metrics reader to fixtures.""" | |
| import storage.forecast_store as store | |
| import analytics.consensus as consensus | |
| monkeypatch.setattr(store, "CONSENSUS_PATH", tmp_path / "consensus.jsonl") | |
| reported = { | |
| "ticker": "NVDA", "period": "Q32027", "period_basis": "quarter", | |
| "report_date": "2026-11-19", "eps": 1.29, | |
| } | |
| monkeypatch.setattr(consensus, "get_all_metrics", lambda t: [reported]) | |
| return store, consensus, reported | |
| def _write(store, **overrides): | |
| rec = { | |
| "ticker": "NVDA", "target_period": "Q32027", "metric": "eps", | |
| "provider": "yfinance", "captured_at": "2026-11-18T21:30:00Z", | |
| "capture_kind": "live_snapshot", "basis": "gaap", "currency": "USD", | |
| "period_basis": "quarter", "value": 1.24, | |
| } | |
| rec.update(overrides) | |
| rec["snapshot_id"] = store.snapshot_id( | |
| rec["ticker"], rec["target_period"], rec["metric"], | |
| rec["provider"], rec["captured_at"], | |
| ) | |
| store.append_snapshot(rec) | |
| return rec | |
| def test_all_conditions_hold_returns_the_snapshot(env): | |
| store, consensus, _ = env | |
| _write(store) | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert status == "ALIGNED" | |
| assert snap["value"] == 1.24 | |
| def test_backfill_records_never_align(env): | |
| """Alpha Vantage gives no estimate timestamp, so it can never prove | |
| the estimate predates the release.""" | |
| store, consensus, _ = env | |
| _write(store, capture_kind="backfill_at_report") | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "no_live_snapshot" | |
| def test_a_capture_after_the_report_date_is_refused(env): | |
| store, consensus, _ = env | |
| _write(store, captured_at="2026-11-20T21:30:00Z") | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "no_pre_release_snapshot" | |
| def test_period_basis_mismatch_is_refused(env): | |
| store, consensus, _ = env | |
| _write(store, period_basis="ttm") | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "period_basis_mismatch" | |
| def test_adjusted_estimate_against_gaap_reported_is_refused(env): | |
| """yfinance estimates are adjusted; metrics.eps comes from the XBRL | |
| EarningsPerShareDiluted concept, which is GAAP. Refusing this is the | |
| expected common case for EPS.""" | |
| store, consensus, _ = env | |
| _write(store, basis="adjusted") | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "basis_mismatch" | |
| def test_non_usd_estimate_is_refused(env): | |
| store, consensus, _ = env | |
| _write(store, currency="EUR") | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "currency_mismatch" | |
| def test_no_snapshot_at_all(env): | |
| store, consensus, _ = env | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "no_live_snapshot" | |
| def test_unknown_period_is_refused(env): | |
| store, consensus, _ = env | |
| _write(store) | |
| snap, status = consensus.aligned_consensus("NVDA", "Q12099", "eps") | |
| assert snap is None | |
| assert status == "no_reported_period" | |
| def test_reported_row_without_a_report_date_is_refused(env, monkeypatch): | |
| store, consensus, reported = env | |
| monkeypatch.setattr( | |
| consensus, "get_all_metrics", | |
| lambda t: [{**reported, "report_date": None}], | |
| ) | |
| _write(store) | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert snap is None | |
| assert status == "no_report_date" | |
| def test_latest_pre_release_capture_wins(env): | |
| """The consensus the market carried into the print is the most recent one | |
| struck before it, not the oldest.""" | |
| store, consensus, _ = env | |
| _write(store, captured_at="2026-10-01T21:30:00Z", value=1.10) | |
| _write(store, captured_at="2026-11-18T21:30:00Z", value=1.24) | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "eps") | |
| assert status == "ALIGNED" | |
| assert snap["value"] == 1.24 | |
| def test_a_live_revenue_snapshot_aligns(env): | |
| """The realistic alignment path: street revenue estimates track reported | |
| GAAP revenue, so a live pre-release revenue snapshot passes every gate.""" | |
| store, consensus, _ = env | |
| _write(store, metric="revenue", basis="gaap", value=54_000_000_000.0) | |
| snap, status = consensus.aligned_consensus("NVDA", "Q32027", "revenue") | |
| assert status == "ALIGNED" | |
| assert snap["value"] == 54_000_000_000.0 | |