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

feat(warning-p0a): 资金缺口阈值外置 + 复算对拍 + 讲人话(数字守恒) + 数据就绪核验
ci/woodpecker/push/main Pipeline failed
Details

Browse Source 
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。
This commit is contained in:
wecom-fullstack
2026-10-17 23:33:51 +08:00
parent a52faa90a6
commit 885cfddbff
6 changed files with 1335 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/scripts/audit_warning_data_readiness.py
+319
View File
@@ -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())
backend/scripts/cash_gap_narrative.py
+408
View File
@@ -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())
backend/scripts/verify_cash_gap_replay.py
+409
View File
@@ -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": "<absent>", "right": right[key], "kind": "missing_left"})
elif key not in right:
diffs.append({"path": f"{path}.{key}", "left": left[key], "right": "<absent>", "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())