diff --git a/backend/app/utils/cash_forecast_engine.py b/backend/app/utils/cash_forecast_engine.py index 6a2ce4a1..c0d27a9a 100644 --- a/backend/app/utils/cash_forecast_engine.py +++ b/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( diff --git a/backend/scripts/verify_cashplan_alert_engine_fix.py b/backend/scripts/verify_cashplan_alert_engine_fix.py index 0be6f0fd..47a9fc7a 100644 --- a/backend/scripts/verify_cashplan_alert_engine_fix.py +++ b/backend/scripts/verify_cashplan_alert_engine_fix.py @@ -212,6 +212,23 @@ def main(): 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)}") + + # ── F4b：summary/suggestions 展示口径与落库口径同源 ── + # （任务 dama-cashplan-zombie-alert-cleanup-20260913 B 段：曾是「写入 0 / 展示 216」） + sm = res["summary"] + check("F4b summary 快照批次不计入逐户口径（count=0）", sm["overdue_receive_count"] == 0, + f"count={sm['overdue_receive_count']} raw={sm['overdue_receive_count_raw']}") + check("F4b 展示口径 ⇔ 落库口径（overdue_receive_count=0 与 new_alerts=0 一致）", + (sm["overdue_receive_count"] == 0) == (res["new_alerts"] == 0), + f"count={sm['overdue_receive_count']} new={res['new_alerts']}") + check("F4b 剔除量显式可见（excluded == skipped_rows == 快照逾期行数）", + sm["snapshot_rows_excluded"] == res["balance_snapshot"]["skipped_rows"] == snap_plans, + f"excluded={sm['snapshot_rows_excluded']} skipped={res['balance_snapshot']['skipped_rows']} expect={snap_plans}") + check("F4b 全集可追溯（raw_count == 快照逾期行数）", sm["overdue_receive_count_raw"] == snap_plans, + f"raw={sm['overdue_receive_count_raw']} expect={snap_plans}") + check("F4b 批次明细可见（bohai_ar_20260630 210 户）", + any(b["source"] == "bohai_ar_20260630" and b["count"] == 210 for b in sm["snapshot_batches"]), + f"batches={sm['snapshot_batches']}") s.close() # ══════════════════════════════════════════════════════════ @@ -282,6 +299,12 @@ def main(): check("F4 aggregate 幂等（二次 0 新增 / 0 变更）", res4["new_alerts"] == 0 and res4["refreshed_alerts"] == 0, f"new={res4['new_alerts']} refreshed={res4['refreshed_alerts']}") + agg_sm = res3["summary"] + check("F4b aggregate 展示口径=落库（业务行 + 每批次 1 条）", + agg_sm["overdue_receive_count"] == + (agg_sm["overdue_receive_count_raw"] - agg_sm["snapshot_rows_total"]) + len(agg_sm["snapshot_batches"]), + f"count={agg_sm['overdue_receive_count']} raw={agg_sm['overdue_receive_count_raw']} " + f"snapshot={agg_sm['snapshot_rows_total']} batches={len(agg_sm['snapshot_batches'])}") s.close() # ══════════════════════════════════════════════════════════ @@ -307,6 +330,11 @@ def main(): 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}万") + sm5 = res5["summary"] + check("F4b per_customer 展示口径=落库（逐户全额 / excluded=0）", + sm5["overdue_receive_count"] == sm5["overdue_receive_count_raw"] and sm5["snapshot_rows_excluded"] == 0, + f"count={sm5['overdue_receive_count']} raw={sm5['overdue_receive_count_raw']} " + f"excluded={sm5['snapshot_rows_excluded']}") s.close() os.environ.pop("CMA_CASH_SNAPSHOT_MODE", None) @@ -347,6 +375,11 @@ def main(): d0.get("balance_snapshot", {}).get("mode") == "skip" and d0.get("balance_snapshot", {}).get("skipped_rows") == snap_plans, f"snapshot={d0.get('balance_snapshot')}") + check("F4b 端点响应 summary=落库口径（count=new_alerts / 快照批次不计入 / excluded 可见）", + d0.get("summary", {}).get("overdue_receive_count") == d0.get("new_alerts") + and d0.get("summary", {}).get("snapshot_rows_excluded") == snap_plans, + f"count={d0.get('summary', {}).get('overdue_receive_count')} new={d0.get('new_alerts')} " + f"excluded={d0.get('summary', {}).get('snapshot_rows_excluded')} expect={snap_plans}") 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'," diff --git a/backend/tests/test_cash_summary_snapshot_alignment_20260913.py b/backend/tests/test_cash_summary_snapshot_alignment_20260913.py new file mode 100644 index 00000000..f7622f69 --- /dev/null +++ b/backend/tests/test_cash_summary_snapshot_alignment_20260913.py @@ -0,0 +1,173 @@ +"""F4b 断言测试：到期未收款「展示口径」与「落库口径」同源（任务 dama-cashplan-zombie-alert-cleanup-20260913 B 段） + +背景：check_cash_alerts() 写库路径已按 get_balance_snapshot_mode() 过滤余额快照批次， +但 forecast_cash_flow_with_plans() 的 summary / suggestions 仍按未过滤集合统计 ⇒ +同一次响应里「写入 0 条」与「展示 216 笔 / 200.72 万」并存（DAMA 复核 fullstack-20260913-0703 §3.3）。 + +本用例锁定： + 1) 默认 skip：summary.overdue_receive_count/amount 不含快照批次行；excluded 字段可见 + 2) per_customer（旧行为）：逐户全额计入 + 3) aggregate（写库侧 1 条/批次）：展示按批次数计 + 4) suggestions 文案与 summary 同口径，且对未计入批次显式标注 + 5) summarize_overdue_receives() 纯函数口径矩阵（含 upcoming_7d 同源） +""" +from datetime import datetime, timedelta + +import pytest +from sqlalchemy.orm import Session + +from app.models import CashPlan +from app.utils.cash_forecast_engine import ( + BALANCE_SNAPSHOT_SOURCES, + forecast_cash_flow_with_plans, + get_balance_snapshot_mode, + summarize_overdue_receives, +) + +SNAP_SRC = "bohai_ar_20260630" # ∈ BALANCE_SNAPSHOT_SOURCES +EID = 1 +TODAY = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + + +def _seed(db: Session): + """业务逾期 2 笔（20 + 30 万）+ 快照批次逾期 3 笔（10 + 20 + 30 万）""" + rows = [ + # 业务行（source 非快照名单） + CashPlan(entity_id=EID, plan_type="receive", status="pending", source="manual", + amount=20.0, counterparty="业务A", plan_date=TODAY - timedelta(days=9)), + CashPlan(entity_id=EID, plan_type="receive", status="pending", source=None, + amount=30.0, counterparty="业务B", plan_date=TODAY - timedelta(days=3)), + # 快照批次行 + CashPlan(entity_id=EID, plan_type="receive", status="pending", source=SNAP_SRC, + amount=10.0, counterparty="快照A", plan_date=TODAY - timedelta(days=75)), + CashPlan(entity_id=EID, plan_type="receive", status="pending", source=SNAP_SRC, + amount=20.0, counterparty="快照B", plan_date=TODAY - timedelta(days=75)), + CashPlan(entity_id=EID, plan_type="receive", status="pending", source=SNAP_SRC, + amount=30.0, counterparty="快照C", plan_date=TODAY - timedelta(days=75)), + ] + for r in rows: + db.add(r) + db.commit() + return rows + + +def _overdue_rows(db: Session): + return db.query(CashPlan).filter( + CashPlan.entity_id == EID, + CashPlan.plan_type == "receive", + CashPlan.status == "pending", + CashPlan.plan_date < TODAY, + ).all() + + +# ── 1. 纯函数口径矩阵 ──────────────────────────────────────────────── +class TestSummarizeOverdueReceives: + + def test_default_is_skip_and_excludes_snapshot(self, db: Session): + assert get_balance_snapshot_mode() == "skip" + rows = _seed(db) + meta = summarize_overdue_receives(_overdue_rows(db)) + assert meta["mode"] == "skip" + assert meta["count"] == 2, "skip 下只计业务行" + assert meta["amount"] == 50.0 + assert meta["raw_count"] == 5 + assert meta["raw_amount"] == 110.0 + assert meta["snapshot_rows_excluded"] == 3 + assert meta["snapshot_rows_excluded_amount"] == 60.0 + assert meta["snapshot_rows_total"] == 3 + assert meta["snapshot_batches"] == [{"source": SNAP_SRC, "count": 3, "amount": 60.0}] + + def test_per_customer_counts_everything(self, db: Session): + _seed(db) + meta = summarize_overdue_receives(_overdue_rows(db), mode="per_customer") + assert meta["count"] == 5 + assert meta["amount"] == 110.0 + assert meta["snapshot_rows_excluded"] == 0 + + def test_aggregate_counts_batches(self, db: Session): + _seed(db) + meta = summarize_overdue_receives(_overdue_rows(db), mode="aggregate") + # 写库侧 aggregate = 1 条/批次 ⇒ 展示 = 2 业务行 + 1 批次 = 3 条，金额全量 110 万 + assert meta["count"] == 3 + assert meta["amount"] == 110.0 + assert meta["snapshot_rows_excluded"] == 2 # 3 行里 2 行被折叠进批次条目 + assert len(meta["snapshot_batches"]) == 1 + + def test_batch_list_is_internal_whitelist(self): + assert SNAP_SRC in BALANCE_SNAPSHOT_SOURCES + + +# ── 2. 端到端：forecast_cash_flow_with_plans 的 summary/suggestions ── +class TestForecastSummaryAlignment: + + def test_skip_mode_summary_excludes_snapshot(self, db: Session): + """默认 skip：summary 口径 = 落库口径（断言「写入 0 条 / 展示 3 笔」不再发生）""" + _seed(db) + result = forecast_cash_flow_with_plans(EID, db, days=30) + s = result["summary"] + assert s["snapshot_mode"] == "skip" + assert s["overdue_receive_count"] == 2 + assert s["overdue_receive_amount"] == 50.0 + assert s["snapshot_rows_excluded"] == 3 + assert s["overdue_receive_count_raw"] == 5 + assert s["overdue_receive_amount_raw"] == 110.0 + + def test_skip_mode_suggestion_discloses_excluded_batch(self, db: Session): + _seed(db) + result = forecast_cash_flow_with_plans(EID, db, days=30) + msgs = [x["message"] for x in result["suggestions"] if "应收款到期未收" in x["message"]] + assert msgs, "应保留业务行催收建议" + msg = msgs[0] + assert msg.startswith("有2笔应收款到期未收，合计50.0万") + assert "余额快照批次3行/60.0万未计入" in msg + + def test_no_business_overdue_means_no_overdue_suggestion(self, db: Session): + """只有快照批次（博海今日实况）：skip 下不产生「到期未收」建议 —— 与 new_alerts=0 一致""" + db.add(CashPlan(entity_id=EID, plan_type="receive", status="pending", source=SNAP_SRC, + amount=198.27, counterparty="快照户", plan_date=TODAY - timedelta(days=75))) + db.commit() + result = forecast_cash_flow_with_plans(EID, db, days=30) + assert result["summary"]["overdue_receive_count"] == 0 + assert result["summary"]["snapshot_rows_excluded"] == 1 + assert not [x for x in result["suggestions"] if "应收款到期未收" in x["message"]] + + def test_per_customer_env_restores_old_display(self, db: Session, monkeypatch): + monkeypatch.setenv("CMA_CASH_SNAPSHOT_MODE", "per_customer") + _seed(db) + result = forecast_cash_flow_with_plans(EID, db, days=30) + s = result["summary"] + assert s["snapshot_mode"] == "per_customer" + assert s["overdue_receive_count"] == 5 + assert s["overdue_receive_amount"] == 110.0 + assert s["snapshot_rows_excluded"] == 0 + + def test_aggregate_env_display_matches_write_side(self, db: Session, monkeypatch): + monkeypatch.setenv("CMA_CASH_SNAPSHOT_MODE", "aggregate") + _seed(db) + result = forecast_cash_flow_with_plans(EID, db, days=30) + s = result["summary"] + assert s["snapshot_mode"] == "aggregate" + assert s["overdue_receive_count"] == 3 + assert s["overdue_receive_amount"] == 110.0 + + def test_upcoming_7d_excludes_snapshot_rows(self, db: Session): + """upcoming_7d 同源：快照行的 plan_date 非真实到期日，skip 下不计入""" + db.add(CashPlan(entity_id=EID, plan_type="receive", status="pending", source="manual", + amount=5.0, counterparty="近7天", plan_date=TODAY + timedelta(days=3))) + db.add(CashPlan(entity_id=EID, plan_type="receive", status="pending", source=SNAP_SRC, + amount=7.0, counterparty="快照7天", plan_date=TODAY + timedelta(days=3))) + db.commit() + result = forecast_cash_flow_with_plans(EID, db, days=30) + assert result["summary"]["upcoming_7d_count"] == 1 + + def test_entity_isolation_and_status_filter(self, db: Session): + """口径修正不影响实体隔离/状态过滤（只算本实体 pending 逾期应收）""" + _seed(db) + db.add(CashPlan(entity_id=2, plan_type="receive", status="pending", source="manual", + amount=999.0, counterparty="博海", plan_date=TODAY - timedelta(days=9))) + db.add(CashPlan(entity_id=EID, plan_type="receive", status="completed", source="manual", + amount=888.0, counterparty="已完成", plan_date=TODAY - timedelta(days=9))) + db.commit() + result = forecast_cash_flow_with_plans(EID, db, days=30) + assert result["summary"]["overdue_receive_count"] == 2 + assert result["summary"]["overdue_receive_amount"] == 50.0