diff --git a/app/core/compare_alert_worker.py b/app/core/compare_alert_worker.py
new file mode 100644
index 0000000..745da2a
--- /dev/null
+++ b/app/core/compare_alert_worker.py
@@ -0,0 +1,196 @@
+"""比价失败报警后台任务:周期扫 comparison_record 新落定记录 → 规则命中 → 飞书汇总。
+
+结构仿 heartbeat_monitor_worker(单实例文件锁 + asyncio 轮询 + 优雅退出);发送与 DB 全同步,
+放 asyncio.to_thread。水位存 app_config(key=compare_alert.last_watermark,值=上次处理的最大
+updated_at ISO 串),查询用 updated_at 自身比较、规避时区。见 spec 第 5/6 节。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy import func, select
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.integrations import feishu_notifier
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services.compare_alert import classify_batch
+from app.services.compare_alert_format import format_alert_message
+
+logger = logging.getLogger("shagua.compare_alert")
+
+WATERMARK_KEY = "compare_alert.last_watermark"
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "compare_alert.lock"
+
+
+def _read_watermark(db) -> datetime | None:
+    row = db.get(AppConfig, WATERMARK_KEY)
+    if row is None or not row.value:
+        return None
+    try:
+        return datetime.fromisoformat(row.value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _write_watermark(db, value: datetime) -> None:
+    iso = value.isoformat()
+    row = db.get(AppConfig, WATERMARK_KEY)
+    if row is None:
+        db.add(AppConfig(key=WATERMARK_KEY, value=iso, updated_by_admin_id=None))
+    else:
+        row.value = iso
+    db.commit()
+
+
+def _send(text: str) -> None:
+    """发飞书(webhook 为空则只打日志、不外发)。失败抛 FeishuNotifyError 由调用方处理。"""
+    webhook = settings.COMPARE_ALERT_FEISHU_WEBHOOK
+    if not webhook:
+        logger.info("[compare-alert] webhook 未配置,仅打印:\n%s", text)
+        return
+    feishu_notifier.send_feishu_text(
+        webhook, text, timeout=settings.COMPARE_ALERT_FEISHU_TIMEOUT_SEC
+    )
+
+
+def _scan_and_alert() -> None:
+    """一轮:读水位 → 查有更新记录 → 规则 → 有命中发飞书 → 成功推进水位。同步,放 to_thread 调。"""
+    with SessionLocal() as db:
+        watermark = _read_watermark(db)
+        if watermark is None:
+            # 冷启动:水位 = 当前 max(updated_at),不回溯历史失败。空表则本轮不建水位、下轮再说——
+            # 不用 datetime.now():那是本地时钟,与 SQLite 的 updated_at(UTC CURRENT_TIMESTAMP)
+            # 不同源、会差 8h,导致新记录永远追不上水位。只用 DB 产出的 updated_at 值。
+            max_updated = db.scalar(select(func.max(ComparisonRecord.updated_at)))
+            if max_updated is None:
+                logger.info("[compare-alert] 冷启动:表空,待有记录后再建水位")
+                return
+            _write_watermark(db, max_updated)
+            logger.info("[compare-alert] 冷启动,水位初始化=%s", max_updated)
+            return
+
+        records = list(
+            db.scalars(
+                select(ComparisonRecord)
+                .where(ComparisonRecord.updated_at > watermark)
+                .order_by(ComparisonRecord.updated_at.asc())
+            )
+        )
+        if not records:
+            return
+        batch_max = max(r.updated_at for r in records)
+
+        hits = classify_batch(
+            records,
+            cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
+            cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
+            timeout_keywords=settings.compare_alert_timeout_keywords,
+            unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
+            biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
+        )
+
+        if hits or settings.COMPARE_ALERT_SEND_EMPTY:
+            label = datetime.now().strftime("%Y-%m-%d %H:%M")
+            text = (
+                format_alert_message(
+                    hits, window_label=label,
+                    max_detail_per_type=settings.COMPARE_ALERT_MAX_DETAIL_PER_TYPE,
+                    max_total=settings.COMPARE_ALERT_MAX_TOTAL,
+                )
+                if hits
+                else f"🚨 比价失败报警 · {label} · 本期无异常"
+            )
+            try:
+                _send(text)
+            except feishu_notifier.FeishuNotifyError:
+                logger.warning("[compare-alert] 发送失败,水位不推进、下轮补发", exc_info=True)
+                return  # 不推进水位
+
+        _write_watermark(db, batch_max)
+        if hits:
+            logger.info("[compare-alert] 本轮命中 %d 条,水位推进到 %s", len(hits), batch_max)
+
+
+# ---- 单实例锁 + 轮询循环(结构同 heartbeat_monitor_worker)---------------------
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+        if fd is None:
+            yield False
+            return
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.COMPARE_ALERT_SCAN_INTERVAL_SEC))
+    lock_stale_after = max(interval * 3, 600)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("compare-alert skipped: another worker owns lock")
+            return
+        logger.info("compare-alert worker started interval=%ss", interval)
+        try:
+            while True:
+                try:
+                    _touch_lock()
+                    await asyncio.to_thread(_scan_and_alert)
+                except SQLAlchemyError:
+                    logger.exception("compare-alert db error")
+                except Exception:  # noqa: BLE001 - 后台任务不因单次异常退出
+                    logger.exception("compare-alert unexpected error")
+                await asyncio.sleep(interval)
+        except asyncio.CancelledError:
+            logger.info("compare-alert worker stopped")
+            raise
+
+
+def start_compare_alert_worker() -> asyncio.Task | None:
+    if not settings.COMPARE_ALERT_ENABLED:
+        logger.info("compare-alert worker disabled")
+        return None
+    return asyncio.create_task(_run_loop(), name="compare-alert-worker")
+
+
+async def stop_compare_alert_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
diff --git a/tests/test_compare_alert_worker.py b/tests/test_compare_alert_worker.py
new file mode 100644
index 0000000..77a235d
--- /dev/null
+++ b/tests/test_compare_alert_worker.py
@@ -0,0 +1,80 @@
+"""扫描 worker 的同步核心 _scan_and_alert:水位读写 / 冷启动 / 发送成功推进、失败不推进。
+
+monkeypatch 掉真正的飞书发送;用真实 SQLite(SessionLocal)插入几条记录。
+"""
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from app.core import compare_alert_worker as w
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+
+WM_KEY = w.WATERMARK_KEY
+
+
+@pytest.fixture()
+def clean_db():
+    db = SessionLocal()
+    db.query(ComparisonRecord).delete()
+    db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
+    db.commit()
+    yield db
+    db.query(ComparisonRecord).delete()
+    db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
+    db.commit()
+    db.close()
+
+
+def _add(db, trace, status, **kw):
+    rec = ComparisonRecord(trace_id=trace, status=status, **kw)
+    db.add(rec)
+    db.commit()
+    db.refresh(rec)
+    return rec
+
+
+def test_cold_start_sets_watermark_no_alert(clean_db, monkeypatch):
+    # 冷启动:表非空 → 水位=当前 max(updated_at),不回溯已有历史失败、不报
+    _add(clean_db, "old-fail", "failed", fail_reason=None, information="比价过程出错")
+    sent = []
+    monkeypatch.setattr(w, "_send", lambda text: sent.append(text))
+    w._scan_and_alert()
+    assert sent == []  # 冷启动不报历史
+    row = clean_db.get(AppConfig, WM_KEY)
+    assert row is not None  # 水位已初始化
+
+
+def test_alerts_on_new_failed_and_advances(clean_db, monkeypatch):
+    # seed 一条 + 冷启动建水位;sleep 1.1s 拉开时间(SQLite 秒级精度,否则新记录同秒、追不上水位)
+    _add(clean_db, "seed", "running")
+    monkeypatch.setattr(w, "_send", lambda text: None)
+    w._scan_and_alert()  # 冷启动,水位=seed.updated_at
+    time.sleep(1.1)
+    sent = []
+    monkeypatch.setattr(w, "_send", lambda text: sent.append(text))
+    _add(clean_db, "new-fail", "failed", fail_reason=None, information="比价过程出错")
+    w._scan_and_alert()
+    assert len(sent) == 1
+    assert "系统技术失败 1 条" in sent[0]
+
+
+def test_send_failure_does_not_advance_watermark(clean_db, monkeypatch):
+    _add(clean_db, "seed2", "running")
+    monkeypatch.setattr(w, "_send", lambda text: None)
+    w._scan_and_alert()  # 冷启动建水位
+    wm_before = clean_db.get(AppConfig, WM_KEY).value
+    time.sleep(1.1)
+    _add(clean_db, "fail-send", "failed", fail_reason=None, information="比价过程出错")
+
+    def boom(text):
+        raise w.feishu_notifier.FeishuNotifyError("down")
+
+    monkeypatch.setattr(w, "_send", boom)
+    w._scan_and_alert()  # 发送失败
+    clean_db.expire_all()
+    wm_after = clean_db.get(AppConfig, WM_KEY).value
+    assert wm_after == wm_before  # 未推进,下轮补发