feat(cma): 口径裁定 2026-09-17 落地(现金流投资=0 二维判定 / 预警单位+期间口径)
ci/woodpecker/push/main Pipeline failed

依据:wecom-finance/reports/CMA_口径裁定_20260917.md(任总 2026-09-17 23:14「按建议执行」)

裁定1(现金流「投资活动净额=0」参与判定)
- classify_enterprise():0 为合法输入;投资恒为 0 → 经营(±)×筹资(±) 二维映射出 2 个
  候选类型(candidate_type_nos/candidates),不再整体判「未判定」;投资栏 investing_note
  标注「本期无投资活动」;经营/筹资亦为 0 → 回退「未判定(无投资活动)」,不报错不静默
- /analysis 增 classification_dimension / investing_note / nets.investing.note;前端出
  二维横幅 + 候选行高亮 + 投资栏「本期无投资活动」(不给单一类型名)
- import_cashflow_statements.py:打印层修 2D(type_no=None)TypeError 崩溃;NA 措辞改为
  「按 0 参与判定」(机读短语 NA_PHRASE_TMPL/parse_na_activities 未动)

裁定2a(比率类单位统一「百分比(%)」口径,先做)
- 新增 app/utils/caliber_units.py:逐 KPI 对照表 RATIO_CALIBER_TABLE(唯一事实源)+ 判定侧
  唯一入口 resolve_caliber();3 项需归一 ×100(流动/速动/毛利率),12 项取数已百分数不归一;
  表外且命中「疑似错配」窄形态(阈值子句量级≥100 且 0<|实测|<10 且单位属比率类)→
  dimension_suspect:不落预警、不推送(唯一规则,防 586/587 假红灯复发)
- alert_generator:判定前归一,文案写「口径归一 ×100(原值 1.29 → 129%)」留审计线索
- 新增 scripts/kpi_unit_crosscheck.py(只读对照表产出器)

裁定2b(期间由「当月硬编码」→「最近可用期间」,后做)
- resolve_latest_period():取该 KPI 自己的最新 YYYY-MM(禁全局 MAX,防演示账套劫持);
  伪期间(2026-H1/H2、2025-FY)在 Python 侧过滤(保「单测=生产」语义)
- 过渡期覆盖度自检扩展:记录回落后的期间分布 + 当月 0 条归因提示
- --dry-run(零写入零推送)作放行前抽查证据;generate_and_push/run_alert_check 支持 dry_run

