Add official DOL and EIA economic calendar feeds

Collect weekly claims and precise commercial petroleum inventory changes with official schedule exceptions, distinct reporting periods, raw evidence, and revisions.

Extend explicit history backfills without changing their default source scope. Report archive gaps and reject mismatched weeks instead of filling current values into history.

backend/src/alphapulse/calendar/backfill.py

@@ -24,7 +24,7 @@ from parsel import Selector
 from sqlalchemy import select, text
 
 from alphapulse.calendar.archives import Announcement, parse_announcements, parse_umich_report
-from alphapulse.calendar import europe
+from alphapulse.calendar import dol, eia, europe
 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
@@ -37,6 +37,8 @@ LOCAL = ZoneInfo("Asia/Shanghai")
 HOSTS = {"bls": {"www.bls.gov", "www.dol.gov"}, "bea": {"www.bea.gov"},
          "fed": {"www.federalreserve.gov"}, "umich": {"www.sca.isr.umich.edu", "data.sca.isr.umich.edu"}}
 US_SOURCES = frozenset(HOSTS)
+HOSTS["dol"] = {"www.dol.gov", "oui.doleta.gov"}
+HOSTS["eia"] = {"www.eia.gov", "ir.eia.gov"}
 HOSTS.update(EUROPE_HOSTS)
 
 
@@ -77,6 +79,8 @@ def family_for(document: Document) -> str | None:
 
 def archive_links(source: str, body: str, base: str, start: date, end: date) -> tuple[list[str], str | None]:
     """只跟随官方索引的实际链接，不按月份合成公告地址。"""
+    if source == "dol":
+        return dol.archive_links(body, start, end), None
     sel = Selector(text=body, type="html")
     links = []
     for node in sel.css("a[href]"):
@@ -157,6 +161,7 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
         indexes = [("bls", f"https://www.bls.gov/bls/news-release/{name}.htm") for name in ("cpi", "ppi", "empsit")]
         indexes += [("bea", "https://www.bea.gov/news/archive?field_related_product_target_id=All&created_1=All&title="),
                     ("fed", "https://www.federalreserve.gov/monetarypolicy/fomccalendars.htm")]
+        indexes += [("dol", dol.SCHEDULE)]
         for source, initial in indexes:
             if source not in selected or source in stopped_sources:
                 continue
@@ -172,6 +177,12 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                     links, following = archive_links(source, index.body.decode("utf-8"), url, start, end)
                     if not links and url == initial:
                         raise ValueError("官方归档未识别到目标公告，覆盖范围尚未核实")
+                    if source == "dol":
+                        years = {int(urlparse(link).path.split("/")[2]) for link in links}
+                        absent = set(range(start.year, end.year + 1)) - years
+                        if absent:
+                            errors.append({"source": source, "url": url,
+                                           "error": f"DOL 归档索引未列出年份 {sorted(absent)}，历史覆盖不完整"})
                     for link in links:
                         if link in known:
                             continue
@@ -186,6 +197,23 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                 except (httpx.HTTPError, ValueError) as exc:
                     report_error(source, str(exc.request.url) if isinstance(exc, httpx.HTTPError) else url, exc)
                     break
+        if "dol" in selected and "dol" not in stopped_sources and dol.LATEST not in known:
+            try:
+                documents.append(await fetch("dol", dol.LATEST))
+            except (httpx.HTTPError, ValueError) as exc:
+                report_error("dol", dol.LATEST, exc)
+        if "eia" in selected:
+            try:
+                documents.append(await fetch("eia", eia.SCHEDULE))
+                index = await fetch("eia", eia.ARCHIVE)
+                for url in eia.archive_links(index.body.decode(), start, end):
+                    page = await fetch("eia", url)
+                    link = eia.csv_link(page.body.decode(), url)
+                    if link not in known:
+                        documents.append(await fetch("eia", link))
+                        known.add(link)
+            except (httpx.HTTPError, ValueError) as exc:
+                report_error("eia", str(exc.request.url) if isinstance(exc, httpx.HTTPError) else eia.ARCHIVE, exc)
         from alphapulse.calendar.europe_archives import collect
         for source in sorted(selected & EUROPE_HOSTS.keys()):
             try:
