sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 资金管理强化—缺口预测+收付款计划+预警

Browse Source
This commit is contained in:
Hermes CI Fix
2026-08-04 14:53:14 +08:00
parent 555d3b621a
commit 8e277de3de
8 changed files with 1351 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/utils/cash_forecast_engine.py
+406 -20
View File
@@ -1,18 +1,21 @@
"""现金流预测引擎 — 根据历史KPI数据推算未来30天现金流"""
import logging
from datetime import datetime, timedelta
from typing import Optional
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编码映射
# 历史KPI编码映射（含候选编码，兼容F_*标准编码）
KPI_CODES = {
"operating_cash_flow": "CASH_FLOW_001", # 经营现金流
"receivables": "AR_001", # 应收账款
@@ -20,14 +23,32 @@ KPI_CODES = {
"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"],
}
def find_kpi(db: Session, entity_id: int, codes: list) -> Optional["KPIDefinition"]:
"""按候选编码列表查找KPI，返回第一个命中的"""
from app.models import KPIDefinition
for code in codes:
kpi = db.query(KPIDefinition).filter(
KPIDefinition.kpi_code == code,
KPIDefinition.entity_id == entity_id,
).first()
if kpi:
return kpi
return None
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 = db.query(KPIDefinition).filter(
KPIDefinition.kpi_code == kpi_code,
KPIDefinition.entity_id == entity_id,
).first()
kpi = find_kpi(db, entity_id, [kpi_code])
if not kpi:
return []
values = db.query(KPIValue).filter(
@@ -51,6 +72,47 @@ def calc_trend(values: list) -> float:
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
kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["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,
@@ -69,31 +131,26 @@ def forecast_cash_flow(
# 获取当前现金余额
if current_cash is None:
cash_kpi = db.query(KPIDefinition).filter(
KPIDefinition.kpi_code == KPI_CODES["cash_balance"],
KPIDefinition.entity_id == entity_id,
).first()
if cash_kpi:
latest_cash = db.query(KPIValue).filter(
KPIValue.kpi_id == cash_kpi.id,
KPIValue.actual_value.isnot(None),
).order_by(KPIValue.period.desc()).first()
base_cash = latest_cash.actual_value if latest_cash else 30.0
else:
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_CODES["operating_cash_flow"], db)
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_CODES["receivables"], db)
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_CODES["payables"], db)
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)
# 计算日均现金变化
@@ -320,3 +377,332 @@ def generate_scenario_suggestion(alert_type: str, kpi_name: str, extra: dict = N
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_receives = 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(
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()
# 整体结论
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_receives:
total_overdue = sum(p.amount for p in overdue_receives)
suggestions.append({
"type": "warning",
"message": f"有{len(overdue_receives)}笔应收款到期未收，合计{total_overdue:.1f}万",
"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),
"overdue_receive_count": len(overdue_receives),
"overdue_receive_amount": round(sum(p.amount for p in overdue_receives), 2),
"upcoming_7d_count": len(upcoming_7d),
},
}
def check_cash_alerts(db: Session, entity_id: int = 1) -> dict:
"""资金预警 — 缺口前3天预警 + 到期未收款提醒，写入预警中心(kpi_alerts)"""
import json as _json
from app.models import KPIAlert, CashPlan
result = forecast_cash_flow_with_plans(entity_id, db, days=30)
new_alerts = []
# 兜底KPI：现金KPI → 经营现金流KPI → 该实体任意KPI
kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"]) or \
find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"]) or \
db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).first()
ar_kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["receivables"]) or kpi
def _exists(msg: str) -> bool:
return db.query(KPIAlert).filter(
KPIAlert.alert_message == msg,
KPIAlert.status.in_(["pending", "processing"]),
).first() is not None
# ── 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(
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()
for p in overdue:
days_late = (today - p.plan_date).days
msg = (f"【到期未收款】应收款{p.counterparty or '客户'} {p.amount:.1f}万 "
f"原计划{p.plan_date.strftime('%Y-%m-%d')}到期，已逾期{days_late}天未收回")
if _exists(msg):
continue
alert = KPIAlert(
kpi_id=ar_kpi.id,
alert_level="red" if days_late >= 7 else "yellow",
alert_message=msg[:500],
alert_type="cash_plan",
status="pending",
suggestion=_json.dumps({
"actions": ["联系客户催收", "评估坏账风险", "调整信用政策"],
"plan_id": p.id,
"days_late": days_late,
}, ensure_ascii=False),
)
db.add(alert)
new_alerts.append({"type": "overdue_receive", "level": alert.alert_level, "message": msg})
logger.info(f"到期未收款提醒: {msg}")
db.commit()
return {
"entity_id": entity_id,
"new_alerts": len(new_alerts),
"alerts": new_alerts,
"gap_dates": result["gap_dates"],
"pre_alerts": result["pre_alerts"],
"summary": result["summary"],
}