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 百分比口径
232 lines
12 KiB
Python
232 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""阈值单位口径修正 + 量纲守卫装配（迁移脚本·准备态）
|
||
|
||
⚠️ **已被取代，不得执行（2026-09-17 首次判定；2026-09-24 复判并补新锚）**：
|
||
本脚本基于 2026-09-16 的口径草案（**把阈值改成「倍」写法** >=2.0/>=1.0/<1.0）。
|
||
**状态更新（2026-09-24）**：
|
||
· 现行有效裁定 = `wecom-finance/reports/CMA_口径裁定_20260924_比率类单位口径_取代2a两指标.md`
|
||
（口径标识 `CMA-CALIBER-v2.1/kpi-unit@2026-09-24`；任总 2026-09-24 16:38 放行 ASK-20260924-11）。
|
||
该件**只取代** `CMA_口径裁定_20260917.md` §2a 中的 **`F_CURRENT_RATIO` / `F_QUICK_RATIO`**
|
||
两项（改「倍」、无量纲直判）；**其余比率类仍按 20260917 裁定2a（百分比 %）执行**。
|
||
· 本脚本 `PLAN` 里这两项的「倍」方向**与 v2.1 一致**，但**已由 ASK-20260924-11 单以受控路径落地**
|
||
（DB 写经受控脚本 + 判定层 `app/utils/caliber_units.py` 同量纲直判 + 代码测试发布），
|
||
**不得再执行本脚本**；本脚本 `PLAN` 另含 **`F_DEBT_RATIO`（不在 v2.1 取代范围内）**，
|
||
执行会越界改动他项 ⇒ 仅保留作审计链（历史草案）。
|
||
· `--resolve-false-reds` 的旧语义（586/587 按「口径错误」**作废**）**已作废**：v2.1 下该两条按
|
||
**口径变更改判**（1.29 倍 → 黄、0.54 倍 → 黄），留痕写明「口径变更所致」，**不删历史行**。
|
||
· 如未来另有阈值单位新裁定，须先改 `PLAN` 与 `--resolve-false-reds` 的语义再执行。
|
||
|
||
原依据：财务口径裁定 finance-20260916-cma-caliber-ruling.md v1.0 §四（裁定③）+ §3.3/§3.4
|
||
· 4.1 单位规范：偿债能力比率=**倍**（无量纲）；资产负债率等=%；返利率=%
|
||
· 4.3 具体阈值裁定（贸易型：白酒经销 / IT 经销）
|
||
· 3.3 09-09 假红灯 586/587 按「口径错误」作废并重算
|
||
· 3.4 切换顺序不可颠倒：先单位 → 再期间 → 后调度（本脚本只做第 1 步）
|
||
|
||
默认 dry-run（只读：列改动 + 影响面复算，不写库）。`--apply` 才写（**需审批**）。
|
||
|
||
用法：
|
||
python3 scripts/migrate_kpi_threshold_unit_20260916.py # dry-run
|
||
python3 scripts/migrate_kpi_threshold_unit_20260916.py --apply # 写阈值/单位（需放行）
|
||
python3 scripts/migrate_kpi_threshold_unit_20260916.py --apply --resolve-false-reds
|
||
# 同时作废假红灯 586/587
|
||
说明：
|
||
· 影响面口径 = 用阈值复算「最近可用期间(<= 最新已入库月)」的命中条数，
|
||
分别给「修正前 / 修正后」两组数字（裁定 §4.5 要求：影响面以修正后复算值为准）。
|
||
· 返利率（F_REBATE_RATE）**不改阈值**（量纲一致、判定有效），仅标注口径待确认（裁定 §4.4）。
|
||
"""
|
||
import argparse
|
||
import sys
|
||
import os
|
||
|
||
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.dimension_guard import check_dimension # noqa: E402
|
||
from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY # noqa: E402
|
||
|
||
ENTITY_ID = 1 # 酣客（白酒经销）；博海（IT）无阈值 KPI 待核
|
||
|
||
#: 迁移计划：kpi_code → 目标 unit / 三档阈值（None = 不改该字段）
|
||
PLAN = {
|
||
"F_CURRENT_RATIO": {"name": "流动比率", "unit": "倍",
|
||
"green": ">=2.0", "yellow": ">=1.0", "red": "<1.0",
|
||
"why": "裁定 §4.3：经验值 2:1；<1 才够紧急（旧 <150 为百分数写法）"},
|
||
"F_QUICK_RATIO": {"name": "速动比率", "unit": "倍",
|
||
"green": ">=1.0", "yellow": ">=0.5", "red": "<0.5",
|
||
"why": "裁定 §4.3：经验值 1:1（旧 <80 为百分数写法）"},
|
||
"F_DEBT_RATIO": {"name": "资产负债率", "unit": "%",
|
||
"green": "<=60", "yellow": ">60", "red": ">70",
|
||
"why": "裁定 §4.3：红 >70 / 黄 60–70 / 绿 ≤60"},
|
||
}
|
||
#: 明确不改、但需标注的项
|
||
UNCHANGED_NOTES = {
|
||
"F_REBATE_RATE": ("返利率", "%", "裁定 §4.4：量纲一致、判定有效，不改阈值；"
|
||
"48.44% 口径本身待确认 → 预警保留并标注「口径待确认」"),
|
||
}
|
||
|
||
#: 假红灯（09-09 单位错配产生）—— 裁定 §3.3：按「口径错误」作废
|
||
FALSE_RED_ALERT_IDS = (586, 587)
|
||
VOID_RESOLUTION = "[口径错误·作废] 阈值与实测值跨量级比较（流动/速动比率单位错配），" \
|
||
"非财务事实；由 finance-20260916-cma-caliber-ruling §3.3 裁定作废，待单位修正后重算。"
|
||
|
||
|
||
def q(sql, params=None):
|
||
with get_engine().connect() as c:
|
||
return [dict(r._mapping) for r in c.execute(text(sql), params or {}).fetchall()]
|
||
|
||
|
||
def exec_sql(sql, params=None):
|
||
with get_engine().begin() as c:
|
||
return c.execute(text(sql), params or {})
|
||
|
||
|
||
def fetch_plan_rows():
|
||
codes = list(PLAN.keys()) + list(UNCHANGED_NOTES.keys())
|
||
ph = ",".join(":%s" % ("c%d" % i) for i in range(len(codes)))
|
||
return q("SELECT id, kpi_code, kpi_name, unit, threshold_green, threshold_yellow, threshold_red, status "
|
||
"FROM kpi_definitions WHERE entity_id=:e AND kpi_code IN (%s) ORDER BY id" % ph,
|
||
dict({"e": ENTITY_ID}, **{("c%d" % i): c for i, c in enumerate(codes)}))
|
||
|
||
|
||
def latest_period():
|
||
r = q("SELECT MAX(period) AS p FROM kpi_values WHERE period REGEXP '^[0-9]{4}-[0-9]{2}$'")
|
||
return r[0]["p"] if r else None
|
||
|
||
|
||
def impact():
|
||
"""影响面复算：最近可用期间上，各 KPI 在「修正前 / 修正后」的命中条数。"""
|
||
p = latest_period()
|
||
rows = q(
|
||
"SELECT d.id, d.kpi_code, d.kpi_name, d.unit, d.threshold_green AS g, "
|
||
"d.threshold_yellow AS y, d.threshold_red AS r, v.actual_value AS actual, v.period "
|
||
"FROM kpi_definitions d JOIN kpi_values v ON v.kpi_id = d.id "
|
||
"WHERE d.entity_id=:e AND d.status='active' AND v.period = :p AND v.actual_value IS NOT NULL "
|
||
"AND (d.threshold_red IS NOT NULL OR d.threshold_yellow IS NOT NULL OR d.threshold_green IS NOT NULL) "
|
||
"ORDER BY d.id", {"e": ENTITY_ID, "p": p})
|
||
|
||
def judge(actual, unit, g, y, r):
|
||
exprs = {"green": g, "yellow": y, "red": r}
|
||
guard_hit = None
|
||
for lv in LEVEL_PRIORITY:
|
||
if exprs[lv]:
|
||
gd = check_dimension(actual, exprs[lv], unit)
|
||
if not gd["ok"]:
|
||
guard_hit = gd
|
||
break
|
||
if guard_hit:
|
||
return "口径错误"
|
||
for lv in LEVEL_PRIORITY:
|
||
if exprs[lv] and eval_threshold(actual, exprs[lv]):
|
||
return lv
|
||
return "info"
|
||
|
||
before, after, detail = {"red": 0, "yellow": 0, "green": 0, "info": 0, "口径错误": 0}, \
|
||
{"red": 0, "yellow": 0, "green": 0, "info": 0, "口径错误": 0}, []
|
||
for x in rows:
|
||
plan = PLAN.get(x["kpi_code"])
|
||
g2 = plan["green"] if plan else x["g"]
|
||
y2 = plan["yellow"] if plan else x["y"]
|
||
r2 = plan["red"] if plan else x["r"]
|
||
u2 = plan["unit"] if plan else x["unit"]
|
||
b = judge(x["actual"], x["unit"], x["g"], x["y"], x["r"])
|
||
a = judge(x["actual"], u2, g2, y2, r2)
|
||
before[b] = before.get(b, 0) + 1
|
||
after[a] = after.get(a, 0) + 1
|
||
if b != a:
|
||
detail.append({"kpi": "%s/%s" % (x["kpi_code"], x["kpi_name"]), "period": x["period"],
|
||
"actual": x["actual"], "unit": x["unit"],
|
||
"before": b, "after": a,
|
||
"thresholds": "%s|%s|%s → %s|%s|%s" % (x["g"], x["y"], x["r"], g2, y2, r2)})
|
||
return p, before, after, detail
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--apply", action="store_true", help="真正写库（缺省=dry-run）")
|
||
ap.add_argument("--resolve-false-reds", action="store_true",
|
||
help="同时把假红灯 %s 按口径错误作废（status=resolved）" % (FALSE_RED_ALERT_IDS,))
|
||
args = ap.parse_args()
|
||
|
||
print("=== 阈值单位口径修正 %s | entity_id=%d ===" % ("APPLY" if args.apply else "DRY-RUN", ENTITY_ID))
|
||
rows = fetch_plan_rows()
|
||
if not rows:
|
||
print("!! 未取到目标 KPI 行（kpi_code 不匹配？）")
|
||
return 2
|
||
|
||
print("\n-- 计划改动 --")
|
||
todo = []
|
||
for x in rows:
|
||
plan = PLAN.get(x["kpi_code"])
|
||
if not plan:
|
||
note = UNCHANGED_NOTES.get(x["kpi_code"])
|
||
print(" [不改] id=%-4s %-6s %-8s unit=%s 阈值=%s|%s|%s ← %s"
|
||
% (x["id"], x["kpi_code"], x["kpi_name"], x["unit"],
|
||
x["threshold_green"], x["threshold_yellow"], x["threshold_red"],
|
||
note[2] if note else ""))
|
||
continue
|
||
changed = (x["unit"] != plan["unit"] or x["threshold_green"] != plan["green"]
|
||
or x["threshold_yellow"] != plan["yellow"] or x["threshold_red"] != plan["red"])
|
||
print(" [%s] id=%-4s %-6s %-8s unit %s→%s 阈值 %s|%s|%s → %s|%s|%s"
|
||
% ("改" if changed else "已符", x["id"], x["kpi_code"], x["kpi_name"],
|
||
x["unit"], plan["unit"], x["threshold_green"], x["threshold_yellow"], x["threshold_red"],
|
||
plan["green"], plan["yellow"], plan["red"]))
|
||
print(" 理由: %s" % plan["why"])
|
||
if changed:
|
||
todo.append((x["id"], plan))
|
||
|
||
print("\n-- 影响面复算（最近可用期间） --")
|
||
p, before, after, detail = impact()
|
||
print(" 期间 P* = %s" % p)
|
||
print(" 修正前: 红 %d / 黄 %d / 绿 %d / 无判定 %d / 口径错误 %d"
|
||
% (before["red"], before["yellow"], before["green"], before["info"], before["口径错误"]))
|
||
print(" 修正后: 红 %d / 黄 %d / 绿 %d / 无判定 %d / 口径错误 %d"
|
||
% (after["red"], after["yellow"], after["green"], after["info"], after["口径错误"]))
|
||
print(" 条数变化（仅列变化项）:")
|
||
for d in detail:
|
||
print(" %-22s %s actual=%s unit=%s %s → %s [%s]"
|
||
% (d["kpi"], d["period"], d["actual"], d["unit"], d["before"], d["after"], d["thresholds"]))
|
||
print(" ※ 影响面以「修正后」为准（裁定 §4.5）；此处红/黄为静态阈值复算上界，"
|
||
"实际落库条数还受期间口径与幂等约束。")
|
||
|
||
if args.resolve_false_reds:
|
||
print("\n-- 假红灯作废 --")
|
||
ph = ",".join(":a%d" % i for i in range(len(FALSE_RED_ALERT_IDS)))
|
||
al = q("SELECT id, kpi_id, alert_level, status, alert_message FROM kpi_alerts "
|
||
"WHERE id IN (%s)" % ph,
|
||
{("a%d" % i): v for i, v in enumerate(FALSE_RED_ALERT_IDS)})
|
||
for a in al:
|
||
print(" id=%s level=%s status=%s msg=%s" % (a["id"], a["alert_level"], a["status"],
|
||
(a["alert_message"] or "")[:70]))
|
||
if not al:
|
||
print(" （未取到该两条预警，可能已清理）")
|
||
|
||
if not args.apply:
|
||
print("\n=== dry-run 结束：未写库（%d 项待改，加 --apply 落库 / 需审批）===" % len(todo))
|
||
return 0
|
||
|
||
for kid, plan in todo:
|
||
exec_sql("UPDATE kpi_definitions SET unit=:u, threshold_green=:g, threshold_yellow=:y, "
|
||
"threshold_red=:r WHERE id=:i AND entity_id=:e",
|
||
{"u": plan["unit"], "g": plan["green"], "y": plan["yellow"],
|
||
"r": plan["red"], "i": kid, "e": ENTITY_ID})
|
||
if args.resolve_false_reds:
|
||
exec_sql("UPDATE kpi_alerts SET status='resolved', resolution=:res, resolved_at=NOW() "
|
||
"WHERE id IN (%s)" % ",".join(":a%d" % i for i in range(len(FALSE_RED_ALERT_IDS))),
|
||
dict({("a%d" % i): v for i, v in enumerate(FALSE_RED_ALERT_IDS)},
|
||
res=VOID_RESOLUTION))
|
||
print("\n=== 已提交：%d 项阈值/单位%s ===" % (len(todo),
|
||
"；已作废假红灯 %s" % (FALSE_RED_ALERT_IDS,) if args.resolve_false_reds else ""))
|
||
print("-- 写入后回查 --")
|
||
for x in fetch_plan_rows():
|
||
if x["kpi_code"] in PLAN:
|
||
print(" id=%-4s %-6s unit=%s 阈值=%s|%s|%s"
|
||
% (x["id"], x["kpi_code"], x["unit"], x["threshold_green"],
|
||
x["threshold_yellow"], x["threshold_red"]))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|