"""Tests for us2kr.collectors.events (FRED + yfinance mocked, no network).""" from __future__ import annotations from datetime import date, datetime from typing import ClassVar from zoneinfo import ZoneInfo import pandas as pd import pytest import requests from us2kr.collectors import events from us2kr.collectors.events import fetch_earnings, fetch_events, next_us_business_day ET = ZoneInfo("America/New_York") SESSION = date(2026, 8, 25) # Tuesday NEXT = date(2026, 8, 26) # --------------------------------------------------------------------------- # fetch_events # --------------------------------------------------------------------------- class _JsonResp: def __init__(self, payload: dict, status: int = 200): self._payload = payload self.status_code = status def raise_for_status(self) -> None: if self.status_code >= 400: raise requests.HTTPError(f"{self.status_code}") def json(self) -> dict: return self._payload FRED_PAYLOAD = { "realtime_start": "2026-08-25", "realtime_end": "2026-08-26", "release_dates": [ {"release_id": 10, "release_name": "Consumer Price Index", "date": "2026-08-25"}, { "release_id": 180, "release_name": "Unemployment Insurance Weekly Claims Report", "date": "2026-08-26", }, {"release_id": 101, "release_name": "FOMC Press Release", "date": "2026-08-26"}, {"release_id": 18, "release_name": "H.15 Selected Interest Rates", "date": "2026-08-25"}, # unmapped {"release_id": 10, "release_name": "Consumer Price Index", "date": "2026-08-25"}, # dup {"release_id": 53, "release_name": "Gross Domestic Product", "date": "2026-08-24"}, # out of range {"release_id": 99, "release_name": "Producer Price Index", "date": "garbage"}, ], } def test_fetch_events_without_key_returns_empty(mocker): get = mocker.patch.object(events.requests, "get") assert fetch_events(SESSION, None) == [] assert fetch_events(SESSION, "") == [] get.assert_not_called() def test_fetch_events_maps_and_classifies(mocker): get = mocker.patch.object(events.requests, "get", return_value=_JsonResp(FRED_PAYLOAD)) out = fetch_events(SESSION, "test-key") assert [e.name for e in out] == ["미국 소비자물가지수(CPI)", "주간 신규 실업수당 청구건수", "FOMC"] cpi, claims, fomc = out assert cpi.kind == "released" and cpi.importance == "high" assert cpi.time == datetime(2026, 8, 25, 8, 30, tzinfo=ET) assert claims.kind == "upcoming" and claims.importance == "medium" assert fomc.kind == "upcoming" and fomc.time == datetime(2026, 8, 26, 14, 0, tzinfo=ET) assert all(e.time is None or e.time.tzinfo is not None for e in out) assert all(e.country == "US" for e in out) params = get.call_args.kwargs["params"] assert params["realtime_start"] == "2026-08-25" assert params["realtime_end"] == "2026-08-26" assert params["include_release_dates_with_no_data"] == "true" assert params["api_key"] == "test-key" assert params["file_type"] == "json" assert get.call_args.args[0] == events.FRED_RELEASE_DATES_URL assert get.call_args.kwargs["timeout"] == events.FRED_TIMEOUT_SECONDS def test_fetch_events_http_error_propagates(mocker): mocker.patch.object(events.requests, "get", return_value=_JsonResp({}, status=500)) with pytest.raises(requests.HTTPError): fetch_events(SESSION, "k") def test_fetch_events_errors_never_expose_the_api_key(mocker): """requests puts the full URL (api_key=...) in its error messages; fetch_events must scrub it.""" key = "0123456789abcdef0123456789abcdef" url = f"https://api.stlouisfed.org/fred/releases/dates?realtime_start=2026-08-25&api_key={key}&file_type=json" class _Resp(_JsonResp): def raise_for_status(self): raise requests.HTTPError(f"400 Client Error: Bad Request for url: {url}", response=self) mocker.patch.object(events.requests, "get", return_value=_Resp({}, status=400)) with pytest.raises(requests.HTTPError) as info: fetch_events(SESSION, key) assert key not in str(info.value) and "url" not in str(info.value) assert "HTTP 400" in str(info.value) and info.value.response is not None assert info.value.__cause__ is None and info.value.__suppress_context__ # original message not chained mocker.patch.object( events.requests, "get", side_effect=requests.ConnectionError(f"Max retries exceeded with url: {url} (Caused by ...)"), ) with pytest.raises(requests.ConnectionError) as info: fetch_events(SESSION, key) assert key not in str(info.value) assert isinstance(info.value, requests.RequestException) # run_safely still records a data gap def test_fetch_events_empty_payload(mocker): mocker.patch.object(events.requests, "get", return_value=_JsonResp({"release_dates": []})) assert fetch_events(SESSION, "k") == [] # --------------------------------------------------------------------------- # fetch_earnings # --------------------------------------------------------------------------- def _earnings_df(rows: list[tuple[datetime, float | None, float | None]]) -> pd.DataFrame: idx = pd.DatetimeIndex([pd.Timestamp(ts) for ts, _, _ in rows], name="Earnings Date") return pd.DataFrame( { "EPS Estimate": [est for _, est, _ in rows], "Reported EPS": [act for _, _, act in rows], "Surprise(%)": [None] * len(rows), }, index=idx, ) class _FakeTicker: registry: ClassVar[dict[str, _FakeTicker]] = {} def __init__(self, ticker: str): self.ticker = ticker spec = self.registry.get(ticker, {}) if spec.get("raise_calendar"): raise RuntimeError("yahoo down") self._calendar = spec.get("calendar", {}) self._df = spec.get("df") self._raise_df = spec.get("raise_df", False) @property def calendar(self) -> dict: return self._calendar def get_earnings_dates(self, limit: int = 12, offset: int = 0): if self._raise_df: raise RuntimeError("scrape failed") return self._df @pytest.fixture def fake_yf(monkeypatch): _FakeTicker.registry = {} monkeypatch.setattr(events.yf, "Ticker", _FakeTicker) monkeypatch.setattr(events, "next_us_business_day", lambda d: NEXT) return _FakeTicker.registry def test_fetch_earnings_matches_session_and_next_day(fake_yf): fake_yf["NVDA"] = { "calendar": {"Earnings Date": [SESSION], "Earnings Average": 1.23, "Revenue Average": 45_000_000_000}, "df": _earnings_df([(datetime(2026, 8, 25, 16, 0, tzinfo=ET), 1.25, 1.31)]), } fake_yf["MU"] = { "calendar": {"Earnings Date": [NEXT, date(2026, 8, 27)]}, "df": _earnings_df([(datetime(2026, 8, 26, 6, 0, tzinfo=ET), 2.0, float("nan"))]), } fake_yf["AAPL"] = {"calendar": {"Earnings Date": [date(2026, 10, 29)]}} fake_yf["TSLA"] = {"calendar": {}} tickers = ["^GSPC", "ES=F", "KRW=X", "BTC-USD", "DX-Y.NYB", "NVDA", "MU", "AAPL", "TSLA"] out = fetch_earnings(SESSION, tickers) assert [e.symbol for e in out] == ["NVDA", "MU"] nvda, mu = out assert nvda.report_date == SESSION and nvda.timing == "amc" assert nvda.name == "엔비디아" assert nvda.eps_estimate == 1.25 and nvda.eps_actual == 1.31 assert nvda.revenue_estimate == 45_000_000_000.0 assert mu.report_date == NEXT and mu.timing == "bmo" assert mu.eps_estimate == 2.0 and mu.eps_actual is None assert mu.revenue_estimate is None def test_earnings_dates_wins_over_local_tz_calendar_date(fake_yf): # yfinance's calendar date is computed in the machine's local zone: on a KST host an amc report # on SESSION appears as SESSION+1. The tz-aware earnings_dates row must take precedence. fake_yf["NVDA"] = { "calendar": {"Earnings Date": [NEXT], "Revenue Average": 92_000_000_000}, "df": _earnings_df([(datetime(2026, 8, 25, 16, 0, tzinfo=ET), 2.09, None)]), } (item,) = fetch_earnings(SESSION, ["NVDA"]) assert item.report_date == SESSION and item.timing == "amc" assert item.eps_estimate == 2.09 and item.revenue_estimate == 92_000_000_000.0 def test_calendar_fallback_when_earnings_dates_empty(fake_yf): fake_yf["MU"] = { "calendar": {"Earnings Date": [NEXT], "Earnings Average": 2.5, "Revenue Average": 9_000_000_000}, "df": pd.DataFrame(columns=["EPS Estimate", "Reported EPS", "Surprise(%)"]), } (item,) = fetch_earnings(SESSION, ["MU"]) assert item.report_date == NEXT and item.timing == "unknown" assert item.eps_estimate == 2.5 and item.revenue_estimate == 9_000_000_000.0 def test_no_match_when_earnings_dates_row_is_elsewhere(fake_yf): # A confirmed row on another date must not fall through to a (possibly stale) calendar entry. fake_yf["AAPL"] = { "calendar": {"Earnings Date": [SESSION]}, "df": _earnings_df([(datetime(2026, 10, 29, 16, 0, tzinfo=ET), 1.0, None)]), } assert fetch_earnings(SESSION, ["AAPL"]) == [] def test_fetch_earnings_per_ticker_isolation_and_unknown_timing(fake_yf): fake_yf["AMD"] = {"raise_calendar": True} fake_yf["AVGO"] = {"calendar": {"Earnings Date": [SESSION]}, "raise_df": True} fake_yf["TSM"] = {"calendar": {"Earnings Date": [SESSION]}, "df": None} out = fetch_earnings(SESSION, ["AMD", "AVGO", "TSM"]) assert [(e.symbol, e.timing) for e in out] == [("AVGO", "unknown"), ("TSM", "unknown")] def test_fetch_earnings_caps_ticker_count(fake_yf, mocker): tickers = [f"T{i}" for i in range(40)] for t in tickers: fake_yf[t] = {"calendar": {"Earnings Date": [SESSION]}, "df": None} spy = mocker.spy(_FakeTicker, "__init__") out = fetch_earnings(SESSION, tickers) assert spy.call_count == events.EARNINGS_MAX_TICKERS assert len(out) == events.EARNINGS_MAX_TICKERS def test_fetch_earnings_respects_time_budget(fake_yf, monkeypatch): tickers = ["A", "B", "C", "D"] for t in tickers: fake_yf[t] = {"calendar": {"Earnings Date": [SESSION]}, "df": None} clock = iter([0.0, 0.0, 5.0, 100.0, 100.0, 100.0, 100.0, 100.0]) monkeypatch.setattr(events.time, "monotonic", lambda: next(clock)) out = fetch_earnings(SESSION, tickers) assert [e.symbol for e in out] == ["A", "B"] def test_fetch_earnings_empty_tickers(fake_yf): assert fetch_earnings(SESSION, []) == [] def test_next_us_business_day_skips_weekend_and_holiday(): assert next_us_business_day(date(2026, 8, 28)) == date(2026, 8, 31) # Fri -> Mon assert next_us_business_day(date(2026, 7, 2)) == date(2026, 7, 6) # Jul 3 2026 observed holiday assert next_us_business_day(SESSION) == NEXT