fix(alerts): F4b 展示口径与落库口径同源 — summary/suggestions/upcoming_7d 按快照批次 mode 过滤 + raw/excluded/batches 可见 (dama-cashplan-zombie-alert-cleanup-20260913)

backend/app/utils/cash_forecast_engine.py

@@ -76,6 +76,82 @@ def get_balance_snapshot_sources() -> set:
     return set(BALANCE_SNAPSHOT_SOURCES)
 
 
+# ── F4b（2026-09-13）：展示口径与落库口径同源 ─────────────────────────
+# 背景：check_cash_alerts() 写库路径已按 get_balance_snapshot_mode() 过滤快照批次，
+# 但 forecast_cash_flow_with_plans() 的 summary/suggestions 仍按未过滤集合统计 ⇒
+# 同一次响应里出现「写入 0 条 / 展示 216 笔 200.72 万」的自相矛盾
+# （任务 dama-cashplan-zombie-alert-cleanup-20260913 B 段 / DAMA 复核 fullstack-20260913-0703）。
+# 约定：展示（summary/suggestions/upcoming）与写库（check_cash_alerts）**同一判据** ——
+#   skip        → 快照批次行不计入（默认）
+#   aggregate   → 快照批次按 source 聚合计数（与写库侧「1 条/批次」一致）
+#   per_customer→ 逐户全额计入（旧行为）
+# 同时保留可追溯字段（raw / excluded / batches），不隐藏数据。
+
+def split_snapshot_plans(rows: list, sources: Optional[set] = None) -> tuple:
+    """把计划行切成 (业务行, 余额快照批次行)，判据与写库侧完全一致（source ∈ 快照名单）"""
+    src = get_balance_snapshot_sources() if sources is None else set(sources)
+    business = [p for p in rows if (p.source or "") not in src]
+    snapshot = [p for p in rows if (p.source or "") in src]
+    return business, snapshot
+
+
+def summarize_overdue_receives(rows: list, mode: Optional[str] = None,
+                               sources: Optional[set] = None) -> dict:
+    """逾期应收展示口径 —— 与 check_cash_alerts() 写库口径同源（F4b）
+
+    rows: 逾期未收款计划（CashPlan，不预先过滤快照批次）
+    返回：
+      count / amount                       计入 summary 与 suggestions 的口径（= 落库口径）
+      raw_count / raw_amount               未过滤全集（含快照批次），保留可追溯
+      snapshot_rows_total / _amount        逾期中的快照批次行数与金额
+      snapshot_rows_excluded / _amount     未按「逐户 1 条」计入展示口径的行数与金额
+      snapshot_batches                     批次分组明细（source/count/amount）
+      mode / sources                       生效口径与名单
+    """
+    mode = mode or get_balance_snapshot_mode()
+    src = get_balance_snapshot_sources() if sources is None else set(sources)
+    business, snapshot = split_snapshot_plans(rows, src)
+
+    batches = {}
+    for p in snapshot:
+        key = p.source or "unknown"
+        b = batches.setdefault(key, {"source": key, "count": 0, "amount": 0.0})
+        b["count"] += 1
+        b["amount"] = round(b["amount"] + (p.amount or 0), 2)
+
+    if mode == "per_customer":
+        display_rows = business + snapshot
+        excluded_rows, excluded_amount = 0, 0.0
+    elif mode == "aggregate":
+        # 与写库侧一致：每 source 聚合为 1 条 → 展示计 len(batches) 条，金额全量计入
+        display_rows = business
+        excluded_rows = len(snapshot) - len(batches)
+        excluded_amount = 0.0
+    else:  # skip（默认）
+        display_rows = business
+        excluded_rows = len(snapshot)
+        excluded_amount = round(sum((p.amount or 0) for p in snapshot), 2)
+
+    count = len(display_rows) + (len(batches) if mode == "aggregate" else 0)
+    amount = round(sum((p.amount or 0) for p in display_rows), 2)
+    if mode == "aggregate":
+        amount = round(amount + sum(b["amount"] for b in batches.values()), 2)
+
+    return {
+        "count": count,
+        "amount": amount,
+        "raw_count": len(rows),
+        "raw_amount": round(sum((p.amount or 0) for p in rows), 2),
+        "snapshot_rows_total": len(snapshot),
+        "snapshot_rows_total_amount": round(sum((p.amount or 0) for p in snapshot), 2),
+        "snapshot_rows_excluded": excluded_rows,
+        "snapshot_rows_excluded_amount": excluded_amount,
+        "snapshot_batches": [batches[k] for k in sorted(batches)],
+        "mode": mode,
+        "sources": sorted(src),
+    }
+
+
 def find_kpi(db: Session, entity_id: int, codes: list, suffixes: Optional[list] = None) -> Optional["KPIDefinition"]:
     """按候选编码列表查找KPI，返回第一个命中的（精确码优先 → 实体前缀后缀码 → None）
 
@@ -628,20 +704,28 @@ def forecast_cash_flow_with_plans(
                 break
 
     # 到期未收款（逾期）
-    overdue_receives = db.query(CashPlan).filter(
+    overdue_all = db.query(CashPlan).filter(
         CashPlan.entity_id == entity_id,
         CashPlan.plan_type == "receive",
         CashPlan.status == "pending",
         CashPlan.plan_date < today,
     ).order_by(CashPlan.plan_date.asc()).all()
 
-    # 未来7天到期
-    upcoming_7d = db.query(CashPlan).filter(
+    # F4b(2026-09-13): 展示口径与落库口径（check_cash_alerts）同源 —— 同一份快照批次判据，
+    # 避免同一次响应内「写入 0 条 / 展示 N 笔」自相矛盾；raw/excluded 字段保留可追溯。
+    overdue_scope = summarize_overdue_receives(overdue_all)
+
+    # 未来7天到期（同样剔除快照批次行：其 plan_date 是导入基准日，不是真实到期日）
+    upcoming_all = db.query(CashPlan).filter(
         CashPlan.entity_id == entity_id,
         CashPlan.status == "pending",
         CashPlan.plan_date >= today,
         CashPlan.plan_date <= today + timedelta(days=7),
     ).order_by(CashPlan.plan_date.asc()).all()
+    if overdue_scope["mode"] == "per_customer":
+        upcoming_7d = upcoming_all
+    else:
+        upcoming_7d = [p for p in upcoming_all if (p.source or "") not in set(overdue_scope["sources"])]
 
     # 整体结论
     min_cash = min(f["predicted_cash"] for f in forecast) if forecast else base_cash
@@ -662,11 +746,16 @@ def forecast_cash_flow_with_plans(
                 "message": f"预计{first_gap['date']}现金余额降至{first_gap['predicted_cash']:.1f}万，低于警戒线{warning_line:.0f}万",
                 "actions": ["加快应收账款回款", "控制采购付款节奏", "评估短期现金流压力"],
             })
-    if overdue_receives:
-        total_overdue = sum(p.amount for p in overdue_receives)
+    if overdue_scope["count"]:
+        total_overdue = overdue_scope["amount"]
+        msg = f"有{overdue_scope['count']}笔应收款到期未收，合计{total_overdue:.1f}万"
+        if overdue_scope["snapshot_rows_excluded"]:
+            msg += (f"（另有余额快照批次{overdue_scope['snapshot_rows_total']}行/"
+                    f"{overdue_scope['snapshot_rows_total_amount']:.1f}万未计入逐户口径："
+                    f"plan_date 为导入基准日，见 balance_snapshot）")
         suggestions.append({
             "type": "warning",
-            "message": f"有{len(overdue_receives)}笔应收款到期未收，合计{total_overdue:.1f}万",
+            "message": msg,
             "actions": ["逐笔催收到期应收账款", "评估客户信用风险"],
         })
 
@@ -688,9 +777,19 @@ def forecast_cash_flow_with_plans(
             "total_planned_in": round(sum(p.amount for p in plans if p.plan_type == "receive"), 2),
             "total_planned_out": round(sum(p.amount for p in plans if p.plan_type == "pay"), 2),
             "plan_count": len(plans),
-            "overdue_receive_count": len(overdue_receives),
-            "overdue_receive_amount": round(sum(p.amount for p in overdue_receives), 2),
+            # F4b: 展示口径 = 落库口径（快照批次按 mode 处理），不注释掉即视为对外契约
+            "overdue_receive_count": overdue_scope["count"],
+            "overdue_receive_amount": overdue_scope["amount"],
             "upcoming_7d_count": len(upcoming_7d),
+            # F4b 透明可追溯字段（不隐藏数据：全集 + 剔除量 + 批次明细）
+            "overdue_receive_count_raw": overdue_scope["raw_count"],
+            "overdue_receive_amount_raw": overdue_scope["raw_amount"],
+            "snapshot_rows_excluded": overdue_scope["snapshot_rows_excluded"],
+            "snapshot_rows_excluded_amount": overdue_scope["snapshot_rows_excluded_amount"],
+            "snapshot_rows_total": overdue_scope["snapshot_rows_total"],
+            "snapshot_rows_total_amount": overdue_scope["snapshot_rows_total_amount"],
+            "snapshot_batches": overdue_scope["snapshot_batches"],
+            "snapshot_mode": overdue_scope["mode"],
         },
     }
 
@@ -792,8 +891,8 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
         ).order_by(CashPlan.plan_date.asc()).all()
 
         # F4: 余额快照批次（plan_date = 导入基准日，非真实到期日）单独处理
-        snapshot_rows = [p for p in overdue if (p.source or "") in snapshot_sources]
-        overdue = [p for p in overdue if (p.source or "") not in snapshot_sources]
+        # F4b: 复用 split_snapshot_plans —— 与 summary/suggestions 展示口径同一判据（单一事实源）
+        overdue, snapshot_rows = split_snapshot_plans(overdue, snapshot_sources)
 
         # 去重索引一次建好（修复原逐行全表扫描），去重键 = suggestion.plan_id / suggestion.snapshot_source
         existing_alerts = db.query(KPIAlert).filter(