wecom-fullstack
a8391a03da
ci/woodpecker/push/main Pipeline failed
- caliber_units: F_CURRENT_RATIO/F_QUICK_RATIO 由 FACTOR_PERCENT(×100) 退场为 FACTOR_IDENTITY；declared_unit=倍 / caliber=倍（无量纲）/ evidence 改引 v2.1 锚；F_GROSS_MARGIN 保持 ×100 不动 - 引用面同步 7 文件：alert_generator / kpi_unit_crosscheck / migrate_...20260916 横幅 / 三个测试套件；self-test 两条改倍口径期望 yellow 口径: CMA-CALIBER-v2.1/kpi-unit@2026-09-24 取代 2a 之两项；其余比率类仍按裁定2a 百分比口径
106 lines
4.4 KiB
Python
106 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""比率类 KPI 单位对照表（只读）—— 口径 = 裁定2a + 2026-09-24 v2.1（取代两项）
|
||
|
||
依据：`wecom-finance/reports/CMA_口径裁定_20260924_比率类单位口径_取代2a两指标.md`
|
||
（`CMA-CALIBER-v2.1/kpi-unit@2026-09-24`）—— 取代 `wecom-finance/reports/CMA_口径裁定_20260917.md`
|
||
§2a 中的 `F_CURRENT_RATIO` / `F_QUICK_RATIO` **两项**（改「倍」、无量纲直判、不再 ×100）；
|
||
其余比率类指标仍按 20260917 裁定2a（百分比 %）执行。
|
||
|
||
用途：逐个比率类 KPI 核对「定义 / 声明单位 / 阈值量纲 / 实测值量纲」，出对照表。
|
||
两类口径在表里分栏可见：`factor=1` + `unit=倍` = 倍口径同量纲直判；`factor=100` = 裁定2a 百分比归一。
|
||
判定侧的事实源是 `app/utils/caliber_units.RATIO_CALIBER_TABLE`（本脚本只核对与呈现，
|
||
不写库、不改阈值）。
|
||
|
||
用法：
|
||
python3 scripts/kpi_unit_crosscheck.py # 全账套
|
||
python3 scripts/kpi_unit_crosscheck.py --entity-id 1 # 单账套
|
||
python3 scripts/kpi_unit_crosscheck.py --json # JSON（机读）
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
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
|
||
from app.utils.caliber_units import resolve_caliber # noqa: E402
|
||
|
||
PERIOD_RE = re.compile(r"^[0-9]{4}-[0-9]{2}$")
|
||
|
||
|
||
def q(sql, **params):
|
||
with get_engine().connect() as c:
|
||
return [dict(r._mapping) for r in c.execute(text(sql), params).fetchall()]
|
||
|
||
|
||
class _K(dict):
|
||
"""把 kpi 行包成 duck type（resolve_caliber 只读属性）"""
|
||
|
||
def __getattr__(self, item):
|
||
return self.get(item)
|
||
|
||
|
||
def latest_value(kpi_id):
|
||
rows = q("SELECT period, actual_value FROM kpi_values WHERE kpi_id=:k AND actual_value IS NOT NULL "
|
||
"ORDER BY period DESC, calculated_at DESC LIMIT 20", k=kpi_id)
|
||
for r in rows:
|
||
if PERIOD_RE.fullmatch(str(r["period"] or "")):
|
||
return r["period"], r["actual_value"]
|
||
return None, None
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--entity-id", type=int, default=None)
|
||
ap.add_argument("--json", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
where = "WHERE d.status='active' AND (d.threshold_red IS NOT NULL OR d.threshold_yellow IS NOT NULL " \
|
||
"OR d.threshold_green IS NOT NULL)"
|
||
params = {}
|
||
if args.entity_id:
|
||
where += " AND d.entity_id=:e"
|
||
params["e"] = args.entity_id
|
||
rows = q("SELECT d.id, d.entity_id, d.kpi_code, d.kpi_name, d.unit, d.threshold_green, "
|
||
"d.threshold_yellow, d.threshold_red FROM kpi_definitions d %s "
|
||
"ORDER BY d.entity_id, d.id" % where, **params)
|
||
|
||
out = []
|
||
for r in rows:
|
||
p, v = latest_value(r["id"])
|
||
cal = resolve_caliber(v, _K(r))
|
||
out.append({
|
||
"kpi_code": r["kpi_code"], "kpi_name": r["kpi_name"], "entity_id": r["entity_id"],
|
||
"declared_unit": r["unit"],
|
||
"thresholds": "%s|%s|%s" % (r["threshold_green"], r["threshold_yellow"], r["threshold_red"]),
|
||
"latest_period": p, "actual": v,
|
||
"caliber_status": cal["status"], "norm_factor": cal["factor"],
|
||
"normalized": cal["value"], "note": cal["note"], "reason": cal["reason"],
|
||
})
|
||
|
||
if args.json:
|
||
print(json.dumps(out, ensure_ascii=False, indent=1, default=str))
|
||
return 0
|
||
|
||
print("=== 比率类 KPI 单位对照表（只读；口径 = 2a + v2.1 取代两项）===")
|
||
print("%-4s %-8s %-22s %-6s %-10s %-12s %-8s %-6s %s" % (
|
||
"eid", "code", "name", "unit", "latest", "actual", "status", "factor", "thresholds"))
|
||
for x in out:
|
||
print("%-4s %-8s %-22s %-6s %-10s %-12s %-18s %-6s %s" % (
|
||
x["entity_id"], x["kpi_code"], x["kpi_name"][:20], x["declared_unit"] or "-",
|
||
x["latest_period"] or "-", x["actual"] if x["actual"] is not None else "-",
|
||
x["caliber_status"], x["norm_factor"], x["thresholds"]))
|
||
n_norm = sum(1 for x in out if x["norm_factor"] != 1)
|
||
n_sus = sum(1 for x in out if x["caliber_status"] == "dimension_suspect")
|
||
print("=== 合计 %d 项：口径归一 ×100 %d 项 / 同量纲直判 %d 项 / 疑似错配降级待核 %d 项 ===" % (
|
||
len(out), n_norm, len(out) - n_norm - n_sus, n_sus))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|