Hermes CI Fix
38a6de9325
ci/woodpecker/push/main Pipeline failed
Task: finance-20261003-dupont-period-basis-design-fullstack (P1; proposer wecom-finance) Basis ruling: finance 2026-10-03 (Boss 09:49 CST DM approval of the design task) - /reports/dupont accepts period/start/end/annualize (default YTD), returns period_start/period_end/basis/annualized metadata; identity guard |ROE - npm x turnover x multiplier| <= 0.05 pp, else identity_broken (no values). - new app/utils/dupont_basis.py = single source of truth (window resolve, flow cumulative via voucher_details, stock average via subject_balances level=1, fail-closed basis_fallback, pure calc + identity check). - scripts/kpi_dupont_deriver.py (idempotent derived writer, dry-run/check/write), scripts/apply_dupont_kpi_defs.py + sql/dupont_kpi_defs_20261003.sql (seed: F_ASSET_TOTAL/F_EQUITY_TOTAL/F_EQUITY_MULTIPLIER as derived; F_ASSET_TURNOVER manual -> derived). Seed already applied under approval 296cc1fd. - scripts/verify_dupont_period_basis.py: read-only acceptance battery A1-A9 (independent SQL recompute). BEFORE (base tree) 1/10 PASS -> AFTER 10/13 PASS; remaining 3 need the deriver value write (approval-gated). - frontend DupontAnalysis.vue: period selector (YTD default) + annualize switch + caliber metadata bar + identity_broken alert. No DDL. No ERP direct connection (old register_dupont_kpis.py caliber dropped).
304 lines
15 KiB
Python
304 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""杜邦分析「区间取值」验收电池 —— 任务 finance-20261003-dupont-period-basis-design-fullstack
|
||
|
||
判别力设计：**改动前必 FAIL、改动后必 PASS**。本脚本直达 `/reports/dupont` 端点函数体
|
||
（绕过 HTTP/鉴权），逐项断言任务件 §三 验收标准；数值侧用**独立 SQL 复算**
|
||
（不复用 app/utils/dupont_basis.py 的取数代码）作比对，避免「自己验自己」。
|
||
|
||
每节用 `with Section(...)` 隔离：改动前（端点不接受 period/annualize）得到的是一张
|
||
全 FAIL 的表而非崩溃堆栈 —— 判别力证据才可比。
|
||
|
||
用法（只读；无任何写路径）:
|
||
venv/bin/python3 scripts/verify_dupont_period_basis.py --check # 全量验收
|
||
venv/bin/python3 scripts/verify_dupont_period_basis.py --check --json # 附 JSON
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import traceback
|
||
|
||
if not any(a in ("--check", "--dry-run") for a in sys.argv[1:]):
|
||
print("拒绝执行：本脚本只读（仅 SELECT/端点只读调用，无 commit），须带显式只读标志 --check",
|
||
file=sys.stderr)
|
||
sys.exit(2)
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.dirname(HERE))
|
||
|
||
from sqlalchemy import text # noqa: E402
|
||
from starlette.requests import Request # noqa: E402
|
||
|
||
from app.database import get_session_local # noqa: E402
|
||
from app.api.reports import get_dupont_analysis # noqa: E402
|
||
|
||
RESULTS: list = []
|
||
|
||
|
||
def check(name: str, ok: bool, detail: str = "") -> bool:
|
||
RESULTS.append({"item": name, "ok": bool(ok), "detail": detail})
|
||
print(f"{'PASS' if ok else 'FAIL'} {name}" + (f" | {detail}" if detail else ""))
|
||
return bool(ok)
|
||
|
||
|
||
class Section:
|
||
"""把一节断言包成「异常 = FAIL」而不是整脚本崩溃（判别力取证的必备形态）。"""
|
||
|
||
def __init__(self, name: str):
|
||
self.name = name
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc, tb):
|
||
if exc is not None:
|
||
check(f"{self.name} —— 本节断言未能执行（改动前端点形态差异必现）", False,
|
||
f"{type(exc).__name__}: {exc}")
|
||
return True
|
||
return False
|
||
|
||
|
||
def stub_request(path: str = "/api/cma/reports/dupont") -> Request:
|
||
return Request({"type": "http", "method": "GET", "path": path, "query_string": b"",
|
||
"headers": [(b"host", b"verify.local")]})
|
||
|
||
|
||
def call(db, **kw):
|
||
"""直达端点函数体（绕过 HTTP 与 router 级 require_role）。
|
||
|
||
⚠️ FastAPI 直调陷阱：省略 kwarg 拿到的是 `Query(None)` 哨兵而非 `None`（会
|
||
`AttributeError: 'Query' object has no attribute ...`）⇒ 探针侧必须显式补齐全部形参。
|
||
"""
|
||
for k, v in (("entity", None), ("period", None), ("start", None), ("end", None),
|
||
("annualize", False), ("guard_probe", None)):
|
||
kw.setdefault(k, v)
|
||
return get_dupont_analysis(request=stub_request(), db=db, **kw)
|
||
|
||
|
||
def scalar(db, sql, **args):
|
||
v = db.execute(text(sql), args).scalar()
|
||
return None if v is None else float(v)
|
||
|
||
|
||
# ────────────────── 独立复算（纯 SQL，不复用 dupont_basis） ──────────────────
|
||
|
||
def independent_recompute(db, entity_id: int, start: str, end: str,
|
||
equity_prefix: str, revenue_codes=("5001",),
|
||
net_profit_mode: str = "carry_forward") -> dict:
|
||
"""独立口径复算（万元）：流量累计 + 存量平均（期初/期末），与端点输出对拍。"""
|
||
codes = ",".join(f"'{c}'" for c in revenue_codes)
|
||
rev = scalar(db, f"""SELECT COALESCE(SUM(credit_amount),0) FROM voucher_details
|
||
WHERE entity_id=:e AND period>=:s AND period<=:e2
|
||
AND carry_forward=0 AND subject_code IN ({codes})""",
|
||
e=entity_id, s=start, e2=end)
|
||
if net_profit_mode == "carry_forward":
|
||
np_flow = scalar(db, """SELECT COALESCE(SUM(credit_amount-debit_amount),0) FROM voucher_details
|
||
WHERE entity_id=:e AND period>=:s AND period<=:e2
|
||
AND carry_forward=1 AND subject_code LIKE :p""",
|
||
e=entity_id, s=start, e2=end, p=f"{equity_prefix}%")
|
||
else:
|
||
np_flow = scalar(db, """SELECT COALESCE(SUM(credit_amount-debit_amount),0) FROM voucher_details
|
||
WHERE entity_id=:e AND period>=:s AND period<=:e2
|
||
AND carry_forward=0 AND subject_code LIKE '6%'""",
|
||
e=entity_id, s=start, e2=end)
|
||
|
||
def bal(period, field, side, prefix):
|
||
ob = "ob_dr-ob_cr" if side == "debit" else "ob_cr-ob_dr"
|
||
cb = "cb_dr-cb_cr" if side == "debit" else "cb_cr-cb_dr"
|
||
expr = {"ob": ob, "cb": cb}[field]
|
||
return scalar(db, f"""SELECT COALESCE(SUM({expr}),0) FROM subject_balances
|
||
WHERE entity_id=:e AND period=:p AND `level`=1
|
||
AND subject_code LIKE :pfx""",
|
||
e=entity_id, p=period, pfx=f"{prefix}%") or 0.0
|
||
|
||
ta_ob, ta_cb = bal(start, "ob", "debit", "1"), bal(end, "cb", "debit", "1")
|
||
eq_ob, eq_cb = bal(start, "ob", "credit", equity_prefix), bal(end, "cb", "credit", equity_prefix)
|
||
ta_avg, eq_avg = (ta_ob + ta_cb) / 2.0, (eq_ob + eq_cb) / 2.0
|
||
np_w, rev_w, ta_w, eq_w = np_flow / 1e4, rev / 1e4, ta_avg / 1e4, eq_avg / 1e4
|
||
return {
|
||
"net_profit": round(np_w, 6), "revenue": round(rev_w, 6),
|
||
"total_assets": round(ta_w, 6), "equity": round(eq_w, 6),
|
||
"total_assets_begin": round(ta_ob / 1e4, 6), "equity_begin": round(eq_ob / 1e4, 6),
|
||
"total_assets_end": round(ta_cb / 1e4, 6), "equity_end": round(eq_cb / 1e4, 6),
|
||
"net_profit_margin": None if not rev_w else round(np_w / rev_w * 100.0, 6),
|
||
"asset_turnover": None if not ta_w else round(rev_w / ta_w, 6),
|
||
"equity_multiplier": None if not eq_w else round(ta_w / eq_w, 6),
|
||
"roe": None if not eq_w else round(np_w / eq_w * 100.0, 6),
|
||
}
|
||
|
||
|
||
def sec_a1(db):
|
||
with Section("A1"):
|
||
d = call(db, entity="hanke")
|
||
meta_ok = all(k in d for k in ("period_start", "period_end", "basis", "annualized"))
|
||
check("A1 无参默认=YTD 且含 period_start/period_end/basis/annualized",
|
||
meta_ok and d.get("period_key") == "ytd"
|
||
and str(d.get("period_start")).endswith("-01"),
|
||
f"period_key={d.get('period_key')} {d.get('period_start')}~{d.get('period_end')} "
|
||
f"basis={d.get('basis')} annualized={d.get('annualized')}")
|
||
check("A1b 默认口径恒等式 PASS（identity.status=pass）",
|
||
d.get("identity", {}).get("status") == "pass",
|
||
f"identity={d.get('identity')}")
|
||
|
||
|
||
def sec_a2(db):
|
||
m = None
|
||
with Section("A2"):
|
||
m = call(db, entity="hanke", period="2026-08", annualize=False)
|
||
exp = independent_recompute(db, 1, "2026-08", "2026-08", "3")
|
||
pairs = []
|
||
for k in ("net_profit", "revenue", "total_assets", "equity"):
|
||
got, want = m["raw_data"][k], exp[k]
|
||
pairs.append((k, got, want, abs((got or 0) - (want or 0)) < 1e-6))
|
||
for k in ("net_profit_margin", "asset_turnover", "equity_multiplier"):
|
||
got, want = m["factors"][k]["value_exact"], exp[k]
|
||
pairs.append((k, got, want, abs((got or 0) - (want or 0)) < 1e-5))
|
||
check("A2 单月 2026-08 非年化：原值+三因子与独立 SQL 复算逐项一致",
|
||
all(p[3] for p in pairs),
|
||
"; ".join(f"{p[0]}={p[1]}/{p[2]}{'✗' if not p[3] else ''}" for p in pairs))
|
||
check("A2b 单月区间 basis=average（期初取同月 ob_*，非回落）",
|
||
m.get("basis") == "average" and m.get("basis_fallback") is False,
|
||
f"basis={m.get('basis')} fallback={m.get('basis_fallback')}")
|
||
return m
|
||
|
||
|
||
def sec_a3(db, m):
|
||
if not m:
|
||
check("A3 年化对比（前序 A2 未取到 payload，节略）", False, "")
|
||
return
|
||
with Section("A3"):
|
||
y = call(db, entity="hanke", period="2026-08", annualize=True)
|
||
k = y["annualize_scale"]
|
||
tol = max(1e-6, 1e-6 * k) # value_exact 已 round(6)，乘 k 后同比例允差
|
||
okat = abs(y["factors"]["asset_turnover"]["value_exact"]
|
||
- m["factors"]["asset_turnover"]["value_exact"] * k) <= tol
|
||
okroe = abs((y["roe_exact"] or 0) - (m["roe_exact"] or 0) * k) <= tol
|
||
oknpm = abs(y["factors"]["net_profit_margin"]["value_exact"]
|
||
- m["factors"]["net_profit_margin"]["value_exact"]) < 1e-9
|
||
okem = abs(y["factors"]["equity_multiplier"]["value_exact"]
|
||
- m["factors"]["equity_multiplier"]["value_exact"]) < 1e-9
|
||
check("A3 年化=true：周转率/ROE ×12/月数、净利率/权益乘数不变（同量纲之比相消）",
|
||
k == 12.0 and okat and okroe and oknpm and okem,
|
||
f"scale={k} at={okat} roe={okroe} npm={oknpm} em={okem}")
|
||
check("A3b 年化后恒等式仍 PASS", y.get("identity", {}).get("status") == "pass",
|
||
f"identity={y.get('identity')}")
|
||
|
||
|
||
def sec_a4(db):
|
||
with Section("A4"):
|
||
g = call(db, entity="hanke", period="2026-08", guard_probe="mismatch")
|
||
check("A4 故意错配区间 → identity_broken 且不返回值（fail-closed）",
|
||
g.get("identity", {}).get("status") == "identity_broken"
|
||
and g.get("identity_broken") is True and g.get("roe") is None
|
||
and "factors" in g,
|
||
f"status={g.get('identity', {}).get('status')} roe={g.get('roe')} "
|
||
f"diff={g.get('identity', {}).get('diff')}")
|
||
|
||
|
||
def sec_a5_a6(db):
|
||
for eid in (1, 2):
|
||
with Section(f"A5 entity{eid}"):
|
||
rows = db.execute(text("""SELECT kpi_code, data_source_type, status FROM kpi_definitions
|
||
WHERE entity_id=:e AND kpi_code IN
|
||
('F_ASSET_TOTAL','F_EQUITY_TOTAL','F_EQUITY_MULTIPLIER')"""),
|
||
{"e": eid}).fetchall()
|
||
got = {r[0]: (r[1], r[2]) for r in rows}
|
||
all_derived = (len(got) == 3 and all(v[0] == "derived" for v in got.values()))
|
||
vals = db.execute(text("""SELECT COUNT(DISTINCT k.kpi_code) FROM kpi_values v
|
||
JOIN kpi_definitions k ON k.id=v.kpi_id
|
||
WHERE k.entity_id=:e AND k.kpi_code IN
|
||
('F_ASSET_TOTAL','F_EQUITY_TOTAL','F_EQUITY_MULTIPLIER')
|
||
AND v.actual_value IS NOT NULL"""), {"e": eid}).scalar()
|
||
check(f"A5 entity{eid} 三码均为 derived 且有 kpi_values（注册 {len(got)}/3，有值 {vals}/3）",
|
||
all_derived and vals == 3, json.dumps(got, ensure_ascii=False))
|
||
with Section("A6"):
|
||
t = db.execute(text("""SELECT data_source_type FROM kpi_definitions
|
||
WHERE entity_id=1 AND kpi_code='F_ASSET_TURNOVER'""")).scalar()
|
||
check("A6 F_ASSET_TURNOVER data_source_type = derived", t == "derived", f"got={t}")
|
||
|
||
|
||
def sec_a7(db):
|
||
with Section("A7"):
|
||
f = call(db, entity="hanke", start="2026-01", end="2026-08", period="custom")
|
||
check("A7 期初缺失（2026-01 无 subject_balances 行）⇒ basis_fallback=true",
|
||
f.get("basis_fallback") is True and f.get("basis") == "ending"
|
||
and "期初" in (f.get("basis_fallback_reason") or ""),
|
||
f"basis={f.get('basis')} reason={f.get('basis_fallback_reason')}")
|
||
|
||
|
||
def sec_a8(db, m):
|
||
if not m:
|
||
check("A8 kpi_refs 镜像对比（前序 A2 未取到 payload，节略）", False, "")
|
||
return
|
||
with Section("A8"):
|
||
refs = m.get("kpi_refs", {})
|
||
exp = independent_recompute(db, 1, "2026-08", "2026-08", "3")
|
||
checks = [
|
||
("F_EQUITY_MULTIPLIER", refs.get("F_EQUITY_MULTIPLIER", {}).get("value"),
|
||
m["factors"]["equity_multiplier"]["value_exact"], "端点权益乘数"),
|
||
("F_ASSET_TURNOVER", refs.get("F_ASSET_TURNOVER", {}).get("value"),
|
||
m["factors"]["asset_turnover"]["value_exact"], "端点周转率"),
|
||
("F_ASSET_TOTAL", refs.get("F_ASSET_TOTAL", {}).get("value"),
|
||
exp["total_assets_end"], "独立复算期末总资产"),
|
||
("F_EQUITY_TOTAL", refs.get("F_EQUITY_TOTAL", {}).get("value"),
|
||
exp["equity_end"], "独立复算期末净资产"),
|
||
]
|
||
drift = [f"{c}:{got}!={want}({src})" for c, got, want, src in checks
|
||
if got is None or abs(float(got) - float(want)) > 0.02]
|
||
check("A8 kpi_refs 派生值与端点/KPI 层同口径一致（乘数/周转率对端点，总量对独立复算；容差 0.02）",
|
||
not drift, ";".join(drift) or "; ".join(
|
||
f"{c}={g}" for c, g, _, _ in checks))
|
||
|
||
|
||
def sec_a9(db):
|
||
with Section("A9"):
|
||
b = call(db, entity="bohai", period="2026-08", annualize=False)
|
||
bexp = independent_recompute(db, 2, "2026-08", "2026-08", "4",
|
||
revenue_codes=("6001", "6051", "6052"),
|
||
net_profit_mode="pnl_net")
|
||
okb = (abs((b["raw_data"]["revenue"] or 0) - bexp["revenue"]) < 1e-6
|
||
and abs((b["raw_data"]["net_profit"] or 0) - bexp["net_profit"]) < 1e-6
|
||
and abs((b["factors"]["equity_multiplier"]["value_exact"] or 0)
|
||
- (bexp["equity_multiplier"] or 0)) < 1e-5)
|
||
check("A9 entity2（新30号准则科目：4xxx权益/6xxx损益）独立复算一致", okb,
|
||
f"rev={b['raw_data']['revenue']}/{bexp['revenue']} np={b['raw_data']['net_profit']}"
|
||
f"/{bexp['net_profit']} em={b['factors']['equity_multiplier']['value_exact']}"
|
||
f"/{bexp['equity_multiplier']}")
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--check", action="store_true")
|
||
ap.add_argument("--json", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
db = get_session_local()()
|
||
try:
|
||
sec_a1(db)
|
||
m = sec_a2(db)
|
||
sec_a3(db, m)
|
||
sec_a4(db)
|
||
sec_a5_a6(db)
|
||
sec_a7(db)
|
||
sec_a8(db, m)
|
||
sec_a9(db)
|
||
except Exception: # 兜底（正常不应触发；各节已自行隔离）
|
||
traceback.print_exc()
|
||
check("顶层执行异常", False, traceback.format_exc().splitlines()[-1][:200])
|
||
|
||
passed = sum(1 for r in RESULTS if r["ok"])
|
||
total = len(RESULTS)
|
||
print(f"\n=== {passed}/{total} PASS ===")
|
||
if args.json:
|
||
print(json.dumps({"passed": passed, "total": total, "results": RESULTS},
|
||
ensure_ascii=False, indent=1))
|
||
db.close()
|
||
return 0 if passed == total else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|