diff --git a/yunke-server/app/management/anomalies.py b/yunke-server/app/management/anomalies.py
new file mode 100644
index 0000000..77c965c
--- /dev/null
+++ b/yunke-server/app/management/anomalies.py
@@ -0,0 +1,234 @@
+"""管理域确定性异常规则(批次B Task B3,v1 §5.4 异常卡片)。
+
+4 条规则,输入=当前+个人基线 PeriodMetrics,输出异常卡片 dict 列表。
+全部确定性——不使用 LLM 推断,reason/title/suggested_action 均为固定
+中文模板填数,reason 只描述数据事实(百分比/样本数/未回串数)。
+
+规则表(触发比较均为严格 > / <):
+- reply_p90_worsened : cur.fr_p90_ms > base×1.5 且 cur.cust_threads≥
+                       soft_threads;>×2.5 升 critical
+- reply_rate_dropped : derive(reply_2h_rate) cur < base 同式×0.7 且
+                       cur.cust_threads≥soft_threads(率由 scoring.
+                       derive_metrics 派生,与评分同一口径)
+- activity_drop      : cur 日均 (msg_sent+msg_recv)/n_days < base 日均
+                       (…/28)×0.5 且 base.days_with_data≥min_days
+                       (基线太稀疏不可比;基线窗口恒 28 天=baseline_window
+                       语义,当前窗口天数由 n_days 传入)
+- workload_overload  : cur.active_friends(周期 distinct) > workload 维
+                       active_friends_band.high(与评分 band 同源阈值)
+
+无基线不判:baseline=None 直接返回 [](新员工无个人对比位即不出卡,
+含与基线无关的 overload——新入职负载爬坡期误报多,不判);某侧值
+None/不可算(无串样本的率、无 fact 的分位、0ms 退化基线)→ 对应规则
+不触发。
+
+纯函数(无 session/IO):B4 dashboard/employees 对全员逐个调用。
+阈值不硬编码:soft_threads/min_days 取 config.sample_guards,过载上限取
+config.dimensions.workload.metrics.active_friends_band.high;乘数
+(1.5/2.5/0.7/0.5)是规则语义本身,定义为模块常量而非可调参数。
+"""
+from __future__ import annotations
+
+from app.management.employee_metrics import PeriodMetrics
+from app.management.scoring import derive_metrics
+
+# 规则乘数(语义常量,非可调参数——调整即换规则版本)
+P90_WARN_MULTIPLIER = 1.5
+P90_CRITICAL_MULTIPLIER = 2.5
+RATE_DROP_MULTIPLIER = 0.7
+ACTIVITY_DROP_MULTIPLIER = 0.5
+BASELINE_WINDOW_DAYS = 28  # 个人基线窗口恒 28 天(employee_metrics.baseline_window)
+
+# ---- 固定文案模板(.format 填数,reason 只含数据事实) ----
+TITLE_P90 = "回复 P90 较个人基线上升 {pct}%"
+TITLE_RATE = "2h 回复率较个人基线下降 {pct}%"
+TITLE_ACTIVITY = "日均消息量较个人基线下降 {pct}%"
+TITLE_OVERLOAD = "活跃客户 {current} 人超过负载上限 {limit} 人"
+
+REASON_P90 = (
+    "近 {n_days} 天首响 P90 为 {cur}(基线前 28 天为 {base}),上升 {pct}%;"
+    "期内 {threads} 串客户消息中 {unreplied} 串未回"
+)
+REASON_RATE = (
+    "近 {n_days} 天 2h 回复率为 {cur:.0%}({cur_cnt}/{cur_thr} 串),"
+    "基线(前 28 天)为 {base:.0%}({base_cnt}/{base_thr} 串);"
+    "期内 {unreplied} 串未回"
+)
+REASON_ACTIVITY = (
+    "近 {n_days} 天日均消息 {cur:.1f} 条/天,基线(前 28 天)日均 {base:.1f} 条/天,"
+    "下降 {pct}%;当前活跃客户 {friends} 人"
+)
+REASON_OVERLOAD = (
+    "近 {n_days} 天活跃客户 {current} 人(周期 distinct 口径),"
+    "超过配置上限 {limit} 人"
+)
+
+ACTION_P90 = "查看超时未回客户明细,与员工确认近期负载与接待安排"
+ACTION_RATE = "查看未回串明细,排查回复流程与时段覆盖问题"
+ACTION_ACTIVITY = "与员工确认在岗状态与客户维护情况,并核对账号数据接入是否正常"
+ACTION_OVERLOAD = "评估客户分配与负载均衡,考虑分流部分客户或安排协助"
+
+
+def _fmt_ms(ms: float) -> str:
+    # 与 scoring._fmt_ms 同式(模块私有不跨文件引用)
+    minutes = ms / 60_000
+    if minutes >= 60 and minutes % 60 == 0:
+        return f"{minutes // 60:g}h"
+    return f"{minutes:g}min"
+
+
+def _pct_up(cur: float, base: float) -> int:
+    """较基线上升百分比(四舍五入取整;触发后调用,base>0 已保证)。"""
+    return int(round((cur - base) / base * 100))
+
+
+def _pct_down(cur: float, base: float) -> int:
+    """较基线下降百分比(触发后调用,base>0 已由触发条件保证)。"""
+    return int(round((base - cur) / base * 100))
+
+
+def detect_anomalies(
+    user_id: str,
+    current: PeriodMetrics,
+    baseline: PeriodMetrics | None,
+    *,
+    n_days: int,
+    config: dict,
+) -> list[dict]:
+    """单员工确定性异常检测(纯函数,卡片结构见 v1 §5.4/模块 docstring)。
+
+    输出按规则表序:p90 恶化 → 回复率骤降 → 活动骤降 → 过载
+    (排序/聚合由 B4 调用方按 severity/affected_count 处理)。
+    """
+    if baseline is None:
+        return []  # 无基线不判(新员工)
+    guards = config.get("sample_guards", {})
+    soft = guards.get("soft_threads", 10)
+    min_days = guards.get("min_days", 5)
+    band_high = (
+        config.get("dimensions", {})
+        .get("workload", {})
+        .get("metrics", {})
+        .get("active_friends_band", {})
+        .get("high")
+    )
+    cards: list[dict] = []
+    unreplied = current.cust_threads - current.threads_answered
+    threads_ok = current.cust_threads >= soft
+
+    # ---- 1. reply_p90_worsened:P90 较个人基线恶化 ×1.5/×2.5 ----
+    if (
+        threads_ok
+        and current.fr_p90_ms is not None
+        and baseline.fr_p90_ms is not None
+        and baseline.fr_p90_ms > 0  # 0ms 基线为退化数据,不可比
+    ):
+        base_p90 = baseline.fr_p90_ms
+        if current.fr_p90_ms > base_p90 * P90_CRITICAL_MULTIPLIER:
+            severity = "critical"
+        elif current.fr_p90_ms > base_p90 * P90_WARN_MULTIPLIER:
+            severity = "warning"
+        else:
+            severity = None
+        if severity is not None:
+            pct = _pct_up(current.fr_p90_ms, base_p90)
+            cards.append({
+                "rule_code": "reply_p90_worsened",
+                "severity": severity,
+                "employee_user_id": user_id,
+                "title": TITLE_P90.format(pct=pct),
+                "current_value": current.fr_p90_ms,
+                "baseline_value": base_p90,
+                "unit": "ms",
+                "sample_size": current.cust_threads,
+                "affected_count": unreplied,
+                "reason": REASON_P90.format(
+                    n_days=n_days, cur=_fmt_ms(current.fr_p90_ms),
+                    base=_fmt_ms(base_p90), pct=pct,
+                    threads=current.cust_threads, unreplied=unreplied,
+                ),
+                "drilldown": {"type": "metric", "query": {
+                    "metric_code": "first_reply_p90", "user_id": user_id}},
+                "suggested_action": ACTION_P90,
+            })
+
+    # ---- 2. reply_rate_dropped:2h 回复率较基线降 30%+ ----
+    if threads_ok:
+        cur_rate = derive_metrics(current)["reply_2h_rate"]
+        base_rate = derive_metrics(baseline)["reply_2h_rate"]
+        if (
+            cur_rate is not None and base_rate is not None
+            and cur_rate < base_rate * RATE_DROP_MULTIPLIER
+        ):
+            cards.append({
+                "rule_code": "reply_rate_dropped",
+                "severity": "warning",
+                "employee_user_id": user_id,
+                "title": TITLE_RATE.format(
+                    pct=_pct_down(cur_rate, base_rate)),
+                "current_value": cur_rate,
+                "baseline_value": base_rate,
+                "unit": "0-1",
+                "sample_size": current.cust_threads,
+                "affected_count": unreplied,
+                "reason": REASON_RATE.format(
+                    n_days=n_days, cur=cur_rate, cur_cnt=current.reply_2h_count,
+                    cur_thr=current.cust_threads, base=base_rate,
+                    base_cnt=baseline.reply_2h_count,
+                    base_thr=baseline.cust_threads, unreplied=unreplied,
+                ),
+                "drilldown": {"type": "metric", "query": {
+                    "metric_code": "unreplied", "user_id": user_id}},
+                "suggested_action": ACTION_RATE,
+            })
+
+    # ---- 3. activity_drop:日均消息量较基线腰斩 ----
+    if n_days > 0 and baseline.days_with_data >= min_days:
+        cur_daily = (current.msg_sent + current.msg_recv) / n_days
+        base_daily = (
+            baseline.msg_sent + baseline.msg_recv
+        ) / BASELINE_WINDOW_DAYS
+        if cur_daily < base_daily * ACTIVITY_DROP_MULTIPLIER:
+            cards.append({
+                "rule_code": "activity_drop",
+                "severity": "warning",
+                "employee_user_id": user_id,
+                "title": TITLE_ACTIVITY.format(
+                    pct=_pct_down(cur_daily, base_daily)),
+                "current_value": cur_daily,
+                "baseline_value": base_daily,
+                "unit": "条",
+                "sample_size": current.active_friends,
+                "affected_count": current.active_friends,
+                "reason": REASON_ACTIVITY.format(
+                    n_days=n_days, cur=cur_daily, base=base_daily,
+                    pct=_pct_down(cur_daily, base_daily),
+                    friends=current.active_friends,
+                ),
+                "drilldown": {"type": "employee",
+                              "query": {"user_id": user_id}},
+                "suggested_action": ACTION_ACTIVITY,
+            })
+
+    # ---- 4. workload_overload:活跃客户超 band 上限 ----
+    if band_high is not None and current.active_friends > band_high:
+        cards.append({
+            "rule_code": "workload_overload",
+            "severity": "warning",
+            "employee_user_id": user_id,
+            "title": TITLE_OVERLOAD.format(
+                current=current.active_friends, limit=band_high),
+            "current_value": current.active_friends,
+            "baseline_value": None,  # 与个人基线无关,阈值在 band 配置
+            "unit": "人",
+            "sample_size": current.active_friends,
+            "affected_count": current.active_friends,
+            "reason": REASON_OVERLOAD.format(
+                n_days=n_days, current=current.active_friends,
+                limit=band_high,
+            ),
+            "drilldown": {"type": "employee", "query": {"user_id": user_id}},
+            "suggested_action": ACTION_OVERLOAD,
+        })
+
+    return cards
diff --git a/yunke-server/app/management/scoring.py b/yunke-server/app/management/scoring.py
index 5b67c5a..f697df3 100644
--- a/yunke-server/app/management/scoring.py
+++ b/yunke-server/app/management/scoring.py
@@ -61,8 +61,11 @@ def normalize(spec: dict, value: float | None) -> float | None:
     raise ValueError(f"未知指标类型 {mtype!r}(须 target/reverse/band)")
 
 