测试:backend/tests/test_caliber_rulings_20260917.py 17 项;全量 1158 passed/1 xfailed
This commit is contained in:
Hermes CI Fix
2026-09-18 00:34:04 +08:00
parent e567d1b94f
commit 3bb55189b1
8 changed files with 923 additions and 79 deletions
+98 -15
View File
@@ -94,9 +94,26 @@ CASHFLOW_TYPE_RULES = [
# 类型 → 规则 快速索引
TYPE_RULE_MAP = {r["type_no"]: r for r in CASHFLOW_TYPE_RULES}
# ============================================================
# 裁定1(2026-09-17 财务口径裁定):投资活动净额 = 0 的合法输入与二维映射
# 「投资活动净额=0」视为**本期无投资活动**,按 0 如实参与八类判定;
# 三项中投资恒为 0 → 按 经营(±) × 筹资(±) **二维映射**到对应类型(候选 2 个,
# 因投资符号未知);投资栏位标注「本期无投资活动」。
# 依据:wecom-finance/reports/CMA_口径裁定_20260917.md 裁定1
# ============================================================
INVESTING_ZERO_NOTE = "本期无投资活动"
#: (经营符号, 筹资符号) → 投资符号未知时的候选类型号(对应类型由 8 类规则表反查)
TWO_DIM_CANDIDATE_MAP = {
("+", "-"): (1, 2),
("+", "+"): (3, 4),
("-", "-"): (5, 8),
("-", "+"): (6, 7),
}
def _sign(value: Optional[float]) -> Optional[str]:
"""净额符号:>0 → +,<0 → -,==0 → '0'(无法落入8类组合),None → None"""
"""净额符号:>0 → +,<0 → -,==0 → '0',None → None"""
if value is None:
return None
if value > 0:
@@ -106,27 +123,86 @@ def _sign(value: Optional[float]) -> Optional[str]:
return "0"
def _rule_brief(type_no: int) -> dict:
"""类型号 → 规则摘要(二维判定用的候选项,字段与三维判定保持一致)"""
r = TYPE_RULE_MAP[type_no]
return {
"type_no": r["type_no"], "name": r["name"], "health_emoji": r["health_emoji"],
"health_label": r["health_label"], "lifecycle": r["lifecycle"],
"o_sign": r["o_sign"], "i_sign": r["i_sign"], "f_sign": r["f_sign"],
"profile": r["profile"], "risk": r["risk"],
}
def classify_enterprise(operating_net: Optional[float],
investing_net: Optional[float],
financing_net: Optional[float]) -> dict:
"""O/I/F 三净额符号 → 8类企业定性判定
返回:
{classified, type_no, name, health_emoji, health_label, risk, profile, lifecycle, o_sign, i_sign, f_sign, reason}
classified=False 时 reason 说明未判定的原因(任一净额缺失或为0)
{classified, type_no, name, health_emoji, health_label, risk, profile, lifecycle,
dimension, o_sign, i_sign, f_sign, reason, investing_note?, candidate_type_nos?, candidates?}
classified=False 时 reason 说明未判定的原因。
2026-09-17 裁定1(财务口径):
· **投资活动净额 = 0 是合法输入**(视为「本期无投资活动」),不再整体判「未判定」;
· 投资为 0 且经营/筹资可判 → **二维映射**(dimension="2D",候选类型 2 个,
投资栏位 investing_note="本期无投资活动");
· 经营/筹资任一亦为 0 或缺失 → 回退「未判定(无投资活动)」,**不报错、不静默**。
"""
o, i, f = _sign(operating_net), _sign(investing_net), _sign(financing_net)
if o is None or i is None or f is None:
missing = [name for sign, name in [(o, "经营"), (i, "投资"), (f, "筹资")] if sign is None]
return {"classified": False, "reason": "以下活动净额缺失:" + "、".join(missing) + "(需官方现金流量表数据)"}
if "0" in (o, i, f):
zero = [name for sign, name in [(o, "经营"), (i, "投资"), (f, "筹资")] if sign == "0"]
return {"classified": False, "reason": "以下活动净额为0(恰好为0无法落入8类正负组合):" + "、".join(zero) + ",请按业务实质人工判定"}
names = {"o": "经营", "i": "投资", "f": "筹资"}
missing = [names[k] for k, s in (("o", o), ("i", i), ("f", f)) if s is None]
if missing:
return {"classified": False, "dimension": "0D",
"reason": "以下活动净额缺失:" + "、".join(missing) + "(需官方现金流量表数据)"}
zeros = [k for k, s in (("o", o), ("i", i), ("f", f)) if s == "0"]
inv_zero = i == "0"
if zeros:
# 恰好为 0 的维度:投资为 0 → 「本期无投资活动」;其余维度为 0 → 仍是有效输入但不可单独归类
zero_names = "、".join(names[k] for k in zeros)
if len(zeros) >= 2 or (o == "0" or f == "0"):
note = "(无投资活动)" if inv_zero else ""
return {
"classified": False, "dimension": "0D",
"o_sign": o, "i_sign": i, "f_sign": f,
"investing_note": INVESTING_ZERO_NOTE if inv_zero else None,
"reason": ("未判定%s:%s活动净额为 0(%s)且不足以二维归类,"
"请按业务实质人工判定" % (
note, zero_names,
"投资活动为 0 = 本期无投资活动" if inv_zero else "该维度无正负号")),
}
# 单维为 0 且该维为投资 → 二维(经营 × 筹资)映射
pair = TWO_DIM_CANDIDATE_MAP.get((o, f)) if (o in ("+", "-") and f in ("+", "-")) else None
if pair:
cands = [_rule_brief(t) for t in pair]
return {
"classified": True,
"dimension": "2D",
"dimension_label": "二维判定(投资活动=0,本期无投资活动)",
"type_no": None,
"name": "/".join("%s %s" % (c["type_no"], c["name"]) for c in cands),
"candidate_type_nos": list(pair),
"candidates": cands,
"health_emoji": "", "health_label": None, "lifecycle": None,
"profile": ";".join("类型%d %s:%s" % (c["type_no"], c["name"], c["profile"]) for c in cands),
"risk": ";".join("类型%d:%s" % (c["type_no"], c["risk"]) for c in cands),
"o_sign": o, "i_sign": i, "f_sign": f,
"investing_note": INVESTING_ZERO_NOTE,
"reason": ("经营(%s) 投资(0=%s) 筹资(%s) → 二维映射候选:%s"
"(投资活动为 0 无法定符号,需结合业务实质在候选中确认)" % (
o, INVESTING_ZERO_NOTE, f,
" / ".join("类型%d %s" % (c["type_no"], c["name"]) for c in cands))),
}
for rule in CASHFLOW_TYPE_RULES:
if rule["o_sign"] == o and rule["i_sign"] == i and rule["f_sign"] == f:
return {
"classified": True,
"dimension": "3D",
"type_no": rule["type_no"],
"name": rule["name"],
"health_emoji": rule["health_emoji"],
@@ -135,9 +211,10 @@ def classify_enterprise(operating_net: Optional[float],
"profile": rule["profile"],
"risk": rule["risk"],
"o_sign": o, "i_sign": i, "f_sign": f,
"investing_note": None,
"reason": "经营(%s) 投资(%s) 筹资(%s) → 类型%d %s" % (o, i, f, rule["type_no"], rule["name"]),
}
return {"classified": False, "reason": "未知组合,请检查数据"}
return {"classified": False, "dimension": "0D", "reason": "未知组合,请检查数据"}
class CashFlowValueIn(BaseModel):
@@ -237,17 +314,21 @@ def get_analysis(
"waterfall": None,
}
classification = classify_enterprise(rec.operating_net, rec.investing_net, rec.financing_net)
i_sign = _sign(rec.investing_net)
# 裁定1:投资活动为 0 → 栏位显式标注「本期无投资活动」,避免读者误判为「收付相抵的 0」
investing_note = classification.get("investing_note") or (
INVESTING_ZERO_NOTE if i_sign == "0" else None)
nets = {
"operating": {"key": "operating", "name": "经营活动", "value": rec.operating_net,
"sign": _sign(rec.operating_net), "unit": "万元"},
"sign": _sign(rec.operating_net), "unit": "万元", "note": None},
"investing": {"key": "investing", "name": "投资活动", "value": rec.investing_net,
"sign": _sign(rec.investing_net), "unit": "万元"},
"sign": i_sign, "unit": "万元", "note": investing_note},
"financing": {"key": "financing", "name": "筹资活动", "value": rec.financing_net,
"sign": _sign(rec.financing_net), "unit": "万元"},
"sign": _sign(rec.financing_net), "unit": "万元", "note": None},
}
classification = classify_enterprise(rec.operating_net, rec.investing_net, rec.financing_net)
# 2. 瀑布图:期初现金 → +经营 → +投资 → +筹资 → 期末现金
waterfall = None
if rec.begin_cash is not None or rec.end_cash is not None:
@@ -279,6 +360,8 @@ def get_analysis(
"remark": rec.remark,
"nets": nets,
"classification": classification,
"classification_dimension": classification.get("dimension"),
"investing_note": investing_note,
"waterfall": waterfall,
}
+288
View File
@@ -0,0 +1,288 @@
# -*- coding: utf-8 -*-
"""比率类 KPI 单位口径归一(取数层) —— 财务口径裁定 裁定2a
依据:`wecom-finance/reports/CMA_口径裁定_20260917.md` 裁定2a
· 比率类指标**统一为「百分比(%)」口径**:取数 ×100,与现有阈值(流动比率红 <150、速动 <80)同量纲;
· 逐个比率类 KPI**核对定义与单位**,出具对照表;
· **单位未核对前的「疑似错配」预警不推送**(降级待核,防假红灯 586/587)。
问题实证(2026-09-17 只读侦查,entity 1 酣客):
`kpi_definitions` 声明单位 = %,但**取数**以比率(倍)存储:流动比率 1.29 / 速动比率 0.54 / 毛利率 0.38,
而阈值按百分数书写(<150 / <80 / <15)→ 跨 2 个数量级比较,判定不可信(09-09 的 586/587 即此形态)。
**注意**:按裁定2a 归一后 流动比率 129% < 150 → **红是合法的**(「流动性偏紧」),586/587 不是假红灯、
不作废;真正的错配处置是「未核对项不推送」。
其余百分数类(返利率 48.44、收入增长率 -82.2、资产负债率 39.9、费用率 30.0、渠补率 75.0 …)
**取数已是百分数**,不得再 ×100(反过来会成为量级越界)。
实现约定(唯一事实源):
· 归一因子表 `RATIO_CALIBER_TABLE`(kpi_code → 口径/因子/证据/是否已核对);
· 判定侧只调 `resolve_caliber(actual, kpi)`,**不自行做字符串/量级判断**;
· 表外且命中「疑似错配形态」的 KPI → status='dimension_suspect' → 调用方**不落预警、不推送**;
· `norm_factor` 只影响**判定/推送口径**,不改 `kpi_values` 存储值、不改 `kpi_definitions` 阈值
(裁定2a 明确:统一取数而非改阈值)。
"""
from __future__ import annotations
from typing import Optional, Sequence
from app.utils.threshold_parser import parse_clauses
#: 百分比口径的归一因子
FACTOR_PERCENT = 100
FACTOR_IDENTITY = 1
#: 比率类 KPI 单位对照表(裁定2a「逐个核对定义与单位」的落地产物)
#: norm_factor=100 → 取数为比率(倍),判定前 ×100 归一到百分数口径
#: norm_factor=1 → 取数已是百分数/同量纲,直接判定
RATIO_CALIBER_TABLE = {
# ---- 需归一(取数=倍,阈值=百分数写法)----
"F_CURRENT_RATIO": {
"name": "流动比率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_PERCENT, "verified": True,
"evidence": "2026-08 实测 1.29(比率量级)→ ×100 = 129%;阈值 <150 为百分数写法(裁定2a 明示示例:129<150 → 红,与「流动性偏紧」一致)",
},
"F_QUICK_RATIO": {
"name": "速动比率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_PERCENT, "verified": True,
"evidence": "2026-08 实测 0.54 → ×100 = 54%;阈值 <80 为百分数写法(经验值 1:1=100%)",
},
"F_GROSS_MARGIN": {
"name": "毛利率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_PERCENT, "verified": True,
"evidence": "2026-08 实测 0.38(同期另有 0.3766 行,源不同)→ ×100 = 38%;阈值 >=20|>=18|<15 为百分数写法(38% 合理,原口径 0.38<15 属假红)",
},
# ---- 无需归一(取数已是百分数/同量纲,×100 会造成反向越界)----
"F_OP_PROFIT_MARGIN": {
"name": "经营利润率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 -161.45 已是百分数(×100 得 -16145,量级越界);阈值 >=10|>=8|<5 同量纲",
},
"F_REBATE_RATE": {
"name": "返利率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 48.44 已是百分数;量纲一致、判定有效(裁定 §4.4:不改阈值,48.44% 口径本身待财务确认 → 保留并标注)",
},
"F_FACTORY_REBATE_RATE": {
"name": "上游厂补率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 278.7 已是百分数(厂补/进货口径 >100% 属正常)",
},
"F_REVENUE_GROWTH": {
"name": "收入增长率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 -82.2 已是百分数",
},
"F_ROE": {
"name": "净资产收益率(ROE)", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 -0.95 已是百分数(小体量亏损月,-0.95% 量级合理)",
},
"F_ROI": {
"name": "总资产报酬率(ROA)", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 -0.57 已是百分数",
},
"F_DEBT_RATIO": {
"name": "资产负债率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 39.9 已是百分数",
},
"F_COST_RATIO": {
"name": "费用率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 30.0 已是百分数(阈值 <=25|<=30|>30)",
},
"C_REBATE_RATE": {
"name": "渠补率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 75.0 已是百分数(阈值含区间子句 52.33~107.67)",
},
"F_BEP_COVER": {
"name": "保本覆盖率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 60.4 已是百分数(阈值 >=100|>=85|<85)",
},
"F_BEP_COVER_CASH": {
"name": "现金口径保本覆盖率", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 85.2 已是百分数",
},
"F_BEP_GAP": {
"name": "保本缺口", "declared_unit": "%", "caliber": "百分比(%)",
"norm_factor": FACTOR_IDENTITY, "verified": True,
"evidence": "2026-08 实测 39.6 已是百分数",
},
}
#: 疑似错配形态的参数:阈值按百分数书写(子句量级 ≥ 100)而实测为比率量级(0 < |v| < 10)
SUSPECT_THRESHOLD_ORDERS = 100.0
SUSPECT_ACTUAL_ABS_MAX = 10.0
#: 参与「疑似错配」判定的声明单位(比率/百分数类)
SUSPECT_UNITS = ("%", "倍", "", "百分比", None)
def _exprs_of(kpi) -> list:
"""取 KPI 三档阈值表达式(ORM 对象或同名 duck type 均可)"""
return [getattr(kpi, "threshold_green", None),
getattr(kpi, "threshold_yellow", None),
getattr(kpi, "threshold_red", None)]
def _clause_values(expr: str) -> list:
"""阈值表达式 → 比较值列表(区间取两端;解析失败返回空表)"""
try:
clauses = parse_clauses(expr)
except Exception:
return []
vals = []
for _op, val in clauses or []:
if isinstance(val, tuple):
vals.extend([v for v in val if isinstance(v, (int, float))])
elif isinstance(val, (int, float)):
vals.append(val)
return vals
def suspected_unit_mismatch(actual, exprs: Sequence, unit=None) -> dict:
"""「疑似单位错配」形态检测(裁定2a 的降级门)。
形态:**所有**可解析阈值子句的量级都 ≥ 100(百分数写法),而实测值量级 < 10(比率写法)
且声明单位属比率/百分数类 → 跨量级比较,判定不可信。
为什么不用通用的「量级差 ≥1 个数量级」守卫:那会把**真实的小值告警**误伤
(实证:总资产周转率 2026-08 = 0.008 次 vs 阈值 1.0 次,量级差 2.1——那是真红,不是错配)。
"""
if actual is None:
return {"suspect": False, "reason": "无实测值"}
try:
a = abs(float(actual))
except (TypeError, ValueError):
return {"suspect": False, "reason": "实测值非数值"}
if a == 0:
return {"suspect": False, "reason": "实测值为 0,量级不可判"}
vals = []
for e in exprs or []:
if e:
vals.extend(_clause_values(e))
if not vals:
return {"suspect": False, "reason": "阈值不可解析"}
all_percent_written = all(abs(float(v)) >= SUSPECT_THRESHOLD_ORDERS for v in vals)
ratio_magnitude = a < SUSPECT_ACTUAL_ABS_MAX
unit_ok = unit in SUSPECT_UNITS
if all_percent_written and ratio_magnitude and unit_ok:
return {"suspect": True, "nearest": min(vals, key=lambda v: abs(a - abs(float(v)))),
"reason": ("疑似单位错配:阈值按百分数书写(子句量级 ≥%g)而实测 %s 为比率量级(|v|<%g)"
",声明单位 %r → 判定不可信,降级待核(不推送)"
% (SUSPECT_THRESHOLD_ORDERS, actual, SUSPECT_ACTUAL_ABS_MAX, unit))}
return {"suspect": False, "reason": "形态不匹配(阈值量级/实测量级/单位 至少一项不符)"}
def resolve_caliber(actual, kpi) -> dict:
"""判定侧唯一入口:返回取数归一结果与可信度状态。
return: {
status: 'ok' | 'normalized' | 'dimension_suspect',
value: 归一后用于判定/展示的数值,
factor: 归一因子(1/100),
verified: 对照表核对状态(表外=False),
kpi_code, note, reason
}
· status='dimension_suspect' → 调用方**不落预警、不推送**(裁定2a 过渡纪律)
"""
code = getattr(kpi, "kpi_code", None)
unit = getattr(kpi, "unit", None)
row = RATIO_CALIBER_TABLE.get(code) if code else None
if row:
factor = row["norm_factor"] if row.get("verified") else FACTOR_IDENTITY
if factor != FACTOR_IDENTITY and actual is not None:
try:
value = float(actual) * factor
except (TypeError, ValueError):
value = actual
else:
value = actual
note = None
if factor != FACTOR_IDENTITY:
note = "口径归一 ×%d(原值 %s %s → %s%%)" % (factor, actual, row["declared_unit"], _fmt(value))
return {"status": "normalized" if factor != FACTOR_IDENTITY else "ok", "value": value,
"factor": factor, "verified": bool(row.get("verified")), "kpi_code": code,
"caliber": row.get("caliber"), "note": note, "reason": row.get("evidence")}
sus = suspected_unit_mismatch(actual, _exprs_of(kpi), unit)
if sus["suspect"]:
return {"status": "dimension_suspect", "value": actual, "factor": FACTOR_IDENTITY,
"verified": False, "kpi_code": code, "note": None, "reason": sus["reason"]}
return {"status": "ok", "value": actual, "factor": FACTOR_IDENTITY, "verified": False,
"kpi_code": code, "note": None, "reason": sus["reason"]}
def _fmt(v) -> str:
try:
return ("%.2f" % float(v)).rstrip("0").rstrip(".")
except (TypeError, ValueError):
return str(v)
class _K:
"""self-test 用的最小 KPI 载体"""
def __init__(self, kpi_code, unit, green=None, yellow=None, red=None):
self.kpi_code, self.unit = kpi_code, unit
self.threshold_green, self.threshold_yellow, self.threshold_red = green, yellow, red
def _self_test() -> int:
"""零库自证(裁定2a 证据):归一、同量纲直判、疑似错配降级、真实小值不误伤"""
from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY
def level(actual, kpi):
cal = resolve_caliber(actual, kpi)
if cal["status"] == "dimension_suspect":
return "dimension_suspect"
v = cal["value"]
exprs = {"green": kpi.threshold_green, "yellow": kpi.threshold_yellow, "red": kpi.threshold_red}
for lv in LEVEL_PRIORITY:
if exprs.get(lv) and eval_threshold(v, exprs[lv]):
return lv
return "info"
cases = (
# (说明, 实测, KPI, 期望)
("流动比率 1.29 → 归一 129% vs 红<150", 1.29,
_K("F_CURRENT_RATIO", "%", ">=200", ">=150", "<150"), "red"),
("速动比率 0.54 → 归一 54% vs 红<80", 0.54,
_K("F_QUICK_RATIO", "%", ">=100", ">=80", "<80"), "red"),
("毛利率 0.38 → 归一 38% vs 绿>=20", 0.38,
_K("F_GROSS_MARGIN", "%", ">=20", ">=18", "<15"), "green"),
("返利率 48.44 已百分数 → 红>12(不 ×100)", 48.44,
_K("F_REBATE_RATE", "%", "<=10", ">10", ">12"), "red"),
("收入增长率 -82.2 已百分数 → 红<15", -82.2,
_K("F_REVENUE_GROWTH", "%", ">=20", ">=15", "<15"), "red"),
("总资产周转率 0.008 次 vs 阈值 1.0(真实小值,不误伤)", 0.008,
_K("F_ASSET_TURNOVER", "次", ">=1.0", ">=0.8", "<0.8"), "red"),
("表外比率 KPI(阈值百分数写法)→ 降级待核不推送", 1.29,
_K("X_UNKNOWN_RATIO", "%", ">=200", ">=150", "<150"), "dimension_suspect"),
("高值不误判:营业收入 14.45 万 vs 阈值 97.33(非比率量级)", 14.45,
_K("F_REVENUE", "万元", ">=97.33", ">=77.87", "<58.4"), "red"),
)
ok = 0
print("=== 比率类单位口径归一 self-test ===")
for desc, actual, kpi, expect in cases:
got = level(actual, kpi)
flag = "PASS" if got == expect else "FAIL"
ok += (got == expect)
cal = resolve_caliber(actual, kpi)
print(" [%s] %-46s → %-18s (期望 %s) value=%s factor=%s"
% (flag, desc, got, expect, cal["value"], cal["factor"]))
print("=== %d/%d PASS | 对照表 %d 项(归一 %d 项 / 同量纲 %d 项)===" % (
ok, len(cases), len(RATIO_CALIBER_TABLE),
sum(1 for v in RATIO_CALIBER_TABLE.values() if v["norm_factor"] != FACTOR_IDENTITY),
sum(1 for v in RATIO_CALIBER_TABLE.values() if v["norm_factor"] == FACTOR_IDENTITY)))
return 0 if ok == len(cases) else 1
if __name__ == "__main__":
raise SystemExit(_self_test())