From 9f5b3550fa6fc5da0a2b73c9b73c169506c88ee6 Mon Sep 17 00:00:00 2001 From: eezd Date: Mon, 26 Oct 2026 17:20:19 +0800 Subject: [PATCH] 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 | 54 ++++++++++--- backend/src/alphapulse/calendar/catalog.py | 1 + backend/tests/test_calendar_history.py | 88 +++++++++++++++++++++ 3 files changed, 134 insertions(+), 9 deletions(-) diff --git a/backend/src/alphapulse/calendar/backfill.py b/backend/src/alphapulse/calendar/backfill.py index a996eaa..cacf700 100644 --- a/backend/src/alphapulse/calendar/backfill.py +++ b/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,10 +327,11 @@ 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] - announcements.append((document, "umich", Announcement(event.at, period, stage, values), event.title)) + 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) if family is None: diff --git a/backend/src/alphapulse/calendar/catalog.py b/backend/src/alphapulse/calendar/catalog.py index 6319f66..c8185ad 100644 --- a/backend/src/alphapulse/calendar/catalog.py +++ b/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", diff --git a/backend/tests/test_calendar_history.py b/backend/tests/test_calendar_history.py index 5643f6e..279330b 100644 --- a/backend/tests/test_calendar_history.py +++ b/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 = ('December Final Results' if year == 2024 else + 'January Preliminary Results' + 'January Final Results') + return httpx.Response(200, text=f'
{year} Reports
' + 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='Historical Prelim/Final Release Dates') + if request.url.path == "/survey-info.php": + return httpx.Response(200, text='2025 Release Dates') + 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'''

Final Results for January 2026

+ + +
JanDec
20262025
Index of Consumer Sentiment56.452.9
''', 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())