fix(alerts): enforce caliber ruling R5-supp (period linkage)

deviation_engine: link kpi_value_id on deviation/trend alert rows; skip when period not attributable or kpi_values row missing; dedupe key now (entity_id,kpi_id,kpi_value_id,alert_type)

deviation_push: same-source R1/R5 guards, no-period request writes nothing

tests: add test_alert_caliber_r5_residual_20260919 (13 cases incl. negative controls); adapt r1r4/r5 fixtures to the supplemental ruling

backend/app/api/deviation_push.py

@@ -8,9 +8,10 @@ from app.database import get_db
 from app.auth_middleware import require_auth, require_role
 from app.deps import get_entity_id
 from app.models import StrategicMap, KPIDefinition, KPIValue, KPIAlert, OperationLog, ActionPlan, BudgetPlan
+# 裁定 R5-补1（2026-09-19 财务口径）：差异反打路径同源接入 R1/R5 守卫
+from app.utils.alert_entity_scope import entity_alert_guard, guard_alert_period
 import json
 import logging
-from datetime import datetime
 
 logger = logging.getLogger("cma.deviation_push")
 
@@ -49,6 +50,22 @@ def push_deviation_to_map(
     if new_value is None:
         raise HTTPException(400, "缺少 newValue")
 
+    # R5-补1（财务裁定 2026-09-19）：「请求体无 period 一律不落库」
+    # 旧码在缺 period 时用当月冒充期间再落库（预警 + kpi_values 双写）——
+    # 该兜底正好绕过 R5「可归期」判据，故此处直接整请求零业务写入返回。
+    if not period:
+        logger.info(
+            "R5-补1 期间守卫：deviation_push 请求无 period，整请求零业务写入"
+            " entity_id=%s map_id=%s node_id=%s", entity_id, map_id, node_id)
+        return {
+            "success": False,
+            "message": "缺少 period：按 R5-补1（期间不可归期）未落库任何数据",
+            "alert_created": False,
+            "alert_skipped_reason": "no_period",
+            "zero_write": True,
+            "review_topic_created": False,
+        }
+
     # 1. 查找战略地图（2026-09-18 派生单: 地图须归属会话账套，NULL 归属按 e1 兼容）
     m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
     if not m or (m.entity_id or 1) != entity_id:
@@ -95,9 +112,8 @@ def push_deviation_to_map(
         raise HTTPException(404, f"KPI {kpi_code} 不存在")
 
     # 3. 更新实际值到 KPIValue 表
-    if not period:
-        period = datetime.now().strftime("%Y-%m")
-
+    #（R5-补1：period 已在上方强制校验，原「缺 period 用当月兜底」分支已删除——
+    #  该兜底会伪造期间，正是 R5「不可归期」的成因之一）
     existing_value = db.query(KPIValue).filter(
         KPIValue.kpi_id == kpi.id,
         KPIValue.period == period,
@@ -106,20 +122,22 @@ def push_deviation_to_map(
     if existing_value:
         existing_value.actual_value = new_value
         existing_value.source_type = "manual"
+        kv_row = existing_value
     else:
-        kv = KPIValue(
+        kv_row = KPIValue(
             kpi_id=kpi.id,
             period=period,
             actual_value=new_value,
             source_type="manual",
         )
-        db.add(kv)
+        db.add(kv_row)
 
     db.flush()
 
     # 4. 检查是否触发预警
     alert_created = False
     alert_id = None
+    alert_skipped_reason = None
     if kpi.target_value and kpi.target_value > 0:
         ratio = new_value / kpi.target_value
         if ratio < 0.7:
@@ -132,17 +150,28 @@ def push_deviation_to_map(
             alert_level = None
 
         if alert_level:
-            alert = KPIAlert(
-                entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属
-                kpi_id=kpi.id,
-                alert_level=alert_level,
-                alert_message=alert_msg,
-                status="pending",
-            )
-            db.add(alert)
-            db.flush()
-            alert_created = True
-            alert_id = alert.id
+            # R1/R5 双守卫（裁定 R5-补1）：demo 账套 / 非规整月期间 → 该 kpi_alerts 行不入库，
+            # 返回体登记跳过原因（KPI 值照写，与 bot_bridge 同口径：裁定只约束预警行）
+            if not entity_alert_guard(db, entity_id, source="deviation_push.push_deviation_to_map",
+                                      detail="kpi=%s" % kpi_code):
+                alert_skipped_reason = "R1_demo_entity_excluded"
+            elif not guard_alert_period(period, source="deviation_push.push_deviation_to_map",
+                                        detail="kpi=%s" % kpi_code):
+                alert_skipped_reason = "R5_non_attributable_period"
+
+            if alert_skipped_reason is None:
+                alert = KPIAlert(
+                    entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属
+                    kpi_id=kpi.id,
+                    kpi_value_id=kv_row.id,         # R5-补1: 期间关联（同 kpi 同期间行 → 可派生）
+                    alert_level=alert_level,
+                    alert_message=alert_msg,
+                    status="pending",
+                )
+                db.add(alert)
+                db.flush()
+                alert_created = True
+                alert_id = alert.id
 
     # 5. 生成战略回顾会议题
     review_topic_created = False
@@ -180,6 +209,7 @@ def push_deviation_to_map(
             "new_value": new_value,
             "period": period,
             "alert_created": alert_created,
+            "alert_skipped_reason": alert_skipped_reason,
             "review_topic_created": review_topic_created,
         }, ensure_ascii=False),
     )
@@ -194,6 +224,7 @@ def push_deviation_to_map(
         "new_value": new_value,
         "alert_created": alert_created,
         "alert_id": alert_id,
+        "alert_skipped_reason": alert_skipped_reason,
         "review_topic_created": review_topic_created,
     }
 

backend/app/utils/deviation_engine.py

@@ -17,6 +17,7 @@ from app.models import KPIDefinition, KPIValue, KPIAlert, BudgetPlan, SystemConf
 # 裁定 R1/R2/R3（2026-09-19 财务口径）：demo 账套排除 / 期间窗口 / 规整月期间
 from app.utils.alert_entity_scope import (
     excluded_entity_ids, kpi_excluded, period_in_window, window_lower_period,
+    guard_alert_period,  # R5-补1（2026-09-19）：期间可归期 + 关联 kpi_values
 )
 
 logger = logging.getLogger("cma.deviation")
@@ -80,13 +81,46 @@ def get_budget_for_kpi(db, kpi_id: int, period: str, version: str = None) -> Opt
     return plan.budget_value if plan else None
 
 
-def get_actual_for_kpi(db, kpi_id: int, period: str) -> Optional[float]:
-    """获取指定KPI在指定期间的实际值"""
-    val = db.query(KPIValue).filter(
+def get_kpi_value_row(db, kpi_id: int, period: str):
+    """取 (kpi_id, period) 的 KPIValue 行（取行口径唯一来源）。
+
+    与 get_actual_for_kpi 同源：预警行关联的 kpi_values 行 = 数值来源那一行。
+    """
+    return db.query(KPIValue).filter(
         KPIValue.kpi_id == kpi_id,
         KPIValue.period == period,
     ).order_by(KPIValue.calculated_at.desc()).first()
-    return val.actual_value if val else None
+
+
+def get_actual_for_kpi(db, kpi_id: int, period: str) -> Optional[float]:
+    """获取指定KPI在指定期间的实际值"""
+    row = get_kpi_value_row(db, kpi_id, period)
+    return row.actual_value if row else None
+
+
+def resolve_alert_kpi_value(db, kpi, period: str):
+    """R5-补1（财务裁定 2026-09-19）：解析差异/趋势预警行必须关联的 `kpi_values` 行。
+
+    裁定原文：「此类预警行必须在落库时写入 `kpi_value_id`，指向 **同 `entity_id` +
+    同 `kpi_id` + 文案所示规整月期间** 的 `kpi_values` 行。文案期间不可解析为 `YYYY-MM`，
+    或该 (kpi_id, period) 无 `kpi_values` 行 → 按 R5 `guard_alert_period` 跳过不入库
+    （不得以 `kpi_value_id=NULL` 落行——『文案里有期间』不构成可归期）。」
+
+    - 账套约束由 `kpi_id` 传递：KPIDefinition 归属单一账套，故 kpi_id 已唯一确定 entity_id；
+      取行口径与 `get_kpi_value_row`（= `get_actual_for_kpi` 的来源行）完全一致，
+      保证「预警关联的那一行」就是该预警数值的来源行，不会关联到别的账套/别的期间。
+    - 返回 `(kpi_value_row | None, skip_reason | None)`；跳过原因供调用点计数与日志。
+    """
+    if not guard_alert_period(period, source="deviation_engine.resolve_alert_kpi_value",
+                              detail="kpi=%s" % getattr(kpi, "kpi_code", "")):
+        return None, "R5_non_attributable_period"
+    row = get_kpi_value_row(db, kpi.id, period)
+    if row is None:
+        logger.info(
+            "R5-补1 期间关联：跳过不入库（该 (kpi, period) 无 kpi_values 行，期间不可派生）"
+            " kpi=%s period=%s", getattr(kpi, "kpi_code", ""), period)
+        return None, "no_kpi_value_row"
+    return row, None
 
 
 def calc_period_deviation(db, kpi_id: int, period: str) -> dict:
@@ -320,6 +354,10 @@ def run_deviation_check(db_session, period: str = None) -> int:
     2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope`）：
       · R1 —— `status='demo'` 的账套不参与遍历（demo 仅演示，不产生业务预警）；
       · R2/R3 —— 期间须为规整月 `YYYY-MM` 且落在窗口（当月往前 6 个自然月，含当月）内。
+      · R5-补1 —— 差异/趋势分支**必须落 `kpi_value_id` 关联**（同 kpi 同期间的 `kpi_values` 行）；
+        文案期间不可解析为 `YYYY-MM` 或该 (kpi, period) 无 `kpi_values` 行 → 跳过不入库；
+        趋势类以「计算基准期（趋势末端当期）」为关联期间；差异类去重键
+        = `(entity_id, kpi_id, kpi_value_id, alert_type)`。
     """
     if period is None:
         period = datetime.now().strftime("%Y-%m")
@@ -335,6 +373,8 @@ def run_deviation_check(db_session, period: str = None) -> int:
 
     new_count = 0
     skipped_demo = 0
+    skipped_period_linkage = 0
+    skipped_linkage_kpis = []
     for kpi in kpis:
         # R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历
         if kpi_excluded(db_session, kpi, excluded):
@@ -349,53 +389,76 @@ def run_deviation_check(db_session, period: str = None) -> int:
                 f"vs 预算{deviation['budget_value']}，"
                 f"差异率{deviation['deviation_rate']}%"
             )
+            alert_type = result["alert_type"] or "actual"
 
-            # 检查是否已有同KPI同期间的差异预警
-            existing = db_session.query(KPIAlert).filter(
-                KPIAlert.kpi_id == kpi.id,
-                KPIAlert.alert_message.contains("[差异预警]"),
-                KPIAlert.alert_message.contains(period),
-                KPIAlert.status.in_(["pending", "processing"]),
-            ).first()
+            # R5-补1（裁定 2026-09-19）：期间可归期 + 必须落 kpi_value_id 关联
+            #（文案里的期间不构成可归期；无 kpi_values 行 → 跳过不入库）
+            linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, period)
+            if linked_kv is None:
+                skipped_period_linkage += 1
+                skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, period))
+            else:
+                # 去重键（R5-补1 变更）：(entity_id, kpi_id, kpi_value_id, alert_type)
+                #（原按 `alert_message contains period` 判断，改关联后按可归期目标去重）
+                existing = db_session.query(KPIAlert).filter(
+                    KPIAlert.entity_id == (kpi.entity_id or 1),
+                    KPIAlert.kpi_id == kpi.id,
+                    KPIAlert.kpi_value_id == linked_kv.id,
+                    KPIAlert.alert_type == alert_type,
+                    KPIAlert.status.in_(["pending", "processing"]),
+                ).first()
 
-            if not existing:
-                alert = KPIAlert(
-                    entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属（原缺省=1 串账套）
-                    kpi_id=kpi.id,
-                    alert_level=result["kpi_alert_level"],
-                    alert_message=f"[差异预警] {alert_msg}",
-                    alert_type=result["alert_type"] or "actual",
-                    suggestion=result["suggestion"],
-                    status="pending",
-                )
-                db_session.add(alert)
-                new_count += 1
-                logger.info(f"  新增差异预警 [{result['kpi_alert_level']}] {kpi.kpi_name}: 差异率{deviation['deviation_rate']}%")
+                if not existing:
+                    alert = KPIAlert(
+                        entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属（原缺省=1 串账套）
+                        kpi_id=kpi.id,
+                        kpi_value_id=linked_kv.id,      # R5-补1: 期间经该行可派生（可归期）
+                        alert_level=result["kpi_alert_level"],
+                        alert_message=f"[差异预警] {alert_msg}",
+                        alert_type=alert_type,
+                        suggestion=result["suggestion"],
+                        status="pending",
+                    )
+                    db_session.add(alert)
+                    new_count += 1
+                    logger.info(f"  新增差异预警 [{result['kpi_alert_level']}] {kpi.kpi_name}: 差异率{deviation['deviation_rate']}%")
 
         # 2. 趋势异常检测（每期检查连续3期）
         trend = check_trend_anomaly(db_session, kpi.id, period, consecutive=3)
         if trend.get("anomaly") and trend.get("level") in ("yellow", "red"):
             trend_alert_msg = trend["message"]
-            existing_trend = db_session.query(KPIAlert).filter(
-                KPIAlert.kpi_id == kpi.id,
-                KPIAlert.alert_message.contains("[趋势预警]"),
-                KPIAlert.status.in_(["pending", "processing"]),
-            ).first()
+            # R5-补1：趋势类以「计算基准期（趋势末端当期）」为关联期间；该期无行 → 跳过
+            trend_periods = trend.get("periods") or []
+            base_period = trend_periods[-1] if trend_periods else period
+            linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, base_period)
+            if linked_kv is None:
+                skipped_period_linkage += 1
+                skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, base_period))
+            else:
+                existing_trend = db_session.query(KPIAlert).filter(
+                    KPIAlert.kpi_id == kpi.id,
+                    KPIAlert.alert_message.contains("[趋势预警]"),
+                    KPIAlert.status.in_(["pending", "processing"]),
+                ).first()
 
-            if not existing_trend:
-                alert = KPIAlert(
-                    entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属
-                    kpi_id=kpi.id,
-                    alert_level=trend["level"],
-                    alert_message=f"[趋势预警] {trend_alert_msg}",
-                    status="pending",
-                )
-                db_session.add(alert)
-                new_count += 1
-                logger.info(f"  新增趋势预警 [{trend['level']}] {trend_alert_msg}")
+                if not existing_trend:
+                    alert = KPIAlert(
+                        entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属
+                        kpi_id=kpi.id,
+                        kpi_value_id=linked_kv.id,      # R5-补1: 趋势末端当期关联
+                        alert_level=trend["level"],
+                        alert_message=f"[趋势预警] {trend_alert_msg}",
+                        status="pending",
+                    )
+                    db_session.add(alert)
+                    new_count += 1
+                    logger.info(f"  新增趋势预警 [{trend['level']}] {trend_alert_msg}")
 
     if skipped_demo:
         logger.info("  差异预警跳过 demo 账套 KPI %s 项（R1，裁定 2026-09-19）", skipped_demo)
+    if skipped_period_linkage:
+        logger.info("  差异/趋势预警跳过（R5-补1 期间不可归期或无 kpi_values 行）KPI %s 项: %s",
+                    skipped_period_linkage, skipped_linkage_kpis)
     db_session.commit()
     return new_count
 

backend/tests/test_alert_caliber_r1r4_20260919.py

@@ -158,10 +158,13 @@ class TestR1DemoEntityExcluded:
         import app.utils.deviation_engine as de
 
         _demo_and_active(db)
-        create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入",
-                        entity_id=6, unit="%", threshold_red=RED_EXPR)
-        create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入",
-                        entity_id=1, unit="%", threshold_red=RED_EXPR)
+        k_demo = create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入",
+                                 entity_id=6, unit="%", threshold_red=RED_EXPR)
+        k_real = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入",
+                                 entity_id=1, unit="%", threshold_red=RED_EXPR)
+        # R5-补1（2026-09-19）：差异预警须落 kpi_value_id 关联，故夹具补该期间的 kpi_values 行
+        #（否则按补1 跳过不入库，本用例将观察不到「遍历范围」这一被测口径）
+        _kpi_value(db, k_real, "2026-08", 14.45, entity_id=1)
         # 让所有 KPI 都「差异触发」，只观察遍历范围（不依赖预算数据形态）
         monkeypatch.setattr(de, "build_deviation_alert",
                             lambda db_s, kpi, period, *a, **kw: {

backend/tests/test_alert_caliber_r5_20260919.py

@@ -15,10 +15,15 @@
 类A（可落地守卫，本套件 RED→GREEN 对象）：期间已知、可能为伪期间的生成路径
   —— 规则引擎（`latest_value.period` 按字符串 desc 取，`2026-H1` 会排在 `2026-09` 之前）、
      Bot 写通道（`bot_bridge` / `bot_bridge_v2` 按显式 period 写）。
-类B（现状登记，见 `TestR5ClassBStatusQuo`）：设计上不携带月度期间的族（差异/趋势/集中度/
-     到期未收款/资金缺口预测），其期间**无从从库派生**（`kpi_value_id IS NULL`）；
-     裁定边界原文「若日后确有『无期间合法预警』的业务需求，由产品侧另提，本裁定不预设」
-     → 本单登记为待产品侧决策项，不做静默屏蔽（会灭掉既有族并触发既有套件失败）。
+类B（现状登记，见 `TestR5ClassBStatusQuo`）：设计上不携带月度期间的族（集中度/到期未收款/
+     资金缺口预测），其期间**无从从库派生**（`kpi_value_id IS NULL`）。
+
+2026-09-19 补充裁定 R5-补（财务，收口两个残口）后本套件的口径变化：
+- **补1**：差异/趋势/差异反打路径改为**必须落 `kpi_value_id`**（不可归期即跳过），
+  差异类去重键改 `(entity_id, kpi_id, kpi_value_id, alert_type)` → **差异/趋势已由类B 转入类A**，
+  其守卫与负对照用例落在 `tests/test_alert_caliber_r5_residual_20260919.py`；
+- **补2**：类B「无月度期间族」**不屏蔽（现状保留）**，登记为已知例外，不再作为待决策项挂账
+  → `TestR5ClassBStatusQuo` 由「待产品侧决策」的登记，改为「按补2 保留」的登记。
 """
 from datetime import datetime
 
@@ -217,31 +222,18 @@ class TestR5BotWritePaths:
 class TestR5ClassBStatusQuo:
     """类B 现状登记（非守卫断言）——1b 实测证据：这些生成路径产出的行**期间无从从库派生**。
 
-    裁定边界原文：「若日后确有『无期间合法预警』的业务需求，由产品侧另提，本裁定不预设」。
-    本单结论：按边界**不做静默屏蔽**（屏蔽会灭掉 423 行既有族 + 触发既有套件失败），
-    登记为待产品侧决策项。若产品侧裁定需屏蔽，本用例应随之更新。
+    裁定演进（2026-09-19 §补充裁定 R5-补2）：类B「无月度期间族」**不屏蔽（现状保留）**，
+    理由 = 驱动量是日期/占比/趋势，可行动性不以月度期间为前提（属 R5 边界条款所指
+    「无期间合法预警」业务需求）；屏蔽会一次性灭掉「到期未收款」等既有能力（存量 423 行族）
+    并触发既有套件失败 → **登记为已知例外，不再作为待决策项挂账**。
+
+    ⚠ 差异路径原属本类，**已由补1 转入类A**（必须落 kpi_value_id，不可归期即跳过）——
+    其守卫与负对照用例见 `tests/test_alert_caliber_r5_residual_20260919.py`，
+    本类不再保留其「无期间关联」的现状断言（该断言与补1 冲突，已随裁定删除）。
+    类B 余族的现状登记同样落在该新套件（`TestR5ResidualClassBStatusQuo`）。
     """
 
-    def test_deviation_path_row_has_no_derivable_period(self, db: Session, monkeypatch):
-        """差异预警路径（cron daily_sync 第 4 步）：期间为规整月，但行内无 kpi_value 关联
-        ⇒ 库内派生期间 = NULL（id 17 同类形态，可复现）"""
-        import app.utils.deviation_engine as de
-
-        _ent(db, 1, "active")
-        create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家")
-        monkeypatch.setattr(de, "build_deviation_alert",
-                            lambda db_s, kpi, period, *a, **kw: {
-                                "triggered": True, "kpi_alert_level": "red", "alert_type": "budget",
-                                "suggestion": None,
-                                "deviation": {"actual_value": 88.8, "budget_value": 10.0,
-                                              "deviation_rate": 788.0}})
-        monkeypatch.setattr(de, "check_trend_anomaly", lambda *a, **kw: {"anomaly": False})
-
-        n = de.run_deviation_check(db, period="2026-07")
-        db.expire_all()
-
-        rows = db.query(KPIAlert).all()
-        assert n == 1 and len(rows) == 1
-        assert rows[0].kpi_value_id is None, "差异预警不落 kpi_value 关联"
-        assert "[差异预警]" in rows[0].alert_message and "2026-07" in rows[0].alert_message, \
-            "期间存在于文案中，但不可由库派生（类B：待产品侧决定是否补关联）"
+    # 原 `test_deviation_path_row_has_no_derivable_period`（差异路径「无期间关联」现状断言）
+    # 已随 §补充裁定 R5-补1 删除：差异路径转为类A，必须落 kpi_value_id，不可归期即跳过。
+    # 替代用例（含负对照：旧逻辑取 NULL / 新逻辑取该期间行）见
+    # tests/test_alert_caliber_r5_residual_20260919.py::TestR5ResidualDeviationLinkage（6 例）。

backend/tests/test_alert_caliber_r5_residual_20260919.py

@@ -0,0 +1,317 @@
+# -*- coding: utf-8 -*-
+"""预警「期间可归期」残口收敛测试（2026-09-19，任务 fullstack-20260919-alert-period-linkage-r5-residual）
+
+依据：`wecom-finance/reports/CMA_预警口径裁定_20260919.md`
+
+- **§R5**（2026-09-19）：预警生效判据 = 期间为规整月 `YYYY-MM`；不可归期者不参与预警。
+- **§补充裁定 R5-补1**（同日，收口残口一）：差异/趋势/差异反打路径**必须落 `kpi_value_id`**
+  （指向同账套 + 同 KPI + 文案所示规整月期间的 `kpi_values` 行）；文案期间不可解析为 `YYYY-MM`
+  或该 (kpi_id, period) 无 `kpi_values` 行 → 跳过不入库；趋势类以**趋势末端当期**为关联期间；
+  差异类去重键改为 `(entity_id, kpi_id, kpi_value_id, alert_type)`；
+  `deviation_push` 同源接入 `guard_alert_period` + `entity_alert_guard`，请求体无 `period` 一律不落库。
+- **§补充裁定 R5-补2**（同日，收口残口二）：类B「无月度期间族」**不屏蔽（现状保留）**，
+  登记为已知例外，不与类A 混算。
+
+本套件每个「期间关联」用例均带**负对照**：夹具同时给出「旧逻辑会取到的目标（未关联 / 别期行）」
+与「新逻辑应取到的目标」，断言新逻辑取新目标、且**不等于**旧目标——否则判据未生效也能通过。
+"""
+from datetime import datetime
+
+import pytest
+from sqlalchemy.orm import Session
+
+from tests.conftest import create_test_kpi, create_test_user
+from app.models import (Entity, KPIValue, KPIAlert, KPIDefinition, StrategicMap,
+                        ActionPlan, OperationLog)
+
+NOW = datetime(2026, 9, 19, 10, 0)
+
+
+def _ent(db, eid, status, name=None):
+    db.add(Entity(id=eid, name=name or ("账套%d" % eid), short_name="e%d" % eid, status=status))
+    db.commit()
+
+
+def _kpi_value(db, kpi, period, value, entity_id=None):
+    row = KPIValue(entity_id=entity_id, kpi_id=kpi.id, period=period, actual_value=value)
+    db.add(row)
+    db.commit()
+    return row
+
+
+def _trigger_deviation(db, monkeypatch):
+    """隔离趋势分支：只观察差异分支的落库形态"""
+    import app.utils.deviation_engine as de
+
+    monkeypatch.setattr(de, "build_deviation_alert",
+                        lambda db_s, kpi, period, *a, **kw: {
+                            "triggered": True, "kpi_alert_level": "red", "alert_type": "budget",
+                            "suggestion": "探针建议",
+                            "deviation": {"actual_value": 88.8, "budget_value": 10.0,
+                                          "deviation_rate": 788.0}})
+    monkeypatch.setattr(de, "check_trend_anomaly", lambda *a, **kw: {"anomaly": False})
+    return de
+
+
+# ─────────────────── 残口一：差异/趋势路径期间关联（补1） ───────────────────
+
+class TestR5ResidualDeviationLinkage:
+    def test_diff_alert_links_current_period_row_not_other_period(self, db: Session, monkeypatch):
+        """差异预警必须关联「文案所示期间」那一行（负对照：不得关联别期行，也不得留空）
+
+        夹具两个 kpi_values 行：2026-06（旧逻辑在别处的可能目标 / 干扰项）与 2026-07（文案期间）。
+        """
+        de = _trigger_deviation(db, monkeypatch)
+        _ent(db, 1, "active")
+        k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家")
+        kv_prev = _kpi_value(db, k, "2026-06", 60.0)
+        kv_cur = _kpi_value(db, k, "2026-07", 88.8)
+
+        n = de.run_deviation_check(db, period="2026-07")
+        db.expire_all()
+
+        rows = db.query(KPIAlert).all()
+        assert n == 1 and len(rows) == 1
+        assert rows[0].kpi_value_id == kv_cur.id, "新逻辑取新目标（文案期间 2026-07 的 kpi_values 行）"
+        assert rows[0].kpi_value_id != kv_prev.id, "负对照：不得关联别期行"
+        assert rows[0].kpi_value_id is not None, "负对照：旧逻辑取旧目标（NULL，期间不可派生）"
+        # 派生期间可复算 = 可归期
+        derived = db.query(KPIValue).filter(KPIValue.id == rows[0].kpi_value_id).first()
+        assert derived.period == "2026-07"
+
+    def test_diff_alert_skipped_when_no_kpi_value_row(self, db: Session, monkeypatch):
+        """该 (kpi, period) 无 kpi_values 行 → 跳过不入库（不得以 NULL 关联落行）"""
+        de = _trigger_deviation(db, monkeypatch)
+        _ent(db, 1, "active")
+        create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家")
+
+        n = de.run_deviation_check(db, period="2026-07")
+        db.expire_all()
+
+        assert n == 0
+        assert db.query(KPIAlert).count() == 0
+
+    def test_trend_alert_links_trend_end_period(self, db: Session, monkeypatch):
+        """趋势类以「计算基准期（趋势末端当期）」为关联期间（负对照：不得关联趋势起点期）"""
+        import app.utils.deviation_engine as de
+
+        monkeypatch.setattr(de, "build_deviation_alert", lambda *a, **kw: {"triggered": False})
+        _ent(db, 1, "active")
+        k = create_test_kpi(db, kpi_code="SALES_TOTAL", kpi_name="销售额", entity_id=1, unit="元")
+        kv_start = _kpi_value(db, k, "2026-05", 30.0)
+        _kpi_value(db, k, "2026-06", 20.0)
+        kv_end = _kpi_value(db, k, "2026-07", 10.0)   # 连续 3 期下滑 → 趋势末端
+
+        n = de.run_deviation_check(db, period="2026-07")
+        db.expire_all()
+
+        rows = db.query(KPIAlert).all()
+        assert n == 1 and len(rows) == 1
+        assert "[趋势预警]" in rows[0].alert_message
+        assert rows[0].kpi_value_id == kv_end.id, "新逻辑：关联趋势末端当期"
+        assert rows[0].kpi_value_id != kv_start.id, "负对照：不得关联趋势起点期"
+        assert rows[0].kpi_value_id is not None, "负对照：旧逻辑取旧目标（NULL）"
+
+    def test_trend_alert_skipped_when_base_period_not_attributable(self, db: Session, monkeypatch):
+        """趋势末端期间不可归期（非 YYYY-MM）→ 跳过不入库（补1：文案期间不可解析即跳过）"""
+        import app.utils.deviation_engine as de
+
+        monkeypatch.setattr(de, "build_deviation_alert", lambda *a, **kw: {"triggered": False})
+        monkeypatch.setattr(de, "check_trend_anomaly",
+                            lambda *a, **kw: {"anomaly": True, "level": "yellow",
+                                              "message": "连续3期下滑", "periods": ["2026-H1"]})
+        _ent(db, 1, "active")
+        create_test_kpi(db, kpi_code="SALES_TOTAL", kpi_name="销售额", entity_id=1, unit="元")
+
+        n = de.run_deviation_check(db, period="2026-07")
+        db.expire_all()
+
+        assert n == 0 and db.query(KPIAlert).count() == 0
+
+    def test_diff_dedupe_key_is_linkage_tuple(self, db: Session, monkeypatch):
+        """去重键改为 (entity_id, kpi_id, kpi_value_id, alert_type)
+
+        负对照：预置一条「文案含同期间、但 kpi_value_id 为空」的历史行（= 旧去重键会命中的形态），
+        新逻辑按可归期目标去重 → 仍应补一条带关联的预警（而非被文案去重吃掉）。
+        """
+        de = _trigger_deviation(db, monkeypatch)
+        _ent(db, 1, "active")
+        k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家")
+        _kpi_value(db, k, "2026-07", 88.8)
+        # 旧形态遗留行：文案带期间、无期间关联
+        db.add(KPIAlert(entity_id=1, kpi_id=k.id, kpi_value_id=None, alert_level="red",
+                        alert_type="budget", status="pending",
+                        alert_message="[差异预警] 新客户数[2026-07] 差异预警: 实际88.8 vs 预算10.0，差异率788.0%"))
+        db.commit()
+
+        n = de.run_deviation_check(db, period="2026-07")
+        db.expire_all()
+
+        rows = db.query(KPIAlert).order_by(KPIAlert.id).all()
+        assert n == 1 and len(rows) == 2, "新去重键按可归期目标判定 → 补一条带关联的行"
+        linked = [r for r in rows if r.kpi_value_id is not None]
+        assert len(linked) == 1 and linked[0].alert_type == "budget"
+
+    def test_diff_dedupe_does_not_duplicate_same_target(self, db: Session, monkeypatch):
+        """正对照：同一可归期目标重复触发不重复入库（幂等）"""
+        de = _trigger_deviation(db, monkeypatch)
+        _ent(db, 1, "active")
+        k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家")
+        _kpi_value(db, k, "2026-07", 88.8)
+
+        assert de.run_deviation_check(db, period="2026-07") == 1
+        assert de.run_deviation_check(db, period="2026-07") == 0
+        assert db.query(KPIAlert).count() == 1
+
+
+# ─────────────────── 残口三：deviation_push 的 R1/R5 缺口（补1） ───────────────────
+
+class TestR5ResidualDeviationPush:
+    def _fixture(self, db, entity_id=1, code="TEST_001"):
+        k = create_test_kpi(db, kpi_code=code, kpi_name="探针KPI", entity_id=entity_id,
+                            unit="%", target_value=100.0)
+        db.add(StrategicMap(entity_id=entity_id, title="探针地图", status="published",
+                            dimensions=[{"key": "finance", "name": "财务维度",
+                                         "objectives": [{"name": "探针目标", "kpis": [code]}]}]))
+        db.commit()
+        m = db.query(StrategicMap).filter(StrategicMap.entity_id == entity_id).first()
+        return k, m
+
+    def _call(self, db, m, data, entity_id=1):
+        from app.api.deviation_push import push_deviation_to_map
+        from app.models import User
+
+        uname = "r5res%d" % entity_id
+        user = db.query(User).filter(User.username == uname).first()
+        if user is None:
+            user = create_test_user(db, username=uname, role="ceo")
+        return push_deviation_to_map(data=data, db=db, entity_id=entity_id, current_user=user)
+
+    def test_demo_entity_alert_skipped_value_still_written(self, db: Session):
+        """R1：demo 会话账套不产生业务预警（KPI 值照写，与 bot_bridge 同口径）"""
+        _ent(db, 6, "demo", "演示企业（虚构数据）")
+        k, m = self._fixture(db, entity_id=6, code="DEMO_PUSH")
+
+        r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0,
+                               "period": "2026-09", "createReviewTopic": False}, entity_id=6)
+        db.expire_all()
+
+        assert r["alert_created"] is False
+        assert r["alert_skipped_reason"] == "R1_demo_entity_excluded"
+        assert db.query(KPIAlert).count() == 0
+        assert db.query(KPIValue).filter(KPIValue.kpi_id == k.id).count() == 1, "值照写"
+
+    def test_pseudo_period_alert_skipped(self, db: Session):
+        """R5：伪期间请求不产生预警行（返回体登记跳过原因）"""
+        _ent(db, 1, "active")
+        k, m = self._fixture(db, code="PSEUDO_PUSH")
+
+        r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0,
+                               "period": "2026-H1", "createReviewTopic": False})
+        db.expire_all()
+
+        assert r["alert_created"] is False
+        assert r["alert_skipped_reason"] == "R5_non_attributable_period"
+        assert db.query(KPIAlert).count() == 0
+
+    def test_missing_period_zero_write(self, db: Session):
+        """补1：「请求体无 period 一律不落库」——整请求零业务写入"""
+        _ent(db, 1, "active")
+        self._fixture(db, code="NOPERIOD_PUSH")
+
+        r = self._call(db, {"id": 1}, {"mapId": 1, "nodeId": "finance-0", "newValue": 10.0})
+        db.expire_all()
+
+        assert r["alert_created"] is False and r["zero_write"] is True
+        assert r["alert_skipped_reason"] == "no_period"
+        assert db.query(KPIValue).count() == 0, "不得用当月兜底冒充期间写值"
+        assert db.query(KPIAlert).count() == 0
+        assert db.query(ActionPlan).count() == 0
+        assert db.query(OperationLog).count() == 0
+
+    def test_valid_request_alert_links_request_period_row(self, db: Session):
+        """正对照 + 期间关联：规整月 + active 账套 → 预警落库且关联本次请求期间的行"""
+        _ent(db, 1, "active")
+        k, m = self._fixture(db, code="VALID_PUSH")
+        kv_other = _kpi_value(db, k, "2026-08", 50.0)   # 干扰项：别期行
+
+        r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0,
+                               "period": "2026-09", "createReviewTopic": False})
+        db.expire_all()
+
+        assert r["alert_created"] is True and r["alert_skipped_reason"] is None
+        alert = db.query(KPIAlert).filter(KPIAlert.id == r["alert_id"]).first()
+        linked = db.query(KPIValue).filter(KPIValue.id == alert.kpi_value_id).first()
+        assert linked is not None and linked.period == "2026-09"
+        assert alert.kpi_value_id != kv_other.id, "负对照：不得关联别期行"
+
+    def test_repeat_push_reuses_value_row_linkage(self, db: Session):
+        """同一 (kpi, period) 二次反打 → kpi_values 行复用（值就地更新，不新增行）
+
+        现状登记：本端点**无预警去重**（第二次反打仍落一条 pending 预警，关联同一 `kpi_values` 行）——
+        裁定 R5-补1 未规定该端点的去重语义（只规定了差异/趋势类的去重键），故按现状登记，
+        见报告「观察项」；本用例只锁「期间关联一致 + kpi_values 不重复」这两条已定口径。
+        """
+        _ent(db, 1, "active")
+        k, m = self._fixture(db, code="REPEAT_PUSH")
+        d = {"mapId": 1, "nodeId": "finance-0", "newValue": 10.0,
+             "period": "2026-09", "createReviewTopic": False}
+
+        r1 = self._call(db, m, dict(d))
+        r2 = self._call(db, m, dict(d))
+        db.expire_all()
+
+        assert db.query(KPIValue).filter(KPIValue.kpi_id == k.id).count() == 1, "值行复用"
+        rows = db.query(KPIAlert).order_by(KPIAlert.id).all()
+        assert len(rows) == 2, "现状登记：该端点无预警去重"
+        assert rows[0].kpi_value_id == rows[1].kpi_value_id, "两次反打关联同一期间行"
+        assert r1["alert_id"] != r2["alert_id"]
+
+
+# ─────────────────── 残口二：类B 不屏蔽（补2，现状登记） ───────────────────
+
+class TestR5ResidualClassBStatusQuo:
+    """补2 裁定：类B「无月度期间族」**不屏蔽（现状保留）**，登记为已知例外。
+
+    本类为**现状登记**（非守卫断言）：这些族的驱动量是日期/占比/趋势，其可行动性不以月度期间
+    为前提；屏蔽会灭掉「到期未收款」等既有能力（存量 423 行族）并触发既有套件失败。
+    若产品侧日后改判屏蔽，本类应随之更新（与 `test_alert_caliber_r5_20260919.py` 同口径）。
+    """
+
+    def test_customer_concentration_site_still_writes_without_linkage(self, db: Session, monkeypatch):
+        """集中度（类B，alert_rules:940）落库仍**不经** R5 期间守卫 → 行内无 kpi_value 关联
+
+        夹具以 monkeypatch 替代占比计算，观察点 = 「落库是否被期间守卫拦截」。
+        """
+        from app.api import alert_rules as ar
+
+        ar.AlertRule.__table__.create(bind=db.get_bind(), checkfirst=True)
+        _ent(db, 1, "active")
+        k = create_test_kpi(db, kpi_code="CONC_KPI", kpi_name="客户集中度", entity_id=1, unit="%")
+        db.add(ar.AlertRule(entity_id=1, kpi_id=k.id, rule_type="customer_concentration", enabled=1,
+                            params={"threshold": 30}))
+        db.commit()
+        monkeypatch.setattr(ar, "_check_customer_concentration",
+                            lambda *a, **kw: ("red", "[客户集中度] 探针 占比 88%"))
+
+        r = ar.run_customer_concentration_check(db=db, entity_id=1)
+        db.expire_all()
+
+        rows = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all()
+        assert r["generated"] == 1 and len(rows) == 1, "类B 现状保留：集中度预警照常入库"
+        assert rows[0].kpi_value_id is None, "类B 行不携带月度期间（登记形态）"
+
+    def test_class_b_creation_sites_have_no_r5_period_guard(self):
+        """登记：类B 其余创建点（集中度/预测值/资金缺口/到期未收款）**未接入** R5 期间守卫
+
+        判据 = 源码级登记（这些文件在本单中零改动），与 `git diff --stat` 互为佐证。
+        """
+        import inspect
+        from app.api import alert_rules as ar
+        from app.utils import cash_forecast_engine as cfe
+
+        for fn in (ar.run_customer_concentration_check,
+                   ar.run_forecast_alert_check,
+                   cfe.check_cash_alerts):
+            src = inspect.getsource(fn)
+            assert "guard_alert_period" not in src, fn.__name__