Fix Jin10 calendar backfill, retries and withdrawn schedules

Refresh historical previews after the local day ends and isolate retries by date without resetting browser scheduling progress.

Preserve withdrawal and restoration revisions, exclude withdrawn fallback values, and cover the behavior with isolated regression checks.

backend/src/alphapulse/calendar/jin10_store.py

@@ -5,7 +5,7 @@ from sqlalchemy import select
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from alphapulse.calendar.jin10 import Entry, day_url
+from alphapulse.calendar.jin10 import LOCAL, Entry, day_url
 from alphapulse.calendar.service import record_revision
 from alphapulse.models import EconomicRelease
 from alphapulse.models.jin10 import Jin10SyncDay
@@ -43,7 +43,7 @@ async def save_entry(session: AsyncSession, entry: Entry, raw_id: int, now: date
         await session.flush()
         record_revision(session, row, raw_id)
         return
-    changed = row.values != values or row.scheduled_at != entry.at or row.title != entry.title
+    changed = row.values != values or row.scheduled_at != entry.at or row.title != entry.title or row.schedule_status != "scheduled"
     row.schedule_checked_at = row.last_checked_at = now
     row.error = None
     if row.first_received_at is None and received:
@@ -51,22 +51,42 @@ async def save_entry(session: AsyncSession, entry: Entry, raw_id: int, now: date
         changed = True
     if changed:
         row.title, row.scheduled_at, row.values = entry.title, entry.at, values
+        row.schedule_status = "scheduled"
         row.schedule_url = day_url(date.fromisoformat(meta["date"]))
         row.revision_no += 1
         record_revision(session, row, raw_id)
 
 
+async def withdraw_missing(session: AsyncSession, day: date, kind: str, entries: list[Entry], raw_id: int, now: datetime) -> None:
+    # Only a fully validated response may withdraw records from its own day/category.
+    seen = {entry.identity for entry in entries}
+    rows = await session.scalars(select(EconomicRelease).where(
+        EconomicRelease.source == "jin10",
+        EconomicRelease.values["_jin10"]["date"].as_string() == day.isoformat(),
+        EconomicRelease.values["_jin10"]["kind"].as_string() == kind,
+        EconomicRelease.schedule_status != "withdrawn",
+    ).with_for_update())
+    for row in rows:
+        if row.external_id not in seen:
+            row.schedule_status = "withdrawn"
+            row.schedule_checked_at = row.last_checked_at = now
+            row.revision_no += 1
+            record_revision(session, row, raw_id)
+
+
 async def mark_day(session: AsyncSession, day: date, counts: dict, now: datetime) -> None:
     stmt = insert(Jin10SyncDay).values(day=day, captured_at=now, counts=counts)
     await session.execute(stmt.on_conflict_do_update(index_elements=["day"], set_={"captured_at": now, "counts": counts}))
 
 
-async def next_history_day(session: AsyncSession, start: date, end: date) -> date | None:
+async def next_history_day(session: AsyncSession, start: date, end: date, *, excluded: set[date] | None = None) -> date | None:
     from datetime import timedelta
-    covered = set(await session.scalars(select(Jin10SyncDay.day).where(Jin10SyncDay.day.between(start, end))))
+    snapshots = await session.scalars(select(Jin10SyncDay).where(Jin10SyncDay.day.between(start, end)))
+    # A preview captured before the local day ended is not a completed historical snapshot.
+    covered = {row.day for row in snapshots if row.captured_at.astimezone(LOCAL).date() > row.day}
     day = start
     while day <= end:
-        if day not in covered:
+        if day not in covered and day not in (excluded or set()):
             return day
         day += timedelta(days=1)
     return None

backend/src/alphapulse/calendar/jin10_view.py

@@ -115,7 +115,10 @@ def jin10_item(row: EconomicRelease, now: datetime) -> dict:
     revised = value.get("previous_revised")
     match = definition(meta)
     eligible = value.get("forecast_before_release", False)
-    if meta["kind"] != "data":
+    if row.schedule_status == "withdrawn":
+        status = "withdrawn"
+        direction, explanation = "uncertain", "金十已移除此日程；已保存值仅供追溯，不作为当前有效日程或方向判断。"
+    elif meta["kind"] != "data":
         status = "recorded"
         direction, explanation = "uncertain", "日程记录不代表事件已经发生；需结合实际消息判断 BTC/ETH 影响。"
     else:
@@ -170,6 +173,8 @@ def merge_items(official: list[tuple[EconomicRelease, dict]], third_party: list[
         item["field_sources"]["scheduled_at"] = provenance(row.source, row.schedule_checked_at, row.schedule_url)
         matches = []
         for extra in third_party:
+            if extra.schedule_status == "withdrawn":
+                continue
             meta = extra.values["_jin10"]
             match = definition(meta)
             if not match or match != (row.family, item["indicator"], row.stage):

backend/src/alphapulse/calendar/jin10_worker.py

@@ -10,7 +10,7 @@ from sqlalchemy import select
 from alphapulse.calendar.jin10 import KINDS, LOCAL, START, day_url, normalize
 from alphapulse.calendar.jin10_auth import Jin10Error
 from alphapulse.calendar.jin10_browser import Jin10Browser
-from alphapulse.calendar.jin10_store import mark_day, next_history_day, save_entry
+from alphapulse.calendar.jin10_store import mark_day, next_history_day, save_entry, withdraw_missing
 from alphapulse.calendar.service import save_raw
 from alphapulse.core.crypto import decrypt
 from alphapulse.core.db import SessionLocal
@@ -51,6 +51,7 @@ async def persist_day(day: date, responses: dict, encrypted: bytes, connected_at
         for kind in KINDS:
             for entry in entries[kind]:
                 await save_entry(session, entry, raw_ids[kind], now)
+            await withdraw_missing(session, day, kind, entries[kind], raw_ids[kind], now)
         counts = {kind: len(entries[kind]) for kind in KINDS}
         await mark_day(session, day, counts, now)
         await session.commit()
@@ -62,6 +63,7 @@ class Jin10Worker:
         self.browser = Jin10Browser()
         self.identity = None
         self.live_due: dict[date, float] = {}
+        self.retry_due: dict[date, float] = {}
         self.state = {"state": "disconnected", "error": None, "last_success_at": None}
 
     async def heartbeat(self) -> None:
@@ -77,6 +79,8 @@ class Jin10Worker:
         if identity is None:
             await self.browser.close()
             self.identity = None
+            self.live_due.clear()
+            self.retry_due.clear()
             self.state.update(state="disconnected", error=None, connected_at=None)
             await asyncio.sleep(5)
             return
@@ -84,8 +88,8 @@ class Jin10Worker:
             self.state = {"state": "connecting", "error": None, "connected_at": identity[1].isoformat(), "last_success_at": None}
             await self.browser.connect(decrypt(identity[0]))
             self.identity = identity
-            self.live_due.clear()
         today = datetime.now(LOCAL).date()
+        self.retry_due = {d: deadline for d, deadline in self.retry_due.items() if deadline > time.monotonic()}
         dates = [today + timedelta(days=i) for i in range(-1, 8)]
         self.live_due = {d: self.live_due.get(d, 0) for d in dates if d >= START}
         due = sorted((deadline, d) for d, deadline in self.live_due.items() if deadline <= time.monotonic())
@@ -95,15 +99,29 @@ class Jin10Worker:
             history = False
         else:
             async with SessionLocal() as session:
-                day = await next_history_day(session, START, today - timedelta(days=1))
+                day = await next_history_day(session, START, today - timedelta(days=1), excluded=set(self.retry_due))
             history = True
         if day is None:
             self.state.update(state="ready", history_next=None)
             await asyncio.sleep(2)
             return
         self.state.update(state="syncing", current_day=day.isoformat(), history=history, error=None)
-        responses = await self.browser.fetch_day(day)
+        try:
-        counts = await persist_day(day, responses, *identity, datetime.now(UTC))
+            responses = await self.browser.fetch_day(day)
+            counts = await persist_day(day, responses, *identity, datetime.now(UTC))
+        except (Jin10Error, ValueError, KeyError) as exc:
+            # A bad day must not reset progress or starve the remaining live/history dates.
+            self.retry_due[day] = time.monotonic() + 300
+            if not history:
+                self.live_due[day] = self.retry_due[day]
+            self.state.update(state="error", error=str(exc))
+            logger.warning("金十日历同步失败（%s）：%s", day, exc)
+            if isinstance(exc, Jin10Error):
+                self.identity = None
+                await self.browser.close()
+            await asyncio.sleep(5)
+            return
+        self.retry_due.pop(day, None)
         if not history:
             self.live_due[day] = time.monotonic() + (10 if day == today else 300)
         self.state.update(state="ready", last_success_at=datetime.now(UTC).isoformat(), counts=counts)

backend/tests/test_jin10_calendar.py

@@ -87,6 +87,7 @@ def test_periods_cross_year_without_inventing_unrecognized_periods():
 
 def stored(entry, id=11, now=None):
     return SimpleNamespace(id=id, source="jin10", family="jin10", title=entry.title, scheduled_at=entry.at,
+                           external_id=entry.identity, schedule_status="scheduled", period=None, stage=None,
                            schedule_url="https://rili.jin10.com/day/2026-11-04/", release_url=None,
                            values=deepcopy(entry.values), first_received_at=now, last_checked_at=now,
                            schedule_checked_at=now, error=None, revision_no=1)
@@ -205,3 +206,108 @@ def test_account_login_does_not_prove_calendar_access():
     response["data"] = {"date": "2025-01-08", "list": []}
     with pytest.raises(Jin10Error, match="读取权限"):
         validate_access(response)
+
+
+@pytest.mark.parametrize("captured_at,completed", [
+    ("2026-11-02T12:00:00+00:00", False),
+    ("2026-11-04T15:59:59+00:00", False),
+    ("2026-11-04T16:00:00+00:00", True),
+])
+def test_history_completion_requires_a_snapshot_after_the_shanghai_day(trading, captured_at, completed):
+    from alphapulse.calendar.jin10_store import next_history_day
+    day = date(2026, 11, 4)
+    snapshot = SimpleNamespace(day=day, captured_at=datetime.fromisoformat(captured_at))
+    session = SimpleNamespace(scalars=AsyncMock(return_value=[snapshot]))
+    assert asyncio.run(next_history_day(session, day, day)) == (None if completed else day)
+    assert asyncio.run(next_history_day(session, day, day, excluded={day})) is None
+
+
+def test_withdrawal_retains_values_and_records_evidence_and_restoration(trading):
+    from alphapulse.calendar.jin10_store import save_entry, withdraw_missing
+    from alphapulse.calendar.jin10_view import jin10_item
+    entry = entries("2026-11-04")[0]
+    now = datetime(2026, 11, 3, tzinfo=UTC)
+    row = stored(entry, now=now)
+    revisions = []
+    session = SimpleNamespace(scalars=AsyncMock(return_value=[row]), scalar=AsyncMock(return_value=row), add=revisions.append)
+    before = deepcopy(row.values)
+    asyncio.run(withdraw_missing(session, date(2026, 11, 4), "data", [], 42, now))
+    assert row.schedule_status == "withdrawn" and row.values == before
+    assert revisions[0].payload["schedule_status"] == "withdrawn" and revisions[0].raw_response_id == 42
+    item = jin10_item(row, now)
+    assert item["status"] == "withdrawn" and item["impact"] == "uncertain"
+    asyncio.run(save_entry(session, entry, 43, now + timedelta(minutes=1)))
+    assert row.schedule_status == "scheduled" and row.revision_no == 3
+    assert row.first_received_at == now
+    assert revisions[-1].payload["schedule_status"] == "scheduled" and revisions[-1].raw_response_id == 43
+
+
+def test_withdrawn_provider_cannot_supply_official_fallback(trading):
+    from alphapulse.calendar.jin10_view import merge_items
+    entry = next(e for e in entries("2026-10-29") if "jin10_79" in e.values)
+    extra = stored(entry)
+    extra.schedule_status = "withdrawn"
+    official = SimpleNamespace(source="dol", family="initial_claims", stage=None, period="2026-10-24", scheduled_at=entry.at,
+                               first_received_at=entry.at, schedule_checked_at=entry.at, release_url="https://www.dol.gov/claims.pdf",
+                               schedule_url="https://oui.doleta.gov/unemploy/archive.asp")
+    original = {"release_id": 1, "indicator": "initial_claims", "actual": "19.40", "previous": None, "forecast": None,
+                "status": "released", "scheduled_at": entry.at}
+    items = merge_items([(official, original)], [extra], entry.at)
+    result = next(item for item in items if item["release_id"] == 1)
+    assert result["previous"] is None and result["forecast"] is None and result["related_release_id"] is None
+    assert next(item for item in items if item["release_id"] == extra.id)["status"] == "withdrawn"
+
+
+@pytest.mark.parametrize("failure_kind", ["contract", "browser", "missing_key"])
+@pytest.mark.parametrize("failure_stage", ["fetch", "persist"])
+def test_failed_days_do_not_starve_live_or_history_and_retry_after_backoff(trading, monkeypatch, failure_kind, failure_stage):
+    from alphapulse.calendar import jin10_worker as worker
+    from alphapulse.calendar.jin10_auth import Jin10Error
+    now = datetime(2026, 11, 4, tzinfo=UTC)
+    today, yesterday = now.date(), now.date() - timedelta(days=1)
+    first_history = date(2025, 1, 1)
+    connection = SimpleNamespace(token_encrypted=b"synthetic", connected_at=now)
+    clock = [0]
+    successful, attempted = set(), []
+    class FrozenDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None): return now.astimezone(tz) if tz else now.replace(tzinfo=None)
+    class Session:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        async def get(self, model, key):
+            return connection if model is worker.Jin10Connection else SimpleNamespace(enabled=True)
+    async def history(session, start, end, *, excluded):
+        return next((first_history + timedelta(days=i) for i in range(3)
+                     if first_history + timedelta(days=i) not in successful | excluded), None)
+    def fail(day):
+        if day in (yesterday, first_history):
+            error = {"contract": ValueError, "browser": Jin10Error, "missing_key": KeyError}[failure_kind]
+            raise error("synthetic day failure")
+    async def fetch(day):
+        attempted.append(day)
+        if failure_stage == "fetch": fail(day)
+        return {}
+    async def persist(day, *args):
+        if failure_stage == "persist": fail(day)
+        successful.add(day)
+        return {}
+    async def sleep(seconds): clock[0] += seconds
+    monkeypatch.setattr(worker, "SessionLocal", Session)
+    monkeypatch.setattr(worker, "decrypt", lambda value: "synthetic")
+    monkeypatch.setattr(worker, "datetime", FrozenDatetime)
+    monkeypatch.setattr(worker.time, "monotonic", lambda: clock[0])
+    monkeypatch.setattr(worker, "next_history_day", history)
+    monkeypatch.setattr(worker, "persist_day", persist)
+    monkeypatch.setattr(worker.asyncio, "sleep", sleep)
+    collector = worker.Jin10Worker()
+    collector.browser = SimpleNamespace(connect=AsyncMock(), close=AsyncMock(), fetch_day=fetch)
+    async def verify():
+        for _ in range(30): await collector.step()
+        assert {today + timedelta(days=i) for i in range(8)} <= successful
+        assert first_history + timedelta(days=1) in successful and first_history + timedelta(days=2) in successful
+        assert attempted.count(yesterday) == attempted.count(first_history) == 1
+        clock[0] += 301
+        for _ in range(30): await collector.step()
+        assert attempted.count(yesterday) == attempted.count(first_history) == 2
+    asyncio.run(verify())

deploy/README.md

@@ -60,7 +60,9 @@ The independent `jin10-calendar` service runs Chromium and Jin10's official brow
 
 After the separately authorized migrations through `a8b1cd465ef3` and deployment, an administrator connects Jin10 in Settings using the Jin10 App QR flow. The account token is encrypted with the existing server encryption key and never returned to the browser. Disconnect retains history and invalidates in-flight writes; reconnect starts a new connection generation. Browser or authentication failures stay visible instead of being recorded as empty days.
 
-The collector prioritizes today, checks yesterday and the next seven days, and fills missing days from 2025-01-01 Asia/Shanghai at a limited pace. A day receives a persistent coverage checkpoint only after all three category responses pass date/structure checks. Historical dates with no checkpoint are coverage gaps, not evidence that no events occurred. A restart resumes the first gap. Failures retain raw responses without advancing coverage. No historical completeness or realtime latency is guaranteed before actual collection finishes.
+The collector prioritizes today, checks yesterday and the next seven days, and fills missing days from 2025-01-01 Asia/Shanghai at a limited pace. A day receives a persistent coverage checkpoint only after all three category responses pass date/structure checks. A checkpoint captured before that local day ended is a preview and must be refreshed for historical completion, including after downtime. Historical dates without a completed snapshot are coverage gaps, not evidence that no events occurred. A restart resumes the first gap. Failed dates retain raw evidence, wait five minutes before retrying, and do not block other dates; browser reconnection preserves scheduling progress. No historical completeness or realtime latency is guaranteed before actual collection finishes.
+
+Each complete daily response withdraws missing records only within its own day and category, retaining their values, raw evidence and a withdrawal revision. Withdrawn records are labeled explicitly and cannot supply fallback fields to an official release. A record that returns under the same source identity is restored with another revision.
 
 Official and Jin10 observations retain separate identities and revisions. The read API merges only verified indicator IDs, definitions, units, reference periods, release stages and exact publication times. Official nonmissing fields win independently; Jin10 fills missing fields and market forecasts with explicit provenance. Unverified definitions remain distinct records. Original and revised previous values remain separate. Forecast observations captured after publication are displayed with that limitation and excluded from historical surprise judgments. Jin10's gold/USD direction flags are not BTC/ETH signals. Tentative times and all-day holidays never display an invented exact midnight time.
 

scripts/smoke_calendar.py

@@ -11,7 +11,7 @@ from sqlalchemy import func, select
 from alphapulse.calendar.jin10 import normalize
 from alphapulse.calendar.jin10_history import calendar_history
 from alphapulse.calendar.jin10_store import next_history_day, save_entry
-from alphapulse.calendar.jin10_view import merge_items
+from alphapulse.calendar.jin10_view import jin10_item, merge_items
 from alphapulse.calendar.jin10_worker import ConnectionChanged, persist_day
 from alphapulse.calendar.parsers import Scheduled
 from alphapulse.calendar.service import release_rows, save_raw, save_result, save_schedule
@@ -67,6 +67,56 @@ async def verify():
         assert await session.scalar(select(func.count()).select_from(RawResponse)) == raw_before + 3
         assert (await session.get(Jin10SyncDay, day)).captured_at == coverage_before
 
+    # Existing preview checkpoints must be revisited after downtime, without a schema migration.
+    async with SessionLocal() as session:
+        checkpoint = await session.get(Jin10SyncDay, day)
+        checkpoint.captured_at = now - timedelta(days=2)
+        await session.commit()
+        assert await next_history_day(session, day, day) == day
+        assert await next_history_day(session, day, day, excluded={day}) is None
+    await persist_day(day, responses, encrypted, connected, now + timedelta(minutes=2))
+    async with SessionLocal() as session:
+        assert await next_history_day(session, day, day) is None
+        other_day_raw = await save_raw(session, "jin10", "https://rili.jin10.com/day/2026-11-03/", 200, {}, b"isolated-other-dates", now)
+        other_entries = [normalize(fixture["2025-01-08"], "data", date(2025, 1, 8))[0],
+                         *normalize(fixture["holiday:2026-11-03"], "holiday", date(2026, 11, 3))]
+        for entry in other_entries:
+            await save_entry(session, entry, other_day_raw, now)
+        await session.commit()
+        revision_count = await session.scalar(select(func.count()).select_from(EconomicRevision))
+
+    # Removing an economic release and event withdraws only those identities, once per change.
+    removed = deepcopy(responses)
+    adp_identity = adp.external_id
+    removed["data"]["list"] = [r for r in removed["data"]["list"] if f"data:{r['data_id']}" != adp_identity]
+    removed_event = removed["event"]["list"].pop(0)
+    removed_ids = {adp_identity, f"event:{removed_event['id']}"}
+    await persist_day(day, removed, encrypted, connected, now + timedelta(minutes=3))
+    async with SessionLocal() as session:
+        rows = list(await session.scalars(select(EconomicRelease).where(EconomicRelease.source == "jin10")))
+        withdrawn = [r for r in rows if r.external_id in removed_ids]
+        assert len(withdrawn) == 2 and all(r.schedule_status == "withdrawn" and r.revision_no == 2 for r in withdrawn)
+        assert all(jin10_item(r, now)["status"] == "withdrawn" for r in withdrawn)
+        assert all(r.schedule_status == "scheduled" and r.revision_no == 1 for r in rows if r.external_id not in removed_ids)
+        evidence = list((await session.execute(select(EconomicRevision, RawResponse).join(
+            RawResponse, RawResponse.id == EconomicRevision.raw_response_id,
+        ).where(EconomicRevision.release_id.in_([r.id for r in withdrawn]), EconomicRevision.revision_no == 2))).all())
+        assert len(evidence) == 2
+        assert all(revision.payload["schedule_status"] == "withdrawn" and raw.url.endswith("#" + owner.external_id.split(":")[0])
+                   for revision, raw in evidence for owner in withdrawn if owner.id == revision.release_id)
+        assert await session.scalar(select(func.count()).select_from(EconomicRevision)) == revision_count + 2
+    await persist_day(day, removed, encrypted, connected, now + timedelta(minutes=4))
+    async with SessionLocal() as session:
+        assert await session.scalar(select(func.count()).select_from(EconomicRevision)) == revision_count + 2
+    await persist_day(day, responses, encrypted, connected, now + timedelta(minutes=5))
+    async with SessionLocal() as session:
+        restored = list(await session.scalars(select(EconomicRelease).where(
+            EconomicRelease.source == "jin10", EconomicRelease.external_id.in_(removed_ids),
+        )))
+        assert all(r.schedule_status == "scheduled" and r.revision_no == 3 for r in restored)
+        assert next(r for r in restored if r.external_id == adp_identity).id == adp.id
+        assert await session.scalar(select(func.count()).select_from(EconomicRevision)) == revision_count + 4
+
     # Join an official weekly release to the exact provider series, including its revised previous value.
     claims_day = date(2026, 10, 29)
     entries = normalize(fixture[str(claims_day)], "data", claims_day)
@@ -92,7 +142,7 @@ async def verify():
         assert len(history) == 1 and history[0]["actual"] == "19.40" and history[0]["forecast"] == "19.9"
         assert history[0]["first_recorded_actual"] == "19.40"
         assert history[0]["first_received_at"] == now
-    print("Calendar acceptance passed: three categories, exact official joins, revisions, raw evidence, late forecasts, coverage and obsolete connection rejection")
+    print("Calendar acceptance passed: three categories, exact official joins, revisions, raw evidence, late forecasts, post-day coverage, scoped withdrawal/restoration and obsolete connection rejection")
 
 
 async def main():