Reject ambiguous PIT facts and separate announcement and report windows

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).
 

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]

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)

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"]
+        )