fix(alerts): cash_plan 到期未收款预警引擎修复 — entity归属/KPI实体感知/余额快照批次/去重刷新/端点鉴权 (dama-cashplan-alert-engine-fix-20260913)

F1 两处 KPIAlert 显式写 entity_id（列默认 1 → 博海预警错挂酣客）；顺带修 deviation_engine/
   alert_generator/deviation_push/bot_bridge/bot_bridge_v2 同表创建点
F2 KPI 选择实体感知：候选码 ∪ 实体前缀后缀码（_AR_BALANCE/_CASH_BALANCE/_AP_BALANCE）
   + 兜底 ORDER BY id（此前命中 uk_entity_kpi 必中 BH_ADM_EXPENSE 437 管理费用）
F3 去重命中改为刷新 alert_message/等级/suggestion.days_late/归属（此前 continue → 文本冻结首版）
F4 余额快照批次（bohai_*_20260630 / receivables_migrate）开关：默认 skip 不逐户，另备
   aggregate(按批次聚合1条) / per_customer(旧行为)
F5 POST /api/cma/admin/alerts/check 补鉴权（无 token 401）+ 可选 entity_id 只跑指定账套

backend/app/api/bot_bridge.py

@@ -713,7 +713,7 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic
                         KPIAlert.kpi_id == kpi_id, KPIAlert.kpi_value_id == kv.id,
                         KPIAlert.alert_level == level, KPIAlert.status == "pending").first()
                     if not dup:
