"""现金流预测引擎 — 根据历史KPI数据推算未来30天现金流""" import logging import os from datetime import datetime, timedelta from typing import Optional, TYPE_CHECKING from sqlalchemy.orm import Session import math import random if TYPE_CHECKING: from app.models import KPIDefinition logger = logging.getLogger("cma.cash_forecast") # 默认现金阈值（万元） DEFAULT_CASH_WARNING = 20.0 # 黄灯 — 低于20万 DEFAULT_CASH_CRITICAL = 10.0 # 红灯 — 低于10万 # 历史KPI编码映射（含候选编码，兼容F_*标准编码） KPI_CODES = { "operating_cash_flow": "CASH_FLOW_001", # 经营现金流 "receivables": "AR_001", # 应收账款 "payables": "AP_001", # 应付账款 "cash_balance": "CASH_001", # 现金余额 } # 候选编码列表 — 依次尝试，找不到则回退 KPI_CODE_CANDIDATES = { "operating_cash_flow": ["CASH_FLOW_001", "F_OP_CFLOW", "F_REVENUE"], "receivables": ["AR_001", "F_AR_DAYS", "C_AR_BALANCE"], "payables": ["AP_001", "F_AP_DAYS"], "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) } # ── 余额快照批次（F4，2026-09-13）──────────────────────────── # 这些 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) # ── 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） 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, ).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 kpi = find_kpi(db, entity_id, [kpi_code]) if not kpi: return [] values = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.period.desc()).limit(limit_months).all() return values def calc_trend(values: list) -> float: """计算趋势系数 — 线性回归斜率 / 均值""" if len(values) < 2: return 0.0 vals = [v.actual_value for v in values] n = len(vals) avg_x = (n - 1) / 2.0 avg_y = sum(vals) / n num = sum((i - avg_x) * (vals[i] - avg_y) for i in range(n)) den = sum((i - avg_x) ** 2 for i in range(n)) slope = num / den if den != 0 else 0 return slope / max(abs(avg_y), 1.0) * 100 # 趋势百分比 def get_current_cash_balance(db: Session, entity_id: int) -> Optional[float]: """获取当前现金余额（万元）— 优先级：KPI实际值 > SystemConfig > None""" from app.models import SystemConfig, KPIValue # C3(2026-09-19 dama-20260919)：补传实体前缀后缀候选（与 F2 的应收/应付路径一致） # 修复前漏传 → 博海/演示企业一条候选都不命中 → 回退 SystemConfig.cash.current_balance(全库唯一键 -9.3) kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"], KPI_CODE_SUFFIXES["cash_balance"]) if kpi: latest_cash = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.period.desc()).first() if latest_cash and latest_cash.actual_value is not None: return float(latest_cash.actual_value) cfg = db.query(SystemConfig).filter( SystemConfig.config_key == "cash.current_balance" ).first() if cfg and cfg.config_value: try: return float(cfg.config_value) except Exception: pass return None def set_current_cash_balance(db: Session, value: float) -> float: """设置当前现金余额（写入SystemConfig，供预测引擎使用）""" from app.models import SystemConfig cfg = db.query(SystemConfig).filter( SystemConfig.config_key == "cash.current_balance" ).first() if cfg: cfg.config_value = str(value) else: cfg = SystemConfig( config_key="cash.current_balance", config_value=str(value), description="当前现金余额（万元），资金预测基线", ) db.add(cfg) db.commit() return value def forecast_cash_flow( entity_id: int, db: Session, days: int = 30, current_cash: Optional[float] = None, ) -> dict: """ 预测未来30天现金流 算法： 1. 获取历史经营现金流、应收、应付趋势 2. 推算每日现金流入/流出 3. 生成每日预测值+置信区间 """ from app.models import KPIDefinition, KPIValue # 获取当前现金余额 if current_cash is None: base_cash = get_current_cash_balance(db, entity_id) if base_cash is None: base_cash = 30.0 # 默认假设30万 else: base_cash = current_cash # 获取经营现金流历史 ocf_history = get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"][0], db) or \ get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"][1], db) or \ get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"][2], db) ocf_trend = calc_trend(ocf_history) # 获取应收历史 ar_history = get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["receivables"][0], db) or \ get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["receivables"][1], db) ar_trend = calc_trend(ar_history) # 获取应付历史 ap_history = get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["payables"][0], db) or \ get_entity_kpi_history(entity_id, KPI_CODE_CANDIDATES["payables"][1], db) ap_trend = calc_trend(ap_history) # 计算日均现金变化 ocf_avg = sum(v.actual_value for v in ocf_history) / max(len(ocf_history), 1) / 30.0 if ocf_history else 0.5 # 预测逻辑：趋势影响 + 季节性（月底回款高峰） forecast = [] cash = base_cash today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) for day_offset in range(1, days + 1): d = today + timedelta(days=day_offset) day_of_month = d.day is_month_end = day_of_month >= 25 # 每日现金变化 = 经营现金流日均值 × (1 + 趋势调整) + 季节因子 trend_factor = 1.0 + ocf_trend / 100.0 daily_change = ocf_avg * trend_factor # 月底回款高峰 if is_month_end: daily_change += ocf_avg * 0.3 # 月底多30%回款 # 周末效应 if d.weekday() >= 5: daily_change *= 0.5 # 周末收支减半 cash += daily_change # 置信区间：随时间增加而扩大 confidence_band = 1.0 + day_offset * 0.08 # 每过1天，区间扩大8% std = max(abs(daily_change) * confidence_band, 0.5) lower = cash - std * 0.5 upper = cash + std * 0.5 # 预警状态 if cash < DEFAULT_CASH_CRITICAL: status = "red" elif cash < DEFAULT_CASH_WARNING: status = "yellow" else: status = "green" forecast.append({ "date": d.strftime("%Y-%m-%d"), "day_offset": day_offset, "predicted_cash": round(cash, 2), "lower_bound": round(max(lower, 0), 2), "upper_bound": round(upper, 2), "alert_status": status, }) # 整体预警 min_cash = min(f["predicted_cash"] for f in forecast) min_date = next(f["date"] for f in forecast if f["predicted_cash"] == min_cash) suggestions = [] if min_cash < DEFAULT_CASH_CRITICAL: suggestions.append({ "type": "critical", "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于警戒线{DEFAULT_CASH_CRITICAL}万", "actions": [ "立即催收大额应收账款", "暂停非必要支出", "准备短期融资安排", ] }) elif min_cash < DEFAULT_CASH_WARNING: suggestions.append({ "type": "warning", "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于关注线{DEFAULT_CASH_WARNING}万", "actions": [ "加快应收账款回款", "控制采购付款节奏", "评估短期现金流压力", ] }) return { "entity_id": entity_id, "base_cash": round(base_cash, 2), "days": days, "forecast": forecast, "min_cash": round(min_cash, 2), "min_cash_date": min_date, "trends": { "operating_cash_flow_trend_pct": round(ocf_trend, 2), "receivables_trend_pct": round(ar_trend, 2), "payables_trend_pct": round(ap_trend, 2), }, "suggestions": suggestions, } def save_forecast_to_db(entity_id: int, forecast_data: dict, db: Session): """将预测结果保存到数据库""" from app.models import CashForecast for f in forecast_data["forecast"]: forecast_date = datetime.strptime(f["date"], "%Y-%m-%d") cf = CashForecast( entity_id=entity_id, forecast_date=forecast_date, predicted_cash=f["predicted_cash"], lower_bound=f["lower_bound"], upper_bound=f["upper_bound"], alert_status=f["alert_status"], ) db.add(cf) db.commit() def calculate_accuracy(entity_id: int, db: Session) -> list: """计算预测准确率 — 对比上期预测 vs 本期实际""" from app.models import CashForecast, KPIDefinition, KPIValue # 获取实体最近的预测 forecasts = db.query(CashForecast).filter( CashForecast.entity_id == entity_id, ).order_by(CashForecast.forecast_date.desc()).limit(90).all() # 获取实际的现金余额KPI值 cash_kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == KPI_CODES["cash_balance"], KPIDefinition.entity_id == entity_id, ).first() if not cash_kpi or not forecasts: return [] actuals = db.query(KPIValue).filter( KPIValue.kpi_id == cash_kpi.id, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.period.desc()).limit(12).all() actual_map = {} for a in actuals: try: # period like "2026-07" -> month approx actual_map[a.period] = a.actual_value except: pass # 按月汇总预测值和实际值，计算准确率 from collections import defaultdict monthly_forecast = defaultdict(list) for f in forecasts: month_key = f.forecast_date.strftime("%Y-%m") monthly_forecast[month_key].append(f.predicted_cash) results = [] for month, f_vals in sorted(monthly_forecast.items()): if month in actual_map: f_avg = sum(f_vals) / len(f_vals) a_val = actual_map[month] mae = abs(f_avg - a_val) mape = abs((f_avg - a_val) / max(abs(a_val), 1)) * 100 results.append({ "period": month, "forecast_value": round(f_avg, 2), "actual_value": round(a_val, 2), "mae": round(mae, 2), "mape": round(mape, 2), }) return results def generate_scenario_suggestion(alert_type: str, kpi_name: str, extra: dict = None) -> dict: """根据预警类型生成情景建议""" extra = extra or {} suggestions = { "cash_low": { "title": "现金流紧张缓解方案", "description": f"现金余额低于阈值，建议加快应收账款催收、控制支出、评估短期融资。", "actions": [ f"催收大额应收账款（预计回款{extra.get('expected_receivables', '待定')}万元）", "暂停非紧急采购和资本性支出", "与供应商协商延长账期", "评估银行短期授信额度", ], "priority": "high", }, "cash_critical": { "title": "现金流危机应对方案", "description": f"现金余额接近断流，需立即采取紧急措施。", "actions": [ "立即催收所有到期应收账款", "暂停所有非必要支出", "紧急联系银行安排短期贷款", "评估资产变现可能性", ], "priority": "high", }, "cost_high": { "title": "成本管控优化方案", "description": f"成本率异常偏高，建议进行成本结构分析和优化。", "actions": [ "逐项分析成本构成，识别异常项", "与供应商重新谈判采购价格", "评估流程优化降本空间", "建立费用审批红线上限", ], "priority": "medium", }, "revenue_drop": { "title": "收入下滑应对方案", "description": f"收入出现下滑趋势，建议分析原因并制定恢复计划。", "actions": [ "分析收入下滑原因（客户流失/价格战/需求变化）", "制定客户留存和挽回计划", "评估新产品/新市场机会", "优化销售激励政策", ], "priority": "high", }, } sug = suggestions.get(alert_type, { "title": "改善建议", "description": "根据预警情况制定改善措施。", "actions": ["分析预警原因", "制定改善计划", "跟踪执行效果"], "priority": "medium", }) if extra: sug["extra"] = extra return sug # ══════════════════════════════════════════════════════════════ # 资金缺口预测 — 趋势引擎 + 收付款计划叠加 (资金管理智能体) # ══════════════════════════════════════════════════════════════ def get_cash_plans(db: Session, entity_id: int, start_date: datetime, end_date: datetime) -> list: """获取指定日期范围内的待执行收付款计划""" from app.models import CashPlan return db.query(CashPlan).filter( CashPlan.entity_id == entity_id, CashPlan.status == "pending", CashPlan.plan_date >= start_date, CashPlan.plan_date <= end_date, ).order_by(CashPlan.plan_date.asc()).all() def _budget_monthly_ocf(db: Session, entity_id: int) -> Optional[float]: """获取本月经营现金流预算（万元/月），用于校准预测基线""" from app.models import BudgetPlan kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"]) if not kpi: return None month_key = datetime.now().strftime("%Y-%m") bp = db.query(BudgetPlan).filter( BudgetPlan.kpi_id == kpi.id, BudgetPlan.period == month_key, BudgetPlan.status == "active", ).order_by(BudgetPlan.version.desc()).first() if bp and bp.budget_value is not None: return float(bp.budget_value) return None def forecast_cash_flow_with_plans( entity_id: int, db: Session, days: int = 30, current_cash: Optional[float] = None, warning_line: float = DEFAULT_CASH_WARNING, critical_line: float = DEFAULT_CASH_CRITICAL, include_plans: bool = True, ) -> dict: """ 资金缺口预测 — 在趋势预测基础上叠加收付款计划： 1. 趋势引擎生成基线预测 2. 叠加 cash_plans 的应收(收) / 应付(付) 3. 识别资金缺口日期（余额 < 警戒线） 4. 生成缺口前3天预警点 """ from collections import defaultdict from app.models import CashPlan base = forecast_cash_flow(entity_id, db, days, current_cash) base_cash = base["base_cash"] # 预算校准：本月经营现金流预算优先作为基线（万元/月） budget_ocf = _budget_monthly_ocf(db, entity_id) if budget_ocf is not None: base["trends"]["budget_monthly_ocf"] = round(budget_ocf, 2) # ── 收付款计划加载 ── today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) end = today + timedelta(days=days) plan_map = defaultdict(lambda: {"in": 0.0, "out": 0.0, "items": []}) plans = [] if include_plans: plans = get_cash_plans(db, entity_id, today, end + timedelta(days=1)) for p in plans: dkey = p.plan_date.strftime("%Y-%m-%d") item = { "id": p.id, "amount": round(p.amount, 2), "counterparty": p.counterparty or "", "description": p.description or "", } if p.plan_type == "receive": plan_map[dkey]["in"] += p.amount plan_map[dkey]["items"].append({"type": "receive", **item}) else: plan_map[dkey]["out"] += p.amount plan_map[dkey]["items"].append({"type": "pay", **item}) # ── 逐日重算余额 ── forecast = [] cash = base_cash prev_predicted = base_cash for i, f in enumerate(base["forecast"]): dkey = f["date"] # 趋势日净变化（与上一天预测值的差） trend_delta = f["predicted_cash"] - prev_predicted prev_predicted = f["predicted_cash"] pin = round(plan_map[dkey]["in"], 2) pout = round(plan_map[dkey]["out"], 2) # 预算校准：有月度预算时用预算日均替代纯趋势增量 if budget_ocf is not None: trend_delta = budget_ocf / 30.0 if f["day_offset"] % 7 in (5, 6): trend_delta *= 0.5 # 周末减半 if dkey[-2:] >= "25": trend_delta *= 1.3 # 月底回款高峰 cash = round(cash + trend_delta + pin - pout, 2) net_flow = round(trend_delta + pin - pout, 2) if cash < critical_line: status = "red" elif cash < warning_line: status = "yellow" else: status = "green" forecast.append({ "date": dkey, "day_offset": f["day_offset"], "predicted_cash": cash, "planned_in": pin, "planned_out": pout, "trend_delta": round(trend_delta, 2), "net_flow": net_flow, "lower_bound": round(max(cash - abs(net_flow) * 0.5 - 0.5, 0), 2), "upper_bound": round(cash + abs(net_flow) * 0.5 + 0.5, 2), "alert_status": status, "gap": cash < warning_line, "plans": plan_map[dkey]["items"], }) # ── 资金缺口日期 ── gap_dates = [ {"date": f["date"], "predicted_cash": f["predicted_cash"], "gap_amount": round(warning_line - f["predicted_cash"], 2), "level": "red" if f["predicted_cash"] < critical_line else "yellow"} for f in forecast if f["gap"] ] # ── 缺口前3天预警（每个连续缺口区间只预警一次，取区间首日） ── pre_alerts = [] prev_was_gap = False for idx, f in enumerate(forecast): is_gap = f["gap"] gap_run_start = is_gap and not prev_was_gap prev_was_gap = is_gap if not gap_run_start: continue # 找到该连续缺口区间的最后一天及区间内最低余额（最严重时点） run_end = forecast[idx] run_min = forecast[idx]["predicted_cash"] run_min_date = forecast[idx]["date"] for j in range(idx + 1, len(forecast)): if forecast[j]["gap"]: run_end = forecast[j] if forecast[j]["predicted_cash"] < run_min: run_min = forecast[j]["predicted_cash"] run_min_date = forecast[j]["date"] else: break for lead in (3, 1): # 缺口前3天(主要)、前1天(紧急) pre_idx = idx - lead if pre_idx < 0: continue pf = forecast[pre_idx] if pf["alert_status"] == "green" or lead == 3: pre_alerts.append({ "alert_date": pf["date"], "alert_offset": pf["day_offset"], "gap_date": f["date"], "gap_cash": f["predicted_cash"], "gap_amount": round(warning_line - f["predicted_cash"], 2), "worst_cash": round(run_min, 2), "worst_date": run_min_date, "lead_days": lead, "level": "red" if run_min < critical_line else "yellow", }) break # 到期未收款（逾期） 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() # 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 min_date = next((f["date"] for f in forecast if f["predicted_cash"] == min_cash), "") suggestions = list(base.get("suggestions", [])) if gap_dates: first_gap = gap_dates[0] if first_gap["level"] == "red": suggestions.insert(0, { "type": "critical", "message": f"预计{first_gap['date']}现金余额降至{first_gap['predicted_cash']:.1f}万，低于警戒线{warning_line:.0f}万，存在资金断流风险", "actions": ["立即催收大额应收账款", "暂停非必要支出", "准备短期融资安排"], }) else: suggestions.insert(0, { "type": "warning", "message": f"预计{first_gap['date']}现金余额降至{first_gap['predicted_cash']:.1f}万，低于警戒线{warning_line:.0f}万", "actions": ["加快应收账款回款", "控制采购付款节奏", "评估短期现金流压力"], }) 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": msg, "actions": ["逐笔催收到期应收账款", "评估客户信用风险"], }) return { "entity_id": entity_id, "base_cash": round(base_cash, 2), "days": days, "warning_line": warning_line, "critical_line": critical_line, "budget_monthly_ocf": budget_ocf, "forecast": forecast, "gap_dates": gap_dates, "pre_alerts": pre_alerts, "min_cash": round(min_cash, 2), "min_cash_date": min_date, "trends": base.get("trends", {}), "suggestions": suggestions, "summary": { "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), # 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"], }, } def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: """资金预警 — 缺口前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，不再逐户生成伪"到期未收款"） R1（裁定 2026-09-19）：demo 演示账套不产生业务预警 → 直接返回 skipped。 """ import json as _json from app.models import KPIAlert, CashPlan, KPIDefinition from app.utils.alert_entity_scope import is_excluded_entity if is_excluded_entity(db, entity_id): logger.info("资金预警检查跳过（demo 账套 R1）: entity_id=%s", entity_id) return {"entity_id": entity_id, "skipped": True, "reason": "R1_demo_entity_excluded", "new_alerts": [], "refreshed": []} 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（F2）：现金KPI → 经营现金流KPI → 该实体 finance 维度首个（全部 ORDER BY id，确定性） 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"], KPI_CODE_SUFFIXES["operating_cash_flow"]) or \ find_fallback_kpi(db, entity_id) 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"]: level = pre["level"] msg = (f"【资金缺口预警】预计{pre['gap_date']}现金余额降至{pre['gap_cash']:.1f}万" f"（低于警戒线{result['warning_line']:.0f}万，缺口{pre['gap_amount']:.1f}万），" f"请于{pre['alert_date']}前（提前{pre['lead_days']}天）安排资金") if _exists(msg): continue alert = KPIAlert( entity_id=entity_id, # F1: 此前未传 → 列默认值 1 kpi_id=kpi.id, alert_level=level, alert_message=msg[:500], alert_type="forecast", status="pending", suggestion=_json.dumps({ "actions": ["加快应收账款回款", "控制付款节奏", "评估短期融资"], "gap_date": pre["gap_date"], "gap_amount": pre["gap_amount"], }, ensure_ascii=False), ) db.add(alert) new_alerts.append({"type": "gap_forecast", "level": level, "message": msg}) logger.info(f"资金缺口预警: {msg}") # ── 2. 到期未收款提醒 ── if ar_kpi: today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) overdue = 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() # F4: 余额快照批次（plan_date = 导入基准日，非真实到期日）单独处理 # 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( KPIAlert.alert_type == "cash_plan", KPIAlert.status.in_(["pending", "processing"]), ).all() by_plan = {} by_snapshot = {} for ea in existing_alerts: es = _to_dict(ea.suggestion) if es.get("plan_id") is not None: by_plan.setdefault(es["plan_id"], ea) if es.get("snapshot_source"): by_snapshot.setdefault(es["snapshot_source"], ea) 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=level, alert_message=msg[:500], alert_type="cash_plan", status="pending", 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) 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": level, "message": msg}) logger.info(f"到期未收款提醒: {msg}") db.commit() return { "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"], }