feat(dupont): period-basis design (cumulative flow x average stock, optional annualize, identity guard)
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).
This commit is contained in:
Hermes CI Fix
2026-10-04 13:35:25 +08:00
parent 1f1385ebbf
commit 38a6de9325
7 changed files with 1435 additions and 166 deletions
+92
View File
@@ -0,0 +1,92 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""杜邦 KPI 定义种子应用器 —— 执行 scripts/sql/dupont_kpi_defs_20261003.sql
任务:finance-20261003-dupont-period-basis-design-fullstack.md
动作:① 两账套注册 F_ASSET_TOTAL/F_EQUITY_TOTAL/F_EQUITY_MULTIPLIER(derived)
② F_ASSET_TURNOVER data_source_type: manual → derived
幂等:SQL 自带条件 INSERT(NOT EXISTS)+ 条件 UPDATE,可安全重跑。
用法(**写库**,须先过 approval-gate 放行):
venv/bin/python3 scripts/apply_dupont_kpi_defs.py --check # 只读预检(不执行任何语句)
venv/bin/python3 scripts/apply_dupont_kpi_defs.py --apply # 执行(需放行)
"""
from __future__ import annotations
import argparse
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import text # noqa: E402
from app.database import get_engine # noqa: E402
SQL_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"sql", "dupont_kpi_defs_20261003.sql")
TARGETS = ("F_ASSET_TOTAL", "F_EQUITY_TOTAL", "F_EQUITY_MULTIPLIER")
TARGET_ENTITIES = (1, 2)
def statements(sql_text: str):
"""按 ASCII ';' 切分(本文件的字符串字面量内不含 ';',已人工核对)"""
out = []
for chunk in sql_text.split(";"):
stmt = chunk.strip()
if stmt and not all(ln.strip().startswith("--") or not ln.strip() for ln in stmt.splitlines()):
out.append(stmt)
return out
def snapshot(conn) -> dict:
rows = conn.execute(text("""SELECT entity_id, kpi_code, data_source_type, status
FROM kpi_definitions
WHERE entity_id IN (1,2) AND kpi_code IN
('F_ASSET_TOTAL','F_EQUITY_TOTAL','F_EQUITY_MULTIPLIER',
'F_ASSET_TURNOVER')
ORDER BY entity_id, kpi_code""")).fetchall()
return {f"{r[0]}/{r[1]}": (r[2], r[3]) for r in rows}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--check", action="store_true", help="只读预检(不执行写语句)")
ap.add_argument("--apply", action="store_true", help="执行种子 SQL(写库)")
args = ap.parse_args()
if not (args.check or args.apply):
print("须指定 --check 或 --apply", file=sys.stderr)
return 2
eng = get_engine()
with open(SQL_FILE, encoding="utf-8") as f:
stmts = statements(f.read())
print(f"SQL 文件: {SQL_FILE}")
print(f"语句数: {len(stmts)}")
for i, s in enumerate(stmts, 1):
head = " ".join(s.split())[:110]
print(f" [{i}] {head}...")
with eng.connect() as conn:
before = snapshot(conn)
print(f"\n写前像({len(before)} 行): {before}")
if args.check:
print("\n[--check] 只读预检:未执行任何语句(写路径未触发)")
return 0
for i, s in enumerate(stmts, 1):
r = conn.execute(text(s))
print(f" [{i}] rowcount={r.rowcount}")
conn.commit()
after = snapshot(conn)
print(f"\n写后像({len(after)} 行): {after}")
added = {k: v for k, v in after.items() if k not in before}
changed = {k: (before[k], v) for k, v in after.items() if k in before and before[k] != v}
print(f"新增: {added}")
print(f"变更: {changed}")
missing = [f"{e}/{c}" for e in TARGET_ENTITIES for c in TARGETS if f"{e}/{c}" not in after]
print(f"缺件: {missing or '无'}")
return 0
if __name__ == "__main__":
sys.exit(main())