-                        db.add(KPIAlert(kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level,
+                        db.add(KPIAlert(entity_id=entity_id, kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level,
                                         alert_message=msg, status="pending", alert_type="bot"))
                         alerts.append({"rule_id": rule.id, "rule_type": rule.rule_type,
                                        "level": level, "message": msg})

backend/app/api/bot_bridge_v2.py

@@ -380,6 +380,7 @@ def receive_mpm_result(
             # 创建预警
             alert_msg = f"⚠️ {kpi_name}异常: {field_value:.1f}% (阈值: {threshold}%) [entity={entity.short_name or entity.name}, period={period}]"
             alert = KPIAlert(
+                entity_id=entity_id,   # 2026-09-13 F1: 显式写归属
                 kpi_id=kpi_id or 1,  # fallback to first KPI if not found
                 alert_level="red",
                 alert_message=alert_msg,

backend/app/api/deviation_push.py

@@ -131,6 +131,7 @@ def push_deviation_to_map(
 
         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,

backend/app/main.py

@@ -1,10 +1,12 @@
 """管理会计OS — 主入口"""
 import logging
-from fastapi import FastAPI, Request, Depends
+from typing import Optional
+from fastapi import FastAPI, Request, Depends, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from dotenv import load_dotenv
-from app.database import init_db
+from app.database import init_db, get_db
 from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json
 from app.utils.cache import clear_all as clear_cache, delete as delete_cache
 from scripts.erp_sync import run_sync as run_erp_sync
@@ -114,6 +116,22 @@ _ERP_CHANNEL_DISABLED_REASON = ("ERP直连同步通道已于2026-08-31停用（e
                                 "财务原始数据真实通道为 zxt 智享通注入（source_type='zxt-sync'），本端点不再可用。")
 
 
+# ── admin 端点鉴权（2026-09-13）──────────────────────────────
+# 未带 token → 显式 401（缺少认证），校验逻辑仍完全委托 require_auth（无效 token → 401）。
+# 说明：FastAPI 的 HTTPBearer(auto_error=True) 对"缺 header"返回 403；本端点验收要求 401，
+# 故用 auto_error=False 的实例自行判空，不改动全局 require_auth（避免其他端点行为漂移）。
+_optional_bearer = HTTPBearer(auto_error=False)
+
+
+def require_auth_401(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_optional_bearer),
+    db=Depends(get_db),
+):
+    if credentials is None or not credentials.credentials:
+        raise HTTPException(status_code=401, detail="未认证：缺少 Bearer token")
+    return require_auth(credentials=credentials, db=db)
+
+
 @app.post("/api/cma/admin/erp-sync")
 def admin_erp_sync(user=Depends(require_auth), kpi_codes: str = None):
     """[已停用] 手动触发ERP数据同步 — ERP 直连通道 2026-08-31 停用，返回 410"""
@@ -127,8 +145,12 @@ def admin_erp_sync_dry_run(user=Depends(require_auth), kpi_codes: str = None):
 
 
 @app.post("/api/cma/admin/alerts/check")
-def admin_check_alerts():
+def admin_check_alerts(entity_id: Optional[int] = None, user=Depends(require_auth_401)):
-    """手动触发预警检查（含资金管理预警：缺口前3天 + 到期未收款）"""
+    """手动触发预警检查（含资金管理预警：缺口前3天 + 到期未收款）
+    - 2026-09-13 安全修复：补鉴权（与同文件 /admin/erp-sync 等 admin 端点一致）；
+      未带 token 显式返回 401（require_auth_401，语义修正：401=未认证 / 403=无权）
+    - 新增可选 entity_id：只跑指定实体（不传=全部 active 实体，兼容原行为）
+    """
     from app.database import get_session_local
     from scripts.alert_generator import generate_and_push
     from app.utils.cash_forecast_engine import check_cash_alerts
@@ -136,9 +158,17 @@ def admin_check_alerts():
     db = get_session_local()()
     try:
         result = generate_and_push(db)
-        # 资金管理预警 — 对每个激活企业执行
+        # 资金管理预警 — 指定实体或全部激活企业
         cash_result = None
         try:
+            if entity_id is not None:
+                target = db.query(Entity).filter(
+                    Entity.id == entity_id, Entity.status == "active").first()
+                if not target:
+                    return JSONResponse(status_code=404,
+                                        content={"detail": f"企业不存在或未激活: entity_id={entity_id}"})
+                entities = [target]
+            else:
                 entities = db.query(Entity).filter(Entity.status == "active").all()
             cash_list = []
             for e in entities:

backend/app/utils/cash_forecast_engine.py

@@ -1,5 +1,6 @@
 """现金流预测引擎 — 根据历史KPI数据推算未来30天现金流"""
 import logging
+import os
 from datetime import datetime, timedelta
 from typing import Optional, TYPE_CHECKING
 from sqlalchemy.orm import Session
@@ -31,20 +32,92 @@ KPI_CODE_CANDIDATES = {
     "cash_balance": ["CASH_001", "CASH_BALANCE", "F_CASH"],
 }
 
+# 实体前缀码后缀模式（F2 实体感知 2026-09-13）——标准码全部未命中时，
+# 按后缀匹配「该实体自己的编码」（博海口径 BH_AR_BALANCE/BH_CASH_BALANCE/BH_AP_BALANCE）。
+# 修复前：博海一条候选都不命中 → 走无 ORDER BY 的兜底 first() →
+# 命中 uk_entity_kpi 索引时取字典序最小的 BH_ADM_EXPENSE(437 管理费用)，即"应收预警挂管理费用"。
+KPI_CODE_SUFFIXES = {
+    "operating_cash_flow": [],            # 无实体前缀变体
+    "receivables": ["_AR_BALANCE"],       # 博海 BH_AR_BALANCE(433)
+    "payables": ["_AP_BALANCE"],          # 博海 BH_AP_BALANCE(441)
+    "cash_balance": ["_CASH_BALANCE"],    # 博海 BH_CASH_BALANCE(434)
+}
 
-def find_kpi(db: Session, entity_id: int, codes: list) -> Optional["KPIDefinition"]:
+# ── 余额快照批次（F4，2026-09-13）────────────────────────────
-    """按候选编码列表查找KPI，返回第一个命中的"""
+# 这些 source 是「期末余额导入」批次：plan_date 是**导入基准日**（如 2026-06-30 期末），
+# 不是真实到期日 → 逐户生成"到期未收款"会永远 overdue，属伪逾期（DAMA 审计 fin-alert-flood-e2）。
+BALANCE_SNAPSHOT_SOURCES = {
+    "bohai_ar_20260630",
+    "bohai_ap_20260630",
+    "bohai_prepay_20260630",
+    "receivables_migrate",
+}
+
+# 处理开关：per_customer(逐户，旧行为) / skip(不生成逐户，默认) / aggregate(按批次聚合1条)
+_SNAPSHOT_MODE_DEFAULT = "skip"
+_SNAPSHOT_MODES = ("per_customer", "skip", "aggregate")
+
+
+def get_balance_snapshot_mode() -> str:
+    """余额快照批次处理模式（环境变量 CMA_CASH_SNAPSHOT_MODE 可覆盖，非法值回落默认）
+    - per_customer: 旧行为，逐户生成到期未收款预警
+    - skip        : 默认。批次行不生成逐户预警（口径待任总拍板）
+    - aggregate   : 按批次聚合 1 条预警（如"应收逾期210户 合计198.27万"）挂应收 KPI
+    """
+    mode = (os.environ.get("CMA_CASH_SNAPSHOT_MODE") or "").strip().lower()
+    return mode if mode in _SNAPSHOT_MODES else _SNAPSHOT_MODE_DEFAULT
+
+
+def get_balance_snapshot_sources() -> set:
+    """余额快照批次 source 集合（环境变量 CMA_CASH_SNAPSHOT_SOURCES 逗号分隔可追加/覆盖）"""
+    env = (os.environ.get("CMA_CASH_SNAPSHOT_SOURCES") or "").strip()
+    if env:
+        return {s.strip() for s in env.split(",") if s.strip()}
+    return set(BALANCE_SNAPSHOT_SOURCES)
+
+
+def find_kpi(db: Session, entity_id: int, codes: list, suffixes: Optional[list] = None) -> Optional["KPIDefinition"]:
+    """按候选编码列表查找KPI，返回第一个命中的（精确码优先 → 实体前缀后缀码 → None）
+
+    F2(2026-09-13): 全部查询显式 ORDER BY id，保证确定性（不再依赖 MySQL 执行计划）。
+    """
     from app.models import KPIDefinition
     for code in codes:
         kpi = db.query(KPIDefinition).filter(
             KPIDefinition.kpi_code == code,
             KPIDefinition.entity_id == entity_id,
-        ).first()
+        ).order_by(KPIDefinition.id.asc()).first()
         if kpi:
             return kpi
+    if suffixes:
+        rows = db.query(KPIDefinition).filter(
+            KPIDefinition.entity_id == entity_id,
+        ).order_by(KPIDefinition.id.asc()).all()
+        for suffix in suffixes:
+            for row in rows:
+                if (row.kpi_code or "").upper().endswith(suffix):
+                    return row
     return None
 
 
+def find_fallback_kpi(db: Session, entity_id: int) -> Optional["KPIDefinition"]:
+    """兜底 KPI（F2）：优先 finance 维度，其次任意；两种都带 ORDER BY id（确定性）
+
+    修复前：`db.query(KPIDefinition).filter(entity_id==entity_id).first()` 无排序 →
+    同一条件两种答案（idx_entity→318 / uk_entity_kpi→437）。
+    """
+    from app.models import KPIDefinition
+    kpi = db.query(KPIDefinition).filter(
+        KPIDefinition.entity_id == entity_id,
+        KPIDefinition.dimension == "finance",
+    ).order_by(KPIDefinition.id.asc()).first()
+    if kpi:
+        return kpi
+    return db.query(KPIDefinition).filter(
+        KPIDefinition.entity_id == entity_id,
+    ).order_by(KPIDefinition.id.asc()).first()
+
+
 def get_entity_kpi_history(entity_id: int, kpi_code: str, db: Session, limit_months: int = 6) -> list:
     """获取实体某个KPI的历史值"""
     from app.models import KPIDefinition, KPIValue
@@ -623,25 +696,65 @@ def forecast_cash_flow_with_plans(
 
 
 def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
-    """资金预警 — 缺口前3天预警 + 到期未收款提醒，写入预警中心(kpi_alerts)"""
+    """资金预警 — 缺口前3天预警 + 到期未收款提醒，写入预警中心(kpi_alerts)
+
+    2026-09-13 修复（任务 dama-cashplan-alert-engine-fix-20260913 / DAMA 审计 fin-alert-flood-e2）：
+      F1 两处 KPIAlert 显式写 entity_id（此前不传 → 列默认值 1 → 博海预警错挂酣客工作台）
+      F2 KPI 选择实体感知（候选码 + 实体前缀后缀码）＋ 兜底显式 ORDER BY id
+         （此前博海走无排序兜底 → 命中 uk_entity_kpi 索引必中 BH_ADM_EXPENSE(437 管理费用)）
+      F3 去重命中时刷新文本/等级/逾期天数/归属（此前直接 continue → 金额与天数永久冻结在首版）
+      F4 余额快照批次行按开关处理（默认 skip，不再逐户生成伪"到期未收款"）
+    """
     import json as _json
     from app.models import KPIAlert, CashPlan, KPIDefinition
 
     result = forecast_cash_flow_with_plans(entity_id, db, days=30)
     new_alerts = []
+    refreshed = []
+    snapshot_skipped = []
+    snapshot_mode = get_balance_snapshot_mode()
+    snapshot_sources = get_balance_snapshot_sources()
 
-    # 兜底KPI：现金KPI → 经营现金流KPI → 该实体任意KPI
+    # 兜底KPI（F2）：现金KPI → 经营现金流KPI → 该实体 finance 维度首个（全部 ORDER BY id，确定性）
-    kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"]) or \
+    kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"], KPI_CODE_SUFFIXES["cash_balance"]) or \
-        find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"]) or \
+        find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"], KPI_CODE_SUFFIXES["operating_cash_flow"]) or \
-        db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).first()
+        find_fallback_kpi(db, entity_id)
-    ar_kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["receivables"]) or kpi
+    ar_kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["receivables"], KPI_CODE_SUFFIXES["receivables"]) or kpi
 
     def _exists(msg: str) -> bool:
         return db.query(KPIAlert).filter(
+            KPIAlert.entity_id == entity_id,
             KPIAlert.alert_message == msg,
             KPIAlert.status.in_(["pending", "processing"]),
         ).first() is not None
 
