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