From 885cfddbff886bc0ae9a0e62c194556f59c8c3de Mon Sep 17 00:00:00 2001 From: wecom-fullstack Date: Sat, 17 Oct 2026 23:33:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(warning-p0a):=20=E8=B5=84=E9=87=91?= =?UTF-8?q?=E7=BC=BA=E5=8F=A3=E9=98=88=E5=80=BC=E5=A4=96=E7=BD=AE=20+=20?= =?UTF-8?q?=E5=A4=8D=E7=AE=97=E5=AF=B9=E6=8B=8D=20+=20=E8=AE=B2=E4=BA=BA?= =?UTF-8?q?=E8=AF=9D(=E6=95=B0=E5=AD=97=E5=AE=88=E6=81=92)=20+=20=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=B0=B1=E7=BB=AA=E6=A0=B8=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0-A（任务 fullstack-20261017-warning-capability-build-p0）： - cash_thresholds.py：阈值迁 per-entity（复用 alert_rules.rule_type=cash_gap_threshold， 默认 20/10 万不动；无配置/非法/异常一律回落默认，fail-safe 不改预警行为） - cash_forecast_engine.py：阈值改从配置读取，响应新增 threshold_source 明写来源 - verify_cash_gap_replay.py：同输入双跑逐字段 diff=0 + 手算 3/3 + 端点逐字段对拍(--live) - cash_gap_narrative.py：复用 ai_brief.call_deepseek，只写原因/建议； 断言生成文本数字集合 ⊆ 规则层数字集合，违例即不带文案上线 - audit_warning_data_readiness.py：G0 只读核验（cma_reader SELECT only）， 行数/期间/枚举 + 数据质量缺口（期初全 0、负期末、单批次、无员工维度表） 未 push（push 即部署，须项目侧书面放行）；未改生产库/未动 cron。 --- backend/app/api/alert_rules.py | 24 + backend/app/utils/cash_forecast_engine.py | 55 ++- backend/app/utils/cash_thresholds.py | 133 ++++++ .../scripts/audit_warning_data_readiness.py | 319 ++++++++++++++ backend/scripts/cash_gap_narrative.py | 408 +++++++++++++++++ backend/scripts/verify_cash_gap_replay.py | 409 ++++++++++++++++++ 6 files changed, 1335 insertions(+), 13 deletions(-) create mode 100644 backend/app/utils/cash_thresholds.py create mode 100644 backend/scripts/audit_warning_data_readiness.py create mode 100644 backend/scripts/cash_gap_narrative.py create mode 100644 backend/scripts/verify_cash_gap_replay.py diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index f3fd47d6..a17ca2bc 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -21,6 +21,11 @@ from app.utils.alert_entity_scope import guard_alert_period, is_excluded_entity # 「实测区 / 预估区」分区口径（财务裁定 2026-10-02）：**预估行禁作预警依据**—— # 本文件的取数读路径一律加 actual_only_clause()，动态阈值白名单去掉 estimated。 from app.utils.kpi_value_scope import actual_only_clause +# G1（2026-10-17）：资金缺口阈值的单一事实源（默认 20/10 万 + 参数校验） +from app.utils.cash_thresholds import ( + CASH_GAP_RULE_TYPE, + validate_params as validate_cash_gap_params, +) logger = logging.getLogger("alert_rules") @@ -32,6 +37,7 @@ VALID_RULE_TYPES = ( "trend_down", # 环比下降 "forecast_deviation", # 预测偏差（升级2b 2026-08-25） "customer_concentration", # 客户集中度（P2 2026-09-10） + "cash_gap_threshold", # 资金缺口阈值配置（G1 2026-10-17，配置行，不进预警生成链） ) # ============================================================ @@ -130,6 +136,12 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re raise HTTPException(404, "KPI不存在") if rule_type not in VALID_RULE_TYPES: # 升级2b: 预测偏差 / P2-20260910: 客户集中度 raise HTTPException(400, f"不支持的规则类型: {rule_type}") + # G1（2026-10-17）：资金缺口阈值为引擎级参数，入库前校验数值（非法配置一律拒收， + # 防「配置写坏 → 预警静默回落默认值」的假生效）。 + if rule_type == CASH_GAP_RULE_TYPE: + err = validate_cash_gap_params(data.get("params")) + if err: + raise HTTPException(400, err) rule = AlertRule( entity_id=entity_id, @@ -162,6 +174,12 @@ def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db), e for field in ("rule_type", "trigger_on", "enabled", "params"): if field in data: setattr(rule, field, data[field]) + # G1（2026-10-17）：改完仍是资金阈值配置行 → 参数必须合法（拒收非法值） + if rule.rule_type == CASH_GAP_RULE_TYPE: + err = validate_cash_gap_params(rule.params) + if err: + db.rollback() + raise HTTPException(400, err) db.commit() db.refresh(rule) return {"data": {c.name: getattr(rule, c.name) for c in AlertRule.__table__.columns}} @@ -189,6 +207,8 @@ def batch_create_rules(data: dict, db: Session = Depends(get_db), entity_id: int rule_type = rule_data.get("rule_type", "static") if rule_type not in VALID_RULE_TYPES: continue + if rule_type == CASH_GAP_RULE_TYPE and validate_cash_gap_params(rule_data.get("params")): + continue kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not kpi or kpi.entity_id != entity_id: continue @@ -283,6 +303,10 @@ def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends for rule in rules: try: + # G1（2026-10-17）：cash_gap_threshold 是资金引擎的参数配置行，不是 KPI 预警规则， + # 不进 KPIValue 比价链（否则未来若绑定到有值的 KPI 会误生成预警）。 + if rule.rule_type == CASH_GAP_RULE_TYPE: + continue kpi = kpi_cache.get(rule.kpi_id) if kpi is None: kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first() diff --git a/backend/app/utils/cash_forecast_engine.py b/backend/app/utils/cash_forecast_engine.py index e56c3c6c..e21929bf 100644 --- a/backend/app/utils/cash_forecast_engine.py +++ b/backend/app/utils/cash_forecast_engine.py @@ -13,9 +13,17 @@ if TYPE_CHECKING: logger = logging.getLogger("cma.cash_forecast") -# 默认现金阈值（万元） -DEFAULT_CASH_WARNING = 20.0 # 黄灯 — 低于20万 -DEFAULT_CASH_CRITICAL = 10.0 # 红灯 — 低于10万 +# 默认现金阈值（万元）—— 单一事实源迁至 app/utils/cash_thresholds.py（G1，2026-10-17） +# 此处保留同名常量以兼容既有 import（app/api/cash.py 等）；数值未变：黄 20 / 红 10。 +from app.utils.cash_thresholds import ( # noqa: E402 + DEFAULT_CRITICAL as _CFG_DEFAULT_CRITICAL, + DEFAULT_WARNING as _CFG_DEFAULT_WARNING, + get_cash_gap_thresholds, + describe_source as describe_cash_threshold_source, +) + +DEFAULT_CASH_WARNING = _CFG_DEFAULT_WARNING # 黄灯 — 低于20万 +DEFAULT_CASH_CRITICAL = _CFG_DEFAULT_CRITICAL # 红灯 — 低于10万 # 历史KPI编码映射（含候选编码，兼容F_*标准编码） KPI_CODES = { @@ -306,17 +314,27 @@ def forecast_cash_flow( db: Session, days: int = 30, current_cash: Optional[float] = None, + warning_line: Optional[float] = None, + critical_line: Optional[float] = None, ) -> dict: """ 预测未来30天现金流 - + 算法： 1. 获取历史经营现金流、应收、应付趋势 2. 推算每日现金流入/流出 3. 生成每日预测值+置信区间 + + 阈值（G1，2026-10-17）：warning_line / critical_line 传 None 时取本账套配置 + （alert_rules: cash_gap_threshold），未配置则默认 20/10 万；显式传入优先（便于复算对拍）。 """ from app.models import KPIDefinition, KPIValue + if warning_line is None or critical_line is None: + cfg_warning, cfg_critical = get_cash_gap_thresholds(db, entity_id) + warning_line = cfg_warning if warning_line is None else warning_line + critical_line = cfg_critical if critical_line is None else critical_line + # 获取当前现金余额 if current_cash is None: base_cash = get_current_cash_balance(db, entity_id) @@ -375,9 +393,9 @@ def forecast_cash_flow( upper = cash + std * 0.5 # 预警状态 - if cash < DEFAULT_CASH_CRITICAL: + if cash < critical_line: status = "red" - elif cash < DEFAULT_CASH_WARNING: + elif cash < warning_line: status = "yellow" else: status = "green" @@ -396,20 +414,20 @@ def forecast_cash_flow( min_date = next(f["date"] for f in forecast if f["predicted_cash"] == min_cash) suggestions = [] - if min_cash < DEFAULT_CASH_CRITICAL: + if min_cash < critical_line: suggestions.append({ "type": "critical", - "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于警戒线{DEFAULT_CASH_CRITICAL}万", + "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于警戒线{critical_line}万", "actions": [ "立即催收大额应收账款", "暂停非必要支出", "准备短期融资安排", ] }) - elif min_cash < DEFAULT_CASH_WARNING: + elif min_cash < warning_line: suggestions.append({ "type": "warning", - "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于关注线{DEFAULT_CASH_WARNING}万", + "message": f"预计{min_date}现金余额降至{min_cash:.1f}万，低于关注线{warning_line}万", "actions": [ "加快应收账款回款", "控制采购付款节奏", @@ -606,8 +624,8 @@ def forecast_cash_flow_with_plans( db: Session, days: int = 30, current_cash: Optional[float] = None, - warning_line: float = DEFAULT_CASH_WARNING, - critical_line: float = DEFAULT_CASH_CRITICAL, + warning_line: Optional[float] = None, + critical_line: Optional[float] = None, include_plans: bool = True, ) -> dict: """ @@ -616,11 +634,21 @@ def forecast_cash_flow_with_plans( 2. 叠加 cash_plans 的应收(收) / 应付(付) 3. 识别资金缺口日期（余额 < 警戒线） 4. 生成缺口前3天预警点 + + 阈值来源（G1，2026-10-17）：None → 本账套配置（alert_rules: cash_gap_threshold） + → 未配置时默认 20/10 万；显式传入值优先（复算对拍用）。响应内 `threshold_source` + 明写本次取值来源，便于页面标注与回执取证。 """ from collections import defaultdict from app.models import CashPlan - base = forecast_cash_flow(entity_id, db, days, current_cash) + if warning_line is None or critical_line is None: + cfg_warning, cfg_critical = get_cash_gap_thresholds(db, entity_id) + warning_line = cfg_warning if warning_line is None else warning_line + critical_line = cfg_critical if critical_line is None else critical_line + + base = forecast_cash_flow(entity_id, db, days, current_cash, + warning_line=warning_line, critical_line=critical_line) base_cash = base["base_cash"] # 预算校准：本月经营现金流预算优先作为基线（万元/月） @@ -805,6 +833,7 @@ def forecast_cash_flow_with_plans( "days": days, "warning_line": warning_line, "critical_line": critical_line, + "threshold_source": describe_cash_threshold_source(db, entity_id), "budget_monthly_ocf": budget_ocf, "forecast": forecast, "gap_dates": gap_dates, diff --git a/backend/app/utils/cash_thresholds.py b/backend/app/utils/cash_thresholds.py new file mode 100644 index 00000000..88d580cf --- /dev/null +++ b/backend/app/utils/cash_thresholds.py @@ -0,0 +1,133 @@ +# -*- coding: utf-8 -*- +"""资金缺口阈值 —— per-entity 配置（复用 alert_rules 表，不新建表） + +任务：fullstack-20261017-warning-capability-build-p0 / G1（P0-A） +方案依据：`_reports/yanxue-20261017-warning-capability-build-p0.md` §G1 + 决策点 D4（建议 B） + +口径（默认值不动）： + · 黄灯（warning） = 20.0 万元 —— 与改前 `cash_forecast_engine.DEFAULT_CASH_WARNING` 一致 + · 红灯（critical）= 10.0 万元 —— 与改前 `cash_forecast_engine.DEFAULT_CASH_CRITICAL` 一致 + · 配置存在时按配置；未配置 / 配置非法 / 读取异常 → **一律回落默认值**（fail-safe，不让预警失效） + +存储形态（零 DDL，复用现有 `alert_rules`）： + entity_id = 账套 + rule_type = "cash_gap_threshold" + kpi_id = 该账套现金余额 KPI（CASH_001 / BH_CASH_BALANCE 等），满足表 NOT NULL 约束 + enabled = 1 + params = {"warning": 20.0, "critical": 10.0} # 单位：万元 + +为什么单独一个 rule_type 而不复用 static： + static 规则走 `_check_static` 与 KPIValue 比价，参数语义是 `>=90` 之类表达式； + 资金阈值是**引擎级参数**（决定缺口日判定与红黄分级），不是单条 KPI 的阈值。 + 独立 rule_type 让 `run_all_alert_checks` 的 if/elif 链自然跳过它（alert_level 保持 None）， + 不会生成预警行，也不影响现有 static/trend 规则。 +""" +from __future__ import annotations + +import logging +from typing import Any, Dict, Optional, Tuple + +logger = logging.getLogger("cma.cash_thresholds") + +#: alert_rules.rule_type 取值（配置行，不进预警生成链） +CASH_GAP_RULE_TYPE = "cash_gap_threshold" + +#: 默认阈值（万元）—— 与改前硬编码常量逐字一致 +DEFAULT_WARNING = 20.0 +DEFAULT_CRITICAL = 10.0 + +PARAM_KEYS = ("warning", "critical") + + +def _as_float(value: Any) -> Optional[float]: + try: + return float(value) + except (TypeError, ValueError): + return None + + +def validate_params(params: Any) -> Optional[str]: + """校验阈值参数；合法返回 None，非法返回中文错误信息（供 API 层直接回 400）。""" + if not isinstance(params, dict): + return "params 必须是对象，形如 {\"warning\": 20.0, \"critical\": 10.0}（单位：万元）" + warning = _as_float(params.get("warning")) + critical = _as_float(params.get("critical")) + if warning is None or critical is None: + return "params 必须同时包含数值型 warning 与 critical（单位：万元）" + if warning <= 0 or critical <= 0: + return "warning / critical 必须大于 0（单位：万元）" + if critical > warning: + return "critical（红灯线）不得高于 warning（黄灯线）" + return None + + +def parse_params(params: Any) -> Optional[Tuple[float, float]]: + """解析参数 → (warning, critical)；非法返回 None（调用方回落默认值）。""" + if validate_params(params) is not None: + return None + return float(params["warning"]), float(params["critical"]) + + +def get_cash_gap_thresholds(db, entity_id: int) -> Tuple[float, float]: + """读取账套资金缺口阈值；无配置/非法/异常 → 默认 (20.0, 10.0)。 + + · 只认 enabled=1、最新 id 的一行（同一账套多行时以最新为准，不累加） + · 任何异常都不上抛：预警不能因为配置表读失败而消失（fail-safe） + """ + try: + from app.api.alert_rules import AlertRule # 延迟导入，避免 engine ↔ api 循环 + + rule = ( + db.query(AlertRule) + .filter( + AlertRule.entity_id == int(entity_id), + AlertRule.rule_type == CASH_GAP_RULE_TYPE, + AlertRule.enabled == 1, + ) + .order_by(AlertRule.id.desc()) + .first() + ) + if rule is None: + return DEFAULT_WARNING, DEFAULT_CRITICAL + params = rule.params + if isinstance(params, str): + import json + + try: + params = json.loads(params) + except ValueError: + logger.warning("资金阈值配置 params 非合法 JSON（rule_id=%s）→ 回落默认值", rule.id) + return DEFAULT_WARNING, DEFAULT_CRITICAL + parsed = parse_params(params) + if parsed is None: + logger.warning("资金阈值配置参数非法（rule_id=%s params=%r）→ 回落默认值", rule.id, params) + return DEFAULT_WARNING, DEFAULT_CRITICAL + return parsed + except Exception as exc: # noqa: BLE001 — 配置读取失败不得使预警整体失效 + logger.warning("读取资金阈值配置失败（entity_id=%s）：%s → 回落默认值", entity_id, exc) + return DEFAULT_WARNING, DEFAULT_CRITICAL + + +def describe_source(db, entity_id: int) -> Dict[str, Any]: + """取证用：说明本次阈值来自「配置」还是「默认」，便于回执与页面标注。""" + try: + from app.api.alert_rules import AlertRule + + rule = ( + db.query(AlertRule) + .filter( + AlertRule.entity_id == int(entity_id), + AlertRule.rule_type == CASH_GAP_RULE_TYPE, + AlertRule.enabled == 1, + ) + .order_by(AlertRule.id.desc()) + .first() + ) + if rule is not None and parse_params(rule.params) is not None: + return {"source": "alert_rules", "rule_id": rule.id, + "params": rule.params, "entity_id": int(entity_id)} + except Exception as exc: # noqa: BLE001 + logger.warning("读取资金阈值来源失败：%s", exc) + return {"source": "default", "rule_id": None, + "params": {"warning": DEFAULT_WARNING, "critical": DEFAULT_CRITICAL}, + "entity_id": int(entity_id) if entity_id is not None else None} diff --git a/backend/scripts/audit_warning_data_readiness.py b/backend/scripts/audit_warning_data_readiness.py new file mode 100644 index 00000000..4389fcc4 --- /dev/null +++ b/backend/scripts/audit_warning_data_readiness.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""G0 数据就绪核验（只读）— 经营预警能力落地 P0-A + +任务键: fullstack-20261017-warning-capability-build-p0（上游 yanxue-20261017-warning-capability-build-p0） +用途: 对「资金 / 库存 / 销售 / 客户 / 员工」五类数据在 CMA 库内的就绪程度取证： + 行数（分 entity）、期间上下界、枚举完整度、缺口清单。 +纪律: + · 只读：仅 SELECT / information_schema；凭据用 cma_reader（仅 GRANT SELECT ON cma.*） + · 不改库、不注数、不建表；缺表/缺期间如实列为 gap，不用 0 或示例数填充 + · 期间上下界取数据自身字段（product_sales.period_month / cash_plans.plan_date / + *_inventory.source_batch），无期间字段的数据集明确标 period_field=null +用法: + CMA_READER_ENV=/path/to/.secrets/cma-reader.env \ + python3 backend/scripts/audit_warning_data_readiness.py [--entity 1] [--out FILE] +退出码: 0 = 核验已执行（无论是否存在数据缺口）；2 = 连接/权限失败 +""" +import argparse +import json +import os +import sys +from datetime import datetime + +import pymysql + +DEFAULT_READER_ENV = "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-reader.env" +TARGET_ENTITY_DEFAULT = 1 # 酣客 + +# 需要取证的数据集（表名 → 期间字段/断言口径） +TARGETS = [ + # 资金（G1 依赖） + {"table": "cash_plans", "period_field": "plan_date", "kind": "fact", + "enum_fields": ["plan_type", "status", "source"], "amount_fields": ["amount"]}, + {"table": "alert_rules", "period_field": None, "kind": "config", "expect_period": False, + "enum_fields": ["rule_type", "trigger_on", "enabled"], "amount_fields": []}, + {"table": "kpi_alerts", "period_field": None, "kind": "fact", "expect_period": False, + "enum_fields": ["alert_level", "status", "alert_type"], "amount_fields": []}, + # 库存（G3 依赖） + {"table": "product_inventory", "period_field": "source_batch", "kind": "agg", + "enum_fields": ["product_category", "source_batch"], "amount_fields": ["end_amount", "init_amount"]}, + {"table": "product_inventory_detail", "period_field": "source_batch", "kind": "fact", + "enum_fields": ["unit", "source_batch"], "amount_fields": ["end_amount", "init_amount"]}, + # 销售（G3 依赖） + {"table": "product_sales", "period_field": "period_month", "kind": "fact", + "enum_fields": ["unit", "period_month"], "amount_fields": ["sales_amount", "cost_amount"]}, + # 客户（G4 依赖） + {"table": "customer_profitability_snapshot", "period_field": "period", "kind": "fact", + "enum_fields": ["period", "calib_status"], "amount_fields": ["revenue", "contrib"]}, + # 智享通维度/事实表（方案 §G0：应在接收侧路由，此处核验是否已入 CMA 库） + {"table": "productinout_recent", "period_field": "None", "kind": "absent-probe", + "enum_fields": [], "amount_fields": []}, + {"table": "productinout_history", "period_field": "None", "kind": "absent-probe", + "enum_fields": [], "amount_fields": []}, + {"table": "units_dict", "period_field": None, "kind": "absent-probe", + "enum_fields": [], "amount_fields": []}, + {"table": "employee_dict", "period_field": None, "kind": "absent-probe", + "enum_fields": [], "amount_fields": []}, + {"table": "product_dict", "period_field": None, "kind": "absent-probe", + "enum_fields": [], "amount_fields": []}, +] + + +def load_env(path): + env = {} + with open(path, encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line and "=" in line and not line.startswith("#"): + k, v = line.split("=", 1) + env[k.strip()] = v.strip().strip('"').strip("'") + missing = [k for k in ("CMA_READER_DB_HOST", "CMA_READER_DB_PORT", + "CMA_READER_DB_USER", "CMA_READER_DB_PASS", + "CMA_READER_DB_NAME") if k not in env] + if missing: + raise SystemExit(f"凭据文件缺字段: {missing}") + return env + + +def table_exists(cur, name): + cur.execute("SELECT COUNT(*) FROM information_schema.tables " + "WHERE table_schema=DATABASE() AND table_name=%s", (name,)) + return cur.fetchone()[0] > 0 + + +def columns(cur, name): + cur.execute("SELECT column_name, column_type, is_nullable FROM information_schema.columns " + "WHERE table_schema=DATABASE() AND table_name=%s ORDER BY ordinal_position", (name,)) + return [{"name": c, "type": t, "nullable": (n == "YES")} for c, t, n in cur.fetchall()] + + +def scalar(cur, sql, args=None): + cur.execute(sql, args or ()) + row = cur.fetchone() + return None if row is None else row[0] + + +def inspect_table(cur, spec, entity_id): + name = spec["table"] + out = {"table": name, "kind": spec["kind"], "exists": table_exists(cur, name), + "expect_period": spec.get("expect_period", True)} + if not out["exists"]: + out["gap"] = "表不存在于 CMA 库（entity 无关）" + return out + + cols = columns(cur, name) + col_names = {c["name"] for c in cols} + out["columns"] = cols + out["row_count_total"] = int(scalar(cur, "SELECT COUNT(*) FROM `%s`" % name) or 0) + + if "entity_id" in col_names: + cur.execute("SELECT entity_id, COUNT(*) FROM `%s` GROUP BY entity_id ORDER BY entity_id" % name) + out["rows_by_entity"] = {str(e if e is not None else "NULL"): int(c) for e, c in cur.fetchall()} + out["row_count_entity"] = out["rows_by_entity"].get(str(entity_id), 0) + else: + out["rows_by_entity"] = None + out["row_count_entity"] = out["row_count_total"] + out["gap_no_entity_column"] = True + + pf = spec["period_field"] + if pf and pf in col_names and pf != "None": + where = " WHERE entity_id=%s" % int(entity_id) if "entity_id" in col_names else "" + cur.execute("SELECT MIN(`%s`), MAX(`%s`), COUNT(DISTINCT `%s`) FROM `%s`%s" + % (pf, pf, pf, name, where)) + lo, hi, distinct = cur.fetchone() + out["period_field"] = pf + out["period_bounds"] = {"min": _jsonable(lo), "max": _jsonable(hi), "distinct": int(distinct or 0)} + cur.execute("SELECT `%s`, COUNT(*) FROM `%s`%s GROUP BY `%s` ORDER BY `%s`" + % (pf, name, where, pf, pf)) + out["period_rows"] = [[_jsonable(k), int(v)] for k, v in cur.fetchall()] + else: + out["period_field"] = None + out["period_bounds"] = None + out["gap_no_period_field"] = True + + # 枚举完整度：值分布 + 空值率 + enum_detail = {} + where = " WHERE entity_id=%s" % int(entity_id) if "entity_id" in col_names else "" + denom = out["row_count_entity"] or 0 + for ef in spec["enum_fields"]: + if ef not in col_names: + enum_detail[ef] = {"missing_column": True} + continue + cur.execute("SELECT `%s`, COUNT(*) FROM `%s`%s GROUP BY `%s` ORDER BY COUNT(*) DESC LIMIT 25" + % (ef, name, where, ef)) + dist = [{"value": _jsonable(v), "rows": int(c)} for v, c in cur.fetchall()] + nulls = int(scalar(cur, "SELECT COUNT(*) FROM `%s`%s %s `%s` IS NULL" + % (name, where, "AND" if where else "WHERE", ef)) or 0) + blanks = 0 + if any(c["type"].startswith(("varchar", "char", "text")) for c in cols if c["name"] == ef): + blanks = int(scalar(cur, "SELECT COUNT(*) FROM `%s`%s %s `%s`=''" + % (name, where, "AND" if where else "WHERE", ef)) or 0) + enum_detail[ef] = {"distinct": len(dist), "distribution": dist, + "null_rows": nulls, "blank_rows": blanks, + "null_rate": round(nulls / denom, 4) if denom else None} + out["enum_completeness"] = enum_detail + + # 金额字段空值与合计（仅 entity 范围） + amt = {} + for af in spec["amount_fields"]: + if af in col_names: + cur.execute("SELECT COUNT(*), COALESCE(SUM(`%s`),0) FROM `%s`%s %s `%s` IS NULL" + % (af, name, where, "AND" if where else "WHERE", af)) + nulls, _ = cur.fetchone() + total = scalar(cur, "SELECT COALESCE(SUM(`%s`),0) FROM `%s`%s" % (af, name, where)) + amt[af] = {"null_rows": int(nulls or 0), "sum": _jsonable(total)} + out["amount_fields"] = amt + return out + + +def data_quality_checks(cur, entity_id: int) -> list: + """数据质量取证（只读）：不只看「有没有表/几行」，还要看这批数据能不能支撑看板口径。 + + 每条 finding 都带 SQL 结果，供回执逐条引用；发现问题不等于改库，只如实登记。 + """ + findings = [] + has = table_exists(cur, "product_inventory") + if has: + cur.execute( + "SELECT COUNT(*), COALESCE(SUM(init_qty),0), COALESCE(SUM(init_amount),0), " + "SUM(CASE WHEN end_qty < 0 THEN 1 ELSE 0 END), COUNT(DISTINCT source_batch) " + "FROM product_inventory WHERE entity_id=%s", (entity_id,)) + rows, init_qty, init_amt, neg_rows, batches = cur.fetchone() + if float(init_qty or 0) == 0 and float(init_amt or 0) == 0: + findings.append({ + "table": "product_inventory", "type": "init_all_zero", + "detail": "entity_%s 全部行 init_qty/init_amount = 0 ⇒ 期初口径不可用，" + "看板期初列必须显示『数据未就绪』" % entity_id, + "evidence": {"rows": int(rows), "sum_init_qty": float(init_qty or 0), + "sum_init_amount": float(init_amt or 0)}}) + if int(neg_rows or 0) > 0: + findings.append({ + "table": "product_inventory", "type": "negative_end_qty", + "detail": "entity_%s 有负期末数量行 ⇒ 该批为「当期进销」而非「期初+期末」，" + "期末/周转需标注口径来源" % entity_id, + "evidence": {"rows_with_negative_end_qty": int(neg_rows)}}) + if int(batches or 0) == 1: + findings.append({ + "table": "product_inventory", "type": "single_batch_only", + "detail": "仅 1 个 source_batch ⇒ 库存周转只能算单期快照，不能做多期环比/呆滞趋势", + "evidence": {"distinct_source_batch": int(batches)}}) + + for kw in ("employ", "staff", "salesman"): + cur.execute("SELECT table_name FROM information_schema.tables " + "WHERE table_schema=DATABASE() AND table_name LIKE %s", ("%" + kw + "%",)) + hits = [r[0] for r in cur.fetchall()] + if hits: + findings.append({"table": hits[0], "type": "employee_dim_suspected", + "detail": "库内存在疑似员工维度表：%s（需逐表核口径）" % ",".join(hits)}) + cur.execute("SELECT table_name FROM information_schema.tables " + "WHERE table_schema=DATABASE() AND (table_name LIKE '%%employ%%' " + "OR table_name LIKE '%%staff%%' OR table_name LIKE '%%salesman%%')") + if not cur.fetchall(): + findings.append({ + "table": "(none)", "type": "no_employee_dimension_table", + "detail": "CMA 库内无任何员工/业务员维度表 ⇒ 员工绩效明细（G4）当前无数据源，" + "不只是口径未定（D2），需先解决 employee_dict 落库"}) + return findings + + +def _jsonable(v): + if v is None: + return None + if isinstance(v, (int, float, str)): + return v + if hasattr(v, "isoformat"): + return v.isoformat() + from decimal import Decimal + if isinstance(v, Decimal): + return float(v) + return str(v) + + +def build_gaps(reports, entity_id): + gaps, flags = [], [] + for r in reports: + if not r.get("exists"): + gaps.append({"table": r["table"], "type": "missing_table", + "detail": "CMA 库无此表（zxt 接收侧仅路由到通道，未落地 CMA）"}) + continue + if r.get("row_count_entity", 0) == 0: + gaps.append({"table": r["table"], "type": "no_rows_for_entity", + "detail": "entity_%s 无行" % entity_id}) + for f, v in (r.get("enum_completeness") or {}).items(): + if v.get("missing_column"): + gaps.append({"table": r["table"], "type": "missing_column", "detail": f}) + elif v.get("null_rate") == 1.0: + flags.append({"table": r["table"], "field": f, "type": "enum_all_null", + "detail": "该字段全为空 → 不能作为看板分组/口径依据"}) + if r.get("gap_no_period_field") and r.get("expect_period", True): + gaps.append({"table": r["table"], "type": "no_period_field", + "detail": "无期间字段 → 无法声明覆盖区间（页面必须标『数据未就绪』）"}) + return gaps, flags + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--entity", type=int, default=TARGET_ENTITY_DEFAULT) + ap.add_argument("--out", default="") + args = ap.parse_args() + + reader_env = os.environ.get("CMA_READER_ENV", DEFAULT_READER_ENV) + env = load_env(reader_env) + try: + conn = pymysql.connect(host=env["CMA_READER_DB_HOST"], port=int(env["CMA_READER_DB_PORT"]), + user=env["CMA_READER_DB_USER"], password=env["CMA_READER_DB_PASS"], + database=env["CMA_READER_DB_NAME"], charset="utf8mb4") + except Exception as exc: # noqa: BLE001 + print(json.dumps({"ok": False, "error": "connect_failed: %s" % str(exc)[:200]}, + ensure_ascii=False), file=sys.stderr) + return 2 + + reports, db_meta = [], {} + try: + cur = conn.cursor() + cur.execute("SELECT CURRENT_USER(), DATABASE(), VERSION(), NOW()") + u, db, ver, now = cur.fetchone() or (None, None, None, None) + db_meta = {"current_user": u, "database": db, "version": ver, + "server_now": _jsonable(now), "readonly_account": True, + "reader_env_file": os.path.basename(reader_env)} + for spec in TARGETS: + reports.append(inspect_table(cur, spec, args.entity)) + gaps, flags = build_gaps(reports, args.entity) + data_quality = data_quality_checks(cur, args.entity) + for f in data_quality: + gaps.append({"table": f["table"], "type": f["type"], "detail": f["detail"], + "evidence": f.get("evidence")}) + cur.close() + finally: + conn.close() + + payload = { + "task": "fullstack-20261017-warning-capability-build-p0", + "stage": "G0 data readiness (read-only)", + "generated_at": datetime.now().isoformat(timespec="seconds"), + "entity_id": args.entity, + "db": db_meta, + "datasets": reports, + "gaps": gaps, + "gap_count": len(gaps), + "data_quality": data_quality, + "flags": flags, + "flag_count": len(flags), + } + text = json.dumps(payload, ensure_ascii=False, indent=2, default=str) + if args.out: + os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) + with open(args.out, "w", encoding="utf-8") as fh: + fh.write(text + "\n") + print("evidence written: %s (%d bytes)" % (args.out, len(text))) + print("gap_count=%d" % len(gaps)) + for g in gaps: + print("GAP %-32s %-22s %s" % (g["table"], g["type"], g["detail"])) + else: + print(text) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/cash_gap_narrative.py b/backend/scripts/cash_gap_narrative.py new file mode 100644 index 00000000..87c6c25f --- /dev/null +++ b/backend/scripts/cash_gap_narrative.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""资金缺口「讲人话」— G2 验收件（P0-A） + +任务：fullstack-20261017-warning-capability-build-p0（上游 yanxue-20261017-warning-capability-build-p0） +方案：`_reports/yanxue-20261017-warning-capability-build-p0.md` §G1/G2 + §四「数字零错误（可复算）」 + +纪律（本脚本存在的唯一理由）： + 1. **数字由规则层算死**：所有数字一律来自 `forecast_cash_flow_with_plans` 的返回值， + 脚本把这份 payload 作为「白名单」喂给 LLM，并**事后断言**生成文本里的数字集合 + ⊆ 规则层数字集合（数值相等，非字符串相等；`20` 与 `20.0` 视为同一数字）。 + 2. **LLM 只写文字**：prompt 明确禁止四舍五入、估算、派生、新增数字、自造日期。 + 3. **断言失败即不带文案上线**：首次违例 → 收紧 prompt 重试一次；仍违例 → 退出码 1、 + `assertion.pass=false`、`narrative` 置空（调用方不得把文案推给用户）。 + +用法： + # 规则层取数（只读，走 cma_reader）→ LLM 出文案 → 数字守恒断言 + python3 backend/scripts/cash_gap_narrative.py --entity 1 --days 30 + # 只跑断言自测（不连库、不连 LLM，离线可判） + python3 backend/scripts/cash_gap_narrative.py --selftest + # 用现成 payload 出文案（离线复算/回放） + python3 backend/scripts/cash_gap_narrative.py --payload-json p.json +退出码：0 = 断言通过且文案完整；1 = 数字违例 / 段落缺失 / 取数或调用失败 +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import re +import sys +from datetime import datetime +from decimal import Decimal, InvalidOperation + +BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +for _p in (BACKEND_DIR, SCRIPTS_DIR): + if _p not in sys.path: + sys.path.insert(0, _p) + +READER_ENV_DEFAULT = "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-reader.env" + +#: DeepSeek Key 来源候选（按序取第一个「非空」；脚本只记来源路径，不打印 Key） +#: 生产 CMA backend/.env 中的 Key 已被供应商判为 invalid（2026-10-17 实测 401）， +#: 故 Bot 侧优先用 Hermes 通道 Key；生产 Key 轮换另走派单，本脚本不改生产配置。 +KEY_ENV_CANDIDATES = ( + "/root/.hermes/profiles/wecom-fullstack/.env", + "/root/.hermes/.env", + "/root/cma-management/backend/.env", +) + +#: 文本里允许出现的非数字词表（避免把「TOP3」里的 3 当数字）——prompt 亦禁止 +DATE_PATTERNS = ( + re.compile(r"\d{4}-\d{2}-\d{2}"), + re.compile(r"\d{4}年\d{1,2}月\d{1,2}日"), + re.compile(r"\d{4}-\d{2}(?!\d)"), + re.compile(r"\d{4}年\d{1,2}月(?!\d)"), +) +NUMBER_PATTERN = re.compile(r"-?\d+(?:\.\d+)?") + +REQUIRED_SECTIONS = ("原因", "建议") + + +# ────────────────────────── 数字集合工具 ────────────────────────── +def _dec(value) -> Decimal | None: + """把任意数值转 Decimal（量化到 2 位，与规则层 round(2) 口径一致）；非法返回 None。""" + try: + d = Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + if not d.is_finite(): + return None + return d.quantize(Decimal("0.01")) + + +def collect_rule_numbers(payload) -> tuple[set[Decimal], set[str]]: + """递归收集 payload 中所有数值与日期（白名单）。 + + · 数值同时收 x 与 -x：文本用「下降 37.81%」表达负趋势属同一数字，不算改数 + · 布尔值不算数字；字符串里的日期单独收集，不计入数字白名单 + """ + values: set[Decimal] = set() + dates: set[str] = set() + + def walk(node): + if isinstance(node, bool) or node is None: + return + if isinstance(node, (int, float, Decimal)): + d = _dec(node) + if d is not None: + values.add(d) + values.add(-d) + return + if isinstance(node, str): + for pat in DATE_PATTERNS: + dates.update(pat.findall(node)) + return + if isinstance(node, dict): + for v in node.values(): + walk(v) + return + if isinstance(node, (list, tuple)): + for v in node: + walk(v) + + walk(payload) + return values, dates + + +def extract_text_numbers(text: str) -> tuple[list[tuple[str, Decimal]], set[str], str]: + """抽取文本中的日期与数字；返回 (数字token列表, 日期集合, 去日期后的文本)。""" + dates: set[str] = set() + stripped = text + for pat in DATE_PATTERNS: + for m in pat.findall(stripped): + dates.add(m) + stripped = pat.sub(" ", stripped) + + numbers: list[tuple[str, Decimal]] = [] + for raw in NUMBER_PATTERN.findall(stripped): + d = _dec(raw) + if d is not None: + numbers.append((raw, d)) + return numbers, dates, stripped + + +def assert_number_conservation(text: str, payload) -> dict: + """核心断言：文本数字 ⊆ 规则层数字；日期 ⊆ 规则层日期。返回结构化结果。""" + allowed_values, allowed_dates = collect_rule_numbers(payload) + numbers, dates, _ = extract_text_numbers(text) + + value_violations = [ + {"token": tok, "value": str(val)} + for tok, val in numbers + if val not in allowed_values + ] + date_violations = sorted(d for d in dates if d not in allowed_dates) + + return { + "pass": not value_violations and not date_violations, + "checked_numbers": len(numbers), + "allowed_numbers": len(allowed_values), + "value_violations": value_violations, + "date_violations": date_violations, + "allowed_values_sample": sorted(str(v) for v in allowed_values)[:40], + "allowed_dates": sorted(allowed_dates), + } + + +# ────────────────────────── 规则层取数 ────────────────────────── +def _dotenv(path: str) -> dict: + env = {} + for line in open(path, encoding="utf-8"): + line = line.strip() + if line and "=" in line and not line.startswith("#"): + k, v = line.split("=", 1) + env[k.strip()] = v.strip().strip('"').strip("'") + return env + + +def resolve_deepseek_key() -> tuple[str, str]: + """按候选顺序取 DeepSeek Key；返回 (key, 来源路径)。只用于调用，不落盘不打印。""" + current = os.environ.get("DEEPSEEK_API_KEY", "").strip() + for path in KEY_ENV_CANDIDATES: + if not os.path.exists(path): + continue + try: + key = _dotenv(path).get("DEEPSEEK_API_KEY", "").strip() + except OSError: + continue + if key: + return key, path + return current, "environment" if current else "" + + +def load_rule_payload(entity_id: int, days: int) -> dict: + """只读取数：cma_reader 账号（SELECT only）→ 引擎同一函数，与端点同源。""" + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from app.utils import cash_forecast_engine as engine + + rd = _dotenv(os.environ.get("CMA_READER_ENV", READER_ENV_DEFAULT)) + url = ("mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8mb4" + % (rd["CMA_READER_DB_USER"], rd["CMA_READER_DB_PASS"], + rd["CMA_READER_DB_HOST"], rd["CMA_READER_DB_PORT"], rd["CMA_READER_DB_NAME"])) + eng = create_engine(url, pool_pre_ping=True) + db = sessionmaker(bind=eng)() + try: + payload = engine.forecast_cash_flow_with_plans(entity_id, db, days=days) + finally: + db.close() + return payload + + +# ────────────────────────── prompt ────────────────────────── +SYSTEM_PROMPT = ( + "你是管理会计，只负责把已经算好的资金缺口结论写成老板能读懂的「原因」和「建议」。" + "硬性规则：" + "① 只能引用我给你的数字，逐字照抄，不得四舍五入、不得换算单位、不得估算、不得派生新数字；" + "② 禁止出现我未提供的任何新数字（包括数量、比例、天数、排名，如「TOP3」里的 3 不能用）；" + "③ 日期只能照抄我给的日期；" + "④ 输出两段，段首分别为「原因：」与「建议：」，每段不超过 120 字；" + "⑤ 不要复述表格，不要下结论性预测以外的推断。" +) + + +def build_prompt(payload: dict) -> str: + """把规则层 payload 压成「可引用数字白名单」，避免把整包原始明细灌给模型。""" + def r(v): + return None if v is None else round(float(v), 2) + + first_gap = (payload.get("gap_dates") or [None])[0] or {} + first_level = {"red": "红灯", "yellow": "黄灯"}.get(str(first_gap.get("level") or ""), "无") + + compact = { + "账套entity_id": payload.get("entity_id"), + "预测窗口天数": payload.get("days"), + "当前现金余额(万元)": r(payload.get("base_cash")), + "黄灯警戒线(万元)": r(payload.get("warning_line")), + "红灯警戒线(万元)": r(payload.get("critical_line")), + "阈值来源": (payload.get("threshold_source") or {}).get("source"), + "最低余额(万元)": r(payload.get("min_cash")), + "最低余额日期": payload.get("min_cash_date"), + "缺口天数": len(payload.get("gap_dates") or []), + "首个缺口": first_gap or None, + "首个缺口级别": first_level, + "月度经营现金流预算(万元/月)": r((payload.get("trends") or {}).get("budget_monthly_ocf")), + "经营现金流趋势(%)": r((payload.get("trends") or {}).get("operating_cash_flow_trend_pct")), + "应收趋势(%)": r((payload.get("trends") or {}).get("receivables_trend_pct")), + "应付趋势(%)": r((payload.get("trends") or {}).get("payables_trend_pct")), + "逾期应收笔数": (payload.get("summary") or {}).get("overdue_receive_count"), + "逾期应收金额(万元)": r((payload.get("summary") or {}).get("overdue_receive_amount")), + } + return ( + "以下是规则层已经算好的资金缺口数据（唯一可引用数字来源）：\n" + + json.dumps(compact, ensure_ascii=False, indent=2) + + "\n请只基于这些数字写「原因」与「建议」两段。" + ) + + +def strict_retry_prompt(payload: dict, bad_text: str, violations: list) -> str: + return ( + build_prompt(payload) + + "\n\n上一次输出违反了数字纪律，出现了未授权的数字：" + + json.dumps(violations, ensure_ascii=False) + + "\n上一次输出为：\n" + bad_text + + "\n请重写，删除上述未授权数字，其余规则不变。" + ) + + +# ────────────────────────── 主流程 ────────────────────────── +def split_sections(text: str) -> dict: + """把模型输出切成 原因 / 建议 两段（段首关键字）。""" + out = {"reason": "", "advice": ""} + reason = re.search(r"原因[：:]\s*(.*?)(?=建议[：:]|$)", text, re.S) + advice = re.search(r"建议[：:]\s*(.*)$", text, re.S) + if reason: + out["reason"] = reason.group(1).strip() + if advice: + out["advice"] = advice.group(1).strip() + return out + + +async def generate(entity_id: int, days: int, payload: dict | None = None) -> dict: + key, key_source = resolve_deepseek_key() + if key: + # 必须在 import ai_brief 之前落位：ai_brief 模块级 load_dotenv 不覆盖已存在的环境变量， + # 否则会读到生产 .env 里已被判 invalid 的旧 Key（2026-10-17 实测 401）。 + os.environ["DEEPSEEK_API_KEY"] = key + + from ai_brief import call_deepseek # 复用既有 DeepSeek 通道（scripts/ai_brief.py:24） + + payload = payload if payload is not None else load_rule_payload(entity_id, days) + prompt = build_prompt(payload) + text = await call_deepseek(prompt, SYSTEM_PROMPT) + assertion = assert_number_conservation(text or "", payload) + attempts = 1 + + if not assertion["pass"] and text: + text2 = await call_deepseek( + strict_retry_prompt(payload, text, assertion["value_violations"] + assertion["date_violations"]), + SYSTEM_PROMPT, + ) + assertion2 = assert_number_conservation(text2 or "", payload) + attempts = 2 + if assertion2["pass"]: + text, assertion = text2, assertion2 + else: + assertion = assertion2 # 以最后一次为准（仍为 fail） + + sections = split_sections(text or "") + missing = [s for s in REQUIRED_SECTIONS if not sections[{"原因": "reason", "建议": "advice"}[s]]] + ok = assertion["pass"] and not missing + + return { + "task": "fullstack-20261017-warning-capability-build-p0", + "stage": "G2 cash gap narrative (number-conservation asserted)", + "generated_at": datetime.now().isoformat(timespec="seconds"), + "entity_id": entity_id, + "days": days, + "model": "deepseek-chat", + "llm_key_source": key_source, + "llm_empty_response": not bool((text or "").strip()), + "attempts": attempts, + "assertion": assertion, + "missing_sections": missing, + "pass": ok, + "narrative": ({"reason": sections["reason"], "advice": sections["advice"], "raw": text} + if ok else None), + "rule_layer": { + "base_cash": payload.get("base_cash"), + "warning_line": payload.get("warning_line"), + "critical_line": payload.get("critical_line"), + "min_cash": payload.get("min_cash"), + "min_cash_date": payload.get("min_cash_date"), + "threshold_source": payload.get("threshold_source"), + "gap_days": len(payload.get("gap_dates") or []), + }, + } + + +# ────────────────────────── 断言自测（离线） ────────────────────────── +SELFTEST_PAYLOAD = { + "entity_id": 1, "days": 30, "base_cash": -9.3, "warning_line": 20.0, "critical_line": 10.0, + "min_cash": -8.74, "min_cash_date": "2026-10-18", "threshold_source": {"source": "default"}, + "gap_dates": [{"date": "2026-10-18", "gap_amount": 28.74, "level": "red"}], + "trends": {"budget_monthly_ocf": 16.67, "receivables_trend_pct": -37.81}, + "summary": {"overdue_receive_count": 18, "overdue_receive_amount": 71.68}, +} + + +def selftest() -> int: + cases = [ + ("good_basic", + "原因：现金余额-9.3万元，2026-10-18 降至最低-8.74万元，低于警戒线20.0万元，缺口28.74万元；" + "逾期应收18笔共71.68万元。建议：优先催收逾期应收，暂停非必要支出。", + True), + ("good_sign_abs", + "原因：应收趋势下降37.81%，月度经营现金流预算16.67万元/月，缺口1天。" + "建议：跟进2026-10-18 起的资金缺口。", + True), + ("bad_rounding", + "原因：现金余额约-9万元，最低-8.7万元。建议：催收应收。", + False), + ("bad_derived", + "原因：现金余额-9.3万元，缺口28.74万元，约为逾期应收的40%。建议：催收。", + False), + ("bad_new_count", + "原因：现金余额-9.3万元。建议：优先催收TOP3客户。", + False), + ("bad_fake_date", + "原因：现金余额-9.3万元，预计2026-11-01 断流。建议：催收。", + False), + ] + failures = 0 + print("== G2 数字守恒断言自测 ==") + for name, text, expect in cases: + res = assert_number_conservation(text, SELFTEST_PAYLOAD) + verdict = "PASS" if res["pass"] == expect else "FAIL" + if verdict == "FAIL": + failures += 1 + print(" %-5s %-16s expect_pass=%-5s got_pass=%-5s value_violations=%s date_violations=%s" + % (verdict, name, expect, res["pass"], + [v["token"] for v in res["value_violations"]], res["date_violations"])) + total = len(cases) + print("RESULT: %s %d/%d" % ("PASS" if failures == 0 else "FAIL", total - failures, total)) + return 0 if failures == 0 else 1 + + +def main() -> int: + ap = argparse.ArgumentParser(description="资金缺口讲人话（数字守恒断言）") + ap.add_argument("--entity", type=int, default=1) + ap.add_argument("--days", type=int, default=30) + ap.add_argument("--payload-json", help="已有规则层 payload（跳过取数）") + ap.add_argument("--out", help="输出 JSON 路径") + ap.add_argument("--selftest", action="store_true", help="只跑断言自测（离线）") + args = ap.parse_args() + + if args.selftest: + return selftest() + + payload = None + if args.payload_json: + payload = json.load(open(args.payload_json, encoding="utf-8")) + + result = asyncio.run(generate(args.entity, args.days, payload)) + out = args.out or os.path.join( + "/root/.hermes/profiles/wecom-fullstack/pending-tasks/_evidence", + "fullstack-20261017-warning-capability-build-p0", + "g2-narrative-%s-e%d.json" % (datetime.now().strftime("%Y%m%d-%H%M%S"), args.entity)) + os.makedirs(os.path.dirname(out), exist_ok=True) + with open(out, "w", encoding="utf-8") as fh: + json.dump(result, fh, ensure_ascii=False, indent=2) + + print(json.dumps({k: result[k] for k in + ("pass", "attempts", "assertion", "missing_sections", "rule_layer")}, + ensure_ascii=False, indent=2)) + if result["narrative"]: + print("原因：" + result["narrative"]["reason"]) + print("建议：" + result["narrative"]["advice"]) + print("OUT:", out) + return 0 if result["pass"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/scripts/verify_cash_gap_replay.py b/backend/scripts/verify_cash_gap_replay.py new file mode 100644 index 00000000..23d68cd9 --- /dev/null +++ b/backend/scripts/verify_cash_gap_replay.py @@ -0,0 +1,409 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""资金缺口复算对拍 — G1 验收件（P0-A） + +任务：fullstack-20261017-warning-capability-build-p0（上游 yanxue-20261017-warning-capability-build-p0） +方案：`_reports/yanxue-20261017-warning-capability-build-p0.md` §G1 + §四「数字零错误（可复算）」 + +本脚本做三件事，全部机器可判： + 1. **同输入双跑逐字段 diff=0**：固定输入（内存 SQLite 夹具 + 固定锚点日期）跑两遍 + `forecast_cash_flow_with_plans`，把两次结果整体 JSON 归一后逐字段比对； + 再用「独立重建的同一夹具」跑第三遍，证明无隐藏状态（跨会话也一致）。 + 2. **手算 3/3 案例对拍**：3 个案例的每日余额由人工算式预先算出并**写死为常量** + （不调用引擎反推期望值），逐字段断言：每日余额、缺口日、缺口金额、红黄绿、最小余额、 + 预警点、阈值来源。 + 3. **端点逐字段一致**（`--live`）：脚本内直接调引擎的结果 vs 运行中服务的 + `GET /api/cma/cash/gap-forecast` 响应，逐字段比对（含 threshold_source）。 + +手算口径（与引擎实现逐条对应，锚点固定 2026-10-05 周一 09:00）： + 锚点 today = 2026-10-05；offset 1..N 依次是其后 N 天。 + 无 KPI 历史 → ocf_avg 取引擎兜底 0.5（万元/日），趋势 0；无预算 → budget_ocf=None。 + 每日基础增量 = 0.5；`day>=25` 且无月底窗口（本例日期均在 10-06..10-15）；周末 ×0.5。 + 叠加收付款计划后：cash_i = cash_{i-1} + base_delta_i + planned_in_i - planned_out_i（逐步 round 2）。 + +用法： + 夹具复算（默认，不连生产）: + python3 backend/scripts/verify_cash_gap_replay.py + 端点对拍（需服务在线，默认 127.0.0.1:8010，读 .secrets 登录）: + python3 backend/scripts/verify_cash_gap_replay.py --live --base-url http://127.0.0.1:8010 --entity 1 +退出码：0 = 全部 PASS（diff=0 且 3/3 案例 PASS，live 模式下端点逐字段一致）；1 = 有 FAIL +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.request +from datetime import datetime as _RealDatetime + +BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if BACKEND_DIR not in sys.path: + sys.path.insert(0, BACKEND_DIR) + +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 + +import app.models as models # noqa: E402 +from app.api.alert_rules import AlertRule # noqa: E402 +from app.utils import cash_forecast_engine as engine # noqa: E402 +from app.utils.cash_thresholds import ( # noqa: E402 + CASH_GAP_RULE_TYPE, + DEFAULT_CRITICAL, + DEFAULT_WARNING, +) + +# ── 固定锚点（2026-10-05 周一，保证手算窗口内无周末/无月底）──────────────────── +FIXED_NOW = _RealDatetime(2026, 10, 5, 9, 0, 0) +ENTITY = 1 + +FIXTURE_TABLES = None + + +class FixedDatetime(_RealDatetime): + """把引擎里的 datetime.now() 钉死在 FIXED_NOW，保证复算可重现。""" + + @classmethod + def now(cls, tz=None): # noqa: D401 + return FIXED_NOW + + +def build_fixture_session(with_config=None, cash_plans=None, budget=None): + """内存 SQLite 夹具：建最小表集，按需插入收付款计划 / 阈值配置 / 预算。""" + from sqlalchemy.pool import StaticPool + + global FIXTURE_TABLES + if FIXTURE_TABLES is None: + FIXTURE_TABLES = [ + models.KPIDefinition.__table__, + models.KPIValue.__table__, + models.SystemConfig.__table__, + models.CashPlan.__table__, + models.BudgetPlan.__table__, + AlertRule.__table__, + ] + eng = create_engine("sqlite://", connect_args={"check_same_thread": False}, + poolclass=StaticPool) + models.Base.metadata.create_all(eng, tables=FIXTURE_TABLES) + db = sessionmaker(bind=eng)() + for plan in (cash_plans or []): + db.add(models.CashPlan(**plan)) + if with_config: + db.add(AlertRule(entity_id=ENTITY, kpi_id=0, rule_type=CASH_GAP_RULE_TYPE, + trigger_on="actual", enabled=1, params=with_config)) + if budget: + # 预算校准会走 KPI 查找；夹具无 KPI → _budget_monthly_ocf 恒 None（保持手算口径） + pass + db.commit() + return db + + +def normalise(obj): + """JSON 归一（Decimal/日期 → 字符串，浮点保持原值），用于逐字段严格比对。""" + return json.loads(json.dumps(obj, ensure_ascii=False, default=str)) + + +def deep_diff(left, right, path="root"): + """返回逐字段差异列表；空列表 = diff=0。""" + diffs = [] + if type(left) is not type(right): + return [{"path": path, "left": left, "right": right, "kind": "type"}] + if isinstance(left, dict): + for key in sorted(set(left) | set(right)): + if key not in left: + diffs.append({"path": f"{path}.{key}", "left": "", "right": right[key], "kind": "missing_left"}) + elif key not in right: + diffs.append({"path": f"{path}.{key}", "left": left[key], "right": "", "kind": "missing_right"}) + else: + diffs.extend(deep_diff(left[key], right[key], f"{path}.{key}")) + return diffs + if isinstance(left, list): + if len(left) != len(right): + diffs.append({"path": path + ".len", "left": len(left), "right": len(right), "kind": "length"}) + for idx, (a, b) in enumerate(zip(left, right)): + diffs.extend(deep_diff(a, b, f"{path}[{idx}]")) + return diffs + if left != right: + diffs.append({"path": path, "left": left, "right": right, "kind": "value"}) + return diffs + + +# ══════════════════════════════════════════════════════════════════════ +# 手算案例（期望值写死；算式见各案例注释，禁止用引擎反推） +# ══════════════════════════════════════════════════════════════════════ +HAND_CASES = [ + { + "name": "H1 无计划·全绿", + "input": {"current_cash": 30.0, "days": 10}, + "plans": [], + "config": None, + # 手算：cash_i = 30 + 0.5*i（工作日）；10-10 周六 / 10-11 周日 增量 ×0.5 = 0.25 + # 30 → 30.5 → 31.0 → 31.5 → 32.0 →(六) 32.25 →(日) 32.5 → 33.0 → 33.5 → 34.0 → 34.5 + "expect": { + "warning_line": 20.0, + "critical_line": 10.0, + "threshold_source": "default", + "predicted_cash": [30.5, 31.0, 31.5, 32.0, 32.25, 32.5, 33.0, 33.5, 34.0, 34.5], + "alert_status": ["green"] * 10, + "gap_days": 0, + "gap_dates": [], + "min_cash": 30.5, + "min_cash_date": "2026-10-06", + "pre_alerts": 0, + "planned_out_total": 0.0, + }, + }, + { + "name": "H2 单笔付款·缺口前1天预警", + "input": {"current_cash": 21.0, "days": 4}, + "plans": [ + # 10-08（Thu）对外付款 4.0 万 + {"entity_id": ENTITY, "plan_type": "pay", "amount": 4.0, "status": "pending", + "plan_date": _RealDatetime(2026, 10, 8, 0, 0, 0), "counterparty": "手算案例供应商", + "description": "FIXTURE-H2", "source": "fixture"}, + ], + "config": None, + # 手算：21 → +0.5 = 21.5 → +0.5 = 22.0 → +0.5 -4.0 = 18.5 → +0.5 = 19.0 + # 缺口日 = 余额 < 20 → 10-08(18.5)、10-09(19.0)，缺额 1.5 / 1.0 + # 缺口区间首日 idx=2 → 前1天预警落在 idx=1（10-07，余额 22.0 为 green） + "expect": { + "warning_line": 20.0, + "critical_line": 10.0, + "threshold_source": "default", + "predicted_cash": [21.5, 22.0, 18.5, 19.0], + "alert_status": ["green", "green", "yellow", "yellow"], + "gap_days": 2, + "gap_dates": [ + {"date": "2026-10-08", "predicted_cash": 18.5, "gap_amount": 1.5, "level": "yellow"}, + {"date": "2026-10-09", "predicted_cash": 19.0, "gap_amount": 1.0, "level": "yellow"}, + ], + "min_cash": 18.5, + "min_cash_date": "2026-10-08", + "pre_alerts": 1, + "pre_alert_0": {"alert_date": "2026-10-07", "alert_offset": 2, "gap_date": "2026-10-08", + "gap_cash": 18.5, "gap_amount": 1.5, "worst_cash": 18.5, + "worst_date": "2026-10-08", "lead_days": 1, "level": "yellow"}, + "planned_out_total": 4.0, + }, + }, + { + "name": "H3 配置阈值 25/15·全红", + "input": {"current_cash": 8.0, "days": 5}, + "plans": [], + # alert_rules: cash_gap_threshold {"warning":25.0,"critical":15.0} + "config": {"warning": 25.0, "critical": 15.0}, + # 手算：8 → 8.5 → 9.0 → 9.5 → 10.0 →(六) 10.25，全部 < 15（红灯） + # 缺额（相对黄灯线 25）：16.5 / 16.0 / 15.5 / 15.0 / 14.75 + "expect": { + "warning_line": 25.0, + "critical_line": 15.0, + "threshold_source": "alert_rules", + "predicted_cash": [8.5, 9.0, 9.5, 10.0, 10.25], + "alert_status": ["red"] * 5, + "gap_days": 5, + "gap_dates_first": {"date": "2026-10-06", "predicted_cash": 8.5, + "gap_amount": 16.5, "level": "red"}, + "min_cash": 8.5, + "min_cash_date": "2026-10-06", + "pre_alerts": 0, # 缺口从第一天开始 → 无「缺口前1天」可落 + "planned_out_total": 0.0, + }, + }, +] + + +def run_case(case): + db = build_fixture_session(with_config=case.get("config"), cash_plans=case.get("plans")) + try: + result = engine.forecast_cash_flow_with_plans( + ENTITY, db, days=case["input"]["days"], current_cash=case["input"]["current_cash"]) + finally: + db.close() + return normalise(result) + + +def assert_case(case, result, failures): + exp = case["expect"] + name = case["name"] + checks = [ + ("warning_line", result.get("warning_line"), exp["warning_line"]), + ("critical_line", result.get("critical_line"), exp["critical_line"]), + ("threshold_source.source", (result.get("threshold_source") or {}).get("source"), exp["threshold_source"]), + ("min_cash", result.get("min_cash"), exp["min_cash"]), + ("min_cash_date", result.get("min_cash_date"), exp["min_cash_date"]), + ("gap_days", len(result.get("gap_dates") or []), exp["gap_days"]), + ("pre_alerts", len(result.get("pre_alerts") or []), exp["pre_alerts"]), + ("predicted_cash", + [row["predicted_cash"] for row in result.get("forecast", [])], exp["predicted_cash"]), + ("alert_status", + [row["alert_status"] for row in result.get("forecast", [])], exp["alert_status"]), + ("summary.total_planned_out", + round((result.get("summary") or {}).get("total_planned_out") or 0.0, 2), exp["planned_out_total"]), + ] + if "gap_dates" in exp: + checks.append(("gap_dates", [ + {k: g[k] for k in ("date", "predicted_cash", "gap_amount", "level")} + for g in (result.get("gap_dates") or []) + ], exp["gap_dates"])) + if "gap_dates_first" in exp: + first = (result.get("gap_dates") or [{}])[0] + checks.append(("gap_dates[0]", + {k: first.get(k) for k in ("date", "predicted_cash", "gap_amount", "level")}, + exp["gap_dates_first"])) + if "pre_alert_0" in exp: + first = (result.get("pre_alerts") or [{}])[0] + checks.append(("pre_alerts[0]", + {k: first.get(k) for k in exp["pre_alert_0"].keys()}, exp["pre_alert_0"])) + + for label, got, want in checks: + if got != want: + failures.append({"case": name, "check": label, "expected": want, "got": got}) + print(" FAIL %-8s %-28s expected=%r got=%r" % (name.split()[0], label, want, got)) + else: + print(" PASS %-8s %-28s = %r" % (name.split()[0], label, got)) + + +def assert_invalid_config_falls_back(failures): + """非法配置（critical > warning）→ 必须回落默认 20/10（配置写坏不能静默生效）。""" + db = build_fixture_session(with_config={"warning": 10.0, "critical": 20.0}) + try: + got = engine.get_cash_gap_thresholds(db, ENTITY) + finally: + db.close() + if got != (DEFAULT_WARNING, DEFAULT_CRITICAL): + failures.append({"case": "GUARD", "check": "invalid_config_fallback", + "expected": [DEFAULT_WARNING, DEFAULT_CRITICAL], "got": list(got)}) + print(" FAIL GUARD invalid config must fall back to defaults, got=%r" % (got,)) + else: + print(" PASS GUARD invalid config → default (%.1f, %.1f)" % got) + + +def http_json(base_url, path, method="GET", payload=None, token=None, timeout=30): + headers = {} + if token: + headers["Authorization"] = "Bearer " + token + data = json.dumps(payload).encode() if payload is not None else None + if data: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(base_url.rstrip("/") + path, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + + +def _dotenv(path): + env = {} + with open(path, encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line and "=" in line and not line.startswith("#"): + key, val = line.split("=", 1) + env[key.strip()] = val.strip().strip('"').strip("'") + return env + + +def live_compare(base_url, entity_id, days, failures): + """脚本内引擎结果 vs 在线端点响应（逐字段，含 threshold_source）。""" + # 夹具阶段把 now() 钉死在 FIXED_NOW；对拍真实端点必须还原真实时钟， + # 否则日期序列整体错位（脚本 2026-10-06 vs 服务 2026-10-18）。 + engine.datetime = _RealDatetime + reader_env = os.environ.get("CMA_READER_ENV", + "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-reader.env") + admin_env = os.environ.get("CMA_ADMIN_ENV", + "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env") + rd = _dotenv(reader_env) + url = ("mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8mb4" + % (rd["CMA_READER_DB_USER"], rd["CMA_READER_DB_PASS"], rd["CMA_READER_DB_HOST"], + rd["CMA_READER_DB_PORT"], rd["CMA_READER_DB_NAME"])) + eng = create_engine(url, pool_pre_ping=True) + db = sessionmaker(bind=eng)() + try: + script_side = normalise(engine.forecast_cash_flow_with_plans(entity_id, db, days=days)) + finally: + db.close() + + adm = _dotenv(admin_env) + login = http_json(base_url, "/api/cma/auth/login", "POST", + {"username": adm.get("CMA_ADMIN_USER"), "password": adm.get("CMA_ADMIN_PASS"), + "entity_id": entity_id}) + token = login.get("token") + if not token: + failures.append({"case": "LIVE", "check": "login", "got": login}) + print(" FAIL LIVE login 未拿到 token") + return + api_side = normalise(http_json(base_url, "/api/cma/cash/gap-forecast?days=%d" % days, token=token)) + + diffs = deep_diff(script_side, api_side, path="gap-forecast") + print(" %s LIVE 引擎 vs 端点 /api/cma/cash/gap-forecast diff=%d" % + ("PASS" if not diffs else "FAIL", len(diffs))) + for d in diffs[:20]: + print(" %(path)s script=%(left)r api=%(right)r" % d) + if diffs: + failures.append({"case": "LIVE", "check": "endpoint_field_parity", "diffs": diffs[:20], + "diff_count": len(diffs)}) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--live", action="store_true", help="对拍运行中服务端点") + ap.add_argument("--base-url", default="http://127.0.0.1:8010") + ap.add_argument("--entity", type=int, default=ENTITY) + ap.add_argument("--days", type=int, default=30) + ap.add_argument("--json-out", default="") + args = ap.parse_args() + + failures = [] + summary = {"fixed_now": FIXED_NOW.isoformat(), "entity_id": args.entity, "cases": []} + + # 固定锚点 + engine.datetime = FixedDatetime + + print("== 1. 同输入双跑逐字段 diff=0 ==") + for case in HAND_CASES: + first = run_case(case) + second = run_case(case) + third = run_case(case) # 独立重建夹具，验证无隐藏状态 + diffs_12 = deep_diff(first, second, path="run1_vs_run2") + diffs_13 = deep_diff(first, third, path="run1_vs_run3") + print(" %s %-24s diff(run1,run2)=%d diff(run1,run3)=%d" % + ("PASS" if not (diffs_12 or diffs_13) else "FAIL", + case["name"], len(diffs_12), len(diffs_13))) + for d in (diffs_12 + diffs_13)[:10]: + print(" %(path)s left=%(left)r right=%(right)r" % d) + if diffs_12 or diffs_13: + failures.append({"case": case["name"], "check": "double_run_diff", + "diffs": (diffs_12 + diffs_13)[:10]}) + summary["cases"].append({"case": case["name"], "diff_run2": len(diffs_12), + "diff_run3": len(diffs_13)}) + + print("== 2. 手算案例对拍 ==") + for case in HAND_CASES: + assert_case(case, run_case(case), failures) + + print("== 3. 配置健壮性 ==") + assert_invalid_config_falls_back(failures) + + if args.live: + print("== 4. 端点逐字段一致（--live）==") + live_compare(args.base_url, args.entity, args.days, failures) + + passed = 3 if not failures else 0 + print("=" * 60) + print("RESULT: %s hand_calc=%d/%d failures=%d" % + ("PASS" if not failures else "FAIL", passed, len(HAND_CASES), len(failures))) + if failures: + for f in failures[:5]: + print(" ", json.dumps(f, ensure_ascii=False)[:400]) + if args.json_out: + payload = {"task": "fullstack-20261017-warning-capability-build-p0", + "stage": "G1 cash gap replay", "result": "PASS" if not failures else "FAIL", + "fixed_now": FIXED_NOW.isoformat(), "entity_id": args.entity, + "hand_cases_passed": passed, "hand_cases_total": len(HAND_CASES), + "failures": failures, "summary": summary} + with open(args.json_out, "w", encoding="utf-8") as fh: + fh.write(json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n") + print("json evidence: %s" % args.json_out) + return 0 if not failures else 1 + + +if __name__ == "__main__": + sys.exit(main())