+    def _to_dict(raw) -> dict:
+        try:
+            d = _json.loads(raw) if raw else {}
+        except Exception:
+            d = {}
+        return d if isinstance(d, dict) else {}
+
+    def _refresh(existing, kpi_id: int, level: str, msg: str, suggestion: str) -> bool:
+        """F3: 按现值刷新已存在的预警（归属/文本/等级/建议），返回是否发生变化"""
+        changed = False
+        if existing.entity_id != entity_id:
+            existing.entity_id = entity_id
+            changed = True
+        if existing.kpi_id != kpi_id:
+            existing.kpi_id = kpi_id
+            changed = True
+        if existing.alert_level != level:
+            existing.alert_level = level
+            changed = True
+        if existing.alert_message != msg[:500]:
+            existing.alert_message = msg[:500]
+            changed = True
+        if existing.suggestion != suggestion:
+            existing.suggestion = suggestion
+            changed = True
+        return changed
+
     # ── 1. 缺口前3天预警 ──
     if kpi:
         for pre in result["pre_alerts"]:
@@ -652,6 +765,7 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
             if _exists(msg):
                 continue
             alert = KPIAlert(
+                entity_id=entity_id,          # F1: 此前未传 → 列默认值 1
                 kpi_id=kpi.id,
                 alert_level=level,
                 alert_message=msg[:500],
@@ -676,40 +790,99 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
             CashPlan.status == "pending",
             CashPlan.plan_date < today,
         ).order_by(CashPlan.plan_date.asc()).all()
-        for p in overdue:
+
-            days_late = (today - p.plan_date).days
+        # F4: 余额快照批次（plan_date = 导入基准日，非真实到期日）单独处理
-            msg = (f"【到期未收款】应收款{p.counterparty or '客户'} {p.amount:.1f}万 "
+        snapshot_rows = [p for p in overdue if (p.source or "") in snapshot_sources]
-                   f"原计划{p.plan_date.strftime('%Y-%m-%d')}到期，已逾期{days_late}天未收回")
+        overdue = [p for p in overdue if (p.source or "") not in snapshot_sources]
-            # 2026-08-30修复: 按plan_id去重（消息含"已逾期X天"动态文本，全文匹配会每天生成重复预警）
+
-            existing_plan_alert = db.query(KPIAlert).filter(
+        # 去重索引一次建好（修复原逐行全表扫描），去重键 = suggestion.plan_id / suggestion.snapshot_source
+        existing_alerts = db.query(KPIAlert).filter(
             KPIAlert.alert_type == "cash_plan",
             KPIAlert.status.in_(["pending", "processing"]),
         ).all()
-            dup = False
+        by_plan = {}
-            for ea in existing_plan_alert:
+        by_snapshot = {}
-                try:
+        for ea in existing_alerts:
-                    es = _json.loads(ea.suggestion) if ea.suggestion else {}
+            es = _to_dict(ea.suggestion)
-                    if isinstance(es, dict) and es.get("plan_id") == p.id:
+            if es.get("plan_id") is not None:
-                        dup = True
+                by_plan.setdefault(es["plan_id"], ea)
-                        break
+            if es.get("snapshot_source"):
-                except Exception:
+                by_snapshot.setdefault(es["snapshot_source"], ea)
-                    continue
+
-            if dup:
+        if snapshot_mode == "per_customer":
+            # 旧行为：批次行也逐户生成（仅显式切换时启用）
+            overdue = overdue + snapshot_rows
+        elif snapshot_mode == "skip":
+            if snapshot_rows:
+                snapshot_skipped = [{"plan_id": p.id, "source": p.source,
+                                     "amount": p.amount, "plan_date": str(p.plan_date)} for p in snapshot_rows]
+                logger.info(f"余额快照批次 {len(snapshot_rows)} 行按 skip 模式不生成逐户预警 "
+                            f"(entity={entity_id}, mode={snapshot_mode})")
+        elif snapshot_mode == "aggregate":
+            grouped = {}
+            for p in snapshot_rows:
+                grouped.setdefault(p.source or "unknown", []).append(p)
+            for src, rows in sorted(grouped.items()):
+                total = sum((p.amount or 0) for p in rows)
+                oldest = min(p.plan_date for p in rows)
+                days_late = (today - oldest).days
+                level = "red" if days_late >= 7 else "yellow"
+                msg = (f"【到期未收款·批次汇总】{src} 应收逾期{len(rows)}户 合计{total:.2f}万 "
+                       f"最早到期{oldest.strftime('%Y-%m-%d')}（{days_late}天）")
+                suggestion = _json.dumps({
+                    "actions": ["核对应收台账", "按客户逐户催收", "确认批次到期口径"],
+                    "snapshot_source": src,
+                    "plan_count": len(rows),
+                    "total_amount": round(total, 2),
+                    "days_late": days_late,
+                }, ensure_ascii=False)
+                existing = by_snapshot.get(src)
+                if existing is not None:
+                    if _refresh(existing, ar_kpi.id, level, msg, suggestion):
+                        refreshed.append({"type": "overdue_snapshot", "snapshot_source": src,
+                                          "alert_id": existing.id, "message": msg})
                     continue
                 alert = KPIAlert(
+                    entity_id=entity_id,
                     kpi_id=ar_kpi.id,
-                alert_level="red" if days_late >= 7 else "yellow",
+                    alert_level=level,
                     alert_message=msg[:500],
                     alert_type="cash_plan",
                     status="pending",
-                suggestion=_json.dumps({
+                    suggestion=suggestion,
+                )
+                db.add(alert)
+                new_alerts.append({"type": "overdue_snapshot", "level": level, "message": msg})
+                logger.info(f"到期未收款批次汇总: {msg}")
+
+        for p in overdue:
+            days_late = (today - p.plan_date).days
+            level = "red" if days_late >= 7 else "yellow"
+            msg = (f"【到期未收款】应收款{p.counterparty or '客户'} {p.amount:.1f}万 "
+                   f"原计划{p.plan_date.strftime('%Y-%m-%d')}到期，已逾期{days_late}天")
+            suggestion = _json.dumps({
                 "actions": ["联系客户催收", "评估坏账风险", "调整信用政策"],
                 "plan_id": p.id,
                 "days_late": days_late,
-                }, ensure_ascii=False),
+            }, ensure_ascii=False)
+            existing = by_plan.get(p.id)
+            if existing is not None:
+                # F3: 命中不 skip —— 按现值刷新金额/逾期天数/等级/归属
+                if _refresh(existing, ar_kpi.id, level, msg, suggestion):
+                    refreshed.append({"type": "overdue_receive", "plan_id": p.id,
+                                      "alert_id": existing.id, "message": msg})
+                continue
+            alert = KPIAlert(
+                entity_id=entity_id,          # F1: 此前未传 → 列默认值 1（错挂酣客工作台）
+                kpi_id=ar_kpi.id,
+                alert_level=level,
+                alert_message=msg[:500],
+                alert_type="cash_plan",
+                status="pending",
+                suggestion=suggestion,
             )
             db.add(alert)
-            new_alerts.append({"type": "overdue_receive", "level": alert.alert_level, "message": msg})
+            new_alerts.append({"type": "overdue_receive", "level": level, "message": msg})
             logger.info(f"到期未收款提醒: {msg}")
 
     db.commit()
@@ -717,6 +890,19 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
         "entity_id": entity_id,
         "new_alerts": len(new_alerts),
         "alerts": new_alerts,
+        "refreshed_alerts": len(refreshed),
+        "refreshed": refreshed,
+        "balance_snapshot": {
+            "mode": snapshot_mode,
+            "skipped_rows": len(snapshot_skipped),
+            "sources": sorted(snapshot_sources),
+        },
+        "kpi_selection": {
+            "ar_kpi_id": ar_kpi.id if ar_kpi else None,
+            "ar_kpi_code": ar_kpi.kpi_code if ar_kpi else None,
+            "cash_kpi_id": kpi.id if kpi else None,
+            "cash_kpi_code": kpi.kpi_code if kpi else None,
+        },
         "gap_dates": result["gap_dates"],
         "pre_alerts": result["pre_alerts"],
         "summary": result["summary"],

backend/app/utils/deviation_engine.py

@@ -341,6 +341,7 @@ def run_deviation_check(db_session, period: str = None) -> int:
 
             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}",
