210 lines
9.0 KiB
Python
210 lines
9.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""量纲守卫（Dimension Guard）—— 阈值与实测值跨量级比较时拒绝判定
|
||
|
||
依据：财务口径裁定 `finance-20260916-cma-caliber-ruling.md` §四（裁定③）
|
||
· 同量纲铁律：阈值表达式须与 kpi_definitions.unit 同量纲
|
||
· 量纲守卫：|log10(阈值)| 与 |log10(实测)| 相差 ≥1 个数量级 → **拒绝判定**，
|
||
标记「口径错误」，**不得输出红/黄/绿**
|
||
· 典型：流动比率实测 1.29（倍）vs 阈值 `<150`（百分数写法）→ 跨 2 个量级 → 假红
|
||
|
||
设计原则（**additive / opt-in**）：
|
||
· 本模块**不修改** `threshold_parser` 的任何既有行为，也不改现有调用方；
|
||
· 判定侧改用 `calc_level_guarded()` 即得守卫（返回 dict，含 level 与守卫明细）；
|
||
· 装配到 alert_generator / kpis.py 属生产行为变更，需审批后单独执行（本单只备代码）。
|
||
|
||
单位维度表（裁定 §4.1）：
|
||
· 倍（无量纲）→ ratio ：流动比率、速动比率、现金比率
|
||
· % → percent ：资产负债率、毛利率、净利率、保本覆盖率、返利率
|
||
· 天 / 次 → days / times：周转天数、周转率
|
||
· 万元 / 元 → currency_wan / currency_yuan
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from typing import Optional, Tuple, Union
|
||
|
||
from app.utils.threshold_parser import LEVEL_PRIORITY, eval_threshold, parse_clauses
|
||
|
||
#: 单位 → 维度
|
||
UNIT_DIMENSION = {
|
||
"倍": "ratio",
|
||
"": "ratio",
|
||
"%": "percent",
|
||
"百分比": "percent",
|
||
"万元": "currency_wan",
|
||
"元": "currency_yuan",
|
||
"天": "days",
|
||
"次": "times",
|
||
"人": "count",
|
||
"个": "count",
|
||
"家": "count",
|
||
"分": "score",
|
||
}
|
||
|
||
#: 默认拒绝阈值：量级差 ≥ 1.0 → 口径错误
|
||
DEFAULT_ORDERS_GAP = 1.0
|
||
|
||
#: 判定返回：口径错误时的 level 常量（不出红/黄/绿）
|
||
LEVEL_DIMENSION_ERROR = "口径错误"
|
||
|
||
|
||
def dimension_of(unit) -> str:
|
||
"""单位字符串 → 维度名（未知单位返回 'unknown'）。"""
|
||
if unit is None:
|
||
return "unknown"
|
||
return UNIT_DIMENSION.get(str(unit).strip(), "unknown")
|
||
|
||
|
||
def _orders(v) -> Optional[float]:
|
||
"""数值量级 = log10(|v|)；0 / None / 非数值 → None（不可判）。"""
|
||
if v is None:
|
||
return None
|
||
try:
|
||
f = float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if f == 0:
|
||
return None
|
||
return math.log10(abs(f))
|
||
|
||
|
||
def _clause_values(clauses) -> list:
|
||
"""子句 → 候选比较值列表（区间取两端）。"""
|
||
vals = []
|
||
for _op, val in clauses:
|
||
if isinstance(val, tuple):
|
||
vals.extend(list(val))
|
||
else:
|
||
vals.append(val)
|
||
return vals
|
||
|
||
|
||
def check_dimension(actual, expr, unit=None, orders_gap: float = DEFAULT_ORDERS_GAP) -> dict:
|
||
"""检查「实测值 vs 阈值表达式」是否同量级。
|
||
|
||
返回 dict：
|
||
ok True=可判定；False=口径错误（拒绝判定）
|
||
verdict 'ok' | 'dimension_error' | 'undecidable'
|
||
gap 与最接近子句的量级差（None=不可判）
|
||
nearest 最接近的子句值
|
||
reason 人类可读说明
|
||
"""
|
||
if actual is None or expr is None or not str(expr).strip():
|
||
return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None,
|
||
"reason": "无实测值或无阈值 → 守卫不介入", "dimension": dimension_of(unit)}
|
||
|
||
a = _orders(actual)
|
||
clauses = parse_clauses(expr)
|
||
if a is None:
|
||
return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None,
|
||
"reason": "实测值为 0（量级不可判）→ 守卫不介入", "dimension": dimension_of(unit)}
|
||
if not clauses:
|
||
return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None,
|
||
"reason": "阈值表达式无法解析 → 守卫不介入（解析器自身会返回 False）",
|
||
"dimension": dimension_of(unit)}
|
||
|
||
best_gap, nearest = None, None
|
||
for v in _clause_values(clauses):
|
||
o = _orders(v)
|
||
if o is None:
|
||
continue
|
||
g = abs(a - o)
|
||
if best_gap is None or g < best_gap:
|
||
best_gap, nearest = g, v
|
||
|
||
if best_gap is None:
|
||
return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None,
|
||
"reason": "阈值子句全为 0 → 守卫不介入", "dimension": dimension_of(unit)}
|
||
|
||
if best_gap >= orders_gap:
|
||
return {"ok": False, "verdict": "dimension_error", "gap": round(best_gap, 3),
|
||
"nearest": nearest,
|
||
"reason": ("量级差 %.2f ≥ %.1f 个数量级：实测 %s（单位 %s）vs 阈值 %r（最近子句 %s）"
|
||
" → 跨量级比较，判定作废" % (best_gap, orders_gap, actual, unit, expr, nearest)),
|
||
"dimension": dimension_of(unit)}
|
||
return {"ok": True, "verdict": "ok", "gap": round(best_gap, 3), "nearest": nearest,
|
||
"reason": "量级差 %.2f < %.1f → 同量级可判" % (best_gap, orders_gap),
|
||
"dimension": dimension_of(unit)}
|
||
|
||
|
||
def calc_level_guarded(actual, kpi, orders_gap: float = DEFAULT_ORDERS_GAP) -> dict:
|
||
"""带量纲守卫的红黄绿判定（不改 threshold_parser 的原行为）。
|
||
|
||
kpi 需含 threshold_green/yellow/red 与 unit 属性（ORM 对象或同名 duck type）。
|
||
返回 {"level": 'red'|'yellow'|'green'|'info'|'口径错误', "guard": {...}, "expr": 命中的表达式}
|
||
口径错误：level = LEVEL_DIMENSION_ERROR，不出红/黄/绿（裁定 §4.2）。
|
||
"""
|
||
exprs = {
|
||
"green": getattr(kpi, "threshold_green", None),
|
||
"yellow": getattr(kpi, "threshold_yellow", None),
|
||
"red": getattr(kpi, "threshold_red", None),
|
||
}
|
||
unit = getattr(kpi, "unit", None)
|
||
if actual is None:
|
||
return {"level": "info", "guard": {"ok": True, "verdict": "undecidable",
|
||
"reason": "无实测值", "dimension": dimension_of(unit)},
|
||
"expr": None}
|
||
|
||
# 逐等级守卫：任一等级被守卫生效（口径错误）→ 不进红/黄/绿
|
||
for level in LEVEL_PRIORITY:
|
||
expr = exprs[level]
|
||
if not expr:
|
||
continue
|
||
g = check_dimension(actual, expr, unit, orders_gap)
|
||
if not g["ok"]:
|
||
return {"level": LEVEL_DIMENSION_ERROR, "guard": g, "expr": expr,
|
||
"level_blocked": level}
|
||
|
||
for level in LEVEL_PRIORITY:
|
||
expr = exprs[level]
|
||
if expr and eval_threshold(actual, expr):
|
||
return {"level": level, "guard": check_dimension(actual, expr, unit, orders_gap),
|
||
"expr": expr}
|
||
return {"level": "info", "guard": {"ok": True, "verdict": "ok", "reason": "未命中任何阈值",
|
||
"dimension": dimension_of(unit)}, "expr": None}
|
||
|
||
|
||
class _K:
|
||
"""self-test 用的最小 KPI 载体。"""
|
||
|
||
def __init__(self, unit, green=None, yellow=None, red=None):
|
||
self.unit, self.threshold_green, self.threshold_yellow, self.threshold_red = \
|
||
unit, green, yellow, red
|
||
|
||
|
||
def _self_test() -> int:
|
||
"""守卫自测（零库连接）：假红必拦、真红必放、真黄必放。"""
|
||
cases: Tuple[Tuple[str, float, "_K", str], ...] = (
|
||
# (说明, 实测, KPI, 期望 level)
|
||
("假红·流动比率 1.29倍 vs <150（跨2量级）", 1.29, _K("%", ">=200", ">=150", "<150"),
|
||
LEVEL_DIMENSION_ERROR),
|
||
("假红·速动比率 0.54倍 vs <80（跨2量级）", 0.54, _K("%", ">=100", ">=80", "<80"),
|
||
LEVEL_DIMENSION_ERROR),
|
||
("真红·返利率 48.44% vs >12（同量级）", 48.44, _K("%", "<=10", ">10", ">12"), "red"),
|
||
("真红·保本覆盖率 60.4% vs <85（同量级）", 60.4, _K("%", ">=100", ">=85", "<85"), "red"),
|
||
("真红·应收周转天数 610.2天 vs >91（同量级）", 610.2, _K("天", None, None, ">91"), "red"),
|
||
("真红·收入增长率 -82.2% vs <15（同量级）", -82.2, _K("%", None, None, "<15"), "red"),
|
||
("修正后·流动比率 1.29倍 vs 红<1.0/黄>=1.0/绿>=2.0", 1.29,
|
||
_K("倍", ">=2.0", ">=1.0", "<1.0"), "yellow"),
|
||
("修正后·速动比率 0.54倍 vs 红<0.5/黄>=0.5/绿>=1.0", 0.54,
|
||
_K("倍", ">=1.0", ">=0.5", "<0.5"), "yellow"),
|
||
("修正后·流动比率 0.8倍 → 红", 0.8, _K("倍", ">=2.0", ">=1.0", "<1.0"), "red"),
|
||
("修正后·流动比率 2.5倍 → 绿", 2.5, _K("倍", ">=2.0", ">=1.0", "<1.0"), "green"),
|
||
("零值不误拦·实测 0 无守卫介入", 0.0, _K("%", ">=100", ">=85", "<85"), "red"),
|
||
("无阈值 → info", 5.0, _K("倍"), "info"),
|
||
)
|
||
ok = 0
|
||
print("=== 量纲守卫 self-test ===")
|
||
for desc, actual, kpi, expect in cases:
|
||
r = calc_level_guarded(actual, kpi)
|
||
flag = "PASS" if r["level"] == expect else "FAIL"
|
||
ok += (r["level"] == expect)
|
||
print(" [%s] %-46s → %-6s (期望 %s) gap=%s" % (
|
||
flag, desc, r["level"], expect, r["guard"].get("gap")))
|
||
print("=== %d/%d PASS ===" % (ok, len(cases)))
|
||
return 0 if ok == len(cases) else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(_self_test())
|