@@ -282,7 +310,7 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
             RawResponse.source_code.in_(selected), RawResponse.status == 200,
             (RawResponse.url.contains("/news.release/") | RawResponse.url.contains("/newsroom/economicdata/") |
              RawResponse.url.contains("/news/202") | RawResponse.url.contains("/pressreleases/monetary") |
-             (RawResponse.source_code == "umich") | RawResponse.source_code.in_(EUROPE_HOSTS)),
+             RawResponse.source_code.in_(("umich", "dol", "eia")) | RawResponse.source_code.in_(EUROPE_HOSTS)),
         ).distinct(RawResponse.source_code, RawResponse.url, RawResponse.body_sha256)
             .order_by(RawResponse.source_code, RawResponse.url, RawResponse.body_sha256, RawResponse.id))).all()
         documents = [Document(r.source_code, r.url, gzip.decompress(body), r.fetched_at, r.id, r.headers) for r, body in rows]
@@ -292,6 +320,8 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
         documents += await fetch_documents(start, end, {d.url for d in documents}, errors, **({"sources": sources} if sources is not None else {}))
     announcements = []
     series_points = []
+    eia_schedules = [doc for doc in documents if doc.source == "eia" and doc.url == eia.SCHEDULE]
+    eia_schedule = max(eia_schedules, key=lambda doc: doc.captured_at) if eia_schedules else None
     # 先处理历史表，再处理年度更正；重跑的稳定身份不依赖日程 PDF 的文件名。
     parsed_schedules = []
     reports = []
@@ -320,6 +350,25 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
         announcements.append((document, "umich", announcement, schedules[identity][0].title))
     for document in sorted(documents, key=lambda d: d.captured_at):
         try:
