Backfill Michigan calendar results from official archives

Fetch annual report indexes, skip stored PDFs, and stop the source on rate limits. Match report periods, stages, and publication dates to official schedules while preserving received results and keeping revised monthly series separate.

backend/src/alphapulse/calendar/backfill.py

@@ -4,6 +4,8 @@ uv run python -m alphapulse.calendar.backfill --start 2025-01-01 [--stored-only]
 按北京时间的公布日期筛选。归档可能已修订，first_received_at 保留真实采集时间，
 不能把本次回补用作当时已知信息。失败与无公告缺口会写入 JSON 输出并返回非零状态。
 写入时应暂停 calendar 进程，完成后恢复，避免与日程同步并发创建同一事件。
+密歇根历史初值/终值取官方年度报告索引与当期 PDF，核对正式日期及日程；
+归档尚未公开的报告仍列为缺失，当前月度修订序列不用于填补当期公告。
 """
 import argparse
 import asyncio
@@ -21,12 +23,12 @@ import httpx
 from parsel import Selector
 from sqlalchemy import select, text
 
-from alphapulse.calendar.archives import Announcement, parse_announcements
+from alphapulse.calendar.archives import Announcement, parse_announcements, parse_umich_report
 from alphapulse.calendar import europe
-from alphapulse.calendar.catalog import UMICH_CSV, UMICH_HISTORY, UMICH_HOME, UMICH_INFO
+from alphapulse.calendar.catalog import UMICH_CSV, UMICH_HISTORY, UMICH_HOME, UMICH_INFO, UMICH_REPORTS
 from alphapulse.calendar.parsers import Scheduled, bea_family, compact
 from alphapulse.calendar.service import record_revision, save_raw, save_result, save_schedule, save_series
-from alphapulse.calendar.umich import current_result, document_links, monthly_series, parse_schedule, pdf_text
+from alphapulse.calendar.umich import current_result, document_links, monthly_series, parse_schedule, pdf_text, report_links
 from alphapulse.core.db import SessionLocal
 from alphapulse.core.official_sources import HOSTS as EUROPE_HOSTS
 from alphapulse.models import EconomicRelease, RawBody, RawResponse
@@ -206,6 +208,27 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                         documents.append(await fetch("umich", link))
             except (httpx.HTTPError, ValueError) as exc:
                 report_error("umich", str(exc.request.url) if isinstance(exc, httpx.HTTPError) else url, exc)
+        # 公告可能在次月公布（如 2025-07 终值在 8 月 1 日）；保留开始月份的前一期，最终按 PDF 正式日期过滤。
+        first_period = (start.replace(day=1) - timedelta(days=1)).strftime("%Y-%m")
+        for year in (range(int(first_period[:4]), end.year + 1) if "umich" in selected else ()):
+            if "umich" in stopped_sources:
+                break
+            url = f"{UMICH_REPORTS}?year={year}"
+            try:
+                index = await fetch("umich", url)
+                documents.append(index)
+                links = report_links(index.body.decode("utf-8"), year)
+                for (period, _), link in links.items():
+                    if first_period <= period <= end.strftime("%Y-%m") and link not in known:
+                        try:
+                            documents.append(await fetch("umich", link))
+                            known.add(link)
+                        except (httpx.HTTPError, ValueError) as exc:
+                            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (429, 503):
+                                raise
+                            report_error("umich", link, exc)
+            except (httpx.HTTPError, ValueError) as exc:
+                report_error("umich", str(exc.request.url) if isinstance(exc, httpx.HTTPError) else url, exc)
     return documents
 
 
@@ -267,22 +290,34 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
             EconomicRelease.source.in_(selected), EconomicRelease.scheduled_at >= begin, EconomicRelease.scheduled_at < finish)))
     if not stored_only:
         documents += await fetch_documents(start, end, {d.url for d in documents}, errors, **({"sources": sources} if sources is not None else {}))
-    schedules: dict[str, tuple[Scheduled, Document]] = {}
     announcements = []
     series_points = []
     # 先处理历史表，再处理年度更正；重跑的稳定身份不依赖日程 PDF 的文件名。
     parsed_schedules = []
+    reports = []
     for document in (d for d in documents if d.source == "umich" and d.body.startswith(b"%PDF")):
         try:
             content = await asyncio.to_thread(pdf_text, document.body)
+            if "Index of Consumer Sentiment" in content:
+                reports.append((document, await asyncio.to_thread(parse_umich_report, content)))
+                continue
             parsed_schedules.append((bool(re.search(r"RELEASE DATES FOR 20\d\d", content)), document,
                                      await asyncio.to_thread(parse_schedule, content)))
         except ValueError as exc:
             errors.append({"url": document.url, "error": str(exc)})
+    all_schedules: dict[str, tuple[Scheduled, Document]] = {}
     for _, document, events in sorted(parsed_schedules, key=lambda item: (item[0], item[1].captured_at)):
         for event in events:
-            if begin <= event.at < finish:
-                schedules[event.external_id] = event, document
+            all_schedules[event.external_id] = event, document
+    schedules = {identity: item for identity, item in all_schedules.items() if begin <= item[0].at < finish}
+    for document, announcement in sorted(reports, key=lambda item: item[0].captured_at):
+        if not begin <= announcement.at < finish:
+            continue
+        identity = f"umich-{announcement.period}-{announcement.stage}"
+        if identity not in schedules or schedules[identity][0].at != announcement.at:
+            errors.append({"url": document.url, "error": "密歇根历史报告的期次、阶段或正式日期与官方日程不符"})
+            continue
+        announcements.append((document, "umich", announcement, schedules[identity][0].title))
     for document in sorted(documents, key=lambda d: d.captured_at):
         try:
             if document.source == "umich":
@@ -292,9 +327,10 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
                 elif document.url.rstrip("/") == UMICH_HOME.rstrip("/"):
                     period, stage, values = current_result(document.body.decode())
                     identity = f"umich-{period}-{stage}"
-                    if identity not in schedules:
+                    if identity not in all_schedules:
                         raise ValueError("密歇根当前期次未找到官方公布日程")
-                    event, _ = schedules[identity]
+                    event, _ = all_schedules[identity]
+                    if begin <= event.at < finish:
                         announcements.append((document, "umich", Announcement(event.at, period, stage, values), event.title))
                 continue
             family = family_for(document)

backend/src/alphapulse/calendar/catalog.py

@@ -87,6 +87,7 @@ UMICH_HOME = "https://www.sca.isr.umich.edu/"
 UMICH_HISTORY = "https://data.sca.isr.umich.edu/technical-docs.php"
 UMICH_INFO = "https://data.sca.isr.umich.edu/survey-info.php"
 UMICH_CSV = "https://www.sca.isr.umich.edu/files/tbmics.csv"
+UMICH_REPORTS = "https://data.sca.isr.umich.edu/reports.php"
 SCHEDULE_URLS = {
     "bls": "https://www.bls.gov/schedule/news_release/bls.ics",
     "bea": "https://www.bea.gov/news/schedule/ics/online-calendar-subscription.ics",

backend/tests/test_calendar_history.py

@@ -263,3 +263,91 @@ def test_archive_limit_stops_entire_source_preserves_documents_and_continues_oth
         assert limited_path in errors[0]["url"]
         assert "/bls/news-release/ppi.htm" not in calls and "/news.release/cpi_03122025.htm" not in calls
     asyncio.run(check())
+
+
+@pytest.mark.parametrize("status", [200, 429, 503])
+def test_umich_report_fetch_skips_stored_documents_and_stops_on_rate_limit(trading, monkeypatch, status):
+    import httpx
+    from unittest.mock import AsyncMock
+    from alphapulse.calendar import backfill
+
+    async def check():
+        calls = []
+        known = {"https://data.sca.isr.umich.edu/fetchdoc.php?docid=1"}
+        def respond(request):
+            calls.append(str(request.url))
+            if request.url.path == "/reports.php":
+                year = int(request.url.params["year"])
+                links = ('<a href="fetchdoc.php?docid=1">December Final Results</a>' if year == 2024 else
+                         '<a href="fetchdoc.php?docid=2">January Preliminary Results</a>'
+                         '<a href="fetchdoc.php?docid=3">January Final Results</a>')
+                return httpx.Response(200, text=f'<div id="report_heading"><span class="title">{year} Reports</span></div>' + links)
+            if request.url.path == "/fetchdoc.php":
+                assert request.url.params["docid"] != "1", "已存报告不应重复下载"
+                if request.url.params["docid"] == "2":
+                    return httpx.Response(status, content=b"%PDF report", headers={"Retry-After": "120"})
+                assert status == 200, "限流后不应继续下载该来源"
+                return httpx.Response(200, content=b"%PDF report")
+            if request.url.path == "/technical-docs.php":
+                return httpx.Response(200, text='<a href="/schedule.pdf">Historical Prelim/Final Release Dates</a>')
+            if request.url.path == "/survey-info.php":
+                return httpx.Response(200, text='<a href="/schedule.pdf">2025 Release Dates</a>')
+            return httpx.Response(200, text="Other official document")
+
+        client = httpx.AsyncClient
+        monkeypatch.setattr(backfill.httpx, "AsyncClient", lambda **kwargs: client(**kwargs, transport=httpx.MockTransport(respond)))
+        monkeypatch.setattr(backfill, "ensure_quiet_window", AsyncMock())
+        monkeypatch.setattr(backfill.asyncio, "sleep", AsyncMock())
+        errors = []
+        docs = await backfill.fetch_documents(date(2025, 1, 1), date(2025, 12, 31), known, errors, sources={"umich"})
+        assert any(d.url.endswith("reports.php?year=2024") for d in docs)
+        report_urls = [d.url for d in docs if "/fetchdoc.php" in d.url]
+        if status == 200:
+            assert len(report_urls) == 2 and not errors
+        else:
+            assert not report_urls and len(errors) == 1
+            assert errors[0]["source"] == "umich" and errors[0]["retry_after"] == "120"
+            assert not any("docid=3" in url for url in calls)
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("schedule_date,valid", [("1/24/2025", True), ("1/23/2025", False)])
+def test_umich_preview_matches_report_date_to_schedule_without_writes(trading, monkeypatch, schedule_date, valid):
+    from unittest.mock import AsyncMock
+    from alphapulse.calendar import backfill
+
+    async def check():
+        now = datetime(2026, 10, 26, tzinfo=UTC)
+        report = (Path(__file__).parent / "fixtures/calendar/umich-2025-01-final.txt").read_text()
+        docs = [backfill.Document("umich", "https://data.sca.isr.umich.edu/fetchdoc.php?docid=1", b"%PDF schedule", now),
+                backfill.Document("umich", "https://data.sca.isr.umich.edu/fetchdoc.php?docid=2", b"%PDF report", now),
+                backfill.Document("umich", backfill.UMICH_HOME, b'''<h1>Final Results for January 2026</h1>
+                <table id="front_table"><tr><td></td><td>Jan</td><td>Dec</td></tr>
+                <tr><td></td><td>2026</td><td>2025</td></tr>
+                <tr><td>Index of Consumer Sentiment</td><td>56.4</td><td>52.9</td></tr></table>''', now)]
+        class Session:
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *_):
+                pass
+
+            async def execute(self, _):
+                return SimpleNamespace(all=lambda: [])
+
+            async def scalars(self, _):
+                return []
+
+        monkeypatch.setattr(backfill, "SessionLocal", Session)
+        monkeypatch.setattr(backfill, "fetch_documents", AsyncMock(return_value=docs))
+        monkeypatch.setattr(backfill, "pdf_text", lambda body: report if body == b"%PDF report" else
+                            f"Jan 2025 1/10/2025 {schedule_date}\nJan 2026 1/9/2026 1/23/2026")
+        save = AsyncMock()
+        monkeypatch.setattr(backfill, "save_announcement", save)
+        result = await backfill.run(date(2025, 1, 1), date(2025, 12, 31), sources={"umich"})
+        assert result["mode"] == "dry-run" and result["changed_announcements"] == 0
+        save.assert_not_awaited()
+        assert result["announcements"].get("umich", 0) == int(valid)
+        assert bool(result["errors"]) is not valid
+        assert {x["stage"] for x in result["missing_announcements"]} == ({"preliminary"} if valid else {"preliminary", "final"})
+    asyncio.run(check())