@@ -364,6 +365,7 @@ def run_deviation_check(db_session, period: str = None) -> int:
 
             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}",

backend/scripts/alert_generator.py

@@ -114,6 +114,7 @@ def run_alert_check(db_session, period: str = None) -> int:
 
                 # 创建新预警
                 alert = KPIAlert(
+                    entity_id=kpi.entity_id or 1,   # 2026-09-13 F1: 显式写归属
                     kpi_id=kpi.id,
                     kpi_value_id=latest.id,
                     alert_level=level,

backend/scripts/verify_cashplan_alert_engine_fix.py

@@ -0,0 +1,423 @@
+#!/usr/bin/env python3
+"""cash_plan「到期未收款」预警引擎修复 — 生产级验收电池
+任务：dama-cashplan-alert-engine-fix-20260913（提出者 wecom-dama / 审计 fin-alert-flood-e2）
+
+覆盖验收项：
+  F1 新建预警 entity_id 恒等于入参 entity（+ 其他创建点复查）
+  F2 KPI 选择实体感知：博海应收=BH_AR_BALANCE(433) 现金=BH_CASH_BALANCE(434)；
+     兜底查询显式 ORDER BY id（不再命中 BH_ADM_EXPENSE 437）
+  F3 去重命中时刷新 alert_message（金额/逾期天数）与 suggestion.days_late
+  F4 余额快照批次开关（默认 skip 不逐户；aggregate 聚合 1 条；per_customer 旧行为）
+  F5 POST /api/cma/admin/alerts/check 无鉴权=401；带 entity_id 只处理该实体
+
+安全设计（零生产写入）：
+  * F1-F4 全部在「生产库副本 cma_cashvfy」上执行（mysqldump 一次性复制）
+  * F3 用副本内合成计划（source='vfy_cashplan_fix'，仅存在于副本）构造"非快照 + 存量陈旧预警"场景
+  * F5 走 in-process TestClient，app.database 的 engine/SessionLocal 重定向到副本
+  * 唯一对生产库的访问是只读 COUNT（前后对比，证明未被写入）
+  * 副本用完即 DROP；不进常驻 cron
+
+用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_cashplan_alert_engine_fix.py
+"""
+import json
+import os
+import re
+import subprocess
+import sys
+from datetime import datetime
+
+BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, BASE)
+
+import pymysql
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+PROD_DB = "cma"
+SCRATCH_DB = "cma_cashvfy"
+SOCKET = "/var/run/mysqld/mysqld.sock"
+SCRATCH_URL = f"mysql+pymysql://root@localhost/{SCRATCH_DB}?unix_socket={SOCKET}&charset=utf8mb4"
+VFY_SOURCE = "vfy_cashplan_fix"
+
+RESULTS = []
+_ENGINES = []
+_SESSIONS = []
+
+
+def check(name, cond, detail=""):
+    RESULTS.append((bool(cond), name, detail))
+    print(f"{'PASS' if cond else 'FAIL'} | {name}" + (f" | {detail}" if detail else ""))
+    return bool(cond)
+
+
+def raw(sql, db=None, fetch=True, args=None):
+    """root socket 直连（副本运维 + 生产只读计数）"""
+    conn = pymysql.connect(unix_socket=SOCKET, user="root", database=db, charset="utf8mb4")
+    try:
+        cur = conn.cursor()
+        cur.execute(sql, args) if args else cur.execute(sql)
+        rows = cur.fetchall() if fetch else None
+        conn.commit()
+        return rows
+    finally:
+        conn.close()
+
+
+def new_session():
+    eng = create_engine(SCRATCH_URL)
+    _ENGINES.append(eng)
+    sess = sessionmaker(bind=eng)()
+    _SESSIONS.append(sess)
+    return sess
+
+
+def prod_counts():
+    q = lambda s: raw(s, db=PROD_DB)[0][0]
+    return {
+        "total": q("SELECT COUNT(*) FROM kpi_alerts"),
+        "e1": q("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id=1"),
+        "e2_pending": q("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id=2 AND status='pending'"),
+        "kpi437_pending": q("SELECT COUNT(*) FROM kpi_alerts WHERE kpi_id=437 AND status='pending'"),
+        "plans_e2_pending": q("SELECT COUNT(*) FROM cash_plans WHERE entity_id=2 AND status='pending'"),
+    }
+
+
+def amount_of(msg):
+    m = re.search(r"([\d.]+)万", msg or "")
+    return float(m.group(1)) if m else None
+
+
+def days_of(msg):
+    m = re.search(r"已逾期(\d+)天", msg or "")
+    return int(m.group(1)) if m else None
+
+
+def parse_sug(sug):
+    try:
+        d = json.loads(sug) if sug else {}
+        return d if isinstance(d, dict) else {}
+    except Exception:
+        return {}
+
+
+def setup_scratch():
+    raw(f"DROP DATABASE IF EXISTS {SCRATCH_DB}", fetch=False)
+    raw(f"CREATE DATABASE {SCRATCH_DB} CHARACTER SET utf8mb4", fetch=False)
+    cmd = f"mysqldump --single-transaction --routines --triggers {PROD_DB} | mysql {SCRATCH_DB}"
+    p = subprocess.run(cmd, shell=True, capture_output=True, text=True)
+    if p.returncode != 0:
+        raise RuntimeError(f"副本复制失败: {p.stderr[:500]}")
+    n = raw("SELECT COUNT(*) FROM kpi_alerts", db=SCRATCH_DB)[0][0]
+    print(f"[setup] cma -> {SCRATCH_DB} 完成, kpi_alerts={n}")
+    # 还原 8/29 洪水现场：副本内把实体的 cash_plan 预警状态恢复成 pending（验证 F3 刷新所需）
+    raw("UPDATE kpi_alerts SET status='pending', resolved_at=NULL, resolution=NULL "
+        "WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
+    return n
+
+
+def seed_synthetic_plans():
+    """副本内合成 3 条非快照逾期应收计划（F1 新建 / F3 刷新场景）"""
+    import datetime as _dt
+    today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    pdate = today.replace(day=1) - _dt.timedelta(days=1)  # 上月最后一天
+    rows = [("VFY-测试客户A", 12.3), ("VFY-测试客户B", 0.4), ("VFY-测试客户C", 4567.89)]
+    for cp, amt in rows:
+        raw("INSERT INTO cash_plans (entity_id, plan_type, amount, plan_date, counterparty, description, "
+            "status, source) VALUES (2, 'receive', %s, %s, %s, '系统验证临时数据', 'pending', %s)",
+            db=SCRATCH_DB, fetch=False, args=(amt, pdate.strftime("%Y-%m-%d"), cp, VFY_SOURCE))
+    ids = [r[0] for r in raw("SELECT id FROM cash_plans WHERE source=%s ORDER BY id",
+                             db=SCRATCH_DB, args=(VFY_SOURCE,))]
+    # 计划A：植入一条「陈旧 + 错挂」预警（entity=1 / kpi=437 / 金额虚增 / 逾期天数过期）
+    stale_msg = f"【到期未收款】应收款VFY-测试客户A 1230000.0万 原计划{pdate:%Y-%m-%d}到期，已逾期10天"
+    raw("INSERT INTO kpi_alerts (entity_id, kpi_id, alert_level, alert_message, alert_type, suggestion, status) "
+        "VALUES (1, 437, 'red', %s, 'cash_plan', %s, 'pending')",
+        db=SCRATCH_DB, fetch=False,
+        args=(stale_msg, json.dumps({"actions": ["联系客户催收"], "plan_id": ids[0], "days_late": 10},
+                                    ensure_ascii=False)))
+    return ids, pdate
+
+
+def main():
+    import app.utils.cash_forecast_engine as eng
+
+    prod_before = prod_counts()
+    print(f"[prod before] {prod_before}")
+
+    # ══════════════════════════════════════════════════════════
+    # F2 — KPI 实体感知 + 兜底确定性
+    # ══════════════════════════════════════════════════════════
+    print("\n=== F2 KPI 实体感知 / ORDER BY 确定性 ===")
+    s = new_session()
+    ar2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["receivables"], eng.KPI_CODE_SUFFIXES["receivables"])
+    cash2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["cash_balance"], eng.KPI_CODE_SUFFIXES["cash_balance"])
+    ap2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["payables"], eng.KPI_CODE_SUFFIXES["payables"])
+    check("F2 博海应收 → BH_AR_BALANCE(433)",
+          bool(ar2) and ar2.kpi_code == "BH_AR_BALANCE" and ar2.id == 433,
+          f"got={getattr(ar2, 'kpi_code', None)}({getattr(ar2, 'id', None)})")
+    check("F2 博海现金 → BH_CASH_BALANCE(434)",
+          bool(cash2) and cash2.kpi_code == "BH_CASH_BALANCE" and cash2.id == 434,
+          f"got={getattr(cash2, 'kpi_code', None)}({getattr(cash2, 'id', None)})")
+    check("F2 博海应付 → BH_AP_BALANCE(441)",
+          bool(ap2) and ap2.kpi_code == "BH_AP_BALANCE" and ap2.id == 441,
+          f"got={getattr(ap2, 'kpi_code', None)}({getattr(ap2, 'id', None)})")
+    fb_ids = {eng.find_fallback_kpi(s, 2).id for _ in range(20)}
+    fb = eng.find_fallback_kpi(s, 2)
+    check("F2 博海兜底 KPI 不等于 437(BH_ADM_EXPENSE)", fb.id != 437, f"fallback={fb.kpi_code}({fb.id})")
+    check("F2 博海兜底 20 次调用恒定（ORDER BY id）", len(fb_ids) == 1, f"ids={fb_ids}")
+    plan_dep = raw("SELECT id FROM kpi_definitions WHERE entity_id=2 LIMIT 1", db=SCRATCH_DB)[0][0]
+    plan_dep_uk = raw("SELECT id FROM kpi_definitions FORCE INDEX(uk_entity_kpi) "
+                      "WHERE entity_id=2 LIMIT 1", db=SCRATCH_DB)[0][0]
+    check("F2 复现旧缺陷：同条件两种答案（执行计划相关）",
+          plan_dep == 318 and plan_dep_uk == 437, f"idx_entity={plan_dep} / FORCE uk={plan_dep_uk}")
+    ar1 = eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["receivables"], eng.KPI_CODE_SUFFIXES["receivables"])
+    cash1 = eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["cash_balance"], eng.KPI_CODE_SUFFIXES["cash_balance"]) or \
+        eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["operating_cash_flow"], eng.KPI_CODE_SUFFIXES["operating_cash_flow"])
+    check("F2 酣客口径零回归（应收=F_AR_DAYS）", bool(ar1) and ar1.kpi_code == "F_AR_DAYS",
+          f"got={getattr(ar1, 'kpi_code', None)}({getattr(ar1, 'id', None)})")
+    check("F2 酣客口径零回归（现金=F_OP_CFLOW）", bool(cash1) and cash1.kpi_code == "F_OP_CFLOW",
+          f"got={getattr(cash1, 'kpi_code', None)}({getattr(cash1, 'id', None)})")
+    s.close()
+
+    # ══════════════════════════════════════════════════════════
+    # F4-skip（默认）+ 快照批次不逐户
+    # ══════════════════════════════════════════════════════════
+    os.environ.pop("CMA_CASH_SNAPSHOT_MODE", None)
+    os.environ.pop("CMA_CASH_SNAPSHOT_SOURCES", None)
+    print("\n=== F4(skip 默认)：余额快照批次不再逐户生成 ===")
+    s = new_session()
+    snap_before = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
+                      "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
+    before_ids = set(r[0] for r in raw("SELECT id FROM kpi_alerts", db=SCRATCH_DB))
+    e1_ids_before = {r[0] for r in raw("SELECT id FROM kpi_alerts WHERE entity_id=1", db=SCRATCH_DB)}
+    snap_plans = raw("SELECT COUNT(*) FROM cash_plans WHERE entity_id=2 AND plan_type='receive' "
+                     "AND status='pending' AND plan_date < CURDATE()", db=SCRATCH_DB)[0][0]
+    res = eng.check_cash_alerts(s, entity_id=2)
+    check("F4 默认模式 = skip", res["balance_snapshot"]["mode"] == "skip",
+          f"mode={res['balance_snapshot']['mode']}")
+    check("F4 快照批次行全部跳过逐户生成", res["balance_snapshot"]["skipped_rows"] == snap_plans,
+          f"skipped={res['balance_snapshot']['skipped_rows']} expect={snap_plans}")
+    check("F4 默认模式零新增预警（不重建 216 条伪逾期）", res["new_alerts"] == 0,
+          f"new_alerts={res['new_alerts']}")
+    check("F2 引擎选中应收 KPI = BH_AR_BALANCE(433)", res["kpi_selection"]["ar_kpi_code"] == "BH_AR_BALANCE",
+          f"ar={res['kpi_selection']['ar_kpi_code']} cash={res['kpi_selection']['cash_kpi_code']}")
+    check("F2 引擎选中现金 KPI = BH_CASH_BALANCE(434)", res["kpi_selection"]["cash_kpi_code"] == "BH_CASH_BALANCE",
+          f"cash={res['kpi_selection']['cash_kpi_code']}")
+    after_ids = set(r[0] for r in raw("SELECT id FROM kpi_alerts", db=SCRATCH_DB))
+    check("F1 未产生任何新预警行", after_ids <= before_ids, f"新增={sorted(after_ids - before_ids)[:5]}")
+    snap_after = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
+                     "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
+    check("F4 skip 模式不触碰存量快照预警（交由财务清理）",
+          [tuple(r) for r in snap_before] == [tuple(r) for r in snap_after],
+          f"before={len(snap_before)} after={len(snap_after)}")
+    e1_ids_after = {r[0] for r in raw("SELECT id FROM kpi_alerts WHERE entity_id=1", db=SCRATCH_DB)}
+    check("F1 酣客(entity1) 预警零变化", e1_ids_before == e1_ids_after,
+          f"{len(e1_ids_before)} -> {len(e1_ids_after)}")
+    s.close()
+
+    # ══════════════════════════════════════════════════════════
+    # F1 + F3 — 非快照计划的「新建归属」与「去重刷新」（副本合成计划）
+    # ══════════════════════════════════════════════════════════
+    print("\n=== F1 新建归属 + F3 去重刷新（副本合成非快照计划）===")
+    ids, pdate = seed_synthetic_plans()
+    exp_days = (datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) - pdate).days
+    s = new_session()
+    res2 = eng.check_cash_alerts(s, entity_id=2)
+    s.expire_all()
+    by_plan = {}
+    for aid, eid, kid, msg, sug, st in raw(
+            "SELECT id, entity_id, kpi_id, alert_message, suggestion, status FROM kpi_alerts "
+            "WHERE alert_type='cash_plan'", db=SCRATCH_DB):
+        pid = parse_sug(sug).get("plan_id")
+        if pid in ids:
+            by_plan[pid] = (aid, eid, kid, msg, parse_sug(sug), st)
+    check("F3 命中去重时刷新而非跳过（陈旧预警已更新）", res2["refreshed_alerts"] == 1,
+          f"refreshed={res2['refreshed_alerts']}")
+    check("F1 新建 2 条（无存量预警的计划）", res2["new_alerts"] == 2, f"new={res2['new_alerts']}")
+    check("F1/F3 3 条计划预警全部 entity_id=2", all(v[1] == 2 for v in by_plan.values()),
+          f"rows={[(k, v[1], v[2]) for k, v in by_plan.items()]}")
+    check("F2 3 条计划预警全部 kpi_id=433", all(v[2] == 433 for v in by_plan.values()),
+          f"rows={[(k, v[1], v[2]) for k, v in by_plan.items()]}")
+    ok_msg, bad = True, []
+    for pid in ids:
+        v = by_plan.get(pid)
+        if not v:
+            bad.append((pid, "缺失"))
+            continue
+        expected_amt = {ids[0]: 12.3, ids[1]: 0.4, ids[2]: 4567.89}[pid]
+        if f"{expected_amt:.1f}万" not in v[3]:
+            bad.append((pid, f"金额未刷新: {v[3][:50]}"))
+        if f"已逾期{exp_days}天" not in v[3]:
+            bad.append((pid, f"逾期天数未刷新: {v[3][:50]}"))
+        if v[4].get("days_late") != exp_days:
+            bad.append((pid, f"days_late={v[4].get('days_late')} expect={exp_days}"))
+    check("F3 文本金额/逾期天数与 suggestion.days_late 全部随现值刷新", not bad, f"异常={bad[:3]}")
+    check("F3 虚增金额（1230000.0万）已被现值覆盖",
+          "1230000.0万" not in "".join(v[3] for v in by_plan.values()), "")
+    check("F3 幂等：二次运行 0 新增 0 刷新",
+          (lambda r: r["new_alerts"] == 0 and r["refreshed_alerts"] == 0)(eng.check_cash_alerts(s, entity_id=2)),
+          "")
+    s.close()
+
+    # ══════════════════════════════════════════════════════════
+    # F4-aggregate — 方案(b)：按批次聚合 1 条
+    # ══════════════════════════════════════════════════════════
+    print("\n=== F4(aggregate) 方案(b)：按批次聚合 ===")
+    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
+    os.environ["CMA_CASH_SNAPSHOT_MODE"] = "aggregate"
+    s = new_session()
+    res3 = eng.check_cash_alerts(s, entity_id=2)
+    check("F4 aggregate 模式生效", res3["balance_snapshot"]["mode"] == "aggregate",
+          f"mode={res3['balance_snapshot']['mode']}")
+    agg = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
+              "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending' "
+              "AND alert_message LIKE '%%批次汇总%%'", db=SCRATCH_DB)
+    check("F4 聚合为每批次 1 条（3 个应收批次）", len(agg) == 3, f"count={len(agg)}")
+    check("F4 聚合行归属正确（entity=2 / kpi=433）", all(r[1] == 433 and r[2] == 2 for r in agg),
+          f"rows={[(r[0], r[1], r[2]) for r in agg]}")
+    ar_agg = [r for r in agg if "bohai_ar_20260630" in r[3]]
+    check("F4 聚合文本含户数与合计（210户 / 198.27万）",
+          len(ar_agg) == 1 and "210户" in ar_agg[0][3] and "198.27万" in ar_agg[0][3],
+          f"msg={ar_agg[0][3] if ar_agg else None}")
+    res4 = eng.check_cash_alerts(s, entity_id=2)
+    check("F4 aggregate 幂等（二次 0 新增 / 0 变更）",
+          res4["new_alerts"] == 0 and res4["refreshed_alerts"] == 0,
+          f"new={res4['new_alerts']} refreshed={res4['refreshed_alerts']}")
+    s.close()
+
+    # ══════════════════════════════════════════════════════════
+    # F4-per_customer — 旧行为开关回归（F1 使其也正确）
+    # ══════════════════════════════════════════════════════════
+    print("\n=== F4(per_customer) 旧行为开关回归 ===")
+    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
+    os.environ["CMA_CASH_SNAPSHOT_MODE"] = "per_customer"
+    s = new_session()
+    res5 = eng.check_cash_alerts(s, entity_id=2)
+    rows5 = raw("SELECT kpi_id, entity_id, alert_message FROM kpi_alerts "
+                "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
+    exp_cnt = snap_plans + 3
+    plan_amts = [r[0] for r in raw(
+        "SELECT amount FROM cash_plans WHERE entity_id=2 AND plan_type='receive' "
+        "AND status='pending' AND plan_date < CURDATE()", db=SCRATCH_DB)]
+    exp_sum = sum(round(a, 1) for a in plan_amts)
+    check("F4 per_customer 复原旧行为（逐户 216+3 条）", len(rows5) == exp_cnt,
+          f"count={len(rows5)} expect={exp_cnt}")
+    check("F1 旧模式下行归属仍正确（entity=2 / kpi=433）",
+          all(r[0] == 433 and r[1] == 2 for r in rows5), "")
+    amts5 = [amount_of(r[2]) or 0 for r in rows5]
+    check("F1/F3 旧模式金额按现值渲染（无 10000 倍放大）",
+          max(amts5) < 5000 and abs(sum(amts5) - exp_sum) < 0.6,
+          f"max={max(amts5)} sum={sum(amts5):.2f}万 expect={exp_sum:.2f}万")
+    s.close()
+    os.environ.pop("CMA_CASH_SNAPSHOT_MODE", None)
+
+    # ══════════════════════════════════════════════════════════
+    # F5 — 端点鉴权 + entity 过滤（in-process TestClient，副本重定向）
+    # ══════════════════════════════════════════════════════════
+    print("\n=== F5 /api/cma/admin/alerts/check 鉴权 + entity 参数 ===")
+    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
+    import app.database as db_module
+    db_module._engine = create_engine(SCRATCH_URL)
+    _ENGINES.append(db_module._engine)
+    db_module._SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=db_module._engine)
+    import app.utils.notifier as notifier
+    notifier.push_pending_alerts = lambda db_session: 0   # 禁止脚本触碰企微推送
+    from app.main import app as fastapi_app
+    from fastapi.testclient import TestClient
+    client = TestClient(fastapi_app)
+    ENDPOINT = "/api/cma/admin/alerts/check"
+
+    r_noauth = client.post(ENDPOINT)
+    check("F5 无鉴权调用返回 401", r_noauth.status_code == 401,
+          f"status={r_noauth.status_code} body={r_noauth.text[:60]}")
+    r_badtoken = client.post(ENDPOINT, headers={"Authorization": "Bearer deadbeef"})
+    check("F5 无效 token 返回 401", r_badtoken.status_code == 401, f"status={r_badtoken.status_code}")
+
+    from app.auth_middleware import create_token
+    token = create_token(1, 2)          # users.id=1 (admin) 在副本中同样存在
+    r_e2 = client.post(ENDPOINT + "?entity_id=2", headers={"Authorization": f"Bearer {token}"})
+    body = r_e2.json() if r_e2.status_code == 200 else {}
+    cash = body.get("cash") or {}
+    details = cash.get("details") or [{}]
+    d0 = details[0] if details else {}
+    check("F5 带 entity_id=2 请求 200", r_e2.status_code == 200,
+          f"status={r_e2.status_code} body={r_e2.text[:100]}")
+    check("F5 只处理指定实体（entities=1）", cash.get("entities") == 1, f"entities={cash.get('entities')}")
+    check("F5 处理后只产出 entity=2 记录", d0.get("entity_id") == 2, f"entity_id={d0.get('entity_id')}")
+    check("F5 修复后快照批次零重建（skipped=216 / 无快照预警）",
+          d0.get("balance_snapshot", {}).get("mode") == "skip"
+          and d0.get("balance_snapshot", {}).get("skipped_rows") == snap_plans,
+          f"snapshot={d0.get('balance_snapshot')}")
+    snap_pids = set(r[0] for r in raw(
+        "SELECT id FROM cash_plans WHERE entity_id=2 AND plan_type='receive' AND status='pending' "
+        "AND plan_date < CURDATE() AND source IN ('bohai_ar_20260630','bohai_ap_20260630',"
+        "'bohai_prepay_20260630','receivables_migrate')", db=SCRATCH_DB))
+    snap_alerts = 0
+    for (sug,) in raw("SELECT suggestion FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan' "
+                      "AND status='pending'", db=SCRATCH_DB):
+        if parse_sug(sug).get("plan_id") in snap_pids:
+            snap_alerts += 1
+    check("F5 调用后不存在指向快照计划的预警（原生 SQL 复核）", snap_alerts == 0, f"snap_alerts={snap_alerts}")
+    check("F5 引擎对博海不再重建伪预警（new = 仅 3 条合成非快照计划）",
+          d0.get("new_alerts") == 3, f"new={d0.get('new_alerts')}")
+    check("F5 响应暴露 KPI 归属便于复核（BH_AR_BALANCE）",
+          d0.get("kpi_selection", {}).get("ar_kpi_code") == "BH_AR_BALANCE",
+          f"kpi_selection={d0.get('kpi_selection')}")
+    r_404 = client.post(ENDPOINT + "?entity_id=9999", headers={"Authorization": f"Bearer {token}"})
+    check("F5 不存在/未激活 entity → 404", r_404.status_code == 404, f"status={r_404.status_code}")
+    leaked = raw("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id<>2 AND status='pending' "
+                 "AND alert_type='cash_plan' AND created_at > NOW() - INTERVAL 5 MINUTE",
+                 db=SCRATCH_DB)[0][0]
+    check("F5 本次调用未向其他实体写入预警", leaked == 0, f"leaked={leaked}")
+
+    # ══════════════════════════════════════════════════════════
+    # 生产库零写入核验
+    # ══════════════════════════════════════════════════════════
+    prod_after = prod_counts()
+    check("生产库 kpi_alerts / cash_plans 零变化", prod_before == prod_after,
+          f"before={prod_before}\n                                                after ={prod_after}")
+
+    ok = sum(1 for c, _, _ in RESULTS if c)
+    print(f"\n{'='*60}\n汇总: {ok}/{len(RESULTS)} PASS")
+    for c, name, detail in RESULTS:
+        if not c:
+            print(f"  FAIL: {name} | {detail}")
+    return 0 if ok == len(RESULTS) else 1
+
+
+def teardown():
+    for es in _SESSIONS:
+        try:
+            es.close()
+        except Exception:
+            pass
+    for en in _ENGINES:
+        try:
+            en.dispose()
+        except Exception:
+            pass
+    for attempt in range(3):
+        try:
+            conn = pymysql.connect(unix_socket=SOCKET, user="root", charset="utf8mb4")
+            cur = conn.cursor()
+            cur.execute("SET SESSION lock_wait_timeout=8")
+            cur.execute(f"DROP DATABASE IF EXISTS {SCRATCH_DB}")
+            conn.close()
+            print(f"[cleanup] {SCRATCH_DB} 已删除")
+            return
+        except Exception as e:
+            print(f"[cleanup] 第{attempt+1}次删除失败: {e}")
+    print(f"[cleanup] 警告：{SCRATCH_DB} 未删除，请手工清理")
+
+
+if __name__ == "__main__":
+    code = 1
+    try:
+        setup_scratch()
+        code = main()
+    except Exception:
+        import traceback
+        traceback.print_exc()
+    finally:
+        if os.environ.get("KEEP_SCRATCH") != "1":
+            teardown()
+    sys.exit(code)