+            if document.source == "eia":
+                if urlparse(document.url).path.endswith("/table1.csv"):
+                    if eia_schedule is None:
+                        raise ValueError("EIA 缺少已采集的公布规则和节假日例外表")
+                    announcement = eia.report(document.body.decode(), eia_schedule.body.decode())
+                    path = urlparse(document.url).path
+                    date_match = re.search(r"/archive/20\d\d/(\d{4}_\d{2}_\d{2})/", path)
+                    if date_match and datetime.strptime(date_match[1], "%Y_%m_%d").date() != announcement.at.astimezone(eia.EASTERN).date():
+                        raise ValueError("EIA 报告统计周与归档公布日期不符")
+                    if begin <= announcement.at < finish:
+                        announcements.append((document, "eia_stocks", announcement, "Weekly Petroleum Status Report"))
+                continue
+            if document.source == "dol":
+                if document.body.startswith(b"%PDF"):
+                    parsed = await asyncio.to_thread(dol.report, await asyncio.to_thread(pdf_text, document.body))
+                    for family, announcement in parsed.items():
+                        if begin <= announcement.at < finish:
+                            announcements.append((document, family, announcement, "Unemployment Insurance Weekly Claims"))
+                continue
             if document.source == "umich":
                 if document.url == UMICH_CSV:
                     series_points.append((document, [(p, v) for p, v in monthly_series(document.body.decode())
@@ -348,6 +397,10 @@ async def run(start: date, end: date, *, commit: bool = False, stored_only: bool
     changed = 0
     if commit:
         await ensure_quiet_window()
+        if eia_schedule is not None and eia_schedule.raw_id is None:
+            async with SessionLocal() as session:
+                await save_raw(session, "eia", eia_schedule.url, 200, eia_schedule.headers, eia_schedule.body, eia_schedule.captured_at)
+                await session.commit()
         batches = {}
         for event, document in schedules.values():
             _, batch = batches.setdefault((document.url, document.captured_at), (document, []))

backend/src/alphapulse/calendar/catalog.py

@@ -37,6 +37,11 @@ INDICATORS = {
     "gdp": Indicator("实际 GDP 年化季率", 4),
     "fed_rate": Indicator("联邦基金利率目标上限", 5, rate_channel="inverse"),
     "umich_sentiment": Indicator("密歇根大学消费者信心指数", 3, "点"),
+    "initial_claims": Indicator("初请失业金人数（季调）", 4, "万人"),
+    "continued_claims": Indicator("续请失业金人数（季调）", 3, "万人"),
+    "eia_crude": Indicator("EIA 商业原油库存变动（不含战略储备）", 4, "万桶"),
+    "eia_gasoline": Indicator("EIA 汽油库存变动", 3, "万桶"),
+    "eia_distillate": Indicator("EIA 精炼油库存变动", 3, "万桶"),
 }
 FAMILIES = {
     "cpi": Family("US", "bls", ("cpi_mom", "cpi_yoy", "core_cpi_mom", "core_cpi_yoy")),
@@ -46,6 +51,9 @@ FAMILIES = {
     "gdp": Family("US", "bea", ("gdp",)),
     "fomc": Family("US", "fed", ("fed_rate",)),
     "umich": Family("US", "umich", ("umich_sentiment",)),
+    "initial_claims": Family("US", "dol", ("initial_claims",)),
+    "continued_claims": Family("US", "dol", ("continued_claims",)),
+    "eia_stocks": Family("US", "eia", ("eia_crude", "eia_gasoline", "eia_distillate")),
 }
 STAGES = {"preliminary": "初值", "final": "终值", "initial": "首发值", "updated": "更新估计", "advance": "初值", "second": "二次估计", "third": "终值"}
 STAGES["flash"] = "快报值"

backend/src/alphapulse/calendar/dol.py

@@ -0,0 +1,100 @@
+"""DOL weekly claims: use published schedules and the seasonally adjusted release."""
+import re
+from datetime import UTC, date, datetime, time, timedelta
+from decimal import Decimal
+from urllib.parse import urljoin, urlparse
+
+from parsel import Selector
+
+from alphapulse.calendar.archives import Announcement, publication_time
+from alphapulse.calendar.parsers import EASTERN, MONTHS, Scheduled, compact
+
+SCHEDULE = "https://oui.doleta.gov/unemploy/archive.asp"
+LATEST = "https://www.dol.gov/ui/data.pdf"
+FAMILIES = ("initial_claims", "continued_claims")
+MONTH_PATTERN = "(?:" + "|".join(MONTHS) + ")"
+
+
+def schedule(body: str, now: datetime) -> list[Scheduled]:
+    sel = Selector(text=body)
+    text = compact(sel.xpath("string(//body)").get() or "")
+    if not re.search(r"Publication Schedule:.*Thursday morning at 8:30am EST", text):
+        raise ValueError("DOL 缺少已验证的每周公布规则")
+    exceptions = {}
+    tables = [table for table in sel.css("table") if {compact(n.xpath("string(.)").get()) for n in table.css("th")} == {"Release Date", "Release Time"}]
+    if len(tables) != 1:
+        raise ValueError("DOL 缺少唯一的节假日例外表")
+    for row in tables[0].css("tbody tr"):
+        cells = [compact(n.xpath("string(.)").get()) for n in row.css("td")]
+        if len(cells) != 2 or not re.fullmatch(r"\w+, " + MONTH_PATTERN + r" \d{1,2}, 20\d\d", cells[0]):
+            raise ValueError("DOL 例外公布日期格式已变化")
+        day = datetime.strptime(cells[0], "%A, %B %d, %Y").date()
+        if cells[1] != "8:30 AM EST":
+            raise ValueError("DOL 例外公布时刻已变化")
+        thursday = day + timedelta(days=3 - day.weekday())
+        if thursday in exceptions:
+            raise ValueError("DOL 同周公布例外不唯一")
+        exceptions[thursday] = day
+    start = now.astimezone(EASTERN).date() - timedelta(days=45)
+    end = now.astimezone(EASTERN).date() + timedelta(days=90)
+    days = {start + timedelta(days=i) for i in range((end - start).days + 1)}
+    result = []
+    for thursday in sorted(day for day in days if day.weekday() == 3):
+        day = exceptions.get(thursday, thursday)
+        at = datetime.combine(day, time(8, 30), EASTERN).astimezone(UTC)
+        for family in FAMILIES:
+            result.append(Scheduled(f"dol-{family}-{day}", family, "Unemployment Insurance Weekly Claims", at))
+    return result
+
+
+def archive_links(body: str, start: date, end: date) -> list[str]:
+    links = []
+    for link in Selector(text=body).css("a[href]::attr(href)").getall():
+        url = urljoin(SCHEDULE, link)
+        parsed = urlparse(url)
+        match = re.fullmatch(r"/press/(20\d\d)/(\d{6})\.pdf", parsed.path)
+        if parsed.scheme != "https" or parsed.hostname != "oui.doleta.gov" or match is None:
+            continue
+        day = datetime.strptime(match[2], "%m%d%y").date()
+        if day.year != int(match[1]):
+            raise ValueError("DOL 归档路径年份与日期不符")
+        if start <= day <= end:
+            links.append(url)
+    if not links:
+        raise ValueError("DOL 归档未列出所选范围的报告，历史覆盖尚未确认")
+    return list(dict.fromkeys(links))
+
+
+def report(text: str) -> dict[str, Announcement]:
+    text = compact(text)
+    if "UNEMPLOYMENT INSURANCE WEEKLY CLAIMS" not in text:
+        raise ValueError("DOL 响应不是失业金周报")
+    at = publication_time(text)
+    match = re.search(r"SEASONALLY ADJUSTED DATA (.*?) UNADJUSTED DATA", text)
+    if match is None:
+        raise ValueError("DOL 缺少独立的季调数据正文")
+    adjusted = match[1]
+    initial = re.search(r"In the week ending (" + MONTH_PATTERN + r" \d{1,2}), the advance figure for seasonally adjusted initial claims was ([\d,]+)", adjusted)
+    continued = re.search(r"advance number for seasonally adjusted insured unemployment during the week ending (" + MONTH_PATTERN + r" \d{1,2}) was ([\d,]+)", adjusted)
+    if initial is None or continued is None:
+        raise ValueError("DOL 缺少初请或续请季调读数")
+    result = {}
+    for family, reading in zip(FAMILIES, (initial, continued), strict=True):
+        # Resolve year rollover from the official publication date, never from collection time.
+        period = datetime.strptime(f"{reading[1]}, {at.astimezone(EASTERN).year}", "%B %d, %Y").date()
+        published = at.astimezone(EASTERN).date()
+        if period > published:
+            period = period.replace(year=period.year - 1)
+        if not 0 <= (published - period).days <= 21:
+            raise ValueError("DOL 统计周与公布日期不符")
+        following = adjusted[reading.end():].split("The 4-week moving average", 1)[0]
+        previous = re.search(r"previous week's level was (?:revised (?:up|down) by [\d,]+ from [\d,]+ to|unrevised at) ([\d,]+)", following)
+        if previous is None:
+            previous = re.search(r"previous week's unrevised level of ([\d,]+)", following)
+        if previous is None:
+            raise ValueError("DOL 缺少本报告列示的可比前值")
+        value = {"actual": format(Decimal(reading[2].replace(",", "")) / 10000, "f"),
+                 "previous": format(Decimal(previous[1].replace(",", "")) / 10000, "f"),
+                 "previous_basis": "DOL 本次季调报告列示的上周值（可能已修订）"}
+        result[family] = Announcement(at, period.isoformat(), None, {family: value})
+    return result

backend/src/alphapulse/calendar/eia.py

@@ -0,0 +1,111 @@
+"""EIA commercial inventory changes from the official, unrounded Table 1 CSV."""
+import csv
+from datetime import UTC, date, datetime, time, timedelta
+from decimal import Decimal
+from io import StringIO
+import re
+from urllib.parse import urljoin, urlparse
+
+from parsel import Selector
+
+from alphapulse.calendar.archives import Announcement
+from alphapulse.calendar.parsers import EASTERN, Scheduled, compact
+
+SCHEDULE = "https://www.eia.gov/petroleum/supply/weekly/schedule.php"
+ARCHIVE = "https://www.eia.gov/petroleum/supply/weekly/archive/"
+LATEST = "https://ir.eia.gov/wpsr/table1.csv"
+ROWS = {"Commercial (Excluding SPR)": "eia_crude", "Total Motor Gasoline": "eia_gasoline", "Distillate Fuel Oil": "eia_distillate"}
+
+
+def exceptions(body: str) -> dict[date, datetime]:
+    sel = Selector(text=body)
+    text = compact(sel.xpath("string(//body)").get() or "")
+    if "10:30 a.m. eastern time on Wednesday" not in text or "All times are eastern" not in text:
+        raise ValueError("EIA 缺少已验证的每周公布规则")
+    tables = sel.css("table.schedule")
+    if len(tables) != 1 or "Data for the week ending" not in tables[0].xpath("string(.)").get():
+        raise ValueError("EIA 缺少唯一的节假日例外表")
+    result = {}
+    for row in tables[0].css("tbody tr"):
+        cells = [compact(n.xpath("string(.)").get()) for n in row.css("th,td")]
+        if len(cells) != 5:
+            raise ValueError("EIA 例外日程列格式已变化")
+        period = datetime.strptime(cells[0], "%B %d, %Y").date()
+        day = datetime.strptime(cells[1], "%B %d, %Y").date()
+        clock = datetime.strptime(cells[3].replace("a.m.", "AM").replace("p.m.", "PM"), "%I:%M %p").time()
+        if cells[2] != day.strftime("%A") or period in result:
+            raise ValueError("EIA 例外日程星期或统计周不符")
+        result[period] = datetime.combine(day, clock, EASTERN).astimezone(UTC)
+    return result
+
+
+def release_at(period: date, changes: dict[date, datetime]) -> datetime:
+    if period.weekday() != 4:
+        raise ValueError("EIA 统计期不是周五")
+    return changes.get(period, datetime.combine(period + timedelta(days=5), time(10, 30), EASTERN).astimezone(UTC))
+
+
+def schedule(body: str, now: datetime) -> list[Scheduled]:
+    changes = exceptions(body)
+    start = now.astimezone(EASTERN).date() - timedelta(days=60)
+    result = []
+    for i in range(160):
+        period = start + timedelta(days=i)
+        if period.weekday() == 4:
+            result.append(Scheduled(f"eia-stocks-{period}", "eia_stocks", "Weekly Petroleum Status Report",
+                                    release_at(period, changes), period=period.isoformat()))
+    return result
+
+
+def archive_links(body: str, start: date, end: date) -> list[str]:
+    result = []
+    for link in Selector(text=body).css("a[href]::attr(href)").getall():
+        url = urljoin(ARCHIVE, link)
+        parsed = urlparse(url)
+        match = re.fullmatch(r"/petroleum/supply/weekly/archive/(20\d\d)/(\d{4}_\d{2}_\d{2})/wpsr_\2\.php", parsed.path)
+        if parsed.scheme != "https" or parsed.hostname != "www.eia.gov" or match is None:
+            continue
+        day = datetime.strptime(match[2], "%Y_%m_%d").date()
+        if day.year != int(match[1]):
+            raise ValueError("EIA 归档年份与日期不符")
+        if start <= day <= end:
+            result.append(url)
+    if not result:
+        raise ValueError("EIA 归档没有所选范围的已公布报告")
+    return list(dict.fromkeys(result))
+
+
+def csv_link(body: str, base: str) -> str:
+    links = [urljoin(base, link) for link in Selector(text=body).css("a[href]::attr(href)").getall()
+             if link == "download/table1.csv"]
+    if len(set(links)) != 1:
+        raise ValueError("EIA 归档未唯一列示 Table 1 CSV")
+    return links[0]
+
+
+def report(body: str, schedule_body: str) -> Announcement:
+    rows = csv.reader(StringIO(body))
+    header = next(rows, [])
+    if len(header) != 8 or header[0] != "STUB_1" or header[3] != "Difference":
+        raise ValueError("EIA Table 1 CSV 表头已变化")
+    period, previous = (datetime.strptime(value, "%m/%d/%y").date() for value in header[1:3])
+    if period - previous != timedelta(days=7):
+        raise ValueError("EIA Table 1 当前周与上周不连续")
+    values = {}
+    for row in rows:
+        if row and row[0] == "STUB_1":
+            break  # Following tables use daily supply rates, not inventory levels.
+        if not row or row[0] not in ROWS:
+            continue
+        if len(row) != 8 or ROWS[row[0]] in values:
+            raise ValueError("EIA 库存数据行重复或缺列")
+        number = row[3].replace(",", "")
+        if not re.fullmatch(r"[+-]?\d+(?:\.\d+)?", number):
+            raise ValueError("EIA 库存变动不是有限数值")
+        change = Decimal(number)
+        # Table 1 inventories are million barrels; the calendar displays ten thousand barrels.
+        # Previous inventory levels are not the previous week's inventory CHANGE.
+        values[ROWS[row[0]]] = {"actual": format(change * 100, "f"), "previous": None}
+    if values.keys() != set(ROWS.values()):
+        raise ValueError("EIA 缺少商业原油、汽油或精炼油库存变动")
+    return Announcement(release_at(period, exceptions(schedule_body)), period.isoformat(), None, values)

backend/src/alphapulse/calendar/service.py

@@ -187,7 +187,10 @@ def release_rows(row: EconomicRelease, now: datetime) -> list[dict]:
     country_name = COUNTRIES[family.country_code]
     period_name = ""
     if row.period is not None:
-        period_name = f"第{row.period[-1]}季度" if "-Q" in row.period else f"{int(row.period[5:7])}月"
+        if len(row.period) == 10:
+            period_name = f"截至{row.period}当周"
+        else:
+            period_name = f"第{row.period[-1]}季度" if "-Q" in row.period else f"{int(row.period[5:7])}月"
     for key in family.indicators:
         indicator = INDICATORS[key]
         value = row.values.get(key, {})

backend/src/alphapulse/calendar/worker.py

@@ -12,7 +12,7 @@ from parsel import Selector
 from sqlalchemy import select
 
 from alphapulse.calendar.catalog import SCHEDULE_URLS, UMICH_CSV, UMICH_HISTORY, UMICH_HOME, UMICH_INFO
-from alphapulse.calendar import europe
+from alphapulse.calendar import dol, eia, europe
 from alphapulse.calendar.parsers import EASTERN, MONTHS, bea_release_link, parse_fed_calendar, parse_ics, parse_values, release_date, bls_periods, release_metadata
 from alphapulse.calendar.service import apply_bls_periods, record_revision, next_poll, save_raw, save_result, save_schedule, save_series
 from alphapulse.calendar.umich import current_result, document_links, monthly_series, parse_schedule, pdf_text
@@ -57,12 +57,14 @@ class CalendarWorker:
         self.http = httpx.AsyncClient(timeout=8, follow_redirects=True, headers={"User-Agent": "AlphaPulse economic calendar"})
         self.bls = httpx.AsyncClient(timeout=8, follow_redirects=True, verify=ctx, headers={"User-Agent": "AlphaPulse economic calendar"})
         self.sources: dict[str, dict] = {}
+        self.eia_schedule_body: str | None = None
         self.fetch_slots = asyncio.Semaphore(2)
 
     async def fetch(self, source: str, url: str, *, query: dict | None = None) -> tuple[httpx.Response, int]:
         # 链接来自公开响应，限制到对应官方域名，避免外部内容扩大访问范围。
         allowed = {"bls": {"www.bls.gov"}, "bea": {"www.bea.gov", "apps.bea.gov"}, "fed": {"www.federalreserve.gov"},
-                   "umich": {"www.sca.isr.umich.edu", "data.sca.isr.umich.edu"}, **HOSTS}
+                   "umich": {"www.sca.isr.umich.edu", "data.sca.isr.umich.edu"}, "dol": {"www.dol.gov", "oui.doleta.gov"},
+                   "eia": {"www.eia.gov", "ir.eia.gov"}, **HOSTS}
         if urlparse(url).scheme != "https" or urlparse(url).hostname not in allowed[source]:
             raise ValueError("公告链接不是对应机构的官方 HTTPS 地址")
         async with self.fetch_slots:
@@ -84,6 +86,8 @@ class CalendarWorker:
         urls += [("bls", f"https://www.bls.gov/schedule/news_release/{code}.htm") for code in ("cpi", "ppi", "empsit")]
         urls += [("umich", UMICH_INFO), ("umich", UMICH_CSV)]
         urls += list(europe.SCHEDULES.items())
+        urls += [("dol", dol.SCHEDULE)]
+        urls += [("eia", eia.SCHEDULE)]
         # 启动时明确各来源尚未就绪；月份滚动后移除不再检查的旧月状态。
         self.sources = {url: self.sources.get(url, {"source": source, "checked_at": now.isoformat(), "error": "尚未同步"})
                         for source, url in urls}
@@ -95,7 +99,14 @@ class CalendarWorker:
                 fetch_url = europe.eurostat_calendar_url(now) if source == "eurostat" else url
                 response, raw_id = await self.fetch(source, fetch_url, query=europe.INSEE_QUERY if source == "insee" else None)
                 response.raise_for_status()
-                if source in europe.SCHEDULES:
+                if source in ("dol", "eia"):
+                    events = (dol if source == "dol" else eia).schedule(response.text, now)
+                    async with SessionLocal() as session:
+                        await save_schedule(session, source, url, events, raw_id, datetime.now(UTC), withdraw_missing=False)
+                        await session.commit()
+                    if source == "eia":
+                        self.eia_schedule_body = response.text
+                elif source in europe.SCHEDULES:
                     events = europe.parse_schedule(source, response.text)
                     async with SessionLocal() as session:
                         # 欧洲公开日程是滚动/分页窗口，缺失不等于撤回。
@@ -168,6 +179,30 @@ class CalendarWorker:
             await session.commit()
 
     async def result_page(self, row: EconomicRelease) -> tuple[httpx.Response, int] | None:
+        if row.source == "eia":
+            if self.eia_schedule_body is None:
+                schedule, _ = await self.fetch("eia", eia.SCHEDULE)
+                schedule.raise_for_status()
+                eia.exceptions(schedule.text)
+                self.eia_schedule_body = schedule.text
+            response, raw_id = await self.fetch("eia", row.release_url or eia.LATEST)
+            response.raise_for_status()
+            announcement = eia.report(response.text, self.eia_schedule_body)
+            if announcement.at < row.scheduled_at or (announcement.at > row.scheduled_at and row.first_received_at is not None):
+                return None
+            if announcement.at != row.scheduled_at or announcement.period != row.period:
+                raise ValueError("EIA 公告统计周或时间与所选日程不符，未填入其他期次的数据")
+            return response, raw_id
+        if row.source == "dol":
+            response, raw_id = await self.fetch("dol", row.release_url or dol.LATEST)
+            response.raise_for_status()
+            announcements = await asyncio.to_thread(dol.report, await asyncio.to_thread(pdf_text, response.content))
+            announcement = announcements[row.family]
+            if announcement.at < row.scheduled_at or (announcement.at > row.scheduled_at and row.first_received_at is not None):
+                return None
+            if announcement.at != row.scheduled_at:
+                raise ValueError("DOL 公告时间与所选日程不符，未填入其他期次的数据")
+            return response, raw_id
         if row.source in europe.SCHEDULES:
             return await self.european_result(row)
         date = row.scheduled_at.astimezone(EASTERN).date()
@@ -275,7 +310,13 @@ class CalendarWorker:
                 page = await self.result_page(row)
                 if page is not None:
                     response, raw_id = page
-                    if row.source in europe.SCHEDULES:
+                    if row.source == "eia":
+                        announcement = eia.report(response.text, self.eia_schedule_body)
+                        period, stage, values = announcement.period, announcement.stage, announcement.values
+                    elif row.source == "dol":
+                        announcement = (await asyncio.to_thread(dol.report, await asyncio.to_thread(pdf_text, response.content)))[row.family]
+                        period, stage, values = announcement.period, announcement.stage, announcement.values
+                    elif row.source in europe.SCHEDULES:
                         announcement = europe.parse_announcement(row.family, response.text)
                         period, stage, values = announcement.period, announcement.stage, announcement.values
                     elif row.family == "umich":