diff --git a/app/services/compare_alert.py b/app/services/compare_alert.py
new file mode 100644
index 0000000..08db7e5
--- /dev/null
+++ b/app/services/compare_alert.py
@@ -0,0 +1,80 @@
+"""比价失败报警规则:一条记录 → 命中的 AlertHit(或 None)。
+
+纯函数、不碰 DB(阈值/关键词由调用方从 config 传入),便于单测与调阈值。判定顺序保证四类互斥:
+  failed → fail_reason 空(且 information 非业务)=T1 / 含未识别=T6 / 含超时词=T2 / 其余业务不报;
+  cancelled 且耗时或步数超阈值=T5(深度放弃);success/running 不报。
+口径依据见 docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md 第 3 节。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+ALERT_TYPE_LABELS: dict[str, str] = {
+    "T1": "系统技术失败",
+    "T6": "商品识别失败",
+    "T2": "启动/超时失败",
+    "T5": "深度放弃(cancelled)",
+}
+
+
+@dataclass(frozen=True)
+class AlertHit:
+    trace_id: str
+    alert_type: str
+    reason: str
+    app_version: str | None
+
+
+def _hit(rec: Any, alert_type: str, reason: str) -> AlertHit:
+    return AlertHit(
+        trace_id=rec.trace_id,
+        alert_type=alert_type,
+        reason=reason,
+        app_version=getattr(rec, "app_version", None),
+    )
+
+
+def classify_record(
+    rec: Any,
+    *,
+    cancelled_ms_threshold: int,
+    cancelled_step_threshold: int,
+    timeout_keywords: tuple[str, ...],
+    unrecognized_keywords: tuple[str, ...],
+    biz_exclude_keywords: tuple[str, ...],
+) -> AlertHit | None:
+    """判定单条记录是否触发报警。rec 需有 status/fail_reason/information/total_ms/step_count/
+    trace_id/app_version 属性(ComparisonRecord 或等价对象)。"""
+    status = rec.status
+    if status == "failed":
+        fail_reason = rec.fail_reason
+        if fail_reason is None:
+            info = (rec.information or "").strip()
+            if info and any(w in info for w in biz_exclude_keywords):
+                return None
+            return _hit(rec, "T1", f"技术失败·{info[:80] or '比价过程出错'}")
+        if any(w in fail_reason for w in unrecognized_keywords):
+            return _hit(rec, "T6", f"识别失败·{fail_reason[:80]}")
+        if any(w in fail_reason for w in timeout_keywords):
+            return _hit(rec, "T2", fail_reason[:80])
+        return None
+    if status == "cancelled":
+        ms = rec.total_ms
+        step = rec.step_count
+        deep = (ms is not None and ms > cancelled_ms_threshold) or (
+            step is not None and step > cancelled_step_threshold
+        )
+        if deep:
+            return _hit(
+                rec, "T5",
+                f"深度放弃·等待 {round((ms or 0) / 1000)}s / {step or 0} 步后退出",
+            )
+        return None
+    return None
+
+
+def classify_batch(records: list, **kwargs) -> list[AlertHit]:
+    """批量分类,过滤掉 None。"""
+    hits = [classify_record(r, **kwargs) for r in records]
+    return [h for h in hits if h is not None]
diff --git a/tests/test_compare_alert_rules.py b/tests/test_compare_alert_rules.py
new file mode 100644
index 0000000..bdd1af0
--- /dev/null
+++ b/tests/test_compare_alert_rules.py
@@ -0,0 +1,61 @@
+"""比价报警规则分类:记录 → AlertHit | None(纯函数,不碰 DB)。"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.services.compare_alert import classify_record
+
+KW = dict(
+    cancelled_ms_threshold=90000,
+    cancelled_step_threshold=30,
+    timeout_keywords=("超时", "启动", "加载"),
+    unrecognized_keywords=("未识别",),
+    biz_exclude_keywords=("未找到", "打烊", "起送", "门店", "店内", "不配送", "这些菜", "未入驻", "休息"),
+)
+
+
+def _rec(**kw):
+    base = dict(
+        status="failed", fail_reason=None, information=None,
+        total_ms=None, step_count=None, trace_id="t", app_version=None, business_type="food",
+    )
+    base.update(kw)
+    return SimpleNamespace(**base)
+
+
+@pytest.mark.parametrize(
+    ("rec", "expected_type"),
+    [
+        (_rec(status="failed", fail_reason=None, information="比价过程出错，请稍后重试"), "T1"),
+        (_rec(status="failed", fail_reason=None, information=None), "T1"),
+        (_rec(status="failed", fail_reason=None, information="美团外卖门店已打烊，无法比价"), None),
+        (_rec(status="failed", fail_reason="未识别到商品"), "T6"),
+        (_rec(status="failed", fail_reason="启动淘宝超时, 请稍后重试"), "T2"),
+        (_rec(status="failed", fail_reason="淘宝闪购店内未找到这些菜品"), None),
+        (_rec(status="cancelled", total_ms=98000, step_count=10), "T5"),
+        (_rec(status="cancelled", total_ms=20000, step_count=31), "T5"),
+        (_rec(status="cancelled", total_ms=20000, step_count=5), None),
+        (_rec(status="cancelled", total_ms=None, step_count=None), None),
+        (_rec(status="success"), None),
+        (_rec(status="running"), None),
+    ],
+)
+def test_classify_record_type(rec, expected_type):
+    hit = classify_record(rec, **KW)
+    assert (hit.alert_type if hit else None) == expected_type
+
+
+def test_t5_boundary_exclusive():
+    assert classify_record(_rec(status="cancelled", total_ms=90000, step_count=30), **KW) is None
+    assert classify_record(_rec(status="cancelled", total_ms=90001, step_count=30), **KW).alert_type == "T5"
+
+
+def test_reason_texts():
+    t1 = classify_record(_rec(status="failed", fail_reason=None, information="  比价过程出错  "), **KW)
+    assert t1.reason == "技术失败·比价过程出错"
+    t1_empty = classify_record(_rec(status="failed", fail_reason=None, information=None), **KW)
+    assert t1_empty.reason == "技术失败·比价过程出错"
+    t5 = classify_record(_rec(status="cancelled", total_ms=98000, step_count=26), **KW)
+    assert t5.reason == "深度放弃·等待 98s / 26 步后退出"