docs/manuals/cash-plan-alerts.md

@@ -0,0 +1,58 @@
+# CMA 模块操作手册：资金收付款计划预警（cash-plan-alerts）
+
+> 用途：每开发/验证完一个 CMA 模块，产出/更新本手册。防重复修改（先查手册再动手）、防重复验证（照清单执行）。
+> 更新触发：每次该模块 bug 修复/功能变更 → 同步更新本手册 ⑤修改记录。
+
+## ① 模块定位
+- **模块名**：资金收付款计划预警（cash_plan 预警引擎）
+- **解决什么问题**：把 cash_plans（收付款计划）里「到期未收款」的计划变成预警中心的待办（kpi_alerts.alert_type='cash_plan'），另有「未来30天资金缺口」预测预警（alert_type='forecast'）。
+- **所属流程环节**：资金管理 → 现金预测 / 预警中心 → 处置
+- **主要角色**：finance（财务，处置预警）、ceo（查看）
+- **生成入口**：`POST /api/cma/admin/alerts/check`（手动/admin）、cron、`/api/cma/cash/*` 相关端点
+
+## ② 操作步骤
+- 财务：预警中心 `/alerts` 查看「到期未收款」pending 预警 → 处理（联系客户/创建改善计划）→ 置 resolved。
+- 运维/Bot：`POST /api/cma/admin/alerts/check`（**需 Bearer token**）；可带 `?entity_id=N` 只跑单个账套。
+- 余额快照批次口径开关（环境变量，改后需重启后端）：
+  - `CMA_CASH_SNAPSHOT_MODE` = `skip`（默认，批次行不生成逐户预警）| `aggregate`（按批次聚合 1 条）| `per_customer`（旧行为，逐户）
+  - `CMA_CASH_SNAPSHOT_SOURCES` = 逗号分隔的 source 名单（覆盖内置名单）
+
+## ③ 验证路径
+- **自动脚本**：`backend/scripts/verify_cashplan_alert_engine_fix.py`（43 项验收电池）
+  - 在**生产库副本**（mysqldump → `cma_cashvfy`）上跑，零生产写入；唯一生产访问是只读计数；跑完自动删副本。
+  - 覆盖：KPI 归属(433/434)/兜底 ORDER BY 确定性/entity_id 归属/去重刷新/快照批次三档开关/端点 401 与 entity 过滤。
+- **期望输出**：`汇总: 43/43 PASS` + `[cleanup] cma_cashvfy 已删除`，退出码 0。
+- **已测基线**：2026-09-13 → 43/43 PASS（副本）；生产端点实测 `?entity_id=2` 200 且 `new_alerts=0`。
+- **手工复核 SQL**（只读）：
+  ```sql
+  -- 博海 pending 红色"到期未收款"应为 0
+  SELECT COUNT(*) FROM cma.kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending';
+  -- 新预警归属：应收应挂 433(BH_AR_BALANCE)，现金应挂 434(BH_CASH_BALANCE)
+  SELECT kpi_id, entity_id, COUNT(*) FROM cma.kpi_alerts WHERE alert_type='cash_plan' GROUP BY kpi_id, entity_id;
+  ```
+
+## ④ 已知边界 / 坑
+- **KPI 选择顺序**：候选标准码（酣客口径 AR_001/F_AR_DAYS/C_AR_BALANCE…）→ **实体前缀后缀码**（`_AR_BALANCE` / `_CASH_BALANCE` / `_AP_BALANCE`，命中博海 BH_*）→ 兜底（finance 维度，`ORDER BY id`）。**任何兜底查询都必须带 ORDER BY**，否则同一条件会出现两种答案（实测 `idx_entity`→318 / `uk_entity_kpi`→437）。
+- **entity_id 必须显式写**：`KPIAlert(entity_id=...)`。该列 default=1，漏传会把博海的预警挂到酣客工作台（2026-09-10 实测 589/590）。
+- **去重键 = suggestion.plan_id**（不是 alert_message，后者含动态"已逾期X天"）。命中后必须**刷新**文本/等级/天数/归属，不能 `continue`。
+- **余额快照批次**（source ∈ bohai_ar_20260630 / bohai_ap_20260630 / bohai_prepay_20260630 / receivables_migrate）：plan_date 是**导入基准日**（期末日），不是到期日 → 逐户生成就会永远 overdue（伪逾期）。默认 `skip` 不逐户生成。
+- **skip 模式不接管存量快照预警**：历史遗留的 pending 快照预警需财务手工置 resolved（引擎不会自动改它们的状态）。
+- **⚠️ 易踩点**：
+  - `LAST_INSERT_ID()` 是**连接级**的——用「新连接」的 raw 查询取不到刚插入的自增 id（验证脚本曾因此把 plan_id 记成 0）。
+  - 金额单位是**万元**：cash_plans.amount 注释=万元，导入票据/台账的「元」值必须先 ÷10000，否则预警文本放大 10000 倍。
+  - FastAPI `HTTPBearer(auto_error=True)` 对「无 Authorization 头」返回 **403**；本端点用 `require_auth_401` 显式返回 **401**（验收要求）。
+- **历史 bug**：2026-08-29 博海 216 条伪"到期未收款"洪水（元当万元 + 基准日当到期日 + 兜底挂 BH_ADM_EXPENSE 437 + 错挂 entity 1）→ 2026-09-13 修复（本手册⑤）。
+
+## ⑤ 修改记录
+| 日期 | 修改人 | 改动 | commit | 影响 |
+|------|--------|------|--------|------|
+| 2026-09-13 | wecom-fullstack | 手册创建；F1 显式写 entity_id（引擎2处+deviation_engine/alert_generator/deviation_push/bot_bridge/bot_bridge_v2）；F2 候选码加实体前缀后缀 + 兜底 ORDER BY；F3 去重命中改为刷新文本/等级/天数/归属；F4 余额快照批次开关（默认 skip，另备 aggregate/per_customer）；F5 admin 端点补鉴权（无 token 401）+ 可选 entity_id | (待填) | 博海不再重建伪逾期预警；预警 KPI 归属 433/434；不再跨账套错挂 |
+
+## ⑥ 关联
+- **依赖表**：`cash_plans`、`kpi_alerts`、`kpi_definitions`、`kpi_values`、`system_config`
+- **依赖接口**：`POST /api/cma/admin/alerts/check[?entity_id=N]`、`GET /api/cma/cash/*`
+- **关联模块**：预警中心（alerts）、预警规则（alert-rules）、资金预测（cash-flow）
+- **代码**：`backend/app/utils/cash_forecast_engine.py::check_cash_alerts`、`backend/app/main.py::admin_check_alerts`
+
+---
+<!-- 注意：更新本手册时必须同时更新 ⑤修改记录；队列状态见 /root/WB_WS/_registry/cma-module-queue.json -->