-def _derive_metrics(pm: PeriodMetrics) -> dict[str, float | None]:
-    """从 PeriodMetrics 派生可评分指标值(键=指标名,None=无值≠0)。"""
+def derive_metrics(pm: PeriodMetrics) -> dict[str, float | None]:
+    """从 PeriodMetrics 派生可评分指标值(键=指标名,None=无值≠0)。
+
+    公有供 anomalies(B3)/dashboard(B4) 复用——评分与异常对比同一派生口径。
+    """
     out: dict[str, float | None] = {
         "fr_p90_ms": pm.fr_p90_ms,
         "active_friends": pm.active_friends,   # band 用,周期 distinct
@@ -121,9 +124,9 @@ def score_employee(
     team 为指标级中位数(键=派生指标名,可 None)。返回结构见模块 docstring
     ——metrics_detail 键=配置里的 metric_code(band 型如 active_friends_band)。
     """
-    cur_vals = _derive_metrics(current)
-    prev_vals = _derive_metrics(prev) if prev is not None else None
-    base_vals = _derive_metrics(baseline) if baseline is not None else None
+    cur_vals = derive_metrics(current)
+    prev_vals = derive_metrics(prev) if prev is not None else None
+    base_vals = derive_metrics(baseline) if baseline is not None else None
     guards = config.get("sample_guards", {})
     soft = guards.get("soft_threads", 10)
     minimum = guards.get("min_threads", 30)
diff --git a/yunke-server/tests/test_management_anomalies.py b/yunke-server/tests/test_management_anomalies.py
new file mode 100644
index 0000000..f52d480
--- /dev/null
+++ b/yunke-server/tests/test_management_anomalies.py
@@ -0,0 +1,229 @@
+"""管理域确定性异常规则判别测试(批次B Task B3,v1 §5.4 异常卡片)。"""
+import json
+
+import pytest
+
+from app.management.anomalies import detect_anomalies
+from app.management.employee_metrics import PeriodMetrics
+from app.management.metric_definitions import DEFAULT_SCORE_CONFIG
+
+
+def _pm(**overrides):
+    """全零/None 默认的 PeriodMetrics 构造器(22 字段,覆盖式改)。
+
+    与 test_management_scoring._pm 同式(测试间不共享 helper,复制保持独立)。
+    """
+    base = {
+        "user_id": "u1", "days_with_data": 0, "msg_sent": 0, "msg_recv": 0,
+        "cust_threads": 0, "threads_answered": 0, "fr_sum_ms": 0,
+        "fr_p50_ms": None, "fr_p90_ms": None,
+        "reply_2h_count": 0, "reply_24h_count": 0,
+        "active_friends": 0, "active_rooms_avg": 0.0,
+        "new_customer": 0, "cooling_customer": 0, "dormant_customer": 0,
+        "end_active_friends": 0, "reactivated_customer": 0,
+        "call_count": 0, "call_seconds": 0, "answered_call": 0,
+        "call_followup": 0,
+    }
+    base.update(overrides)
+    return PeriodMetrics(**base)
+
+
+def _detect(cur, base, *, n_days=7, user_id="u1", config=None):
+    return detect_anomalies(
+        user_id, cur, base, n_days=n_days,
+        config=config if config is not None else DEFAULT_SCORE_CONFIG)
+
+
+# ---- 1. reply_p90_worsened ----
+
+def test_p90_worsened_warning_with_pct():
+    """×1.6 → warning;标题百分比=(160-100)/100=60%;affected=未回串数。"""
+    cur = _pm(cust_threads=20, threads_answered=12, fr_p90_ms=9_600_000)   # 160min
+    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)  # 100min
+    cards = _detect(cur, base)
+    assert [c["rule_code"] for c in cards] == ["reply_p90_worsened"]
+    c = cards[0]
+    assert c["severity"] == "warning"
+    assert c["title"] == "回复 P90 较个人基线上升 60%"
+    assert c["current_value"] == 9_600_000
+    assert c["baseline_value"] == 6_000_000
+    assert c["unit"] == "ms"
+    assert c["sample_size"] == 20
+    assert c["affected_count"] == 8  # 20 串 - 12 已回 = 期内未回串数
+    assert c["drilldown"] == {"type": "metric",
+                              "query": {"metric_code": "first_reply_p90",
+                                        "user_id": "u1"}}
+
+
+def test_p90_worsened_critical_at_3x():
+    cur = _pm(cust_threads=20, threads_answered=20, fr_p90_ms=18_000_000)
+    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)
+    cards = _detect(cur, base)
+    assert len(cards) == 1
+    assert cards[0]["severity"] == "critical"
+    assert "上升 200%" in cards[0]["title"]
+
+
+def test_p90_boundaries_strict_compare():
+    """×1.5 恰好 → 严格大于不成立不触发;×2.5 恰好 → warning 非 critical;
+    threads 恰好 = soft(10) → ≥soft 触发。"""
+    base = _pm(cust_threads=40, fr_p90_ms=6_000_000)
+    assert _detect(_pm(cust_threads=20, fr_p90_ms=9_000_000), base) == []
+    cards = _detect(_pm(cust_threads=20, fr_p90_ms=15_000_000), base)
+    assert [c["severity"] for c in cards] == ["warning"]
+    cards = _detect(_pm(cust_threads=10, fr_p90_ms=9_600_000), base)
+    assert [c["rule_code"] for c in cards] == ["reply_p90_worsened"]
+
+
+def test_p90_below_soft_threads_not_fired():
+    cur = _pm(cust_threads=5, fr_p90_ms=9_600_000)  # <soft 10
+    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)
+    assert _detect(cur, base) == []
+
+
+# ---- 2. 基线缺失 ----
+
+def test_baseline_none_all_rules_silent():
+    """新员工无个人基线:即便 p90 恶化/活动骤降/过载形态俱在也不出卡。"""
+    cur = _pm(cust_threads=20, threads_answered=5, fr_p90_ms=30_000_000,
+              active_friends=85, msg_sent=10, msg_recv=10)
+    assert _detect(cur, None) == []
+
+
+# ---- 3. reply_rate_dropped ----
+
+def test_reply_rate_dropped_warning():
+    """0.8→0.5(0.5 < 0.7×0.8=0.56)→ warning;百分比 37.5%→38%。"""
+    cur = _pm(cust_threads=20, reply_2h_count=10, threads_answered=15)   # 0.5
+    base = _pm(cust_threads=40, reply_2h_count=32, threads_answered=40)  # 0.8
+    cards = _detect(cur, base)
+    assert [c["rule_code"] for c in cards] == ["reply_rate_dropped"]
+    c = cards[0]
+    assert c["severity"] == "warning"
+    assert c["title"] == "2h 回复率较个人基线下降 38%"
+    assert c["current_value"] == pytest.approx(0.5)
+    assert c["baseline_value"] == pytest.approx(0.8)
+    assert c["unit"] == "0-1"
+    assert c["sample_size"] == 20
+    assert c["affected_count"] == 5
+    assert c["drilldown"] == {"type": "metric",
+                              "query": {"metric_code": "unreplied",
+                                        "user_id": "u1"}}
+
+
+def test_reply_rate_small_drop_not_fired():
+    """0.8→0.6(0.6 ≥ 0.56)→ 不触发。"""
+    cur = _pm(cust_threads=20, reply_2h_count=12)   # 0.6
+    base = _pm(cust_threads=40, reply_2h_count=32)  # 0.8
+    assert _detect(cur, base) == []
+
+
+def test_rate_baseline_no_threads_not_fired():
+    """基线无串样本 → 基线率 None → 不判(基线值 None 不触发)。"""
+    cur = _pm(cust_threads=20, reply_2h_count=0)
+    base = _pm(cust_threads=0)  # derive → reply_2h_rate None
+    assert _detect(cur, base) == []
+
+
+# ---- 4. activity_drop ----
+
+def test_activity_drop_triggered():
+    """当前 7 天日均 10 < 基线 28 天日均 30×0.5=15 → warning。"""
+    cur = _pm(msg_sent=40, msg_recv=30, active_friends=25)      # 70/7=10 条/天
+    base = _pm(msg_sent=600, msg_recv=240, days_with_data=20)   # 840/28=30 条/天
+    cards = _detect(cur, base)
+    assert [c["rule_code"] for c in cards] == ["activity_drop"]
+    c = cards[0]
+    assert c["severity"] == "warning"
+    assert c["title"] == "日均消息量较个人基线下降 67%"  # 66.7→67
+    assert c["current_value"] == pytest.approx(10.0)
+    assert c["baseline_value"] == pytest.approx(30.0)
+    assert c["unit"] == "条"
+    assert c["sample_size"] == 25
+    assert c["affected_count"] == 25
+    assert c["drilldown"] == {"type": "employee", "query": {"user_id": "u1"}}
+
+
+def test_activity_drop_sparse_baseline_not_fired():
+    """基线 days_with_data=3 < min_days(5) → 基线不可比,不触发。"""
+    base = _pm(msg_sent=600, msg_recv=240, days_with_data=3)
+    cur = _pm(msg_sent=40, msg_recv=30)
+    assert _detect(cur, base) == []
+
+
+# ---- 5. workload_overload ----
+
+def test_workload_overload_triggered():
+    cards = _detect(_pm(active_friends=85), _pm())
+    assert [c["rule_code"] for c in cards] == ["workload_overload"]
+    c = cards[0]
+    assert c["severity"] == "warning"
+    assert c["title"] == "活跃客户 85 人超过负载上限 80 人"
+    assert c["current_value"] == 85
+    assert c["baseline_value"] is None  # 规则与个人基线无关,阈值在 band 配置
+    assert c["unit"] == "人"
+    assert c["sample_size"] == 85
+    assert c["affected_count"] == 85
+    assert c["drilldown"] == {"type": "employee", "query": {"user_id": "u1"}}
+
+
+def test_workload_overload_at_limit_not_fired():
+    assert _detect(_pm(active_friends=80), _pm()) == []  # 严格大于上限
+
+
+# ---- 6. 配置驱动与组合 ----
+
+def test_thresholds_read_from_config():
+    """soft_threads/band high 改配置即改判——阈值不硬编码。"""
+    cfg = json.loads(json.dumps(DEFAULT_SCORE_CONFIG))
+    cfg["sample_guards"]["soft_threads"] = 25
+    cfg["dimensions"]["workload"]["metrics"]["active_friends_band"]["high"] = 90
+    cur = _pm(cust_threads=20, fr_p90_ms=9_600_000, active_friends=85)
+    base = _pm(cust_threads=40, fr_p90_ms=6_000_000)
+    # threads 20 < soft25 → p90/rate 不评;85 ≤ 90 → 不 overload
+    assert _detect(cur, base, config=cfg) == []
+
+
+def test_multiple_rules_stack_in_order():
+    """p90 ×5 critical + activity 骤降 + overload 90>80 同时触发,按规则表序输出;
+    两侧回复率同为 0(0<0 不成立)→ rate 规则不触发。"""
+    cur = _pm(cust_threads=30, threads_answered=10, fr_p90_ms=30_000_000,
+              active_friends=90, msg_sent=10, msg_recv=10)
+    base = _pm(cust_threads=60, threads_answered=60, fr_p90_ms=6_000_000,
+               msg_sent=600, msg_recv=240, days_with_data=20)
+    cards = _detect(cur, base)
+    assert [c["rule_code"] for c in cards] == [
+        "reply_p90_worsened", "activity_drop", "workload_overload"]
+    assert cards[0]["severity"] == "critical"
+
+
+def test_card_structure_all_fields():
+    """单卡 12 字段齐全且类型正确(v1 §5.4)。"""
+    cur = _pm(cust_threads=20, reply_2h_count=20, threads_answered=12,
+              fr_p90_ms=9_600_000, active_friends=30,
+              msg_sent=80, msg_recv=30)  # 日均 110/7≈15.7 ≥ 15 → activity 不触发
+    base = _pm(cust_threads=40, reply_2h_count=40, threads_answered=40,
+               fr_p90_ms=6_000_000, active_friends=30,
+               msg_sent=600, msg_recv=240, days_with_data=20)
+    cards = _detect(cur, base, user_id="u9")
+    assert len(cards) == 1
+    c = cards[0]
+    assert set(c) == {
+        "rule_code", "severity", "employee_user_id", "title",
+        "current_value", "baseline_value", "unit", "sample_size",
+        "affected_count", "reason", "drilldown", "suggested_action",
+    }
+    assert isinstance(c["rule_code"], str)
+    assert c["severity"] in ("warning", "critical")
+    assert c["employee_user_id"] == "u9"
+    assert isinstance(c["title"], str) and c["title"]
+    assert isinstance(c["current_value"], (int, float))
+    assert c["baseline_value"] is None or isinstance(
+        c["baseline_value"], (int, float))
+    assert isinstance(c["unit"], str)
+    assert isinstance(c["sample_size"], int)
+    assert isinstance(c["affected_count"], int)
+    assert isinstance(c["reason"], str) and c["reason"]
+    assert set(c["drilldown"]) == {"type", "query"}
+    assert isinstance(c["drilldown"]["query"], dict)
+    assert isinstance(c["suggested_action"], str) and c["suggested_action"]