diff --git a/README.md b/README.md index 6b47357..3fa26ba 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,12 @@ # quant-data-kit +`point_in_time_join` rejects different facts with the same entity/availability timestamp; +exact repeats are deduplicated. Resolve conflicting source revisions explicitly before joining. +`fetch_earnings_forecasts(start_date, end_date)` filters the announcement window and queries +all report quarters from the prior calendar year through the end year. Supply `report_periods` +explicitly for older revisions. Its next-weekday effective date remains an approximation, +not an exchange-calendar or full historical PIT certification. + Immutable downstream revision tracing and frozen paired result comparison are available through [Revision impact](docs/REVISION_IMPACT.md). diff --git a/src/quant_data_kit/providers/earnings_forecast.py b/src/quant_data_kit/providers/earnings_forecast.py index 324d8df..20ec37a 100644 --- a/src/quant_data_kit/providers/earnings_forecast.py +++ b/src/quant_data_kit/providers/earnings_forecast.py @@ -63,10 +63,34 @@ def fetch_earnings_forecasts( end_date: str, fetch_fn: Callable[[str], pd.DataFrame] | None = None, sleep_seconds: float = 0.2, + *, + report_periods: list[str] | None = None, ) -> pd.DataFrame: - """Fetch quarterly earnings forecasts; effective_date is next business day after announce.""" + """Fetch announcements in the inclusive date window, querying report periods separately. + + Default coverage is all quarters of the preceding calendar year through the + end year (including future quarter ends for early forecasts). Explicit + ``report_periods`` supports older restatements; this is not full PIT coverage. + ``effective_date`` retains the legacy weekday approximation, not an exchange calendar. + """ + start, end = parse_date(start_date), parse_date(end_date) + if start > end: + raise ValueError("start_date must not be after end_date") + periods = ( + report_periods + if report_periods is not None + else _quarter_end_dates(f"{start.year - 1}-01-01", f"{end.year}-12-31") + ) + if any( + not isinstance(p, str) + or len(p) != 8 + or not p.isdigit() + or not pd.Timestamp(p).is_quarter_end + for p in periods + ): + raise ValueError("report_periods must contain YYYYMMDD quarter ends") frames: list[pd.DataFrame] = [] - for period in _quarter_end_dates(start_date, end_date): + for period in sorted(set(periods)): if fetch_fn is not None: raw = fetch_fn(period) else: @@ -110,11 +134,16 @@ def fetch_earnings_forecasts( ) out["forecast_score"] = out["forecast_type"].map(_score_forecast_type) out["effective_date"] = out["announce_date"] + pd.offsets.BDay(1) + out = out[out["announce_date"].between(start, end)] frames.append(out) if not frames: return pd.DataFrame(columns=EARNINGS_COLUMNS) result = pd.concat(frames, ignore_index=True) - result = result.sort_values(["symbol", "effective_date"]).reset_index(drop=True) + result = ( + result.drop_duplicates() + .sort_values(["symbol", "effective_date", "report_period"]) + .reset_index(drop=True) + ) return result[EARNINGS_COLUMNS] diff --git a/src/quant_data_kit/temporal.py b/src/quant_data_kit/temporal.py index de0fba4..d2c2721 100644 --- a/src/quant_data_kit/temporal.py +++ b/src/quant_data_kit/temporal.py @@ -71,6 +71,12 @@ def point_in_time_join( if missing_facts: raise ValidationError(f"Missing fact columns: {missing_facts}") + # An availability timestamp is not a revision tie-breaker. Identical selected + # facts are harmless; conflicting values must never depend on input order. + right = right[[*by, availability_output, *selected]].drop_duplicates() + if right.duplicated([*by, availability_output], keep=False).any(): + raise ValidationError("Conflicting facts for the same point-in-time key and availability") + tolerance = pd.Timedelta(max_age) if max_age is not None else None parts: list[pd.DataFrame] = [] group_key: str | list[str] = by[0] if len(by) == 1 else list(by) diff --git a/tests/test_temporal_closeout.py b/tests/test_temporal_closeout.py new file mode 100644 index 0000000..60dfde1 --- /dev/null +++ b/tests/test_temporal_closeout.py @@ -0,0 +1,56 @@ +import pandas as pd +import pytest + +from quant_data_kit.exceptions import ValidationError +from quant_data_kit.providers.earnings_forecast import fetch_earnings_forecasts +from quant_data_kit.temporal import point_in_time_join + + +def test_conflicting_same_time_facts_are_rejected_in_both_orders(): + obs = pd.DataFrame({"date": ["2026-01-10"], "symbol": ["A"]}) + facts = pd.DataFrame({"available_at": ["2026-01-01"] * 2, "symbol": ["A"] * 2, "pe": [10, 20]}) + for value in (facts, facts.iloc[::-1]): + with pytest.raises(ValidationError, match="Conflicting"): + point_in_time_join(obs, value) + facts["pe"] = 10 + assert point_in_time_join(obs, facts)["pe"].tolist() == [10] + + +def test_announcement_window_queries_previous_and_forward_report_periods(): + called = [] + + def fetch(period): + called.append(period) + if period not in {"20251231", "20260331"}: + return pd.DataFrame() + return pd.DataFrame( + { + "股票代码": ["000001"] * 3, + "公告日期": ["2025-12-31", "2026-01-15", "2026-02-16"], + "预告类型": ["预增"] * 3, + } + ) + + result = fetch_earnings_forecasts("2026-01-01", "2026-02-15", fetch_fn=fetch) + assert {"20251231", "20260331"} <= set(called) + assert len(result) == 2 + assert set(result["announce_date"]) == {pd.Timestamp("2026-01-15")} + with pytest.raises(ValueError, match="start_date"): + fetch_earnings_forecasts("2026-02-15", "2026-01-01", fetch_fn=fetch) + + +def test_explicit_report_periods_allow_old_revisions_and_deduplicate_calls(): + calls = [] + + def fetch(period): + calls.append(period) + return pd.DataFrame() + + fetch_earnings_forecasts( + "2026-01-01", "2026-02-15", fetch_fn=fetch, report_periods=["20231231", "20231231"] + ) + assert calls == ["20231231"] + with pytest.raises(ValueError, match="quarter ends"): + fetch_earnings_forecasts( + "2026-01-01", "2026-02-15", fetch_fn=fetch, report_periods=["20230101"] + )